Files
espresso_frame/server/app/immich_client.py
T
tfaour a358045cea
Build and push server image / build-and-push (push) Successful in 32s
Fix "Show next" staleness bug; add location/date/share-QR to manage overlay
Two changes, bundled since they landed in the same session and touch
overlapping files:

1. Fix: "Show next" sent the browser's full queue snapshot to
   POST /api/queue/reorder, which hard-rejected if the server's queue
   had shifted since the last fetch (e.g. right after a queue-length
   trim). New POST /api/queue/promote moves one photo to the front
   authoritatively, with no dependency on client staleness. /reorder
   itself is now tolerant too -- unrecognized IDs are dropped and
   missing ones appended, instead of rejecting the whole request.

2. Feature: the manage button's overlay now also shows the photo's
   location (top-left, only if Immich reverse-geocoded it from GPS
   EXIF), the date it was taken (bottom-right), and a QR code (bottom-
   left) linking to a 30-minute public Immich share link -- created
   lazily when someone actually scans it, not when the button's
   pressed. New server endpoints GET /frame/photo-info and
   GET /frame/share/{asset_id} (scoped to the frame's current/queued
   photos, not any arbitrary Immich asset). Firmware-side, the overlay
   mechanism generalizes from one spliced region to up to four
   (manage_qr_overlay.c), each its own small buffer, still never
   holding the full frame in RAM.
2026-07-19 01:28:25 -04:00

107 lines
4.1 KiB
Python

"""Thin wrapper around the bits of the Immich API this project needs."""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
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")
def get_asset(self, asset_id: str) -> dict:
"""Full asset metadata, including embedded exifInfo (location,
capture date) -- used for the manage-button overlay's location/
date-taken text."""
resp = httpx.get(f"{self.base_url}/api/assets/{asset_id}", headers=self._headers, timeout=10)
resp.raise_for_status()
return resp.json()
def create_share_link(self, asset_id: str, expires_in_s: int) -> str:
"""Creates a public, view-only Immich share link for a single
asset, expiring expires_in_s seconds from now, and returns its
public URL. Used by the manage-button overlay's share QR --
created lazily (only when someone actually scans it), not when
the button's pressed, so the expiry clock starts when it's
actually used."""
expires_at = (datetime.now(timezone.utc) + timedelta(seconds=expires_in_s)).isoformat()
resp = httpx.post(
f"{self.base_url}/api/shared-links",
headers=self._headers,
json={
"type": "INDIVIDUAL",
"assetIds": [asset_id],
"expiresAt": expires_at,
"allowUpload": False,
"allowDownload": True,
"showMetadata": True,
},
timeout=10,
)
resp.raise_for_status()
key = resp.json()["key"]
return f"{self.base_url}/share/{key}"