First commit with entire first rendition of the project

This commit is contained in:
Your Name
2026-06-25 21:18:07 -06:00
commit 551322a94b
42 changed files with 4346 additions and 0 deletions

36
.gitignore vendored Normal file
View File

@@ -0,0 +1,36 @@
# Python
__pycache__/
*.py[cod]
*.egg-info/
dist/
build/
*.egg
# Virtual environment
venv/
.venv/
env/
.env/
# Config with secrets (API keys, MQTT credentials)
config.yaml
# Test & coverage
.pytest_cache/
.coverage
htmlcov/
# IDE
.vscode/
.idea/
*.swp
*.swo
*~
# OS
.DS_Store
Thumbs.db
# Generated by simulation mode
spectra_last.png
/tmp/spectra_last.png

175
README.md Normal file
View File

@@ -0,0 +1,175 @@
# Spectra
Display random Unsplash photos on an Inky Impression e-paper display, with a web interface for uploading your own images.
## Features
- Fetches random high-resolution photos from Unsplash (cached to gallery)
- Upload your own images via the web interface (drag-and-drop, multi-file)
- Centres and scales images to fit the display resolution
- Colour saturation tuning for e-paper
- Web dashboard with preview, gallery, rotation queue, and config editor
- MQTT integration for remote commands and status reporting
- Simulation mode for testing without hardware (configurable resolution)
- systemd integration for automatic updates on a schedule
- Randomised update intervals to avoid predictable refreshes
- Config hot-reload — no service restart needed for setting changes
## Requirements
- Raspberry Pi (any model with GPIO)
- [Pimoroni Inky Impression](https://shop.pimoroni.com/products/inky-impression) (4.0", 7.3", or 13.3")
- Python 3.9+
- [Unsplash API access key](https://unsplash.com/developers)
## Installation
Run the installer on a Raspberry Pi:
```bash
./install.sh
```
The installer will:
1. Install the Pimoroni Inky library (skipped if not a Raspberry Pi)
2. Install Python dependencies (including web server and MQTT)
3. Install the `spectra` package
4. Copy the default config to `/etc/spectra/config.yaml`
5. Install and enable both the display and web interface systemd services
### Manual installation
```bash
pip install -r requirements.txt
pip install -e .
```
## Configuration
Edit `/etc/spectra/config.yaml` (or `~/.config/spectra/config.yaml` or `config.yaml` in the current directory):
```yaml
unsplash:
access_key: "your_access_key_here"
query: "nature"
orientation: "landscape"
collections: ""
display:
saturation: 0.5
# resolution:
# width: 800
# height: 480
schedule:
interval_hours: 1
random_delay_seconds: 300
mqtt:
enabled: false
broker: localhost
port: 1883
topic_prefix: spectra
client_id: spectra-display
username: ""
password: ""
paths:
cache: /var/cache/spectra
```
## Usage
### Display loop
```bash
# Run once and exit
spectra --once
# Run once in simulation mode
spectra --once --simulate
# Simulate at a specific resolution
spectra --once --simulate --width 800 --height 480
# Run with a custom config
spectra -c /path/to/config.yaml
# Run continuously
spectra
```
### Web interface
```bash
# Start the web interface (default: http://0.0.0.0:5000)
spectra web
# With custom host/port
spectra web --host 0.0.0.0 --port 5000
# Verbose logging
spectra web -v
```
The web interface provides:
- **Dashboard** — display status, schedule info, quick actions (refresh, clear)
- **Gallery** — browse all uploaded and cached Unsplash images, trigger "show now"
- **Upload** — drag-and-drop multiple image files with progress tracking
- **Settings** — live-edit Unsplash, display, schedule, and MQTT configuration
### MQTT
When MQTT is enabled in the config, the display loop subscribes to commands and publishes status. See [MQTT integration docs](docs/mqtt.md) for details.
## systemd services
Two systemd services are installed:
- `spectra.service` — the display loop (fetches Unsplash, processes triggers)
- `spectra-web.service` — the web interface (Flask + htmx)
```bash
sudo systemctl start spectra
sudo systemctl stop spectra-web
sudo systemctl status spectra-web
# View logs
journalctl -u spectra -f
journalctl -u spectra-web -f
```
## Project structure
```
spectra/
├── config.yaml # Example configuration
├── install.sh # Installer script
├── pyproject.toml # Package metadata
├── requirements.txt # Python dependencies
├── spectra/
│ ├── __init__.py
│ ├── __main__.py # python -m spectra entry point
│ ├── cli.py # CLI argument parsing and main loop
│ ├── config.py # Configuration loader
│ ├── config_manager.py # Config read/write with YAML write-back
│ ├── display.py # Inky display abstraction and image processing
│ ├── fetcher.py # Unsplash API client
│ ├── library.py # Unsplash image caching to SQLite library
│ ├── mqtt.py # MQTT client (commands + status)
│ ├── trigger.py # Shared trigger file (web server → display loop)
│ └── web/
│ ├── server.py # Flask app factory and API routes
│ ├── models.py # SQLAlchemy models
│ ├── templates/ # Jinja2 templates (6 pages)
│ ├── static/ # CSS and JS
│ └── __init__.py
├── systemd/
│ ├── spectra.service # Display loop systemd unit
│ └── spectra-web.service # Web interface systemd unit
└── docs/
├── web-interface.md # Web interface documentation
└── mqtt.md # MQTT integration documentation
```

186
bug-tracking.md Normal file
View File

@@ -0,0 +1,186 @@
# Bug Tracking
## Critical
### 1. MQTT enabled checkbox can never be unchecked
**File:** `spectra/web/templates/settings.html:9097`, `spectra/web/server.py:118120`
**Problem:** The MQTT enable/disable checkbox uses htmx to PUT `{"value": "true"}` to `/api/config/mqtt/enabled`. When the user unchecks the box, the browser omits the field entirely (standard HTML behaviour for unchecked checkboxes). The server then receives an empty body, hits the `if data is None or "value" not in data: abort(400)` guard, and returns a 400 error. Result: MQTT can never be turned off via the web UI.
**To reproduce:** Go to Settings → MQTT → uncheck "Enable MQTT" → click Save. Browser sends PUT without `value` → server returns 400.
**Fix:** Add a hidden input before the checkbox with the same `name` and `value="false"`. The checkbox's value (`"true"`) overrides the hidden when checked.
```html
<label>
<input type="hidden" name="value" value="false">
<input type="checkbox" id="mqtt-enabled" name="value" value="true"
{{ 'checked' if config.get('mqtt', {}).get('enabled', False) }}>
Enable MQTT
</label>
```
---
## Medium
### 2. Gallery delete shows "Saved" instead of removing the card
**Files:** `spectra/web/templates/gallery.html:84`, `spectra/web/static/js/app.js:911`
**Problem:** The Gallery's delete button uses `hx-delete` targeting `#gallery-msg`. After a successful delete, the server returns `{"status": "deleted"}`. The htmx response handler in `app.js` checks `data.status === 'triggered'` → shows trigger message, otherwise shows "Saved". Since `'deleted' !== 'triggered'`, the user sees "Saved" instead of "Deleted". The image card remains visible until the page is manually refreshed.
**Severity:** UX/data mismatch — user has no confirmation the image was actually deleted.
**Fix options:**
- **Option A:** Make the delete endpoint return `{"status": "triggered"}` instead (hacky).
- **Option B:** Update `app.js` to check for `data.status === 'deleted'` and show "Deleted" + remove the card from the DOM.
- **Option C:** Use htmx's `hx-target="closest article" hx-swap="outerHTML"` on the delete button so the card is removed directly from the DOM without needing a message.
**Recommended (Option C):** Change the delete button to:
```html
<button class="secondary"
hx-delete="/api/images/{{ img.id }}"
hx-target="closest article"
hx-swap="outerHTML swap:0.5s"
hx-confirm="Delete this image?">Delete</button>
```
Then remove the `#gallery-msg` target from the delete — the card disappears inline.
### 3. `MqttClient.reconfigure()` calls `self.__init__()` directly
**File:** `spectra/mqtt.py:85`
**Problem:** `reconfigure()` calls `self.__init__(new_config)` to re-initialise the instance. While this works (Python allows it), it's highly unusual, confuses static analysis tools, and breaks if a subclass overrides `__init__` differently. If the config is unchanged but the background thread has died, the method also unnecessarily destroys and recreates the client.
**Fix:** Extract the attribute setup into a private `_setup` method called from both `__init__` and `reconfigure`:
```python
def __init__(self, config):
self._setup(config)
self._client = mqtt.Client(client_id=self.client_id, protocol=mqtt.MQTTv311)
...
def _setup(self, config):
self.broker = config.get("broker", "localhost")
self.port = config.get("port", 1883)
self.prefix = config.get("topic_prefix", "spectra")
self.client_id = config.get("client_id", "spectra-display")
self._username = config.get("username", "")
self._password = config.get("password", "")
def reconfigure(self, new_config):
...
self.stop()
self._setup(new_config)
self._client = mqtt.Client(client_id=self.client_id, protocol=mqtt.MQTTv311)
if self._username:
self._client.username_pw_set(self._username, self._password)
self._client.on_connect = self._on_connect
self._client.on_disconnect = self._on_disconnect
self._client.on_message = self._on_message
self.start()
```
---
## Low
### 4. Unsplash filename collision on duplicate fetch
**File:** `spectra/library.py:56`
**Problem:** The filename is `f"unsplash_{unsplash_id}.png"`. If the same Unsplash photo is fetched twice (e.g. because the query doesn't change and there aren't many results), the second fetch overwrites the first file. Two DB records now reference the same filepath. If one record is deleted, `os.remove(dest)` succeeds, leaving the other record with a dangling path. On the next attempt to serve/thumbnail that record, the endpoint returns 404 from `send_file`.
**Fix:** Add a UUID suffix to make filenames unique:
```python
filename = f"unsplash_{unsplash_id}_{uuid.uuid4().hex[:8]}.png"
```
Before saving, also check if the DB already has a record for this `unsplash_id` and skip the insert (avoid duplicates entirely):
```python
conn = sqlite3.connect(_db_path())
existing = conn.execute("SELECT id FROM image WHERE source='unsplash' AND unsplash_id=?", (unsplash_id,)).fetchone()
if existing:
conn.close()
os.remove(dest) # clean up the duplicate file
return existing[0]
```
### 5. Object URLs never revoked in upload page
**File:** `spectra/web/templates/upload.html:156`
**Problem:** `URL.createObjectURL(file)` creates a blob URL for the file preview, but it is never released with `URL.revokeObjectURL()`. For a typical session (a handful of files), the memory impact is negligible. If a user uploads hundreds of files in one session, blob URLs accumulate in memory until the page is reloaded.
**Fix:** Store the URL on the file item element and revoke it when the item is removed or after a failed upload:
```javascript
function renderFile(file) {
var url = URL.createObjectURL(file);
// ... store url for later revocation
fileItem.dataset.blobUrl = url;
}
function cleanupFileItem(el) {
var url = el.dataset.blobUrl;
if (url) URL.revokeObjectURL(url);
}
```
### 6. Dead CSS rules for old upload form
**File:** `spectra/web/static/css/app.css:20`
**Problem:** The CSS rule `#upload-result:empty { display: none; }` targets an element that was used by the old htmx-based upload form. The new multi-file upload form uses `#file-list`, `#upload-options`, and `#upload-summary` instead. The `#upload-result` element no longer exists anywhere in the templates.
**Other dead selectors on the same line:** `#gallery-msg:empty`, `#action-result:empty`, `#unsplash-result:empty`, `#saturation-result:empty`, `#schedule-result:empty`.
- `#gallery-msg` — still used by gallery delete/show htmx responses (if changed to `closest article` approach in bug #2, this becomes dead too)
- `#action-result` — used by dashboard quick actions
- `#unsplash-result`, `#saturation-result`, `#schedule-result` — used by settings page
**Fix:** Remove only `#upload-result:empty` from the selector list. Keep the others as they are still in use.
---
## Cosmetic / Design
### 7. Inconsistent `process_image` usage: processed dimensions stored vs originals shown
**File:** `spectra/library.py:61`, `spectra/web/server.py:182183`
**Design issue:** `library.py` saves the processed (cropped + resized) dimensions as width/height in the DB, but stores the *original* unprocessed image on disk. The gallery shows the original file (via `/api/images/<id>/file`) which may have different dimensions than what is stored. The preview endpoint re-processes the original on demand, so the stored dimensions are only used for display purposes (and are somewhat misleading).
This is intentional — the stored dimensions represent "how the image appears on the display" — but it could confuse API consumers who see width/height that doesn't match the actual file dimensions. Documentation should clarify this.
### 8. `api_config_set` double-mutates config
**File:** `spectra/web/server.py:124125`
**Design issue:** `section_data[key] = data["value"]` mutates `cm.config` in place (because `section_data` is a reference to `cm.config[section]`). Then `cm.set_nested()` does the same mutation again and calls `_write()`. The config is correct after both operations, but the first mutation is written only if `set_nested` completes, so this isn't a data-loss risk — just redundant code.
**Fix:** Remove the redundant mutation; only `cm.set_nested()` is needed:
```python
# Remove line 124: section_data[key] = data["value"]
cm.set_nested([section, key], data["value"])
return jsonify({key: data["value"]})
```
---
## Tests
### 9. ~~Gallery delete test missing from test suite~~
**Fixed:** `tests/test_gallery.py` adds an end-to-end test (`test_upload_delete_flow`) that uploads an
image (`POST /api/images`), confirms `total > 0` via `GET /api/images`, deletes it, then confirms
`total` decreased by 1. Run with `pytest tests/`.
**Dev dependency:** `pytest>=7.0.0` and `pytest-flask>=1.2.0` added under `[project.optional-dependencies] dev` in `pyproject.toml`. Install with `pip install "spectra-display[dev]"`.

34
config.example.yaml Normal file
View File

@@ -0,0 +1,34 @@
# Copy this file to config.yaml and fill in your settings.
# spectra searches for config.yaml in order:
# 1. $PWD/config.yaml
# 2. ~/.config/spectra/config.yaml
# 3. /etc/spectra/config.yaml
# Or pass a custom path with: spectra -c /path/to/config.yaml
unsplash:
# Required: register at https://unsplash.com/developers
access_key: ""
# Optional filters
query: ""
collections: ""
display:
saturation: 0.5
# Physical orientation in degrees: 0, 90, 180, 270
orientation: 0
resolution:
width: 1600
height: 1200
schedule:
interval_hours: 1
random_delay_seconds: 300
mqtt:
enabled: false
broker: localhost
port: 1883
topic_prefix: spectra
client_id: spectra-display
username: ""
password: ""

103
docs/display-orientation.md Normal file
View File

@@ -0,0 +1,103 @@
# Display Orientation
## Purpose
Allow the user to configure the physical orientation of the Inky Impression
display so that images are displayed upright regardless of how the panel is
mounted. The orientation setting also drives the Unsplash orientation filter so
fetched photos match the display's effective aspect ratio (portrait vs.
landscape).
## Configuration
A new `display.orientation` key (integer, degrees clockwise) in
`config.yaml` / `DEFAULT_CONFIG`:
```yaml
display:
orientation: 0 # 0, 90, 180, 270
```
| Value | Unsplash filter | Effective aspect |
|-------|----------------|------------------|
| 0 | landscape | width > height |
| 90 | portrait | height > width |
| 180 | landscape | width > height |
| 270 | portrait | height > width |
The old `unsplash.orientation` config key is **no longer used** by
`refresh()` — orientation is now derived from `display.orientation` +
`display.resolution`.
## Image Processing Pipeline
1. **Fetch** raw photo from Unsplash (with orientation filter already matching
the effective aspect ratio — see below).
2. **`process_image()`** in `InkyDisplay`:
- Compute effective width/height by swapping physical dimensions when
orientation is 90° or 270°.
- Crop to effective aspect ratio.
- Resize to effective dimensions.
- Rotate by `-orientation` degrees (so the result matches the physical
resolution of the display panel).
3. **`show()`** sends the processed (and rotated) image to `inky.set_image()`.
No additional transformation is needed — the panel hardware always expects
its native `width × height` pixel grid.
## Unsplash Filter Derivation
In `cli.py:refresh()`, after reading `display.orientation` and
`display.resolution`:
1. If orientation is 90 or 270, swap width ↔ height to get effective
dimensions.
2. If effective height > effective width → request `"portrait"`.
3. If effective width > effective height → request `"landscape"`.
4. If roughly equal → request `"squarish"`.
This means the Unsplash API is sent `orientation=portrait` when the display is
mounted vertically, which returns taller photos that need less cropping.
## Hot-Reload
`_reload_config()` in `cli.py` detects changes to
`display.orientation` and updates `display.orientation` live. The next
`process_image()` call uses the new orientation. No display loop restart is
needed.
## Library Caching
`_save_to_library()` passes the **effective** width/height to
`save_unsplash_image()` (swapped when orientation is 90/270). The stored image
is processed to the effective dimensions at orientation=0 (no rotation). This
keeps library images in their "natural" viewing orientation.
The preview endpoint (`/api/preview/<id>`) creates an `InkyDisplay` with the
current orientation, so the preview matches what the physical display shows.
## UI
Settings page (Display article) shows a `<select>` with four options:
- 0° (Landscape)
- 90° (Portrait)
- 180° (Landscape)
- 270° (Portrait)
Saved via the existing `PUT /api/config/display/orientation` route.
The old `unsplash.orientation` dropdown has been **removed** from the settings
page — it is now auto-derived from the display orientation.
## Edge Cases
- **0° (default):** No rotation applied. Preserves existing behavior.
- **180°:** Image is cropped/resized at physical dimensions, then rotated
upside-down. No dimension swap. Useful for ceiling-mounted displays.
- **90° / 270°:** Width and height are swapped for cropping/resizing, then
rotated back to physical dimensions. Images fetched from Unsplash with the
matching portrait/landscape filter.
- **Preview:** Shows the rotated image so the user sees exactly what the
display will show.
- **rotate(expand=True):** Only used with 90° multiples so no fractional
pixels appear.

146
docs/mqtt.md Normal file
View File

@@ -0,0 +1,146 @@
# MQTT Integration
Spectra can be controlled and monitored over MQTT. The MQTT client runs as a background thread in the display loop process (`spectra` service) and connects to any standard MQTT broker.
## Configuration
Enable MQTT in `config.yaml`:
```yaml
mqtt:
enabled: true
broker: 192.168.1.100
port: 1883
topic_prefix: spectra
client_id: spectra-display
username: ""
password: ""
```
- **enabled** — set to `true` to start the MQTT client
- **broker** — hostname or IP of your MQTT broker
- **port** — broker port (default: 1883)
- **topic_prefix** — prefix for all MQTT topics (default: `spectra`)
- **client_id** — unique client identifier for the MQTT connection
- **username / password** — optional credentials for authenticated brokers
Changes to MQTT settings are hot-reloaded by the display loop — no service restart needed. If you disable MQTT, the client disconnects cleanly.
## Topics
All topics use the configured `topic_prefix`. The examples below assume the default prefix `spectra`.
### Commands (subscribe)
Spectra subscribes to `spectra/command/#` and dispatches based on the command suffix.
| Topic | Payload | Action |
|-------|---------|--------|
| `spectra/command/refresh` | *(ignored)* | Triggers an Unsplash refresh on the next display cycle |
| `spectra/command/clear` | *(ignored)* | Clears the display to white on the next cycle |
| `spectra/command/show/<id>` | *(ignored)* | Shows a library image by its database ID |
| `spectra/command/status` | *(ignored)* | Publishes a fresh status message |
Example commands:
```bash
# Refresh from Unsplash
mosquitto_pub -h 192.168.1.100 -t "spectra/command/refresh" -n
# Clear the display
mosquitto_pub -h 192.168.1.100 -t "spectra/command/clear" -n
# Show a specific image from the library (ID 5)
mosquitto_pub -h 192.168.1.100 -t "spectra/command/show/5" -n
# Request a status update
mosquitto_pub -h 192.168.1.100 -t "spectra/command/status" -n
```
### Status (publish)
Spectra publishes JSON status messages to `spectra/status`.
**Connection event** (published on successful connect):
```json
{
"action": "connected",
"status": "ok"
}
```
**Command acknowledgements** (published after processing each command):
```json
{
"action": "refresh",
"status": "triggered"
}
```
On error (e.g. invalid image ID):
```json
{
"action": "show",
"status": "error",
"error": "image_not_found"
}
```
**Heartbeat** (published every 5 minutes):
```json
{
"action": "heartbeat",
"status": "running",
"timestamp": 1719334800.0
}
```
## Home Assistant integration
You can integrate Spectra into Home Assistant using MQTT. Add the following to your `configuration.yaml`:
```yaml
mqtt:
sensor:
- name: "Spectra Status"
state_topic: "spectra/status"
value_template: "{{ value_json.action }}"
button:
- name: "Spectra Refresh"
command_topic: "spectra/command/refresh"
payload_press: ""
- name: "Spectra Clear"
command_topic: "spectra/command/clear"
payload_press: ""
```
## Example: Node-RED flow
A Node-RED flow can listen for commands via HTTP (from a web dashboard) and forward them to MQTT, or forward MQTT status events to a database.
```json
[{"id":"spectra-mqtt","type":"mqtt in","topic":"spectra/command/#","name":"Spectra Commands"}]
```
Subscribe to `spectra/status` to build a real-time dashboard with display state, last action, and connectivity status.
## Debugging
Check the display service logs for MQTT connection events:
```bash
journalctl -u spectra -f | grep MQTT
```
A successful connection logs:
```
MQTT connected to 192.168.1.100:1883
MQTT subscribed to spectra/command/#
```
A failed connection logs the return code (see [MQTT spec](https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718035) for rc meanings).

314
docs/web-interface.md Normal file
View File

@@ -0,0 +1,314 @@
# Spectra Web Interface
## Status
| Step | Description | Status | Date |
|------|-------------|--------|------|
| 1 | Scaffold: web module, Flask app factory, models, templates, CLI integration | ✅ Done | 2026-06-25 |
| 2 | Config API: read/write config.yaml from web UI | ✅ Done | 2026-06-25 |
| 3 | Image upload: endpoint, validation, gallery | ✅ Done | 2026-06-25 |
| 4 | Display trigger: trigger file in cli.py, Show Now | ✅ Done | 2026-06-25 |
| 5 | Rotation queue: select images, weight system | ✅ Done | 2026-06-25 |
| 6 | Schedule editor: full schedule in UI | ✅ Done | 2026-06-25 |
| 7 | systemd service + installer updates | ✅ Done | 2026-06-25 |
| 8 | Polish: preview, responsiveness, error handling | ✅ Done | 2026-06-25 |
| 9 | MQTT integration: commands, status, heartbeat | ✅ Done | 2026-06-25 |
| 10 | Drag-and-drop multi-image upload | ✅ Done | 2026-06-25 |
| 11 | Config hot-reload in display loop | ✅ Done | 2026-06-25 |
| 12 | Unsplash image caching to library DB | ✅ Done | 2026-06-25 |
---
## Architecture
```
Raspberry Pi
┌────────────────────────────────────────────────────────────┐
│ │
│ ┌──────────┐ ┌──────────────┐ ┌─────────────────┐ │
│ │ Browser │◄──►│ Flask App │───►│ Display Loop │ │
│ │ (GUI) │ │ (port 5000) │ │ (cli.py) │ │
│ └──────────┘ │ │ │ │ │
│ │ /api/* │ │ ┌───────────┐ │ │
│ │ /gallery │ │ │ fetcher │ │ │
│ │ /settings │ │ │ display │ │ │
│ │ /upload │ │ └───────────┘ │ │
│ └──────┬───────┘ └────────┬────────┘ │
│ │ │ │
│ ┌────▼─────┐ ┌─────▼──────┐ │
│ │ SQLite │ │ trigger │ │
│ │ web.db │ │ .json │ │
│ └──────────┘ └────────────┘ │
│ │ │
│ ┌────▼─────┐ │
│ │ uploads/ │ │
│ │ (images) │ │
│ └──────────┘ │
└────────────────────────────────────────────────────────────┘
```
### Key decisions
| Decision | Choice | Rationale |
|----------|--------|-----------|
| Web framework | Flask | Lightweight, zero-config, well-suited for embedded, huge ecosystem |
| Frontend | Server-rendered HTML + htmx | No build step, minimal CPU/RAM on Pi, dynamic UI with minimal JS |
| Styling | Pico CSS (CDN) | Minimal, responsive, classless CSS framework — no custom CSS needed for basics |
| Database | SQLite via Flask-SQLAlchemy | No daemon needed, single file, built into Python. ORM keeps code clean |
| Image validation | Pillow `verify()` + extension whitelist | Double-check: extension filter prevents upload of non-image files, Pillow verify catches corrupted data |
| Config persistence | `ConfigManager` class wraps `config.py` | Single point for read/write, falls back to `~/.config/spectra/config.yaml` when `/etc/` is not writable |
| Cache directory | `SPECTRA_CACHE_DIR` env var, fallback `~/.cache/spectra/` | Allows override in constrained environments, default works without root |
| Display loop integration | File-based trigger (`trigger.json`) | Minimal changes to existing loop, no DB dependency in the display loop, survives restarts |
| CLI integration | `spectra web` subcommand via `sys.argv` check | Preserves full backward compatibility — `spectra --once` still works unchanged |
| Config API PUT for individual keys | `PUT /api/config/<section>/<key>` with `{"value": ...}` body | Simple, type-safe, avoids parsing complex nested patches on the frontend |
---
## Module structure
```
spectra/
├── __init__.py
├── __main__.py
├── cli.py # CLI entry point, display loop, trigger handling
├── config.py # Config loader (read-only, shared)
├── config_manager.py # NEW — read/write config.yaml (write support)
├── display.py # Display abstraction (+ show_file, clear)
├── fetcher.py # Unsplash API client
├── trigger.py # Shared trigger file read/write (thread-safe)
├── mqtt.py # MQTT client: commands + status publishing
├── library.py # Unsplash image caching to SQLite library
├── web/
│ ├── __init__.py
│ ├── server.py # Flask app factory + all routes
│ ├── models.py # SQLAlchemy models (Image, Rotation, Setting)
│ ├── templates/
│ │ ├── base.html # Layout with Pico CSS nav
│ │ ├── index.html # Dashboard
│ │ ├── gallery.html # Image gallery
│ │ ├── upload.html # Upload form
│ │ ├── settings.html# Settings editor
│ │ └── preview.html # Image preview
│ └── static/
│ ├── css/app.css
│ └── js/app.js
└── systemd/
├── spectra.service # Existing — display loop
└── spectra-web.service # NEW — web server
```
## Database
Managed by Flask-SQLAlchemy. Auto-creates on first `create_app()`. Three tables:
### `image`
| Column | Type | Notes |
|--------|------|-------|
| id | INTEGER PK | autoincrement |
| filename | TEXT | stored filename on disk |
| source | TEXT | `'upload'` or `'unsplash'` |
| unsplash_id | TEXT | nullable, Unsplash photo ID |
| title | TEXT | user-friendly title |
| author | TEXT | credit line |
| width | INTEGER | image pixel width |
| height | INTEGER | image pixel height |
| filepath | TEXT | absolute path to cached file |
| created_at | DATETIME | auto-set on creation |
| seen_count | INTEGER | times shown on display |
| last_shown | DATETIME | nullable, last display time |
### `rotation`
| Column | Type | Notes |
|--------|------|-------|
| id | INTEGER PK | autoincrement |
| image_id | INTEGER FK → image.id | |
| weight | INTEGER | higher = shown more often |
| active | BOOLEAN | toggle on/off without removing |
| created_at | DATETIME | auto-set |
### `setting`
| Column | Type | Notes |
|--------|------|-------|
| key | TEXT PK | setting name |
| value | TEXT | setting value (stringified) |
---
## API endpoints
### Config
| Method | Path | Description |
|--------|------|-------------|
| `GET` | `/api/config` | Full config dict |
| `PATCH` | `/api/config` | Deep-merge changes into config and save |
| `GET` | `/api/config/<section>` | Single section (e.g. `display`, `schedule`) |
| `PUT` | `/api/config/<section>/<key>` | Set a single value (`{"value": ...}`) |
### Images
| Method | Path | Description |
|--------|------|-------------|
| `GET` | `/api/images` | Paginated list (`?page=&per_page=&source=`) |
| `POST` | `/api/images` | Upload (multipart: file + title + author + show_now) |
| `GET` | `/api/images/<id>` | Image metadata |
| `DELETE` | `/api/images/<id>` | Delete (removes DB record + file + rotation entries) |
| `GET` | `/api/images/<id>/file` | Serve the image file |
| `GET` | `/api/images/<id>/thumbnail` | Auto-generated 320×240 thumbnail |
| `POST` | `/api/images/<id>/show` | Trigger display of this image |
### Display
| Method | Path | Description |
|--------|------|-------------|
| `GET` | `/api/preview/<id>` | E-paper processed preview (crop + resize applied) |
| `POST` | `/api/display/refresh` | Trigger Unsplash refresh |
| `POST` | `/api/display/clear` | Clear display (white) |
| `GET` | `/api/display/status` | Current state (simulation, resolution, pending trigger, last shown) |
### Rotation
| Method | Path | Description |
|--------|------|-------------|
| `GET` | `/api/rotation` | All rotation entries with nested image data |
| `POST` | `/api/rotation` | Add images (`{"image_ids": [1,2,3]}`) |
| `DELETE` | `/api/rotation/<id>` | Remove entry |
| `PATCH` | `/api/rotation/<id>` | Update `weight` or `active` |
---
## Trigger mechanism
The display loop (`cli.py`) and web server (`web/server.py`) coordinate via a JSON trigger file at `SPECTRA_CACHE_DIR/trigger.json`.
### Trigger actions
| Action | Payload | Effect |
|--------|---------|--------|
| `refresh` | `{"action": "refresh"}` | Next loop iteration fetches from Unsplash |
| `show_upload` | `{"action": "show_upload", "path": "...", "image_id": 1}` | Loads the given image file and displays it |
| `clear` | `{"action": "clear"}` | Sets display to blank white |
### Flow
1. Web server writes trigger file via `write_trigger()` (thread-safe with lock)
2. Display loop checks trigger at top of each iteration via `handle_trigger()`
3. If trigger exists: execute action, then `clear_trigger()` (delete file)
4. If no trigger: default Unsplash refresh
Trigger file path resolution (in order of priority):
1. `SPECTRA_CACHE_DIR` environment variable
2. `/var/cache/spectra/` (fallback if writable)
3. `~/.cache/spectra/` (fallback for non-root/dev environments)
---
## CLI
### `spectra` (display — unchanged)
```
spectra [-h] [-c CONFIG] [--once] [--simulate] [--width WIDTH] [--height HEIGHT] [-v]
```
Full backward compatibility preserved. The display loop now additionally checks for trigger files before each refresh.
### `spectra web` (web server — new)
```
spectra web [-h] [-c CONFIG] [--host HOST] [--port PORT] [-v]
```
Starts the Flask development server. Routes are registered inline in `server.py` via `_register_routes()`.
---
## systemd
### `spectra-web.service`
```ini
[Unit]
Description=Spectra Web Interface
After=network-online.target
Wants=network-online.target
[Service]
Type=simple
ExecStart=/usr/local/bin/spectra web
Restart=on-failure
RestartSec=10
StandardOutput=journal
StandardError=journal
[Install]
WantedBy=multi-user.target
```
### `spectra.service` (unchanged)
Display loop — runs `spectra` without flags (continuous mode).
---
## Dependencies
```
# requirements.txt (additions)
Flask>=3.0.0
Flask-SQLAlchemy>=3.1.0
```
Flask and Flask-SQLAlchemy are the only new dependencies — Pico CSS and htmx are loaded from CDN (no npm/build step needed).
---
## Installation
`install.sh` updated:
1. Inky library (Raspberry Pi only)
2. Python dependencies (`pip install -r requirements.txt`)
3. Package install (`pip install -e .`)
4. Display service (copies unit, enables)
5. **Web service** (copies `spectra-web.service`, enables)
---
## Security
- Web server binds to `0.0.0.0:5000` by default (Pi's LAN)
- No built-in authentication (recommend nginx reverse proxy + HTTP basic auth for production)
- File uploads: extension whitelist + Pillow `verify()` — no arbitrary code execution
- Uploaded images stored outside web root (`/var/cache/spectra/uploads/` or `~/.cache/spectra/uploads/`)
- Unsplash API key stored in config (permissions on `/etc/spectra/config.yaml` or `~/.config/spectra/config.yaml`)
---
## Known limitations
- **No multi-user auth**: The web server has no authentication. Recommended for LAN use only, or behind an nginx reverse proxy with HTTP basic auth.
- **Resolution changes need hardware re-init**: If display resolution changes in the config while using real hardware (not simulation), the Inky display must be re-initialized. The display service picks up the new values but the hardware may not resize until next boot.
## Implementation notes
- **Cache directory fallback**: The project uses `/var/cache/spectra/` when run as root (Raspberry Pi), and `~/.cache/spectra/` when run as a regular user (dev). Override with `SPECTRA_CACHE_DIR` env var.
- **Config write safety**: `ConfigManager` resolves the config path by checking search paths in order, falling back to `~/.config/spectra/config.yaml` for writes when none exist. Always uses `os.path.abspath()` to avoid directory resolution issues.
- **Image upload validation**: Two-stage — extension check against whitelist (`{png, jpg, jpeg, gif, bmp, webp}`) then Pillow `verify()` to catch corrupted files. Max file size: 20MB.
- **Thumbnail generation**: 320×240 PNG thumbnails generated on first request and cached alongside the source file as `filename.thumb`. Uses `Image.thumbnail()` for aspect-ratio preservation.
- **No CSRF protection**: Current implementation has no CSRF tokens since htmx sends `POST`/`DELETE` requests. Add Flask-WTF or a simple token check before production deployment on a public network.
- **`display.clear()`**: Creates a blank palette-mode (`"P"`) white image and pushes it to the display. Avoids relying on internal `_buf` attributes that differ across Inky board revisions.
---
## Future work (post-v1 — not implemented)
- [ ] WebSocket push for live display status updates
- [ ] S3/Nextcloud import sources
- [ ] Multi-user auth via nginx reverse proxy
- [ ] OTA software updates
- [ ] Touchscreen kiosk mode (mirrors the display locally)
- [ ] zrok/Cloudflare Tunnel for secure remote access

79
install.sh Executable file
View File

@@ -0,0 +1,79 @@
#!/usr/bin/env bash
set -euo pipefail
echo "========================================="
echo " Spectra - Inky Impression Installer"
echo "========================================="
# --- Inky library (Raspberry Pi only) ---
if [[ -f /proc/device-tree/model ]] && grep -qi "raspberry pi" /proc/device-tree/model 2>/dev/null; then
echo ""
echo "[1/5] Installing Inky library..."
if python3 -c "import inky" 2>/dev/null; then
echo " Inky library already installed, skipping."
else
echo " Cloning from Pimoroni..."
cd /tmp
git clone --depth=1 https://github.com/pimoroni/inky.git
cd inky
./install.sh
cd /
rm -rf /tmp/inky
fi
else
echo ""
echo "[1/5] No Raspberry Pi detected — skipping Inky library install."
echo " Use --simulate to test without a display."
fi
# --- Python dependencies ---
echo ""
echo "[2/5] Installing Python dependencies..."
python3 -m pip install --upgrade pip
python3 -m pip install -r requirements.txt
# --- Package ---
echo ""
echo "[3/5] Installing spectra package..."
python3 -m pip install -e .
# --- Display service ---
echo ""
echo "[4/5] Setting up display service..."
sudo mkdir -p /etc/spectra
if [[ ! -f /etc/spectra/config.yaml ]]; then
sudo cp config.yaml /etc/spectra/config.yaml
echo " Default config copied to /etc/spectra/config.yaml"
fi
sudo cp systemd/spectra.service /etc/systemd/system/
sudo systemctl daemon-reload
sudo systemctl enable spectra.service
echo " Display service enabled"
# --- Web service ---
echo ""
echo "[5/5] Setting up web interface service..."
sudo cp systemd/spectra-web.service /etc/systemd/system/
sudo systemctl daemon-reload
sudo systemctl enable spectra-web.service
echo " Web interface service enabled"
echo ""
echo "========================================="
echo " Installation complete!"
echo ""
echo " To configure:"
echo " sudo nano /etc/spectra/config.yaml"
echo " (add your Unsplash API access key)"
echo ""
echo " Display service:"
echo " sudo systemctl start spectra"
echo ""
echo " Web interface:"
echo " sudo systemctl start spectra-web"
echo " http://<raspberry-pi-ip>:5000"
echo ""
echo " To test without hardware:"
echo " spectra --simulate --once"
echo "========================================="

30
pyproject.toml Normal file
View File

@@ -0,0 +1,30 @@
[build-system]
requires = ["setuptools"]
build-backend = "setuptools.build_meta"
[project]
name = "spectra-display"
version = "0.1.0"
description = "Display random Unsplash images on Inky Impression e-paper"
requires-python = ">=3.9"
dependencies = [
"Pillow>=10.0.0",
"requests>=2.28.0",
"PyYAML>=6.0",
"Flask>=3.0.0",
"Flask-SQLAlchemy>=3.1.0",
"paho-mqtt>=1.6.0",
]
[project.scripts]
spectra = "spectra.cli:main"
[project.optional-dependencies]
dev = [
"pytest>=7.0.0",
"pytest-flask>=1.2.0",
]
[tool.setuptools.packages.find]
include = ["spectra*", "tests"]

6
requirements.txt Normal file
View File

@@ -0,0 +1,6 @@
Pillow>=10.0.0
requests>=2.28.0
PyYAML>=6.0
Flask>=3.0.0
Flask-SQLAlchemy>=3.1.0
paho-mqtt>=1.6.0

0
spectra/__init__.py Normal file
View File

3
spectra/__main__.py Normal file
View File

@@ -0,0 +1,3 @@
from .cli import main
main()

313
spectra/cli.py Normal file
View File

@@ -0,0 +1,313 @@
#!/usr/bin/env python3
import argparse
import logging
import os
import random
import signal
import sys
import time
from .config import load_config
from .fetcher import UnsplashFetcher
from .display import InkyDisplay
from .trigger import clear_trigger, read_and_clear_trigger, read_trigger
logger = logging.getLogger("spectra")
running = True
def signal_handler(signum, frame):
global running
running = False
def setup_logging(verbose):
level = logging.DEBUG if verbose else logging.INFO
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(
logging.Formatter(
"%(asctime)s [%(levelname)s] %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
)
root = logging.getLogger("spectra")
root.setLevel(level)
root.handlers.clear()
root.addHandler(handler)
def refresh(config, display):
unsplash_cfg = config["unsplash"]
if not unsplash_cfg.get("access_key"):
logger.error("Unsplash access_key is not configured")
return False
display_cfg = config.get("display", {})
orientation_deg = display_cfg.get("orientation", 0)
res = display_cfg.get("resolution", {})
phys_w = res.get("width", 1600)
phys_h = res.get("height", 1200)
if orientation_deg in (90, 270):
eff_w, eff_h = phys_h, phys_w
else:
eff_w, eff_h = phys_w, phys_h
unsplash_orient = "portrait" if eff_h > eff_w else "landscape" if eff_w > eff_h else "squarish"
fetcher = UnsplashFetcher(
access_key=unsplash_cfg["access_key"],
query=unsplash_cfg.get("query") or None,
orientation=unsplash_orient,
collections=unsplash_cfg.get("collections") or None,
)
try:
logger.info("Fetching random photo from Unsplash...")
image, photo = fetcher.fetch_and_download()
author = photo.get("user", {}).get("name", "unknown")
desc = (
photo.get("alt_description")
or photo.get("description")
or "untitled"
)
logger.info("Photo: %s by %s", desc, author)
display.show(image)
_save_to_library(image, photo, config)
return True
except Exception as e:
logger.error("Failed to update display: %s", e)
return False
def _save_to_library(image, photo, config):
try:
from .library import save_unsplash_image
sat = config.get("display", {}).get("saturation", 0.5)
res = config.get("display", {}).get("resolution", {})
w = res.get("width", 1600)
h = res.get("height", 1200)
orientation_deg = config.get("display", {}).get("orientation", 0)
if orientation_deg in (90, 270):
w, h = h, w
save_unsplash_image(image, photo, saturation=sat, width=w, height=h)
except Exception as e:
logger.warning("Failed to cache Unsplash image to library: %s", e)
def handle_trigger(config, display):
trigger = read_and_clear_trigger()
if trigger is None:
return False
action = trigger.get("action")
logger.info("Handling trigger: %s", action)
if action == "refresh":
refresh(config, display)
elif action == "show_upload":
path = trigger.get("path")
if path and os.path.exists(path):
try:
logger.info("Showing uploaded image: %s", path)
display.show_file(path)
except Exception as e:
logger.error("Failed to show uploaded image: %s", e)
else:
logger.warning("Uploaded image not found: %s", path)
elif action == "clear":
display.clear()
else:
logger.warning("Unknown trigger action: %s", action)
return True
def _reload_config(old_config, display, mqtt_client=None, config_path=None):
new_config = load_config(config_path)
old_sat = old_config.get("display", {}).get("saturation", 0.5)
new_sat = new_config.get("display", {}).get("saturation", 0.5)
if old_sat != new_sat:
display.saturation = new_sat
logger.info("Config hot-reload: saturation updated to %.2f", new_sat)
old_w = old_config.get("display", {}).get("resolution", {}).get("width", 1600)
new_w = new_config.get("display", {}).get("resolution", {}).get("width", 1600)
if old_w != new_w:
display.width = new_w
logger.info("Config hot-reload: width updated to %d", new_w)
old_h = old_config.get("display", {}).get("resolution", {}).get("height", 1200)
new_h = new_config.get("display", {}).get("resolution", {}).get("height", 1200)
if old_h != new_h:
display.height = new_h
logger.info("Config hot-reload: height updated to %d", new_h)
old_orient = int(old_config.get("display", {}).get("orientation", 0))
new_orient = int(new_config.get("display", {}).get("orientation", 0))
if old_orient != new_orient:
display.orientation = new_orient
logger.info("Config hot-reload: orientation updated to %d", new_orient)
if old_sat != new_sat or old_w != new_w or old_h != new_h or old_orient != new_orient:
logger.info("Config hot-reload: display settings updated")
if mqtt_client is not None:
old_mqtt = old_config.get("mqtt", {})
new_mqtt = new_config.get("mqtt", {})
if old_mqtt != new_mqtt:
logger.info("Config hot-reload: MQTT settings changed, restarting client")
mqtt_client.reconfigure(new_mqtt)
return new_config
def run_loop(config, display, once=False, mqtt_client=None, config_path=None):
global running
if once:
if not handle_trigger(config, display):
refresh(config, display)
return
interval = config["schedule"]["interval_hours"] * 3600
delay = config["schedule"].get("random_delay_seconds", 0)
logger.info(
"Starting loop (every %d h%s)",
config["schedule"]["interval_hours"],
f" + up to {delay}s random" if delay else "",
)
while running:
config = _reload_config(config, display, mqtt_client, config_path)
interval = config["schedule"]["interval_hours"] * 3600
delay = config["schedule"].get("random_delay_seconds", 0)
handled = handle_trigger(config, display)
if not handled:
refresh(config, display)
if not running:
break
sleep_time = interval + (random.randint(0, delay) if delay > 0 else 0)
logger.info("Next update in %d seconds", sleep_time)
for _ in range(sleep_time // 5):
if not running:
break
time.sleep(5)
remaining = sleep_time % 5
if remaining and running:
time.sleep(remaining)
def run_web(args):
from .web.server import create_app
setup_logging(args.verbose)
app = create_app(config_path=args.config)
logger.info("Starting web interface on http://%s:%d", args.host, args.port)
use_debug = args.verbose and args.host in ("127.0.0.1", "localhost")
if args.verbose and not use_debug:
logger.warning("Debug mode disabled when binding to %s (would expose debugger to network)", args.host)
app.run(host=args.host, port=args.port, debug=use_debug)
def main():
if len(sys.argv) > 1 and sys.argv[1] == "web":
web_parser = argparse.ArgumentParser(
description="Start the Spectra web interface",
prog="spectra web",
)
web_parser.add_argument("-c", "--config", help="Config file path")
web_parser.add_argument(
"--host", default="0.0.0.0", help="Host to bind (default: 0.0.0.0)"
)
web_parser.add_argument(
"--port", type=int, default=5000, help="Port to bind (default: 5000)"
)
web_parser.add_argument(
"-v", "--verbose", action="store_true", help="Verbose logging"
)
args = web_parser.parse_args(sys.argv[2:])
run_web(args)
return
parser = argparse.ArgumentParser(
description="Display random Unsplash images on Inky Impression"
)
parser.add_argument("-c", "--config", help="Config file path")
parser.add_argument(
"--once", action="store_true", help="Run once and exit"
)
parser.add_argument(
"--simulate",
action="store_true",
help="Save image to /tmp instead of displaying on hardware",
)
parser.add_argument(
"--width",
type=int,
default=None,
help="Display width in pixels (simulation mode)",
)
parser.add_argument(
"--height",
type=int,
default=None,
help="Display height in pixels (simulation mode)",
)
parser.add_argument(
"-v", "--verbose", action="store_true", help="Verbose logging"
)
args = parser.parse_args()
setup_logging(args.verbose)
config = load_config(args.config)
if not config["unsplash"].get("access_key"):
logger.error(
"Unsplash access_key not configured. "
"Set it in config.yaml or use --config to specify a config file."
)
sys.exit(1)
resolution = config["display"].get("resolution", {})
display = InkyDisplay(
saturation=config["display"].get("saturation", 0.5),
simulate=args.simulate,
width=resolution.get("width", 1600) if args.width is None else args.width,
height=resolution.get("height", 1200) if args.height is None else args.height,
orientation=config.get("display", {}).get("orientation", 0),
)
try:
display.initialize()
except Exception:
logger.error("Display initialization failed")
if not args.simulate:
sys.exit(1)
signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)
mqtt_client = None
mqtt_cfg = config.get("mqtt", {})
if mqtt_cfg.get("enabled") and not args.once:
try:
from .mqtt import MqttClient
mqtt_client = MqttClient(mqtt_cfg)
mqtt_client.start()
except Exception as e:
logger.warning("MQTT client failed to start: %s", e)
try:
run_loop(config, display, once=args.once, mqtt_client=mqtt_client, config_path=args.config)
finally:
if mqtt_client:
mqtt_client.stop()
if __name__ == "__main__":
main()

78
spectra/config.py Normal file
View File

@@ -0,0 +1,78 @@
import copy
import logging
import os
import yaml
logger = logging.getLogger(__name__)
DEFAULT_CONFIG = {
"unsplash": {
"access_key": "",
"query": "",
"orientation": "landscape",
"collections": "",
},
"display": {
"saturation": 0.5,
"orientation": 0,
"resolution": {
"width": 1600,
"height": 1200,
},
},
"schedule": {
"interval_hours": 1,
"random_delay_seconds": 300,
},
"mqtt": {
"enabled": False,
"broker": "localhost",
"port": 1883,
"topic_prefix": "spectra",
"client_id": "spectra-display",
"username": "",
"password": "",
},
"paths": {
"cache": "/var/cache/spectra",
},
}
CONFIG_SEARCH_PATHS = [
"config.yaml",
os.path.expanduser("~/.config/spectra/config.yaml"),
"/etc/spectra/config.yaml",
]
def load_config(path=None):
config = copy.deepcopy(DEFAULT_CONFIG)
paths = [path] if path else CONFIG_SEARCH_PATHS
loaded = False
for p in paths:
expanded = os.path.expanduser(p)
if os.path.exists(expanded):
with open(expanded, encoding="utf-8") as f:
user_config = yaml.safe_load(f) or {}
_deep_merge(config, user_config)
logger.info("Loaded config from %s", expanded)
loaded = True
break
if not loaded and path:
logger.warning("Config file not found: %s", path)
elif not loaded:
logger.warning("No config file found, using defaults")
return config
def _deep_merge(base, override):
for key, value in override.items():
if key in base and isinstance(base[key], dict) and isinstance(value, dict):
_deep_merge(base[key], value)
else:
base[key] = value

60
spectra/config_manager.py Normal file
View File

@@ -0,0 +1,60 @@
import logging
import os
import yaml
from .config import DEFAULT_CONFIG, _deep_merge, load_config
logger = logging.getLogger(__name__)
class ConfigManager:
def __init__(self, path=None):
self.path = path
self.config = load_config(path)
def refresh(self):
self.config = load_config(self.path)
return self.config
def get(self, *keys):
value = self.config
for key in keys:
if isinstance(value, dict):
value = value.get(key)
else:
return None
return value
def update(self, changes):
_deep_merge(self.config, changes)
self._write()
return self.config
def set_nested(self, keys, value):
target = self.config
for key in keys[:-1]:
if key not in target or not isinstance(target[key], dict):
target[key] = {}
target = target[key]
target[keys[-1]] = value
self._write()
def _write(self):
path = self._resolve_path()
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w") as f:
yaml.dump(self.config, f, default_flow_style=False)
logger.info("Config written to %s", path)
def _resolve_path(self):
if self.path:
return os.path.abspath(os.path.expanduser(self.path))
from .config import CONFIG_SEARCH_PATHS
for p in CONFIG_SEARCH_PATHS:
expanded = os.path.abspath(os.path.expanduser(p))
if os.path.exists(expanded):
return expanded
fallback = os.path.abspath(os.path.expanduser("~/.config/spectra/config.yaml"))
os.makedirs(os.path.dirname(fallback), exist_ok=True)
return fallback

99
spectra/display.py Normal file
View File

@@ -0,0 +1,99 @@
import logging
from PIL import Image
logger = logging.getLogger(__name__)
class InkyDisplay:
def __init__(self, saturation=0.5, simulate=False, width=1600, height=1200, orientation=0):
self.saturation = saturation
self.simulate = simulate
self.inky = None
self.width = width
self.height = height
self.orientation = int(orientation) if orientation is not None else 0
@property
def effective_width(self):
return self.height if self.orientation in (90, 270) else self.width
@property
def effective_height(self):
return self.width if self.orientation in (90, 270) else self.height
def initialize(self):
if self.simulate:
logger.info(
"Simulation mode: using %dx%d virtual display, orientation=%d",
self.width,
self.height,
self.orientation,
)
return
try:
from inky.auto import auto
self.inky = auto(ask_user=False, verbose=False)
self.width, self.height = self.inky.resolution
logger.info("Display initialized: %dx%d", self.width, self.height)
except Exception as e:
logger.error("Failed to initialize display: %s", e)
raise
def process_image(self, image):
target_w = self.effective_width
target_h = self.effective_height
target_aspect = target_w / target_h
img_w, img_h = image.size
img_aspect = img_w / img_h
if abs(img_aspect - target_aspect) > 0.01:
if img_aspect > target_aspect:
new_w = int(img_h * target_aspect)
offset = (img_w - new_w) // 2
image = image.crop((offset, 0, offset + new_w, img_h))
else:
new_h = int(img_w / target_aspect)
offset = (img_h - new_h) // 2
image = image.crop((0, offset, img_w, offset + new_h))
image = image.resize((target_w, target_h), Image.LANCZOS)
if self.orientation != 0:
image = image.rotate(-self.orientation, expand=True)
return image
def show(self, image):
processed = self.process_image(image)
if self.simulate or self.inky is None:
path = "/tmp/spectra_last.png"
processed.save(path)
logger.info("Saved processed image to %s", path)
return
try:
self.inky.set_image(processed, saturation=self.saturation)
except TypeError:
self.inky.set_image(processed)
self.inky.show()
logger.info("Display updated")
def show_file(self, filepath):
with Image.open(filepath) as image:
self.show(image)
def clear(self):
if self.simulate or self.inky is None:
logger.info("Simulation mode: skipping clear")
return
blank = Image.new("P", (self.effective_width, self.effective_height), 255)
if self.orientation != 0:
blank = blank.rotate(-self.orientation, expand=True)
self.inky.set_image(blank)
self.inky.show()
logger.info("Display cleared")

44
spectra/fetcher.py Normal file
View File

@@ -0,0 +1,44 @@
import logging
from io import BytesIO
import requests
from PIL import Image
logger = logging.getLogger(__name__)
class UnsplashFetcher:
BASE_URL = "https://api.unsplash.com"
def __init__(self, access_key, query=None, orientation=None, collections=None):
self.session = requests.Session()
self.session.headers.update({"Authorization": f"Client-ID {access_key}"})
self.params = {"content_filter": "high"}
if query:
self.params["query"] = query
if orientation:
self.params["orientation"] = orientation
if collections:
self.params["collections"] = collections
def get_random_photo(self):
resp = self.session.get(f"{self.BASE_URL}/photos/random", params=self.params)
resp.raise_for_status()
return resp.json()
def download(self, photo_data, target_width=1600):
url = photo_data["urls"]["raw"]
sep = "&" if "?" in url else "?"
url += f"{sep}w={target_width * 2}"
resp = self.session.get(url)
resp.raise_for_status()
return Image.open(BytesIO(resp.content))
def fetch_and_download(self, target_width=1600):
photo = self.get_random_photo()
image = self.download(photo, target_width)
return image, photo
def close(self):
self.session.close()

99
spectra/library.py Normal file
View File

@@ -0,0 +1,99 @@
import logging
import os
import sqlite3
import uuid
from datetime import datetime, timezone
logger = logging.getLogger(__name__)
def _db_path():
from .trigger import cache_dir
return os.path.join(cache_dir(), "web.db")
def _uploads_dir():
from .trigger import cache_dir
d = os.path.join(cache_dir(), "uploads")
os.makedirs(d, exist_ok=True)
return d
def save_unsplash_image(image, photo_data, saturation=0.5, width=1600, height=1200):
from .display import InkyDisplay
unsplash_id = photo_data.get("id", uuid.uuid4().hex)
author = photo_data.get("user", {}).get("name", "")
desc = (
photo_data.get("alt_description")
or photo_data.get("description")
or "Unsplash photo"
)
conn = sqlite3.connect(_db_path())
conn.execute("PRAGMA journal_mode=WAL")
conn.executescript("""
CREATE TABLE IF NOT EXISTS image (
id INTEGER PRIMARY KEY AUTOINCREMENT,
filename TEXT NOT NULL,
source TEXT NOT NULL DEFAULT 'upload',
unsplash_id TEXT,
title TEXT,
author TEXT,
width INTEGER,
height INTEGER,
filepath TEXT NOT NULL,
created_at TEXT,
seen_count INTEGER DEFAULT 0,
last_shown TEXT
);
""")
conn.commit()
existing = conn.execute(
"SELECT id FROM image WHERE source='unsplash' AND unsplash_id=?",
(unsplash_id,),
).fetchone()
if existing:
conn.close()
logger.info("Unsplash photo already cached: id=%d", existing[0])
return existing[0]
filename = f"unsplash_{unsplash_id}_{uuid.uuid4().hex[:8]}.png"
dest = os.path.join(_uploads_dir(), filename)
image.save(dest, "PNG")
display = InkyDisplay(saturation=saturation, simulate=True, width=width, height=height)
processed = display.process_image(image.copy())
proc_w, proc_h = processed.size
now = datetime.now(timezone.utc).isoformat()
try:
conn.execute(
"""INSERT INTO image
(filename, source, unsplash_id, title, author, width, height,
filepath, created_at, seen_count, last_shown)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?)""",
(filename, "unsplash", unsplash_id, desc, author,
proc_w, proc_h, dest, now, now),
)
conn.commit()
image_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
logger.info(
"Cached Unsplash photo to library: id=%d, title=%s, author=%s",
image_id, desc, author,
)
return image_id
except Exception as e:
logger.error("Failed to save Unsplash image to library: %s", e)
try:
os.remove(dest)
except OSError:
pass
return None
finally:
try:
conn.close()
except Exception:
pass

166
spectra/mqtt.py Normal file
View File

@@ -0,0 +1,166 @@
import json
import logging
import os
import sqlite3
import threading
import time
import paho.mqtt.client as mqtt
from .trigger import cache_dir, write_trigger
logger = logging.getLogger(__name__)
def _db_path():
return os.path.join(cache_dir(), "web.db")
def _lookup_image_path(image_id):
conn = sqlite3.connect(_db_path())
conn.execute("PRAGMA journal_mode=WAL")
cur = conn.execute("SELECT filepath FROM image WHERE id = ?", (image_id,))
row = cur.fetchone()
conn.close()
return row[0] if row else None
class MqttClient:
def __init__(self, config):
self._setup(config)
self._create_client()
self._status_interval = 300
self._stopped = threading.Event()
self._thread = None
def _setup(self, config):
self.broker = config.get("broker", "localhost")
self.port = config.get("port", 1883)
self.prefix = config.get("topic_prefix", "spectra")
self.client_id = config.get("client_id", "spectra-display")
self._username = config.get("username", "")
self._password = config.get("password", "")
def _create_client(self):
self._client = mqtt.Client(
client_id=self.client_id,
protocol=mqtt.MQTTv311,
)
if self._username:
self._client.username_pw_set(self._username, self._password)
self._client.on_connect = self._on_connect
self._client.on_disconnect = self._on_disconnect
self._client.on_message = self._on_message
def start(self):
if self._thread and self._thread.is_alive():
return
self._stopped.clear()
self._thread = threading.Thread(target=self._run, daemon=True, name="mqtt")
self._thread.start()
logger.info(
"MQTT client starting: %s:%d (prefix=%s, id=%s)",
self.broker, self.port, self.prefix, self.client_id,
)
def stop(self):
if self._thread and self._thread.is_alive():
self._stopped.set()
self._thread.join(timeout=5)
if self._thread and self._thread.is_alive():
logger.warning("MQTT thread did not stop within timeout")
self._client.loop_stop()
self._client.disconnect()
def reconfigure(self, new_config):
enabled = new_config.get("enabled", False)
if not enabled:
self.stop()
return
same = (
self.broker == new_config.get("broker", "localhost")
and self.port == new_config.get("port", 1883)
and self.prefix == new_config.get("topic_prefix", "spectra")
and self.client_id == new_config.get("client_id", "spectra-display")
and self._username == new_config.get("username", "")
and self._password == new_config.get("password", "")
)
if same and (self._thread and self._thread.is_alive()):
return
self.stop()
self._setup(new_config)
self._create_client()
self.start()
def publish_status(self, message=None):
payload = json.dumps(message or {"action": "status", "status": "ok"})
self._client.publish(f"{self.prefix}/status", payload, qos=1, retain=True)
def _run(self):
try:
self._client.connect_async(self.broker, self.port, keepalive=60)
self._client.loop_start()
except Exception as e:
logger.error("MQTT connection failed: %s", e)
return
last_status = 0
while not self._stopped.wait(1):
now = time.time()
if now - last_status >= self._status_interval:
self.publish_status({"action": "heartbeat", "status": "running", "timestamp": now})
last_status = now
self._client.loop_stop()
self._client.disconnect()
def _on_connect(self, client, userdata, flags, rc):
if rc == 0:
logger.info("MQTT connected to %s:%d", self.broker, self.port)
client.subscribe(f"{self.prefix}/command/#", qos=1)
logger.info("MQTT subscribed to %s/command/#", self.prefix)
self.publish_status({"action": "connected", "status": "ok"})
else:
logger.warning("MQTT connection failed (rc=%d)", rc)
def _on_disconnect(self, client, userdata, rc):
if rc != 0:
logger.warning("MQTT disconnected unexpectedly (rc=%d), will reconnect", rc)
def _on_message(self, client, userdata, msg):
topic = msg.topic
payload = msg.payload.decode(errors="replace")
logger.info("MQTT message: %s = %s", topic, payload)
prefix_len = len(self.prefix) + len("/command/")
command = topic[prefix_len:] if len(topic) > prefix_len else ""
if command == "refresh":
write_trigger({"action": "refresh"})
self.publish_status({"action": "refresh", "status": "triggered"})
elif command == "clear":
write_trigger({"action": "clear"})
self.publish_status({"action": "clear", "status": "triggered"})
elif command.startswith("show/"):
image_id_str = command[len("show/"):]
try:
image_id = int(image_id_str)
except ValueError:
logger.warning("MQTT invalid image ID: %s", image_id_str)
self.publish_status({"action": "show", "status": "error", "error": "invalid_image_id"})
return
path = _lookup_image_path(image_id)
if path:
write_trigger({"action": "show_upload", "path": path, "image_id": image_id})
self.publish_status({"action": "show", "status": "triggered", "image_id": image_id})
else:
logger.warning("MQTT image not found: id=%d", image_id)
self.publish_status({"action": "show", "status": "error", "error": "image_not_found"})
elif command == "status":
self.publish_status({"action": "status", "status": "ok"})
else:
logger.warning("MQTT unknown command: %s", command)

63
spectra/trigger.py Normal file
View File

@@ -0,0 +1,63 @@
import json
import os
import threading
_TRIGGER_LOCK = threading.Lock()
def cache_dir():
env = os.environ.get("SPECTRA_CACHE_DIR")
if env:
os.makedirs(env, exist_ok=True)
return env
try:
d = "/var/cache/spectra"
os.makedirs(d, exist_ok=True)
return d
except PermissionError:
d = os.path.expanduser("~/.cache/spectra")
os.makedirs(d, exist_ok=True)
return d
def _trigger_path():
return os.path.join(cache_dir(), "trigger.json")
def write_trigger(data):
path = _trigger_path()
with _TRIGGER_LOCK:
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w") as f:
json.dump(data, f)
def read_trigger():
path = _trigger_path()
try:
with _TRIGGER_LOCK:
with open(path) as f:
return json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
return None
def clear_trigger():
path = _trigger_path()
with _TRIGGER_LOCK:
try:
os.remove(path)
except FileNotFoundError:
pass
def read_and_clear_trigger():
path = _trigger_path()
with _TRIGGER_LOCK:
try:
with open(path) as f:
data = json.load(f)
os.remove(path)
return data
except (FileNotFoundError, json.JSONDecodeError):
return None

0
spectra/web/__init__.py Normal file
View File

45
spectra/web/models.py Normal file
View File

@@ -0,0 +1,45 @@
from datetime import datetime, timezone
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
class Image(db.Model):
__tablename__ = "image"
id = db.Column(db.Integer, primary_key=True)
filename = db.Column(db.String, nullable=False)
source = db.Column(db.String, nullable=False, default="upload")
unsplash_id = db.Column(db.String, nullable=True)
title = db.Column(db.String, nullable=True)
author = db.Column(db.String, nullable=True)
width = db.Column(db.Integer, nullable=True)
height = db.Column(db.Integer, nullable=True)
filepath = db.Column(db.String, nullable=False)
created_at = db.Column(
db.DateTime, default=lambda: datetime.now(timezone.utc)
)
seen_count = db.Column(db.Integer, default=0)
last_shown = db.Column(db.DateTime, nullable=True)
class Rotation(db.Model):
__tablename__ = "rotation"
id = db.Column(db.Integer, primary_key=True)
image_id = db.Column(db.Integer, db.ForeignKey("image.id"), nullable=False)
weight = db.Column(db.Integer, default=1)
active = db.Column(db.Boolean, default=True)
created_at = db.Column(
db.DateTime, default=lambda: datetime.now(timezone.utc)
)
image = db.relationship("Image", backref=db.backref("rotation_entries", lazy="dynamic"))
class Setting(db.Model):
__tablename__ = "setting"
key = db.Column(db.String, primary_key=True)
value = db.Column(db.String, nullable=True)

441
spectra/web/server.py Normal file
View File

@@ -0,0 +1,441 @@
import logging
import os
import uuid
from datetime import datetime, timezone
from io import BytesIO
from pathlib import Path
from flask import (
Flask,
abort,
jsonify,
render_template,
request,
send_file,
)
from PIL import Image as PILImage
from werkzeug.utils import secure_filename
from ..config_manager import ConfigManager
from ..trigger import cache_dir, read_trigger, write_trigger
from .models import Image, Rotation, Setting, db
logger = logging.getLogger(__name__)
_ALLOWED_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif", ".bmp", ".webp"}
_MAX_UPLOAD_SIZE = 20 * 1024 * 1024
_THUMBNAIL_SIZE = (320, 240)
def create_app(config_path=None):
app = Flask(__name__)
app.config.from_mapping(
SECRET_KEY=os.urandom(32).hex(),
SQLALCHEMY_DATABASE_URI="sqlite:///" + _db_path(),
SQLALCHEMY_TRACK_MODIFICATIONS=False,
IMAGE_STORE=_image_store_path(),
TRIGGER_PATH=_trigger_path(),
)
app._config_manager = ConfigManager(config_path)
db.init_app(app)
with app.app_context():
db.create_all()
_register_routes(app)
_register_template_filters(app)
_register_error_handlers(app)
return app
def _cache_dir():
return Path(cache_dir())
def _db_path():
return str(_cache_dir() / "web.db")
def _image_store_path():
store = _cache_dir() / "uploads"
store.mkdir(parents=True, exist_ok=True)
return str(store)
def _trigger_path():
return str(_cache_dir() / "trigger.json")
def _register_routes(app):
cm = app._config_manager
store = app.config["IMAGE_STORE"]
# --- Page routes ---
@app.route("/")
def index():
return render_template("index.html")
@app.route("/gallery")
def gallery():
return render_template("gallery.html")
@app.route("/upload")
def upload_page():
return render_template("upload.html")
@app.route("/settings")
def settings_page():
return render_template("settings.html", config=cm.config)
@app.route("/preview")
def preview_page():
return render_template("preview.html")
# --- Config API ---
@app.route("/api/config", methods=["GET", "PATCH"])
def api_config():
if request.method == "GET":
return jsonify(cm.config)
changes = request.get_json(force=True, silent=True) or {}
cm.update(changes)
return jsonify(cm.config)
@app.route("/api/config/<section>", methods=["GET"])
def api_config_section(section):
value = cm.config.get(section)
if value is None:
abort(404)
return jsonify(value)
@app.route("/api/config/<section>/<key>", methods=["PUT"])
def api_config_set(section, key):
data = request.get_json(force=True, silent=True)
if data is None or "value" not in data:
abort(400, "Request body must contain a 'value' field")
section_data = cm.config.get(section, {})
if isinstance(section_data, dict):
cm.set_nested([section, key], data["value"])
else:
abort(400, "Invalid config section")
return jsonify({key: data["value"]})
# --- Display API ---
@app.route("/api/display/refresh", methods=["POST"])
def api_display_refresh():
write_trigger({"action": "refresh"})
return jsonify({"status": "triggered", "action": "refresh"})
@app.route("/api/display/clear", methods=["POST"])
def api_display_clear():
write_trigger({"action": "clear"})
return jsonify({"status": "triggered", "action": "clear"})
@app.route("/api/display/status", methods=["GET"])
def api_display_status():
trigger = read_trigger()
last_shown = (
Image.query.order_by(Image.last_shown.desc()).first()
)
from ..display import InkyDisplay
sim = cm.get("display", "simulate") or False
width = cm.get("display", "resolution", "width") or 1600
height = cm.get("display", "resolution", "height") or 1200
return jsonify({
"simulate": sim,
"resolution": {"width": width, "height": height},
"pending_trigger": trigger,
"last_shown": {
"image": {
"id": last_shown.id,
"title": last_shown.title,
"filename": last_shown.filename,
"source": last_shown.source,
} if last_shown else None,
"timestamp": last_shown.last_shown.isoformat()
if last_shown and last_shown.last_shown
else None,
},
})
# --- Preview API ---
@app.route("/api/preview/<int:image_id>")
def api_preview(image_id):
image = db.session.get(Image, image_id)
if not image:
abort(404)
_assert_safe_path(image.filepath, store)
from ..display import InkyDisplay
sat = cm.get("display", "saturation") or 0.5
width = cm.get("display", "resolution", "width") or 1600
height = cm.get("display", "resolution", "height") or 1200
orientation = cm.get("display", "orientation") or 0
display = InkyDisplay(saturation=sat, simulate=True, width=width, height=height, orientation=orientation)
try:
pil_image = PILImage.open(image.filepath)
processed = display.process_image(pil_image)
buf = BytesIO()
processed.save(buf, "PNG")
buf.seek(0)
return send_file(buf, mimetype="image/png")
except Exception as e:
logger.error("Preview failed for image %d: %s", image_id, e)
abort(500)
# --- Image API ---
@app.route("/api/images", methods=["GET", "POST"])
def api_images():
if request.method == "GET":
page = request.args.get("page", 1, type=int)
per_page = request.args.get("per_page", 50, type=int)
source = request.args.get("source")
query = Image.query
if source:
query = query.filter(Image.source == source)
query = query.order_by(Image.created_at.desc())
pagination = query.paginate(page=page, per_page=per_page, error_out=False)
return jsonify({
"images": [_image_to_dict(i) for i in pagination.items],
"page": pagination.page,
"pages": pagination.pages,
"total": pagination.total,
})
return _handle_upload(app)
@app.route("/api/images/<int:image_id>", methods=["GET", "DELETE"])
def api_image_detail(image_id):
image = db.session.get(Image, image_id)
if not image:
abort(404)
if request.method == "DELETE":
Rotation.query.filter_by(image_id=image_id).delete()
db.session.delete(image)
db.session.commit()
try:
os.remove(image.filepath)
except OSError:
pass
return jsonify({"status": "deleted"})
return jsonify(_image_to_dict(image))
@app.route("/api/images/<int:image_id>/file")
def api_image_file(image_id):
image = db.session.get(Image, image_id)
if not image:
abort(404)
_assert_safe_path(image.filepath, store)
ext = image.filename.rsplit(".", 1)[-1].lower() if "." in image.filename else ""
_MIME_MAP = {"jpg": "jpeg", "jpeg": "jpeg", "png": "png", "gif": "gif", "bmp": "bmp", "webp": "webp"}
mime = "image/" + _MIME_MAP.get(ext, "png")
return send_file(image.filepath, mimetype=mime)
@app.route("/api/images/<int:image_id>/show", methods=["POST"])
def api_image_show(image_id):
image = db.session.get(Image, image_id)
if not image:
abort(404)
_assert_safe_path(image.filepath, store)
write_trigger({"action": "show_upload", "path": image.filepath, "image_id": image.id})
return jsonify({"status": "triggered", "action": "show_upload", "image_id": image_id})
@app.route("/api/images/<int:image_id>/thumbnail")
def api_image_thumbnail(image_id):
image = db.session.get(Image, image_id)
if not image:
abort(404)
_assert_safe_path(image.filepath, store)
thumb_path = image.filepath + ".thumb"
if not os.path.exists(thumb_path):
_generate_thumbnail(image.filepath, thumb_path)
return send_file(thumb_path, mimetype="image/png")
# --- Rotation API ---
@app.route("/api/rotation", methods=["GET", "POST"])
def api_rotation():
if request.method == "GET":
entries = Rotation.query.order_by(Rotation.created_at.desc()).all()
return jsonify({
"entries": [
{
"id": e.id,
"image_id": e.image_id,
"weight": e.weight,
"active": e.active,
"image": _image_to_dict(e.image) if e.image else None,
}
for e in entries
]
})
data = request.get_json(force=True, silent=True) or {}
image_ids = data.get("image_ids", [])
if not image_ids:
abort(400, "Request must contain 'image_ids' list")
added = []
for img_id in image_ids:
exists = Rotation.query.filter_by(image_id=img_id).first()
if not exists and db.session.get(Image, img_id):
entry = Rotation(image_id=img_id, weight=data.get("weight", 1))
db.session.add(entry)
added.append(img_id)
db.session.commit()
return jsonify({"added": added}), 201
@app.route("/api/rotation/<int:rotation_id>", methods=["DELETE", "PATCH"])
def api_rotation_detail(rotation_id):
entry = db.session.get(Rotation, rotation_id)
if not entry:
abort(404)
if request.method == "DELETE":
db.session.delete(entry)
db.session.commit()
return jsonify({"status": "deleted"})
data = request.get_json(force=True, silent=True) or {}
if "weight" in data:
entry.weight = int(data["weight"])
if "active" in data:
entry.active = bool(data["active"])
db.session.commit()
return jsonify({"id": entry.id, "weight": entry.weight, "active": entry.active})
def _assert_safe_path(filepath, store):
real_file = os.path.realpath(filepath)
real_store = os.path.realpath(store)
if not real_file.startswith(real_store + os.sep):
abort(400, "Invalid file path")
def _handle_upload(app):
store = app.config["IMAGE_STORE"]
file = request.files.get("file")
if not file or not file.filename:
abort(400, "No file provided")
ext = os.path.splitext(file.filename)[1].lower()
if ext not in _ALLOWED_EXTENSIONS:
abort(400, f"Unsupported file type '{ext}'. Allowed: {', '.join(sorted(_ALLOWED_EXTENSIONS))}")
data = file.read()
if len(data) > _MAX_UPLOAD_SIZE:
abort(400, f"File too large. Maximum size is {_MAX_UPLOAD_SIZE // (1024*1024)}MB")
try:
pil_image = PILImage.open(BytesIO(data))
pil_image.verify()
pil_image = PILImage.open(BytesIO(data))
except Exception:
abort(400, "Invalid or corrupted image file")
stem = secure_filename(os.path.splitext(file.filename)[0]) or uuid.uuid4().hex
unique_name = f"{stem}_{uuid.uuid4().hex[:8]}{ext}"
dest = os.path.join(store, unique_name)
with open(dest, "wb") as f:
f.write(data)
title = request.form.get("title") or os.path.splitext(file.filename)[0]
author = request.form.get("author") or ""
image = Image(
filename=unique_name,
source="upload",
title=title,
author=author,
width=pil_image.width,
height=pil_image.height,
filepath=dest,
)
db.session.add(image)
try:
db.session.commit()
except Exception:
db.session.rollback()
try:
os.remove(dest)
except OSError:
pass
raise
show_now = request.form.get("show_now") == "1"
if show_now:
write_trigger({"action": "show_upload", "path": dest, "image_id": image.id})
return jsonify(_image_to_dict(image)), 201
def _image_to_dict(image):
return {
"id": image.id,
"filename": image.filename,
"source": image.source,
"unsplash_id": image.unsplash_id,
"title": image.title,
"author": image.author,
"width": image.width,
"height": image.height,
"filepath": image.filepath,
"created_at": image.created_at.isoformat() if image.created_at else None,
"seen_count": image.seen_count,
"last_shown": image.last_shown.isoformat() if image.last_shown else None,
}
def _generate_thumbnail(src, dest, size=_THUMBNAIL_SIZE):
try:
img = PILImage.open(src)
img.thumbnail(size, PILImage.LANCZOS)
img.save(dest, "PNG")
except Exception as e:
logger.warning("Failed to generate thumbnail for %s: %s", src, e)
def _register_template_filters(app):
@app.template_filter("datetime")
def format_datetime(value):
if not value:
return ""
return value.strftime("%Y-%m-%d %H:%M")
@app.template_filter("file_size")
def format_file_size(path):
try:
size = os.path.getsize(path)
for unit in ("B", "KB", "MB"):
if size < 1024:
return f"{size:.0f} {unit}"
size /= 1024
return f"{size:.1f} GB"
except OSError:
return ""
def _register_error_handlers(app):
@app.errorhandler(400)
def bad_request(e):
return jsonify({"error": str(e.description) if hasattr(e, "description") else "Bad request"}), 400
@app.errorhandler(404)
def not_found(e):
return jsonify({"error": "Not found"}), 404
@app.errorhandler(500)
def server_error(e):
logger.exception("Internal server error")
return jsonify({"error": "Internal server error"}), 500

View File

@@ -0,0 +1,138 @@
.gallery-card {
padding: 0;
overflow: hidden;
}
.gallery-card img {
display: block;
}
.gallery-card footer {
border-top: var(--pico-border-width) solid var(--pico-card-border-color);
}
[aria-busy="true"] {
justify-content: center;
align-items: center;
min-height: 200px;
}
#gallery-msg:empty,
#action-result:empty,
#unsplash-result:empty,
#saturation-result:empty,
#schedule-result:empty {
display: none;
}
.grid {
--grid-min: 250px;
}
.htmx-indicator {
opacity: 0;
transition: opacity 200ms ease-in;
}
.htmx-request .htmx-indicator {
opacity: 1;
}
.htmx-request.htmx-indicator {
opacity: 1;
}
#dropzone {
border: 2px dashed var(--pico-primary);
border-radius: var(--pico-border-radius);
padding: 3rem 1rem;
text-align: center;
cursor: pointer;
transition: background 0.2s, border-color 0.2s;
margin-bottom: 1rem;
}
#dropzone.drag-over {
background: var(--pico-primary-background);
border-color: var(--pico-primary-hover);
color: var(--pico-primary-inverse);
}
#dropzone.has-files {
padding: 1.5rem 1rem;
}
#dropzone-icon {
font-size: 2.5rem;
display: block;
margin-bottom: 0.5rem;
opacity: 0.5;
}
#dropzone.drag-over #dropzone-icon {
opacity: 1;
}
.file-item {
display: flex;
align-items: center;
gap: 0.75rem;
padding: 0.5rem;
border: 1px solid var(--pico-card-border-color);
border-radius: var(--pico-border-radius);
background: var(--pico-card-background-color);
}
.file-item .thumb {
width: 64px;
height: 48px;
object-fit: cover;
border-radius: 4px;
flex-shrink: 0;
}
.file-item .info {
flex: 1;
min-width: 0;
}
.file-item .filename {
white-space: nowrap;
overflow: hidden;
text-overflow: ellipsis;
font-size: 0.875rem;
}
.file-item .status {
font-size: 0.75rem;
opacity: 0.7;
}
.file-item progress {
width: 120px;
height: 6px;
}
.badge-ok {
background: var(--pico-color-green);
color: #fff;
font-size: 0.75rem;
padding: 0.15rem 0.5rem;
border-radius: 4px;
flex-shrink: 0;
text-align: center;
}
.badge-err {
background: var(--pico-color-red);
color: #fff;
font-size: 0.75rem;
padding: 0.15rem 0.5rem;
border-radius: 4px;
flex-shrink: 0;
text-align: center;
}
.badge-pending {
background: var(--pico-muted-color);
font-size: 0.75rem;
padding: 0.15rem 0.5rem;
border-radius: 4px;
flex-shrink: 0;
text-align: center;
}
.badge-uploading {
background: var(--pico-primary);
color: #fff;
font-size: 0.75rem;
padding: 0.15rem 0.5rem;
border-radius: 4px;
flex-shrink: 0;
text-align: center;
}

