Files
espresso_frame/server/app/immich_client.py
T
tfaour d395cf3bb9
Build and push server image / build-and-push (push) Successful in 35s
Add two physical buttons: factory-reset and next-photo
Factory-reset (GPIO3, hold 10s): clears stored WiFi/server config and
restarts into provisioning -- the deliberate, USB-free replacement for
the earlier reverted RST-based auto-reprovisioning idea.

Next-photo (GPIO2, tap): wakes the device and forces the server to
advance immediately via a new POST /frame/advance, instead of waiting
for the refresh interval. Both buttons arm themselves as deep-sleep GPIO
wakeup sources so a press is noticed promptly even while asleep.

Also makes GET /frame/image side-effect-free: it now only advances once
refresh_interval_s has elapsed since the current photo was set (tracked
server-side), so a device reboot for any reason just redisplays the
current photo instead of silently skipping ahead. The server maintains a
small reorderable upcoming-photos queue, viewable and rearrangeable from
the web UI.
2026-07-18 23:28:36 -04:00

72 lines
2.7 KiB
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 list_album_assets(self, album_id: str) -> list[dict]:
"""GET /api/albums/{id} doesn't embed assets in this Immich version
(AlbumResponseDto only has assetCount) -- assets live behind the
general search API instead, filtered by albumIds.
Only returns the first page. Fine for a photo frame cycling
through an album; worth adding nextPage pagination if someone
points this at an album large enough to need it.
"""
resp = httpx.post(
f"{self.base_url}/api/search/metadata",
headers=self._headers,
json={"albumIds": [album_id]},
timeout=15,
)
resp.raise_for_status()
return resp.json().get("assets", {}).get("items", [])
def get_asset_faces(self, asset_id: str) -> list[dict]:
"""Face bounding boxes Immich already computed for its own People
feature -- reused here to bias cropping toward keeping faces on
screen instead of running our own detection."""
resp = httpx.get(
f"{self.base_url}/api/faces",
params={"id": asset_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
def download_asset_thumbnail(self, asset_id: str) -> tuple[bytes, str]:
"""Smaller than download_asset_preview -- used for the web UI's
upcoming-photos list, not the actual rendered frame. Returns
(content, content_type) since this one gets proxied straight to a
browser <img> tag and needs a correct Content-Type header."""
resp = httpx.get(
f"{self.base_url}/api/assets/{asset_id}/thumbnail",
params={"size": "thumbnail"},
headers=self._headers,
timeout=30,
)
resp.raise_for_status()
return resp.content, resp.headers.get("content-type", "image/jpeg")