Files
espresso_frame/server/app/immich_client.py
T
tfaour 5e7c47e2eb Add FastAPI server: pulls from Immich, pre-processes for the panel
Implements the server side of the architecture decided on: the ESP32-C6
has no PSRAM and a tight RAM budget, so all the heavy lifting (JPEG
decode, resize, Floyd-Steinberg dithering, 6-color quantization, 4bpp
packing) happens here instead of on-device. The frame just does a single
GET and streams the response straight to SPI.

- GET /frame/image: looks up the current cursor's asset in the configured
  Immich album, downloads its preview thumbnail, and returns it packed
  into the panel's exact 800x480/4bpp/2px-per-byte format
  (application/octet-stream, always exactly 192,000 bytes).
- GET / + POST /api/config + GET /api/albums: a small web UI for entering
  the Immich URL/API key and picking an album, rather than cramming that
  into the ESP32's captive portal form.
- Config (Immich creds, selected album, cursor) persists to a JSON file
  via a docker-compose volume mount.

Verified locally with a venv (Docker isn't available in this environment):
unit-tested image_pipeline against a synthetic image (exact byte count,
valid panel color codes only), and ran a full end-to-end pass against a
mock Immich HTTP server exercising the real /frame/image path.

Pinned dependency versions in requirements.txt after hitting a real bug
with unpinned floors: the latest starlette (1.3.1) resolved by `pip
install fastapi` breaks Jinja2Templates outright.

Not yet wired to the ESP32 side (task 6) or authenticated -- /frame/image
is unauthenticated for now, fine on a trusted LAN but worth revisiting
once the firmware sends a shared device token.
2026-07-18 14:35:14 -04:00

32 lines
1014 B
Python

"""Thin wrapper around the bits of the Immich API this project needs."""
from __future__ import annotations
import httpx
class ImmichClient:
def __init__(self, base_url: str, api_key: str):
self.base_url = base_url.rstrip("/")
self._headers = {"x-api-key": api_key}
def list_albums(self) -> list[dict]:
resp = httpx.get(f"{self.base_url}/api/albums", headers=self._headers, timeout=10)
resp.raise_for_status()
return resp.json()
def get_album(self, album_id: str) -> dict:
resp = httpx.get(f"{self.base_url}/api/albums/{album_id}", headers=self._headers, timeout=10)
resp.raise_for_status()
return resp.json()
def download_asset_preview(self, asset_id: str) -> bytes:
resp = httpx.get(
f"{self.base_url}/api/assets/{asset_id}/thumbnail",
params={"size": "preview"},
headers=self._headers,
timeout=30,
)
resp.raise_for_status()
return resp.content