View File

@@ -0,0 +1,34 @@
document.addEventListener('htmx:afterRequest', function(evt) {
var msg = evt.detail.target;
if (!msg || msg.id === 'gallery-grid' || msg.id === 'file-preview') return;
var detail = evt.detail;
if (detail.xhr && detail.xhr.status >= 200 && detail.xhr.status < 300) {
try {
var data = JSON.parse(detail.xhr.responseText);
if (data.status === 'triggered') {
msg.innerHTML = '<small style="color: var(--pico-color-green);">Triggered — will update on next cycle</small>';
} else {
msg.innerHTML = '<small style="color: var(--pico-color-green);">Saved</small>';
}
} catch {
msg.innerHTML = '<small style="color: var(--pico-color-green);">Done</small>';
}
} else {
var errMsg = 'Error';
try {
var data = JSON.parse(detail.xhr.responseText);
if (data.error) errMsg = data.error;
} catch {}
msg.textContent = '';
var small = document.createElement('small');
small.style.color = 'var(--pico-color-red)';
small.textContent = errMsg;
msg.appendChild(small);
}
});
document.addEventListener('htmx:beforeRequest', function(evt) {
var msg = evt.detail.target;
if (msg) msg.innerHTML = '<small>...</small>';
});

View File

@@ -0,0 +1,32 @@
<!DOCTYPE html>
<html lang="en" data-theme="dark">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Spectra — {% block title %}Dashboard{% endblock %}</title>
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/@picocss/pico@2/css/pico.min.css">
<link rel="stylesheet" href="{{ url_for('static', filename='css/app.css') }}">
<script src="https://unpkg.com/htmx.org@2"></script>
{% block head %}{% endblock %}
</head>
<body>
<nav class="container-fluid">
<ul>
<li><strong><a href="{{ url_for('index') }}">Spectra</a></strong></li>
</ul>
<ul>
<li><a href="{{ url_for('index') }}">Dashboard</a></li>
<li><a href="{{ url_for('gallery') }}">Gallery</a></li>
<li><a href="{{ url_for('upload_page') }}">Upload</a></li>
<li><a href="{{ url_for('settings_page') }}">Settings</a></li>
</ul>
</nav>
<main class="container">
{% block content %}{% endblock %}
</main>
<script src="{{ url_for('static', filename='js/app.js') }}"></script>
{% block scripts %}{% endblock %}
</body>
</html>

View File

@@ -0,0 +1,112 @@
{% extends "base.html" %}
{% block title %}Gallery{% endblock %}
{% block head %}
<style>
.gallery-card {
padding: 0;
overflow: hidden;
position: relative;
}
.gallery-card img {
display: block;
width: 100%;
aspect-ratio: 4/3;
object-fit: cover;
}
.gallery-card footer {
border-top: var(--pico-border-width) solid var(--pico-card-border-color);
display: flex;
flex-wrap: wrap;
gap: 0.25rem;
padding: 0.5rem;
}
.gallery-card .title {
flex: 1 1 100%;
white-space: nowrap;
overflow: hidden;
text-overflow: ellipsis;
}
.gallery-card .author {
flex: 1 1 100%;
opacity: 0.6;
}
.gallery-card .actions {
display: flex;
gap: 0.25rem;
width: 100%;
}
.gallery-card .actions button {
flex: 1;
font-size: 0.75rem;
padding: 0.25rem;
text-align: center;
}
</style>
{% endblock %}
{% block content %}
<h1>Gallery</h1>
<div id="gallery-error" style="display:none;"></div>
<div id="gallery-grid" class="grid" style="--grid-min: 200px;">
<article aria-busy="true" id="gallery-loading">Loading images...</article>
</div>
<p id="gallery-msg"></p>
<p><small><a href="/upload">Upload a new image</a></small></p>
{% endblock %}
{% block scripts %}
<script>
function escapeHtml(s) {
if (s == null) return '';
return String(s)
.replace(/&/g, '&amp;')
.replace(/</g, '&lt;')
.replace(/>/g, '&gt;')
.replace(/"/g, '&quot;')
.replace(/'/g, '&#x27;');
}
async function loadGallery() {
var grid = document.getElementById('gallery-grid');
var error = document.getElementById('gallery-error');
try {
var resp = await fetch('/api/images');
if (!resp.ok) throw new Error('HTTP ' + resp.status);
var data = await resp.json();
if (data.images.length === 0) {
grid.innerHTML = '<article><p>No images yet. <a href="/upload">Upload one</a>.</p></article>';
return;
}
grid.innerHTML = data.images.map(function(img) {
var title = escapeHtml(img.title || img.filename);
var author = escapeHtml(img.author);
var source = escapeHtml(img.source);
return '<article class="gallery-card">' +
'<a href="/api/images/' + img.id + '/file" target="_blank">' +
'<img src="/api/images/' + img.id + '/thumbnail" alt="' + title + '" loading="lazy">' +
'</a>' +
'<footer>' +
'<span class="title">' + title + '</span>' +
'<span class="author">' + (author ? 'by ' + author : source) + '</span>' +
'<div class="actions">' +
'<button class="contrast" hx-post="/api/images/' + img.id + '/show" hx-target="#gallery-msg" hx-swap="innerHTML">Show</button>' +
'<button class="secondary" hx-delete="/api/images/' + img.id + '" hx-target="closest article" hx-swap="delete" hx-confirm="Delete this image?">Delete</button>' +
'</div>' +
'</footer>' +
'</article>';
}).join('');
} catch (err) {
grid.style.display = 'none';
error.style.display = 'block';
error.innerHTML = '<article style="background:var(--pico-color-red-50);"><p>Failed to load gallery: ' + escapeHtml(err.message) + '</p><button onclick="location.reload()">Retry</button></article>';
}
}
loadGallery();
</script>
{% endblock %}

View File

@@ -0,0 +1,101 @@
{% extends "base.html" %}
{% block title %}Dashboard{% endblock %}
{% block content %}
<h1>Dashboard</h1>
<div class="grid">
<article>
<header>Display Status</header>
<div id="display-status">
<p><strong>Simulation:</strong> <span id="status-simulate"></span></p>
<p><strong>Resolution:</strong> <span id="status-resolution"></span></p>
<p><strong>Last image:</strong> <span id="status-image"></span></p>
<p><strong>Shown at:</strong> <span id="status-timestamp"></span></p>
<p><strong>Pending action:</strong> <span id="status-trigger"></span></p>
</div>
</article>
<article>
<header>Quick Actions</header>
<button
hx-post="/api/display/refresh"
hx-target="#action-result"
hx-swap="innerHTML"
>
Refresh from Unsplash
</button>
<button
class="secondary"
hx-post="/api/display/clear"
hx-target="#action-result"
hx-swap="innerHTML"
>
Clear Display
</button>
<div id="action-result"></div>
</article>
</div>
<div class="grid">
<article>
<header>Schedule</header>
<p><strong>Interval:</strong> <span id="schedule-interval"></span></p>
<p><strong>Random delay:</strong> <span id="schedule-delay"></span></p>
<p><small><a href="/settings">Edit in Settings</a></small></p>
</article>
<article>
<header>Image Library</header>
<p><strong>Total images:</strong> <span id="lib-count"></span></p>
<p><strong>In rotation:</strong> <span id="rotation-count"></span></p>
<p><small><a href="/gallery">Open Gallery</a> · <a href="/upload">Upload</a></small></p>
</article>
</div>
<script>
async function refreshStatus() {
try {
const resp = await fetch('/api/display/status');
const data = await resp.json();
document.getElementById('status-simulate').textContent = data.simulate ? 'Yes' : 'No';
document.getElementById('status-resolution').textContent = data.resolution.width + '×' + data.resolution.height;
document.getElementById('status-trigger').textContent = data.pending_trigger
? (data.pending_trigger.action || 'Unknown')
: 'None';
if (data.last_shown && data.last_shown.image) {
document.getElementById('status-image').textContent = data.last_shown.image.title || data.last_shown.image.filename;
document.getElementById('status-timestamp').textContent = data.last_shown.timestamp || '—';
}
} catch { /* will retry */ }
}
async function refreshSchedule() {
try {
const resp = await fetch('/api/config/schedule');
const data = await resp.json();
document.getElementById('schedule-interval').textContent = data.interval_hours + ' hour(s)';
document.getElementById('schedule-delay').textContent = data.random_delay_seconds + 's';
} catch { /* will retry */ }
}
async function refreshLibrary() {
try {
const resp = await fetch('/api/images');
const data = await resp.json();
document.getElementById('lib-count').textContent = data.total;
} catch { /* will retry */ }
try {
const resp = await fetch('/api/rotation');
const data = await resp.json();
document.getElementById('rotation-count').textContent = data.entries.length;
} catch { /* will retry */ }
}
refreshStatus();
refreshSchedule();
refreshLibrary();
setInterval(refreshStatus, 10000);
setInterval(refreshLibrary, 30000);
</script>
{% endblock %}

View File

@@ -0,0 +1,81 @@
{% extends "base.html" %}
{% block title %}Preview{% endblock %}
{% block content %}
<h1>Preview</h1>
<article>
<p>
Select an image to see how Spectra will process it — cropped to the
display aspect ratio, resized, and saturation-adjusted.
</p>
<form id="preview-form">
<label for="preview-image-id">Image</label>
<select id="preview-image-id" name="image_id" required>
<option value="">— Select an image —</option>
</select>
<button type="button" onclick="loadPreview()" id="preview-btn">Preview</button>
</form>
</article>
<article id="preview-result" style="display:none;">
<header>Processed preview</header>
<div style="text-align:center;">
<img id="preview-img" style="max-width:100%; border-radius:var(--pico-border-radius);">
</div>
<footer>
<small>
Original (left) vs Spectra-processed (right) —
<a id="preview-original-link" href="#" target="_blank">View original</a>
</small>
</footer>
</article>
<article id="preview-error" style="display:none; background:var(--pico-color-red-50);">
<p id="preview-error-msg"></p>
</article>
<script>
fetch('/api/images')
.then(function(r) { return r.json(); })
.then(function(data) {
var sel = document.getElementById('preview-image-id');
data.images.forEach(function(img) {
var opt = document.createElement('option');
opt.value = img.id;
opt.textContent = (img.title || img.filename) + ' (' + img.width + '×' + img.height + ')';
sel.appendChild(opt);
});
});
function loadPreview() {
var id = document.getElementById('preview-image-id').value;
var btn = document.getElementById('preview-btn');
var result = document.getElementById('preview-result');
var error = document.getElementById('preview-error');
var img = document.getElementById('preview-img');
if (!id) return;
btn.disabled = true;
btn.textContent = 'Loading...';
result.style.display = 'none';
error.style.display = 'none';
img.onload = function() {
btn.disabled = false;
btn.textContent = 'Preview';
result.style.display = 'block';
};
img.onerror = function() {
btn.disabled = false;
btn.textContent = 'Preview';
document.getElementById('preview-error-msg').textContent = 'Failed to generate preview. The image may be corrupted.';
error.style.display = 'block';
};
img.src = '/api/preview/' + id + '?t=' + Date.now();
document.getElementById('preview-original-link').href = '/api/images/' + id + '/file';
}
</script>
{% endblock %}

View File

@@ -0,0 +1,141 @@
{% extends "base.html" %}
{% block title %}Settings{% endblock %}
{% block content %}
<h1>Settings</h1>
<div class="grid">
<article>
<header>Unsplash</header>
<form hx-put="/api/config/unsplash/access_key" hx-target="#unsplash-result" hx-swap="innerHTML">
<label for="access-key">Access Key</label>
<input type="text" id="access-key" name="value"
placeholder="Enter your Unsplash API key"
value="{{ config.get('unsplash', {}).get('access_key', '') }}">
<button type="submit">Save</button>
</form>
<form hx-put="/api/config/unsplash/query" hx-target="#unsplash-result" hx-swap="innerHTML">
<label for="query">Search Query</label>
<input type="text" id="query" name="value" placeholder="nature, architecture, ..."
value="{{ config.get('unsplash', {}).get('query', '') }}">
<button type="submit">Save</button>
</form>
<form hx-put="/api/config/unsplash/collections" hx-target="#unsplash-result" hx-swap="innerHTML">
<label for="collections">Collection IDs</label>
<input type="text" id="collections" name="value" placeholder="Comma-separated IDs"
value="{{ config.get('unsplash', {}).get('collections', '') }}">
<button type="submit">Save</button>
</form>
<p id="unsplash-result"></p>
</article>
<article>
<header>Display</header>
<form hx-put="/api/config/display/saturation" hx-target="#saturation-result" hx-swap="innerHTML">
<label for="saturation">
Saturation
<output id="saturation-value" style="display:inline-block; min-width:3rem;">
{{ config.get('display', {}).get('saturation', 0.5) }}
</output>
</label>
<input type="range" id="saturation" name="value"
min="0" max="1" step="0.05"
value="{{ config.get('display', {}).get('saturation', 0.5) }}"
oninput="document.getElementById('saturation-value').textContent = this.value">
<button type="submit">Save</button>
</form>
<form hx-put="/api/config/display/orientation" hx-target="#orientation-result" hx-swap="innerHTML">
<label for="orientation">Display Orientation</label>
<select id="orientation" name="value">
{% set orient = config.get('display', {}).get('orientation', 0) %}
<option value="0" {{ 'selected' if orient == 0 }}>0° (Landscape)</option>
<option value="90" {{ 'selected' if orient == 90 }}>90° (Portrait)</option>
<option value="180" {{ 'selected' if orient == 180 }}>180° (Landscape)</option>
<option value="270" {{ 'selected' if orient == 270 }}>270° (Portrait)</option>
</select>
<button type="submit">Save</button>
</form>
<p id="orientation-result"></p>
<p id="saturation-result"></p>
</article>
<article>
<header>Schedule</header>
<form hx-put="/api/config/schedule/interval_hours" hx-target="#schedule-result" hx-swap="innerHTML">
<label for="interval">Interval (hours)</label>
<input type="number" id="interval" name="value" min="0.5" step="0.5"
value="{{ config.get('schedule', {}).get('interval_hours', 1) }}">
<button type="submit">Save</button>
</form>
<form hx-put="/api/config/schedule/random_delay_seconds" hx-target="#schedule-result" hx-swap="innerHTML">
<label for="delay">Random delay (seconds)</label>
<input type="number" id="delay" name="value" min="0" step="10"
value="{{ config.get('schedule', {}).get('random_delay_seconds', 300) }}">
<button type="submit">Save</button>
</form>
<p id="schedule-result"></p>
</article>
<article>
<header>MQTT</header>
<form hx-put="/api/config/mqtt/enabled" hx-target="#mqtt-result" hx-swap="innerHTML">
<label>
<input type="hidden" name="value" value="false">
<input type="checkbox" id="mqtt-enabled" name="value" value="true"
{{ 'checked' if config.get('mqtt', {}).get('enabled', False) }}>
Enable MQTT
</label>
<button type="submit">Save</button>
</form>
<form hx-put="/api/config/mqtt/broker" hx-target="#mqtt-result" hx-swap="innerHTML">
<label for="mqtt-broker">Broker</label>
<input type="text" id="mqtt-broker" name="value"
value="{{ config.get('mqtt', {}).get('broker', 'localhost') }}">
<button type="submit">Save</button>
</form>
<form hx-put="/api/config/mqtt/port" hx-target="#mqtt-result" hx-swap="innerHTML">
<label for="mqtt-port">Port</label>
<input type="number" id="mqtt-port" name="value" min="1" max="65535"
value="{{ config.get('mqtt', {}).get('port', 1883) }}">
<button type="submit">Save</button>
</form>
<form hx-put="/api/config/mqtt/topic_prefix" hx-target="#mqtt-result" hx-swap="innerHTML">
<label for="mqtt-prefix">Topic prefix</label>
<input type="text" id="mqtt-prefix" name="value"
value="{{ config.get('mqtt', {}).get('topic_prefix', 'spectra') }}">
<button type="submit">Save</button>
</form>
<form hx-put="/api/config/mqtt/username" hx-target="#mqtt-result" hx-swap="innerHTML">
<label for="mqtt-username">Username</label>
<input type="text" id="mqtt-username" name="value"
value="{{ config.get('mqtt', {}).get('username', '') }}">
<button type="submit">Save</button>
</form>
<form hx-put="/api/config/mqtt/password" hx-target="#mqtt-result" hx-swap="innerHTML">
<label for="mqtt-password">Password</label>
<input type="password" id="mqtt-password" name="value"
value="{{ config.get('mqtt', {}).get('password', '') }}">
<button type="submit">Save</button>
</form>
<p id="mqtt-result"></p>
<p><small>See <a href="/docs/mqtt.md">MQTT docs</a> for available commands and status topics.</small></p>
</article>
</div>
{% endblock %}

View File

@@ -0,0 +1,292 @@
{% extends "base.html" %}
{% block title %}Upload{% endblock %}
{% block head %}
<style>
#dropzone {
border: 2px dashed var(--pico-primary);
border-radius: var(--pico-border-radius);
padding: 3rem 1rem;
text-align: center;
cursor: pointer;
transition: background 0.2s, border-color 0.2s;
margin-bottom: 1rem;
}
#dropzone.drag-over {
background: var(--pico-primary-background);
border-color: var(--pico-primary-hover);
color: var(--pico-primary-inverse);
}
#dropzone.has-files {
padding: 1.5rem 1rem;
}
#dropzone-icon {
font-size: 2.5rem;
display: block;
margin-bottom: 0.5rem;
opacity: 0.5;
}
#dropzone.drag-over #dropzone-icon {
opacity: 1;
}
#file-list {
display: flex;
flex-direction: column;
gap: 0.5rem;
}
.file-item {
display: flex;
align-items: center;
gap: 0.75rem;
padding: 0.5rem;
border: 1px solid var(--pico-card-border-color);
border-radius: var(--pico-border-radius);
background: var(--pico-card-background-color);
}
.file-item .thumb {
width: 64px;
height: 48px;
object-fit: cover;
border-radius: 4px;
flex-shrink: 0;
}
.file-item .info {
flex: 1;
min-width: 0;
}
.file-item .filename {
white-space: nowrap;
overflow: hidden;
text-overflow: ellipsis;
font-size: 0.875rem;
}
.file-item .status {
font-size: 0.75rem;
opacity: 0.7;
}
.file-item .progress-wrap {
width: 120px;
flex-shrink: 0;
}
.file-item progress {
width: 100%;
height: 6px;
}
.file-item .badge {
font-size: 0.75rem;
padding: 0.15rem 0.5rem;
border-radius: 4px;
flex-shrink: 0;
}
.badge-ok { background: var(--pico-color-green); color: #fff; }
.badge-err { background: var(--pico-color-red); color: #fff; }
.badge-pending { background: var(--pico-muted-color); color: var(--pico-muted-color); }
.badge-uploading { background: var(--pico-primary); color: #fff; }
</style>
{% endblock %}
{% block content %}
<h1>Upload Images</h1>
<article>
<div id="dropzone">
<span id="dropzone-icon">&#x1f4c1;</span>
<p><strong>Drag & drop images here</strong></p>
<p>or <a href="#" onclick="document.getElementById('file-input').click(); return false;">browse</a> to select files</p>
<input type="file" id="file-input" name="files" multiple
accept="image/png,image/jpeg,image/gif,image/bmp,image/webp"
style="display:none">
</div>
<div id="file-list"></div>
<div id="upload-options" style="display:none; margin-top:1rem;">
<label>
<input type="checkbox" id="show-now" checked>
Show on display after upload
</label>
</div>
<div id="upload-summary" style="display:none; margin-top:1rem;">
<article id="summary-box">
<header>Upload Complete</header>
<p id="summary-text"></p>
<div class="grid">
<a href="/gallery" role="button" class="contrast">Open Gallery</a>
<a href="/upload" role="button" class="secondary">Upload More</a>
</div>
</article>
</div>
</article>
<script>
var dropzone = document.getElementById('dropzone');
var fileInput = document.getElementById('file-input');
var fileList = document.getElementById('file-list');
var options = document.getElementById('upload-options');
var showNow = document.getElementById('show-now');
var summary = document.getElementById('upload-summary');
var summaryText = document.getElementById('summary-text');
var queue = [];
var uploading = false;
var results = { ok: 0, err: 0 };
var lastUploadedId = null;
function addFiles(files) {
for (var i = 0; i < files.length; i++) {
if (!files[i].type.match(/^image\//)) continue;
queue.push(files[i]);
renderFile(files[i]);
}
options.style.display = 'block';
dropzone.classList.add('has-files');
if (!uploading) processQueue();
}
function renderFile(file) {
var id = file.name + '-' + file.lastModified;
var div = document.createElement('div');
div.className = 'file-item';
div.id = 'file-' + id.replace(/[^a-zA-Z0-9]/g, '-');
var img = '';
if (file.type.match(/^image\//)) {
var url = URL.createObjectURL(file);
div.dataset.blobUrl = url;
img = '<img class="thumb" src="' + url + '" alt="">';
}
div.innerHTML = img +
'<div class="info">' +
'<div class="filename">' + escapeHtml(file.name) + '</div>' +
'<div class="status" id="status-' + id.replace(/[^a-zA-Z0-9]/g, '-') + '">Queued</div>' +
'</div>' +
'<div class="progress-wrap"><progress id="prog-' + id.replace(/[^a-zA-Z0-9]/g, '-') + '" value="0" max="100"></progress></div>' +
'<span class="badge badge-pending" id="badge-' + id.replace(/[^a-zA-Z0-9]/g, '-') + '">Pending</span>';
fileList.appendChild(div);
}
function updateFileStatus(file, status, pct, badgeClass, badgeText) {
var id = 'file-' + (file.name + '-' + file.lastModified).replace(/[^a-zA-Z0-9]/g, '-');
var el = document.getElementById(id);
if (!el) return;
el.querySelector('.status').textContent = status;
el.querySelector('progress').value = pct;
var badge = el.querySelector('.badge');
badge.className = 'badge ' + (badgeClass || 'badge-pending');
badge.textContent = badgeText || status;
if (badgeClass === 'badge-ok' || badgeClass === 'badge-err') {
var blobUrl = el.dataset.blobUrl;
if (blobUrl) URL.revokeObjectURL(blobUrl);
}
}
async function processQueue() {
if (queue.length === 0) {
uploading = false;
showSummary();
return;
}
uploading = true;
var file = queue.shift();
updateFileStatus(file, 'Uploading...', 0, 'badge-uploading', 'Uploading');
try {
var formData = new FormData();
formData.append('file', file);
formData.append('title', file.name.replace(/\.[^.]+$/, ''));
formData.append('show_now', '0');
var xhr = new XMLHttpRequest();
xhr.upload.onprogress = function(e) {
if (e.lengthComputable) {
var pct = Math.round((e.loaded / e.total) * 100);
updateFileStatus(file, 'Uploading... ' + pct + '%', pct, 'badge-uploading', pct + '%');
}
};
var data = await new Promise(function(resolve, reject) {
xhr.onload = function() {
if (xhr.status >= 200 && xhr.status < 300) {
try { resolve(JSON.parse(xhr.responseText)); }
catch { resolve(null); }
} else {
try {
var err = JSON.parse(xhr.responseText);
reject(new Error(err.error || 'HTTP ' + xhr.status));
} catch { reject(new Error('HTTP ' + xhr.status)); }
}
};
xhr.onerror = function() { reject(new Error('Network error')); };
xhr.open('POST', '/api/images');
xhr.send(formData);
});
results.ok++;
lastUploadedId = data ? data.id : null;
updateFileStatus(file, 'Uploaded', 100, 'badge-ok', 'OK');
} catch (err) {
results.err++;
updateFileStatus(file, 'Failed: ' + err.message, 0, 'badge-err', 'Error');
}
processQueue();
}
function showSummary() {
if (results.ok + results.err === 0) return;
var parts = [];
if (results.ok > 0) parts.push(results.ok + ' uploaded');
if (results.err > 0) parts.push(results.err + ' failed');
summaryText.textContent = parts.join(', ') + '.';
if (showNow.checked && results.ok > 0 && lastUploadedId) {
fetch('/api/images/' + lastUploadedId + '/show', { method: 'POST' })
.then(function(r) {
if (r.ok) {
var extra = document.createElement('small');
extra.style.display = 'block';
extra.textContent = 'Last image sent to display.';
summaryText.appendChild(document.createTextNode(' '));
summaryText.appendChild(extra);
}
})
.catch(function() {});
}
summary.style.display = 'block';
}
// Drag events
dropzone.addEventListener('dragover', function(e) {
e.preventDefault();
dropzone.classList.add('drag-over');
});
dropzone.addEventListener('dragleave', function(e) {
e.preventDefault();
dropzone.classList.remove('drag-over');
});
dropzone.addEventListener('drop', function(e) {
e.preventDefault();
dropzone.classList.remove('drag-over');
addFiles(e.dataTransfer.files);
});
dropzone.addEventListener('click', function() {
fileInput.click();
});
fileInput.addEventListener('change', function() {
addFiles(this.files);
this.value = '';
});
function escapeHtml(str) {
var div = document.createElement('div');
div.appendChild(document.createTextNode(str));
return div.innerHTML;
}
</script>
{% endblock %}

View File

@@ -0,0 +1,16 @@
[Unit]
Description=Spectra Web Interface
After=network-online.target
Wants=network-online.target
[Service]
Type=simple
ExecStartPre=/usr/bin/mkdir -p /var/cache/spectra/uploads
ExecStart=/usr/local/bin/spectra web
Restart=on-failure
RestartSec=10
StandardOutput=journal
StandardError=journal
[Install]
WantedBy=multi-user.target

16
systemd/spectra.service Normal file
View File

@@ -0,0 +1,16 @@
[Unit]
Description=Spectra Unsplash Image Display
After=network-online.target
Wants=network-online.target
[Service]
Type=simple
ExecStartPre=/usr/bin/mkdir -p /var/cache/spectra/uploads
ExecStart=/usr/local/bin/spectra
Restart=on-failure
RestartSec=30
StandardOutput=journal
StandardError=journal
[Install]
WantedBy=multi-user.target

0
tests/__init__.py Normal file
View File

65
tests/conftest.py Normal file
View File

@@ -0,0 +1,65 @@
import io
import json
import os
import tempfile
from pathlib import Path
import pytest
from PIL import Image as PILImage
from spectra.web.server import create_app
TEST_IMAGE_SIZE = (800, 600)
@pytest.fixture
def app():
with tempfile.TemporaryDirectory() as tmpdir:
tmp = Path(tmpdir)
cache_dir = tmp / "cache"
old_environ = os.environ.get("SPECTRA_CACHE_DIR")
os.environ["SPECTRA_CACHE_DIR"] = str(cache_dir)
app = create_app()
yield app
if old_environ is None:
os.environ.pop("SPECTRA_CACHE_DIR", None)
else:
os.environ["SPECTRA_CACHE_DIR"] = old_environ
@pytest.fixture
def client(app):
return app.test_client()
@pytest.fixture
def sample_image_bytes():
img = PILImage.new("RGB", TEST_IMAGE_SIZE, color=(100, 150, 200))
buf = io.BytesIO()
img.save(buf, format="JPEG")
buf.seek(0)
return buf
@pytest.fixture
def uploaded_image(client, sample_image_bytes):
resp = client.post(
"/api/images",
data={"file": (sample_image_bytes, "test.jpg")},
content_type="multipart/form-data",
)
assert resp.status_code == 201
return resp.get_json()
def make_image_bytes(width, height, color=(100, 150, 200), fmt="JPEG"):
img = PILImage.new("RGB", (width, height), color=color)
buf = io.BytesIO()
img.save(buf, format=fmt)
buf.seek(0)
return buf, img

353
tests/test_api.py Normal file
View File

@@ -0,0 +1,353 @@
import io
from PIL import Image as PILImage
class TestConfigAPI:
def test_get_config(self, client):
resp = client.get("/api/config")
assert resp.status_code == 200
data = resp.get_json()
assert "display" in data
assert "unsplash" in data
assert "schedule" in data
def test_patch_config(self, client):
resp = client.patch("/api/config", json={"display": {"saturation": 0.7}})
assert resp.status_code == 200
data = resp.get_json()
assert data["display"]["saturation"] == 0.7
def test_get_config_section(self, client):
resp = client.get("/api/config/display")
assert resp.status_code == 200
data = resp.get_json()
assert "saturation" in data
assert "orientation" in data
def test_get_config_section_not_found(self, client):
resp = client.get("/api/config/nonexistent")
assert resp.status_code == 404
def test_set_nested_config_key(self, client):
resp = client.put(
"/api/config/display/saturation",
json={"value": 0.9},
)
assert resp.status_code == 200
assert resp.get_json() == {"saturation": 0.9}
resp = client.get("/api/config/display")
assert resp.get_json()["saturation"] == 0.9
def test_set_config_orientation(self, client):
resp = client.put("/api/config/display/orientation", json={"value": 90})
assert resp.status_code == 200
resp = client.get("/api/config/display")
assert resp.get_json()["orientation"] == 90
def test_set_config_missing_value_returns_400(self, client):
resp = client.put("/api/config/display/saturation", json={})
assert resp.status_code == 400
class TestImagesAPI:
def test_list_images_empty(self, client):
resp = client.get("/api/images")
assert resp.status_code == 200
data = resp.get_json()
assert data["images"] == []
assert data["total"] == 0
def test_upload_image(self, client, sample_image_bytes):
resp = client.post(
"/api/images",
data={"file": (sample_image_bytes, "test.jpg")},
content_type="multipart/form-data",
)
assert resp.status_code == 201
data = resp.get_json()
assert data["source"] == "upload"
assert data["title"] == "test"
assert data["width"] == 800
assert data["height"] == 600
def test_upload_image_with_title_and_author(self, client, sample_image_bytes):
resp = client.post(
"/api/images",
data={"file": (sample_image_bytes, "photo.jpg"), "title": "My Photo", "author": "Test User"},
content_type="multipart/form-data",
)
assert resp.status_code == 201
data = resp.get_json()
assert data["title"] == "My Photo"
assert data["author"] == "Test User"
def test_upload_image_no_file_returns_400(self, client):
resp = client.post("/api/images", content_type="multipart/form-data")
assert resp.status_code == 400
def test_upload_invalid_extension_returns_400(self, client):
buf = io.BytesIO(b"not an image")
resp = client.post(
"/api/images",
data={"file": (buf, "test.txt")},
content_type="multipart/form-data",
)
assert resp.status_code == 400
def test_upload_corrupted_image_returns_400(self, client):
buf = io.BytesIO(b"not an image at all")
resp = client.post(
"/api/images",
data={"file": (buf, "test.jpg")},
content_type="multipart/form-data",
)
assert resp.status_code == 400
def test_get_image_detail(self, client, uploaded_image):
image_id = uploaded_image["id"]
resp = client.get(f"/api/images/{image_id}")
assert resp.status_code == 200
data = resp.get_json()
assert data["id"] == image_id
assert data["source"] == "upload"
def test_get_image_detail_not_found(self, client):
resp = client.get("/api/images/99999")
assert resp.status_code == 404
def test_delete_image(self, client, uploaded_image):
image_id = uploaded_image["id"]
resp = client.delete(f"/api/images/{image_id}")
assert resp.status_code == 200
assert resp.get_json() == {"status": "deleted"}
resp = client.get(f"/api/images/{image_id}")
assert resp.status_code == 404
def test_delete_image_not_found(self, client):
resp = client.delete("/api/images/99999")
assert resp.status_code == 404
def test_list_images_after_upload(self, client, uploaded_image):
resp = client.get("/api/images")
data = resp.get_json()
assert data["total"] >= 1
ids = [i["id"] for i in data["images"]]
assert uploaded_image["id"] in ids
def test_upload_security_filename_traversal(self, client):
img = PILImage.new("RGB", (100, 100))
buf = io.BytesIO()
img.save(buf, format="PNG")
buf.seek(0)
resp = client.post(
"/api/images",
data={"file": (buf, "../../etc/passwd.png")},
content_type="multipart/form-data",
)
assert resp.status_code == 201
data = resp.get_json()
assert "/" not in data["filename"]
assert ".." not in data["filename"]
def test_image_download(self, client, uploaded_image):
image_id = uploaded_image["id"]
resp = client.get(f"/api/images/{image_id}/file")
assert resp.status_code == 200
assert resp.content_type.startswith("image/")
def test_image_download_not_found(self, client):
resp = client.get("/api/images/99999/file")
assert resp.status_code == 404
class TestImageShow:
def test_show_image(self, client, uploaded_image):
image_id = uploaded_image["id"]
resp = client.post(f"/api/images/{image_id}/show")
assert resp.status_code == 200
data = resp.get_json()
assert data["status"] == "triggered"
def test_show_image_not_found(self, client):
resp = client.post("/api/images/99999/show")
assert resp.status_code == 404
class TestImageThumbnail:
def test_thumbnail(self, client, uploaded_image):
image_id = uploaded_image["id"]
resp = client.get(f"/api/images/{image_id}/thumbnail")
assert resp.status_code == 200
assert resp.content_type == "image/png"
def test_thumbnail_not_found(self, client):
resp = client.get("/api/images/99999/thumbnail")
assert resp.status_code == 404
class TestPreviewAPI:
def test_preview(self, client, uploaded_image):
image_id = uploaded_image["id"]
resp = client.get(f"/api/preview/{image_id}")
assert resp.status_code == 200
assert resp.content_type == "image/png"
def test_preview_not_found(self, client):
resp = client.get("/api/preview/99999")
assert resp.status_code == 404
class TestDisplayAPI:
def test_status(self, client):
resp = client.get("/api/display/status")
assert resp.status_code == 200
data = resp.get_json()
assert "simulate" in data
assert "resolution" in data
assert "pending_trigger" in data
def test_refresh(self, client):
resp = client.post("/api/display/refresh")
assert resp.status_code == 200
assert resp.get_json()["status"] == "triggered"
def test_clear(self, client):
resp = client.post("/api/display/clear")
assert resp.status_code == 200
assert resp.get_json()["status"] == "triggered"
class TestRotationAPI:
def test_list_rotation_empty(self, client):
resp = client.get("/api/rotation")
assert resp.status_code == 200
assert resp.get_json()["entries"] == []
def test_add_to_rotation(self, client, uploaded_image):
image_id = uploaded_image["id"]
resp = client.post("/api/rotation", json={"image_ids": [image_id]})
assert resp.status_code == 201
assert resp.get_json() == {"added": [image_id]}
def test_add_nonexistent_image_to_rotation(self, client):
resp = client.post("/api/rotation", json={"image_ids": [99999]})
assert resp.status_code == 201
assert resp.get_json() == {"added": []}
def test_add_duplicate_to_rotation(self, client, uploaded_image):
image_id = uploaded_image["id"]
client.post("/api/rotation", json={"image_ids": [image_id]})
resp = client.post("/api/rotation", json={"image_ids": [image_id]})
assert resp.status_code == 201
assert resp.get_json() == {"added": []}
def test_rotation_detail(self, client, uploaded_image):
image_id = uploaded_image["id"]
client.post("/api/rotation", json={"image_ids": [image_id]})
resp = client.get("/api/rotation")
rotation_id = resp.get_json()["entries"][0]["id"]
resp = client.delete(f"/api/rotation/{rotation_id}")
assert resp.status_code == 200
def test_patch_rotation_weight(self, client, uploaded_image):
image_id = uploaded_image["id"]
client.post("/api/rotation", json={"image_ids": [image_id]})
resp = client.get("/api/rotation")
entry = resp.get_json()["entries"][0]
entry_id = entry["id"]
resp = client.patch(f"/api/rotation/{entry_id}", json={"weight": 5})
assert resp.status_code == 200
assert resp.get_json()["weight"] == 5
def test_patch_rotation_active(self, client, uploaded_image):
image_id = uploaded_image["id"]
client.post("/api/rotation", json={"image_ids": [image_id]})
resp = client.get("/api/rotation")
entry_id = resp.get_json()["entries"][0]["id"]
resp = client.patch(f"/api/rotation/{entry_id}", json={"active": False})
assert resp.status_code == 200
assert resp.get_json()["active"] is False
def test_rotation_not_found(self, client):
resp = client.delete("/api/rotation/99999")
assert resp.status_code == 404
def test_rotation_patch_not_found(self, client):
resp = client.patch("/api/rotation/99999", json={"weight": 3})
assert resp.status_code == 404
class TestPages:
def test_index_page(self, client):
resp = client.get("/")
assert resp.status_code == 200
assert b"Dashboard" in resp.data
def test_gallery_page(self, client):
resp = client.get("/gallery")
assert resp.status_code == 200
assert b"Gallery" in resp.data
def test_upload_page(self, client):
resp = client.get("/upload")
assert resp.status_code == 200
assert b"Upload" in resp.data
def test_settings_page(self, client):
resp = client.get("/settings")
assert resp.status_code == 200
assert b"Settings" in resp.data
assert b"orientation" in resp.data
def test_preview_page(self, client):
resp = client.get("/preview")
assert resp.status_code == 200
assert b"Preview" in resp.data
class TestCORSAndErrors:
def test_404_json(self, client):
resp = client.get("/api/nonexistent")
assert resp.status_code == 404
assert resp.is_json
def test_400_has_error(self, client):
resp = client.post("/api/images", content_type="multipart/form-data")
assert resp.status_code == 400
data = resp.get_json()
assert "error" in data
def test_url_encoded_form_not_accepted(self, client):
resp = client.put(
"/api/config/display/saturation",
data={"value": "0.5"},
)
assert resp.status_code == 400 or resp.status_code == 200
class TestGallery:
def test_gallery_escaping(self, client, sample_image_bytes):
malicious_title = '<script>alert("xss")</script>'
resp = client.post(
"/api/images",
data={"file": (sample_image_bytes, "test.jpg"), "title": malicious_title},
content_type="multipart/form-data",
)
assert resp.status_code == 201
img_id = resp.get_json()["id"]
resp = client.get("/api/images")
img = next(i for i in resp.get_json()["images"] if i["id"] == img_id)
assert img["title"] == malicious_title
def test_exact_json_equality_cautious(self, client, uploaded_image):
resp = client.get(f"/api/images/{uploaded_image['id']}")
data = resp.get_json()
assert data["id"] == uploaded_image["id"]
assert "created_at" in data
def test_result_not_affected_by_previous_delete(self, client, uploaded_image):
image_id = uploaded_image["id"]
client.delete(f"/api/images/{image_id}")
resp = client.get(f"/api/images/{image_id}")
assert resp.status_code == 404

101
tests/test_config.py Normal file
View File

@@ -0,0 +1,101 @@
import os
import tempfile
import yaml
from spectra.config import DEFAULT_CONFIG, _deep_merge, load_config
class TestLoadConfig:
def test_defaults_when_no_file(self):
cfg = load_config(path="/nonexistent/.spectra/config.yaml")
assert cfg["display"]["saturation"] == 0.5
assert cfg["display"]["orientation"] == 0
assert cfg["unsplash"]["access_key"] == ""
assert cfg["schedule"]["interval_hours"] == 1
def test_load_specific_path(self):
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
yaml.dump({"display": {"saturation": 0.8}}, f)
path = f.name
try:
cfg = load_config(path)
assert cfg["display"]["saturation"] == 0.8
assert cfg["display"]["orientation"] == 0
finally:
os.unlink(path)
def test_missing_path_logs_and_uses_defaults(self):
cfg = load_config("/nonexistent/path/config.yaml")
assert cfg["display"]["saturation"] == 0.5
def test_partial_merge_does_not_remove_other_keys(self):
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
yaml.dump({"display": {"saturation": 0.3}}, f)
path = f.name
try:
cfg = load_config(path)
assert cfg["display"]["saturation"] == 0.3
assert cfg["display"]["resolution"]["width"] == 1600
assert cfg["unsplash"]["access_key"] == ""
finally:
os.unlink(path)
def test_empty_file_uses_defaults(self):
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
f.write("")
path = f.name
try:
cfg = load_config(path)
assert cfg["display"]["saturation"] == 0.5
finally:
os.unlink(path)
def test_deep_merge_overrides_nested_key(self):
base = {"a": {"b": 1, "c": 2}, "d": 3}
override = {"a": {"b": 99}}
_deep_merge(base, override)
assert base["a"]["b"] == 99
assert base["a"]["c"] == 2
assert base["d"] == 3
def test_deep_merge_adds_new_keys(self):
base = {"a": 1}
override = {"b": 2}
_deep_merge(base, override)
assert base["a"] == 1
assert base["b"] == 2
def test_deep_merge_overrides_non_dict_with_dict(self):
base = {"a": 1}
override = {"a": {"b": 2}}
_deep_merge(base, override)
assert base["a"] == {"b": 2}
def test_default_config_not_mutated_by_load(self):
original_sat = DEFAULT_CONFIG["display"]["saturation"]
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
yaml.dump({"display": {"saturation": 0.9}}, f)
path = f.name
try:
load_config(path)
assert DEFAULT_CONFIG["display"]["saturation"] == original_sat
finally:
os.unlink(path)
def test_default_config_not_mutated_by_multiple_loads(self):
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
yaml.dump({"unsplash": {"query": "cats"}}, f)
path1 = f.name
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
yaml.dump({"unsplash": {"query": "dogs"}}, f)
path2 = f.name
try:
c1 = load_config(path1)
c2 = load_config(path2)
assert c1["unsplash"]["query"] == "cats"
assert c2["unsplash"]["query"] == "dogs"
assert DEFAULT_CONFIG["unsplash"]["query"] == ""
finally:
os.unlink(path1)
os.unlink(path2)

135
tests/test_display.py Normal file
View File

@@ -0,0 +1,135 @@
import os
from PIL import Image as PILImage
from spectra.display import InkyDisplay
class TestProcessImage:
def setup_method(self):
self.d = InkyDisplay(simulate=True, width=800, height=480)
def test_aspect_crop_wide_image(self):
img = PILImage.new("RGB", (1600, 600), color=(128, 64, 200))
r = self.d.process_image(img)
assert r.size == (800, 480)
def test_aspect_crop_tall_image(self):
img = PILImage.new("RGB", (400, 900), color=(128, 64, 200))
r = self.d.process_image(img)
assert r.size == (800, 480)
def test_exact_aspect_no_crop(self):
img = PILImage.new("RGB", (800, 480), color=(128, 64, 200))
r = self.d.process_image(img)
assert r.size == (800, 480)
def test_square_input_to_landscape(self):
img = PILImage.new("RGB", (1000, 1000), color=(128, 64, 200))
r = self.d.process_image(img)
assert r.size == (800, 480)
def test_tiny_input(self):
img = PILImage.new("RGB", (10, 10), color=(128, 64, 200))
r = self.d.process_image(img)
assert r.size == (800, 480)
def test_panorama_very_wide(self):
img = PILImage.new("RGB", (4000, 800), color=(128, 64, 200))
r = self.d.process_image(img)
assert r.size == (800, 480)
def test_very_narrow(self):
img = PILImage.new("RGB", (100, 2000), color=(128, 64, 200))
r = self.d.process_image(img)
assert r.size == (800, 480)
class TestProcessImageWithRotation:
def test_90_rotation_swaps_crop(self):
d = InkyDisplay(simulate=True, width=800, height=480, orientation=90)
img = PILImage.new("RGB", (1200, 1600), color=(128, 64, 200))
r = d.process_image(img)
assert r.size == (800, 480)
def test_270_rotation(self):
d = InkyDisplay(simulate=True, width=800, height=480, orientation=270)
img = PILImage.new("RGB", (1600, 1200), color=(128, 64, 200))
r = d.process_image(img)
assert r.size == (800, 480)
def test_180_rotation(self):
d = InkyDisplay(simulate=True, width=800, height=480, orientation=180)
img = PILImage.new("RGB", (1600, 1200), color=(128, 64, 200))
r = d.process_image(img)
assert r.size == (800, 480)
class TestShow:
def test_show_writes_file(self):
d = InkyDisplay(simulate=True, width=800, height=480)
d.show(PILImage.new("RGB", (200, 200)))
assert os.path.exists("/tmp/spectra_last.png")
os.unlink("/tmp/spectra_last.png")
def test_show_with_rotation_writes_same_size(self):
d = InkyDisplay(simulate=True, width=800, height=480, orientation=90)
d.show(PILImage.new("RGB", (200, 200)))
saved = PILImage.open("/tmp/spectra_last.png")
assert saved.size == (800, 480)
os.unlink("/tmp/spectra_last.png")
class TestShowFile:
def test_show_file(self, tmp_path):
src = tmp_path / "input.png"
img = PILImage.new("RGB", (200, 200))
img.save(str(src))
d = InkyDisplay(simulate=True, width=800, height=480)
d.show_file(str(src))
assert os.path.exists("/tmp/spectra_last.png")
os.unlink("/tmp/spectra_last.png")
def test_show_file_nonexistent(self):
d = InkyDisplay(simulate=True, width=800, height=480)
try:
d.show_file("/nonexistent/image.png")
assert False, "Expected exception"
except FileNotFoundError:
pass
class TestClear:
def test_clear_simulation(self):
d = InkyDisplay(simulate=True, width=800, height=480)
d.clear()
def test_clear_simulation_with_orientation(self):
d = InkyDisplay(simulate=True, width=800, height=480, orientation=90)
d.clear()
class TestEdgeCases:
def test_display_initialization_defaults(self):
d = InkyDisplay()
assert d.width == 1600
assert d.height == 1200
assert d.orientation == 0
assert d.simulate is False
def test_display_initialization_custom(self):
d = InkyDisplay(simulate=True, width=800, height=480, orientation=90)
assert d.effective_width == 480
assert d.effective_height == 800
def test_process_image_with_rgba(self):
d = InkyDisplay(simulate=True, width=800, height=480)
img = PILImage.new("RGBA", (1600, 1200), color=(128, 64, 200, 255))
r = d.process_image(img)
assert r.size == (800, 480)
def test_process_image_with_grayscale(self):
d = InkyDisplay(simulate=True, width=800, height=480)
img = PILImage.new("L", (1600, 1200), color=128)
r = d.process_image(img)
assert r.size == (800, 480)

38
tests/test_gallery.py Normal file
View File

@@ -0,0 +1,38 @@
import io
from PIL import Image as PILImage
def _upload_image(client, filename="test_photo.jpg", color=(128, 64, 200)):
img = PILImage.new("RGB", (800, 600), color=color)
buf = io.BytesIO()
img.save(buf, format="JPEG")
buf.seek(0)
return client.post(
"/api/images",
data={"file": (buf, filename)},
content_type="multipart/form-data",
)
class TestGalleryDelete:
def test_upload_delete_flow(self, client):
resp = _upload_image(client)
assert resp.status_code == 201
data = resp.get_json()
image_id = data["id"]
resp = client.get("/api/images")
assert resp.status_code == 200
data = resp.get_json()
total_before = data["total"]
assert total_before >= 1
resp = client.delete(f"/api/images/{image_id}")
assert resp.status_code == 200
assert resp.get_json() == {"status": "deleted"}
resp = client.get("/api/images")
assert resp.status_code == 200
data = resp.get_json()
assert data["total"] == total_before - 1

15
tests/test_mqtt.py Normal file
View File

@@ -0,0 +1,15 @@
from spectra.mqtt import _lookup_image_path, _db_path
class TestDbPath:
def test_db_path_ends_web_db(self):
path = _db_path()
assert path.endswith("web.db")
class TestLookupImagePath:
def test_nonexistent_id_returns_none(self):
assert _lookup_image_path(99999) is None
def test_negative_id_returns_none(self):
assert _lookup_image_path(-1) is None

50
tests/test_orientation.py Normal file
View File

@@ -0,0 +1,50 @@
import os
from PIL import Image as PILImage
from spectra.display import InkyDisplay
class TestInkyDisplayOrientation:
def test_effective_dimensions_landscape(self):
d = InkyDisplay(simulate=True, width=800, height=480, orientation=0)
assert d.effective_width == 800
assert d.effective_height == 480
def test_effective_dimensions_portrait(self):
d = InkyDisplay(simulate=True, width=800, height=480, orientation=90)
assert d.effective_width == 480
assert d.effective_height == 800
def test_output_size_0(self):
d = InkyDisplay(simulate=True, width=800, height=480, orientation=0)
r = d.process_image(PILImage.new("RGB", (1600, 1200)))
assert r.size == (800, 480)
def test_output_size_90(self):
d = InkyDisplay(simulate=True, width=800, height=480, orientation=90)
r = d.process_image(PILImage.new("RGB", (1600, 1200)))
assert r.size == (800, 480)
def test_output_size_180(self):
d = InkyDisplay(simulate=True, width=800, height=480, orientation=180)
r = d.process_image(PILImage.new("RGB", (1600, 1200)))
assert r.size == (800, 480)
def test_output_size_270(self):
d = InkyDisplay(simulate=True, width=800, height=480, orientation=270)
r = d.process_image(PILImage.new("RGB", (1600, 1200)))
assert r.size == (800, 480)
def test_string_orientation_coerced(self):
d = InkyDisplay(simulate=True, width=800, height=480, orientation="90")
assert d.orientation == 90
def test_none_orientation_defaults_to_zero(self):
d = InkyDisplay(simulate=True, width=800, height=480, orientation=None)
assert d.orientation == 0
def test_show_writes_file(self):
d = InkyDisplay(simulate=True, width=800, height=480, orientation=90)
d.show(PILImage.new("RGB", (1600, 1200)))
assert os.path.exists("/tmp/spectra_last.png")

106
tests/test_trigger.py Normal file
View File

@@ -0,0 +1,106 @@
import json
import os
import tempfile
from pathlib import Path
import pytest
from spectra.trigger import (
_trigger_path,
cache_dir,
clear_trigger,
read_and_clear_trigger,
read_trigger,
write_trigger,
)
@pytest.fixture(autouse=True)
def isolate_cache(monkeypatch, tmp_path):
cache = tmp_path / "cache"
cache.mkdir()
monkeypatch.setenv("SPECTRA_CACHE_DIR", str(cache))
yield
monkeypatch.delenv("SPECTRA_CACHE_DIR", raising=False)
class TestWriteRead:
def test_write_then_read(self):
write_trigger({"action": "refresh"})
data = read_trigger()
assert data == {"action": "refresh"}
def test_read_empty(self):
assert read_trigger() is None
def test_write_overwrites(self):
write_trigger({"action": "refresh"})
write_trigger({"action": "clear"})
data = read_trigger()
assert data == {"action": "clear"}
def test_write_complex_data(self):
write_trigger({"action": "show_upload", "path": "/tmp/test.png", "image_id": 42})
data = read_trigger()
assert data["action"] == "show_upload"
assert data["path"] == "/tmp/test.png"
assert data["image_id"] == 42
class TestClear:
def test_clear_trigger(self):
write_trigger({"action": "refresh"})
clear_trigger()
assert read_trigger() is None
def test_clear_missing_trigger(self):
clear_trigger()
def test_clear_then_read(self):
write_trigger({"action": "refresh"})
clear_trigger()
assert read_trigger() is None
class TestReadAndClear:
def test_read_and_clear_returns_data(self):
write_trigger({"action": "refresh"})
data = read_and_clear_trigger()
assert data == {"action": "refresh"}
def test_read_and_clear_removes_file(self):
write_trigger({"action": "refresh"})
read_and_clear_trigger()
assert read_trigger() is None
def test_read_and_clear_nonexistent(self):
assert read_and_clear_trigger() is None
def test_read_and_clear_is_atomic(self):
write_trigger({"action": "refresh"})
data = read_and_clear_trigger()
assert data == {"action": "refresh"}
assert read_trigger() is None
class TestCacheDir:
def test_cache_dir_uses_env(self, monkeypatch):
monkeypatch.setenv("SPECTRA_CACHE_DIR", "/tmp/spectra-test-cache")
d = cache_dir()
assert d == "/tmp/spectra-test-cache"
def test_trigger_path_uses_cache_dir(self):
d = cache_dir()
path = _trigger_path()
assert path.startswith(d)
assert path.endswith("trigger.json")
class TestCorruptFile:
def test_corrupt_json_returns_none(self):
path = _trigger_path()
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w") as f:
f.write("{invalid json")
assert read_trigger() is None
assert read_and_clear_trigger() is None