tfaour
fdcb444d57
Replaces the ad-hoc inline styling with a shared base template driven by CSS custom properties (light/dark palettes), a card-based two-column layout, and a persistent dark-mode toggle. Also moves quiet-hours' timezone from the container's TZ env var into a proper web UI setting (zoneinfo-backed), so it no longer needs a docker-compose.yml edit and restart to change.
802 lines
32 KiB
Python
802 lines
32 KiB
Python
"""ESPresso Frame server: pulls photos from Immich, pre-processes them for
|
|
the panel, and serves the ESP32 a ready-to-display frame."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import logging
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
from zoneinfo import ZoneInfo, available_timezones
|
|
|
|
import httpx
|
|
from fastapi import Depends, FastAPI, File, HTTPException, Form, Request, UploadFile
|
|
from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse, Response
|
|
from fastapi.templating import Jinja2Templates
|
|
from PIL import Image
|
|
from pydantic import BaseModel
|
|
|
|
from . import config, photo_queue
|
|
from .face_labels import compute_face_labels
|
|
from .image_pipeline import render_frame
|
|
from .immich_client import ImmichClient
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
app = FastAPI(title="ESPresso Frame Server")
|
|
templates = Jinja2Templates(directory="app/templates")
|
|
|
|
MIN_REFRESH_INTERVAL_S = 60
|
|
MAX_REFRESH_INTERVAL_S = 86400
|
|
MIN_QUEUE_TARGET_LEN = 5
|
|
MAX_QUEUE_TARGET_LEN = 5000
|
|
|
|
ORIENTATIONS = ("landscape", "portrait", "landscape_flipped", "portrait_flipped")
|
|
|
|
# Populated once from the OS's zoneinfo database (installed via the
|
|
# `tzdata` apt package in the Dockerfile) and offered as a <select> in the
|
|
# web UI's "Timezone" field -- see api_config_save/index below.
|
|
ALL_TIMEZONES = sorted(available_timezones())
|
|
|
|
MANAGEMENT_TOKEN_COOKIE = "mgmt_token"
|
|
|
|
# Battery-history / estimate tuning (see /frame/battery and _battery_estimate).
|
|
BATTERY_HISTORY_MAX = 500 # ~20 days at hourly reports
|
|
BATTERY_LOG_MAX = 20000 # ~2 years at hourly reports -- see FrameConfig.battery_log
|
|
RECHARGE_JUMP_PCT = 5 # a report this much above the previous one = battery was recharged
|
|
MIN_ESTIMATE_SPAN_S = 2 * 3600 # need at least this much observed time...
|
|
MIN_ESTIMATE_DROP_PCT = 2 # ...and this much observed drop before estimating
|
|
|
|
# "Overdue" threshold multiplier: the device should check in roughly every
|
|
# refresh_interval_s; give it half again as long before flagging it.
|
|
OVERDUE_FACTOR = 1.5
|
|
|
|
|
|
def _valid_hhmm(s: str) -> bool:
|
|
try:
|
|
datetime.strptime(s, "%H:%M")
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def _zoneinfo(name: str) -> ZoneInfo:
|
|
"""Falls back to UTC for an unrecognized zone name -- defensive only;
|
|
api_config_save already validates against ALL_TIMEZONES before saving,
|
|
so this only matters for a config.json hand-edited or written by an
|
|
older version of this file."""
|
|
try:
|
|
return ZoneInfo(name)
|
|
except Exception:
|
|
return ZoneInfo("UTC")
|
|
|
|
|
|
def _quiet_hours_state(now: datetime, start_str: str, end_str: str) -> tuple[bool, datetime | None]:
|
|
"""Whether `now` falls inside the quiet-hours window, and the next
|
|
boundary: if inside, when it ends; if outside, when it next starts.
|
|
Handles a window that wraps past midnight (e.g. 22:00-07:00). Returns
|
|
(False, None) for a degenerate window (start == end)."""
|
|
sh, sm = (int(x) for x in start_str.split(":"))
|
|
eh, em = (int(x) for x in end_str.split(":"))
|
|
start = now.replace(hour=sh, minute=sm, second=0, microsecond=0)
|
|
end = now.replace(hour=eh, minute=em, second=0, microsecond=0)
|
|
|
|
if start == end:
|
|
return False, None
|
|
|
|
if start < end:
|
|
# Same-day window, e.g. 13:00-15:00. End is exclusive, so being
|
|
# exactly at `end` counts as already outside the window.
|
|
if start <= now < end:
|
|
return True, end
|
|
if now < start:
|
|
return False, start
|
|
return False, start + timedelta(days=1)
|
|
|
|
# Wraps midnight, e.g. 22:00-07:00.
|
|
if now >= start:
|
|
return True, end + timedelta(days=1)
|
|
if now < end:
|
|
return True, end
|
|
return False, start
|
|
|
|
|
|
def _quiet_hours_span_s(start_str: str, end_str: str) -> int:
|
|
"""Duration of the quiet-hours window in seconds, wrap-aware."""
|
|
sh, sm = (int(x) for x in start_str.split(":"))
|
|
eh, em = (int(x) for x in end_str.split(":"))
|
|
span_min = ((eh * 60 + em) - (sh * 60 + sm)) % (24 * 60)
|
|
return span_min * 60
|
|
|
|
|
|
def _effective_refresh_interval_s(cfg: config.FrameConfig) -> int:
|
|
"""The refresh interval actually handed to the device: its configured
|
|
value, unless quiet hours are enabled, in which case it's clamped so
|
|
the device sleeps through the whole window instead of waking inside
|
|
it. A device already mid-sleep when quiet hours begin can still land
|
|
one wake inside the window (nothing server-side can prevent that
|
|
without touching the firmware) -- but from that wake on, it's told to
|
|
sleep exactly until the window ends."""
|
|
if not cfg.quiet_hours_enabled:
|
|
return cfg.refresh_interval_s
|
|
now = datetime.now(_zoneinfo(cfg.timezone))
|
|
in_quiet, boundary = _quiet_hours_state(now, cfg.quiet_hours_start, cfg.quiet_hours_end)
|
|
if boundary is None:
|
|
return cfg.refresh_interval_s
|
|
seconds_to_boundary = max(1, int((boundary - now).total_seconds()))
|
|
if in_quiet:
|
|
return seconds_to_boundary
|
|
return min(cfg.refresh_interval_s, seconds_to_boundary)
|
|
|
|
|
|
def _max_expected_gap_s(cfg: config.FrameConfig) -> int:
|
|
"""Longest gap between wakes the device might legitimately have --
|
|
normally just refresh_interval_s, but quiet hours can make the real
|
|
gap much longer, and the "overdue" check (see api_queue) shouldn't
|
|
mistake a device quietly sleeping through the night for a dead one."""
|
|
gap = cfg.refresh_interval_s
|
|
if cfg.quiet_hours_enabled:
|
|
gap = max(gap, _quiet_hours_span_s(cfg.quiet_hours_start, cfg.quiet_hours_end))
|
|
return gap
|
|
|
|
|
|
def _touch_last_seen() -> None:
|
|
"""Records that the device just made contact. Called by every
|
|
/frame/* route -- a handful of extra config writes per wake cycle,
|
|
which is nothing at hourly wakes."""
|
|
with config.locked():
|
|
cfg = config.load()
|
|
cfg.last_seen = time.time()
|
|
config.save(cfg)
|
|
|
|
|
|
def _token_valid(request: Request, cfg: config.FrameConfig) -> bool:
|
|
"""No management_token configured (MANAGEMENT_TOKEN env var, see
|
|
docker-compose.yml.example) means the whole server stays open on a
|
|
trusted LAN, matching this project's original default. Once one's
|
|
set, a request is authorized by either a ?token= query param (what
|
|
the ESP32 sends on every device request, and what the manage-menu/
|
|
share QR codes embed for a human scanning them) or the cookie
|
|
index() sets after a valid query-param hit (so the web UI's own
|
|
fetch()/<img> calls, which carry no query string, stay authorized
|
|
for the rest of that browsing visit)."""
|
|
if not cfg.management_token:
|
|
return True
|
|
supplied = request.query_params.get("token") or request.cookies.get(MANAGEMENT_TOKEN_COOKIE)
|
|
return supplied is not None and supplied == cfg.management_token
|
|
|
|
|
|
def require_access_token(request: Request) -> None:
|
|
"""Dependency for every route except / and /health: the web UI's
|
|
/api/* and every device-facing /frame/*. index() handles the
|
|
unauthorized case itself (a friendlier HTML prompt, not a bare 401)
|
|
since that's the one route a human is actually meant to land on with
|
|
no token yet; the ESP32 sends its token as ?token= on every request
|
|
it makes (see frame_client.c's build_url()), so device endpoints
|
|
just 401 outright on a missing/wrong one. /health stays open -- it
|
|
reveals nothing but process liveness, and gating it would break
|
|
plain infra/uptime monitoring for no real security benefit."""
|
|
if not _token_valid(request, config.load()):
|
|
raise HTTPException(401, "Missing or invalid access token")
|
|
|
|
|
|
@app.get("/health")
|
|
def health() -> dict:
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.get("/frame/config", dependencies=[Depends(require_access_token)])
|
|
def frame_config(request: Request):
|
|
"""Device-facing settings, polled by the frame alongside its
|
|
reachability check. Always returns 200 with current settings
|
|
(defaults if nothing's been saved yet) -- no Immich-configured gate,
|
|
since this doubles as the "is the server up" signal. Also captures
|
|
the device's running firmware version (X-Frame-Version header) and
|
|
advertises the uploaded OTA image's version, so the device's update
|
|
check costs zero extra round trips."""
|
|
reported_version = request.headers.get("X-Frame-Version", "")
|
|
with config.locked():
|
|
cfg = config.load()
|
|
cfg.last_seen = time.time()
|
|
if reported_version:
|
|
cfg.device_firmware_version = reported_version
|
|
config.save(cfg)
|
|
return {
|
|
"refresh_interval_s": _effective_refresh_interval_s(cfg),
|
|
"firmware_version": cfg.firmware_available_version or None,
|
|
}
|
|
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
def index(request: Request):
|
|
cfg = config.load()
|
|
if not _token_valid(request, cfg):
|
|
supplied = request.query_params.get("token")
|
|
return templates.TemplateResponse(
|
|
"token_prompt.html", {"request": request, "wrong": supplied is not None}
|
|
)
|
|
|
|
response = templates.TemplateResponse(
|
|
"index.html", {"request": request, "cfg": cfg, "timezones": ALL_TIMEZONES}
|
|
)
|
|
supplied = request.query_params.get("token")
|
|
if cfg.management_token and supplied == cfg.management_token:
|
|
# Query-param access (typically the manage-menu QR code) earns a
|
|
# cookie so the rest of this visit's fetch()/<img> calls -- which
|
|
# never carry the query string -- stay authorized too.
|
|
response.set_cookie(
|
|
MANAGEMENT_TOKEN_COOKIE, supplied, max_age=86400 * 365, httponly=True, samesite="lax"
|
|
)
|
|
return response
|
|
|
|
|
|
@app.get("/api/albums", dependencies=[Depends(require_access_token)])
|
|
def api_albums():
|
|
cfg = config.load()
|
|
if not cfg.immich_url or not cfg.immich_api_key:
|
|
raise HTTPException(400, "Immich URL/API key not configured yet")
|
|
try:
|
|
albums = ImmichClient(cfg.immich_url, cfg.immich_api_key).list_albums()
|
|
except httpx.HTTPError as e:
|
|
raise HTTPException(502, f"Could not reach Immich at {cfg.immich_url}: {e}") from e
|
|
return [{"id": a["id"], "name": a["albumName"], "count": a.get("assetCount", 0)} for a in albums]
|
|
|
|
|
|
@app.post("/api/config", dependencies=[Depends(require_access_token)])
|
|
def api_config_save(
|
|
album_id: str = Form(""),
|
|
order: str = Form("sequential"),
|
|
refresh_interval_s: int = Form(3600),
|
|
smart_crop_faces: bool = Form(True),
|
|
queue_target_len: int = Form(20),
|
|
orientation: str = Form("landscape"),
|
|
quiet_hours_enabled: bool = Form(False),
|
|
quiet_hours_start: str = Form("22:00"),
|
|
quiet_hours_end: str = Form("07:00"),
|
|
timezone: str = Form("UTC"),
|
|
):
|
|
# Immich URL/API key are env-var only (IMMICH_URL/IMMICH_API_KEY, see
|
|
# docker-compose.yml.example) -- config.load() already applies them,
|
|
# and this handler doesn't touch cfg.immich_url/immich_api_key at all,
|
|
# so there's nothing here that could overwrite or clear them.
|
|
with config.locked():
|
|
cfg = config.load()
|
|
if album_id != cfg.album_id:
|
|
# A newly selected album starts clean -- the old current photo and
|
|
# queue don't mean anything in the new album's context.
|
|
cfg.current_asset_id = ""
|
|
cfg.current_asset_set_at = 0.0
|
|
cfg.queue = []
|
|
cfg.queue_cursor = 0
|
|
cfg.history = []
|
|
cfg.excluded_asset_ids = []
|
|
cfg.album_id = album_id
|
|
cfg.order = order if order in ("sequential", "shuffle") else "sequential"
|
|
cfg.refresh_interval_s = max(MIN_REFRESH_INTERVAL_S, min(MAX_REFRESH_INTERVAL_S, refresh_interval_s))
|
|
cfg.smart_crop_faces = smart_crop_faces
|
|
cfg.queue_target_len = max(MIN_QUEUE_TARGET_LEN, min(MAX_QUEUE_TARGET_LEN, queue_target_len))
|
|
cfg.orientation = orientation if orientation in ORIENTATIONS else "landscape"
|
|
cfg.quiet_hours_enabled = quiet_hours_enabled
|
|
if _valid_hhmm(quiet_hours_start):
|
|
cfg.quiet_hours_start = quiet_hours_start
|
|
if _valid_hhmm(quiet_hours_end):
|
|
cfg.quiet_hours_end = quiet_hours_end
|
|
if timezone in ALL_TIMEZONES:
|
|
cfg.timezone = timezone
|
|
config.save(cfg)
|
|
return {"status": "saved"}
|
|
|
|
|
|
def _require_configured(cfg: config.FrameConfig) -> None:
|
|
if not cfg.immich_url or not cfg.immich_api_key:
|
|
raise HTTPException(400, "Immich URL/API key not configured yet")
|
|
if not cfg.album_id:
|
|
raise HTTPException(400, "No album configured yet")
|
|
|
|
|
|
def _list_assets(client: ImmichClient, cfg: config.FrameConfig) -> list[dict]:
|
|
try:
|
|
assets = client.list_album_assets(cfg.album_id)
|
|
except httpx.HTTPError as e:
|
|
raise HTTPException(502, f"Could not reach Immich at {cfg.immich_url}: {e}") from e
|
|
if not assets:
|
|
raise HTTPException(404, "Album has no photos")
|
|
return assets
|
|
|
|
|
|
def _render_asset(client: ImmichClient, cfg: config.FrameConfig, asset_id: str) -> bytes:
|
|
try:
|
|
jpeg_bytes = client.download_asset_preview(asset_id)
|
|
except httpx.HTTPError as e:
|
|
raise HTTPException(502, f"Could not download asset from Immich: {e}") from e
|
|
|
|
faces = None
|
|
if cfg.smart_crop_faces:
|
|
try:
|
|
faces = client.get_asset_faces(asset_id)
|
|
except httpx.HTTPError as e:
|
|
# A faces lookup hiccup shouldn't block showing a photo at
|
|
# all -- just fall back to a plain center-crop this cycle.
|
|
logger.warning("Could not fetch faces for asset %s: %s", asset_id, e)
|
|
|
|
source = Image.open(io.BytesIO(jpeg_bytes))
|
|
return render_frame(source, faces=faces, orientation=cfg.orientation)
|
|
|
|
|
|
@app.get("/frame/image", dependencies=[Depends(require_access_token)])
|
|
def frame_image():
|
|
"""Returns the current photo. Idempotent: only actually advances to
|
|
the next photo once refresh_interval_s has elapsed since the current
|
|
one was set (see app/photo_queue.py) -- safe to call as often as the
|
|
device wants, including after an unplanned reboot, without skipping
|
|
ahead in the album."""
|
|
_touch_last_seen()
|
|
cfg = config.load()
|
|
_require_configured(cfg)
|
|
|
|
client = ImmichClient(cfg.immich_url, cfg.immich_api_key)
|
|
assets = _list_assets(client, cfg)
|
|
|
|
with config.locked():
|
|
cfg = config.load() # re-read: state may have changed since the unlocked read above
|
|
if photo_queue.get_current(cfg, assets):
|
|
config.save(cfg)
|
|
|
|
return Response(content=_render_asset(client, cfg, cfg.current_asset_id), media_type="application/octet-stream")
|
|
|
|
|
|
@app.post("/frame/advance", dependencies=[Depends(require_access_token)])
|
|
def frame_advance():
|
|
"""Forces an immediate advance to the next photo, ignoring
|
|
refresh_interval_s, and resets the interval clock from now. Used by
|
|
the device's next-photo button."""
|
|
_touch_last_seen()
|
|
cfg = config.load()
|
|
_require_configured(cfg)
|
|
|
|
client = ImmichClient(cfg.immich_url, cfg.immich_api_key)
|
|
assets = _list_assets(client, cfg)
|
|
|
|
with config.locked():
|
|
cfg = config.load() # re-read: state may have changed since the unlocked read above
|
|
photo_queue.advance_forced(cfg, assets)
|
|
config.save(cfg)
|
|
|
|
return Response(content=_render_asset(client, cfg, cfg.current_asset_id), media_type="application/octet-stream")
|
|
|
|
|
|
@app.post("/frame/back", dependencies=[Depends(require_access_token)])
|
|
def frame_back():
|
|
"""Returns to the previously-current photo (the mirror image of
|
|
/frame/advance -- see photo_queue.back_forced()), and resets the
|
|
interval clock from now. A no-op (still 200, current photo
|
|
unchanged) if there's no history to go back to -- same "always
|
|
returns something displayable" contract as /frame/advance, rather
|
|
than erroring. Used by the device's back-photo button."""
|
|
_touch_last_seen()
|
|
cfg = config.load()
|
|
_require_configured(cfg)
|
|
|
|
client = ImmichClient(cfg.immich_url, cfg.immich_api_key)
|
|
assets = _list_assets(client, cfg)
|
|
|
|
with config.locked():
|
|
cfg = config.load() # re-read: state may have changed since the unlocked read above
|
|
photo_queue.back_forced(cfg, assets)
|
|
config.save(cfg)
|
|
|
|
return Response(content=_render_asset(client, cfg, cfg.current_asset_id), media_type="application/octet-stream")
|
|
|
|
|
|
class BatteryReport(BaseModel):
|
|
percent: int
|
|
|
|
|
|
@app.post("/frame/battery", dependencies=[Depends(require_access_token)])
|
|
def frame_battery(body: BatteryReport):
|
|
"""Battery level reported by the device (only when running on battery
|
|
-- it stays silent on mains, where the charging voltage would read
|
|
misleadingly full). Stored with a timestamp plus a per-discharge-
|
|
cycle history that feeds the Device panel's "on battery for" and
|
|
"estimated remaining" numbers."""
|
|
if not 0 <= body.percent <= 100:
|
|
raise HTTPException(400, "percent must be 0-100")
|
|
now = time.time()
|
|
with config.locked():
|
|
cfg = config.load()
|
|
if cfg.battery_history and body.percent >= cfg.battery_history[-1][1] + RECHARGE_JUMP_PCT:
|
|
# Percent jumped up meaningfully -- the battery was recharged
|
|
# (or swapped). Start a fresh discharge cycle so runtime and
|
|
# discharge-rate estimates never span a charge.
|
|
cfg.battery_history = []
|
|
cfg.battery_history.append([now, body.percent])
|
|
cfg.battery_history = cfg.battery_history[-BATTERY_HISTORY_MAX:]
|
|
cfg.battery_log.append([now, body.percent])
|
|
cfg.battery_log = cfg.battery_log[-BATTERY_LOG_MAX:]
|
|
cfg.battery_percent = body.percent
|
|
cfg.battery_as_of = now
|
|
cfg.last_seen = now
|
|
config.save(cfg)
|
|
return {"status": "saved"}
|
|
|
|
|
|
# ESP-IDF app images embed an esp_app_desc_t at byte offset 32 (24-byte
|
|
# image header + 8-byte first-segment header): magic word, then version
|
|
# (32 bytes, NUL-padded) at +16 and project name (32 bytes) at +48 --
|
|
# verified against this project's real build artifact.
|
|
APP_DESC_OFFSET = 32
|
|
APP_DESC_MAGIC = 0xABCD5432
|
|
EXPECTED_PROJECT_NAME = "espresso_frame"
|
|
|
|
|
|
def _firmware_path():
|
|
return config.CONFIG_PATH.parent / "firmware.bin"
|
|
|
|
|
|
def _parse_app_version(data: bytes) -> str:
|
|
"""Extracts the embedded version from an ESP-IDF app image, raising
|
|
HTTPException(400) for anything that isn't this project's firmware."""
|
|
if len(data) < APP_DESC_OFFSET + 80:
|
|
raise HTTPException(400, "File is too small to be a firmware image")
|
|
magic = int.from_bytes(data[APP_DESC_OFFSET : APP_DESC_OFFSET + 4], "little")
|
|
if magic != APP_DESC_MAGIC:
|
|
raise HTTPException(400, "Not an ESP-IDF application image")
|
|
version = data[APP_DESC_OFFSET + 16 : APP_DESC_OFFSET + 48].split(b"\x00")[0].decode(errors="replace")
|
|
project = data[APP_DESC_OFFSET + 48 : APP_DESC_OFFSET + 80].split(b"\x00")[0].decode(errors="replace")
|
|
if project != EXPECTED_PROJECT_NAME:
|
|
raise HTTPException(400, f"Image is for project '{project}', not '{EXPECTED_PROJECT_NAME}'")
|
|
if not version:
|
|
raise HTTPException(400, "Image has no embedded version")
|
|
return version
|
|
|
|
|
|
@app.post("/api/firmware", dependencies=[Depends(require_access_token)])
|
|
def api_firmware_upload(file: UploadFile = File(...)):
|
|
"""Uploads a firmware image for OTA. The version is parsed out of the
|
|
image itself (esp_app_desc_t) rather than trusted from a filename or
|
|
form field, and the project name is checked so an unrelated .bin
|
|
can't be pushed to the frame by mistake."""
|
|
data = file.file.read()
|
|
version = _parse_app_version(data)
|
|
_firmware_path().write_bytes(data)
|
|
with config.locked():
|
|
cfg = config.load()
|
|
cfg.firmware_available_version = version
|
|
config.save(cfg)
|
|
return {"status": "saved", "version": version, "size": len(data)}
|
|
|
|
|
|
@app.get("/frame/firmware", dependencies=[Depends(require_access_token)])
|
|
def frame_firmware():
|
|
"""The uploaded OTA image, streamed to the device (esp_https_ota).
|
|
404 until something has been uploaded."""
|
|
_touch_last_seen()
|
|
path = _firmware_path()
|
|
if not path.exists():
|
|
raise HTTPException(404, "No firmware uploaded")
|
|
return FileResponse(path, media_type="application/octet-stream")
|
|
|
|
|
|
def _battery_estimate_s(cfg: config.FrameConfig) -> int | None:
|
|
"""Linear remaining-time estimate from the current discharge cycle's
|
|
observed rate, or None when there's not enough signal to be honest
|
|
about (too little time observed, or too little drop -- a flat line
|
|
extrapolates to garbage)."""
|
|
hist = cfg.battery_history
|
|
if len(hist) < 2:
|
|
return None
|
|
first_ts, first_pct = hist[0]
|
|
last_ts, last_pct = hist[-1]
|
|
span = last_ts - first_ts
|
|
drop = first_pct - last_pct
|
|
if span < MIN_ESTIMATE_SPAN_S or drop < MIN_ESTIMATE_DROP_PCT:
|
|
return None
|
|
rate = drop / span # percent per second
|
|
return int(last_pct / rate)
|
|
|
|
|
|
LOCATION_LINE_MAX_LEN = 14
|
|
|
|
US_STATE_ABBR = {
|
|
"alabama": "AL", "alaska": "AK", "arizona": "AZ", "arkansas": "AR", "california": "CA",
|
|
"colorado": "CO", "connecticut": "CT", "delaware": "DE", "florida": "FL", "georgia": "GA",
|
|
"hawaii": "HI", "idaho": "ID", "illinois": "IL", "indiana": "IN", "iowa": "IA",
|
|
"kansas": "KS", "kentucky": "KY", "louisiana": "LA", "maine": "ME", "maryland": "MD",
|
|
"massachusetts": "MA", "michigan": "MI", "minnesota": "MN", "mississippi": "MS", "missouri": "MO",
|
|
"montana": "MT", "nebraska": "NE", "nevada": "NV", "new hampshire": "NH", "new jersey": "NJ",
|
|
"new mexico": "NM", "new york": "NY", "north carolina": "NC", "north dakota": "ND", "ohio": "OH",
|
|
"oklahoma": "OK", "oregon": "OR", "pennsylvania": "PA", "rhode island": "RI", "south carolina": "SC",
|
|
"south dakota": "SD", "tennessee": "TN", "texas": "TX", "utah": "UT", "vermont": "VT",
|
|
"virginia": "VA", "washington": "WA", "west virginia": "WV", "wisconsin": "WI", "wyoming": "WY",
|
|
"district of columbia": "DC",
|
|
}
|
|
|
|
CA_PROVINCE_ABBR = {
|
|
"alberta": "AB", "british columbia": "BC", "manitoba": "MB", "new brunswick": "NB",
|
|
"newfoundland and labrador": "NL", "northwest territories": "NT", "nova scotia": "NS",
|
|
"nunavut": "NU", "ontario": "ON", "prince edward island": "PE", "quebec": "QC",
|
|
"saskatchewan": "SK", "yukon": "YT",
|
|
}
|
|
|
|
US_COUNTRY_NAMES = {"united states", "united states of america", "usa", "us"}
|
|
CA_COUNTRY_NAMES = {"canada"}
|
|
|
|
|
|
def _truncate(text: str, max_len: int) -> str:
|
|
if len(text) <= max_len:
|
|
return text
|
|
return text[: max_len - 3] + "..."
|
|
|
|
|
|
def _format_location(exif: dict) -> tuple[str, str] | None:
|
|
"""Returns (city_line, region_line), each independently truncated to
|
|
fit its own corner-overlay line, or None if Immich hasn't geocoded
|
|
this photo. region_line is the abbreviated state/province for US/CAN
|
|
locations (e.g. "CA", "ON"), else the full country name."""
|
|
city = exif.get("city")
|
|
if not city:
|
|
return None
|
|
|
|
state = exif.get("state")
|
|
country = exif.get("country")
|
|
country_key = (country or "").strip().lower()
|
|
|
|
if state and country_key in US_COUNTRY_NAMES:
|
|
region = US_STATE_ABBR.get(state.strip().lower(), state)
|
|
elif state and country_key in CA_COUNTRY_NAMES:
|
|
region = CA_PROVINCE_ABBR.get(state.strip().lower(), state)
|
|
elif country:
|
|
region = country
|
|
elif state:
|
|
region = state
|
|
else:
|
|
region = ""
|
|
|
|
return _truncate(city, LOCATION_LINE_MAX_LEN), _truncate(region, LOCATION_LINE_MAX_LEN)
|
|
|
|
|
|
def _format_taken_at(exif: dict) -> str | None:
|
|
raw = exif.get("dateTimeOriginal")
|
|
if not raw:
|
|
return None
|
|
try:
|
|
return datetime.fromisoformat(raw.replace("Z", "+00:00")).strftime("%m/%d/%y")
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
@app.get("/frame/photo-info", dependencies=[Depends(require_access_token)])
|
|
def frame_photo_info():
|
|
"""Location/date-taken text for the manage-button overlay, plus the
|
|
asset id used to build the share-QR's target URL. Read-only, same
|
|
idempotent current-photo semantics as /frame/image -- doesn't advance
|
|
anything."""
|
|
_touch_last_seen()
|
|
cfg = config.load()
|
|
_require_configured(cfg)
|
|
|
|
client = ImmichClient(cfg.immich_url, cfg.immich_api_key)
|
|
assets = _list_assets(client, cfg)
|
|
|
|
with config.locked():
|
|
cfg = config.load() # re-read: state may have changed since the unlocked read above
|
|
if photo_queue.get_current(cfg, assets):
|
|
config.save(cfg)
|
|
|
|
if not cfg.current_asset_id:
|
|
raise HTTPException(404, "No current photo")
|
|
|
|
try:
|
|
asset = client.get_asset(cfg.current_asset_id)
|
|
except httpx.HTTPError as e:
|
|
raise HTTPException(502, f"Could not reach Immich: {e}") from e
|
|
|
|
exif = asset.get("exifInfo") or {}
|
|
location = _format_location(exif)
|
|
return {
|
|
"asset_id": cfg.current_asset_id,
|
|
"location_line1": location[0] if location else None,
|
|
"location_line2": location[1] if location and location[1] else None,
|
|
"taken_at": _format_taken_at(exif),
|
|
}
|
|
|
|
|
|
@app.get("/frame/share/{asset_id}", dependencies=[Depends(require_access_token)])
|
|
def frame_share(asset_id: str):
|
|
"""Creates a 30-minute public Immich share link for asset_id and
|
|
redirects to it -- what the manage overlay's bottom-left QR code
|
|
points to (the firmware bakes ?token= into that QR the same way it
|
|
does for the management QR, see frame_client.c's build_url()). The
|
|
link is created lazily, when this actually gets hit (i.e. when
|
|
someone scans it), not when the manage button was pressed, so the
|
|
30-minute window starts when it's actually used. Also scoped to the
|
|
photo currently showing or queued -- not any arbitrary Immich asset
|
|
id -- as a second layer even a leaked token wouldn't bypass."""
|
|
_touch_last_seen()
|
|
cfg = config.load()
|
|
_require_configured(cfg)
|
|
|
|
if asset_id != cfg.current_asset_id and asset_id not in cfg.queue:
|
|
raise HTTPException(404, "That photo isn't currently showing or queued on this frame")
|
|
|
|
client = ImmichClient(cfg.immich_url, cfg.immich_api_key)
|
|
try:
|
|
share_url = client.create_share_link(asset_id, expires_in_s=1800)
|
|
except httpx.HTTPError as e:
|
|
raise HTTPException(502, f"Could not create share link: {e}") from e
|
|
|
|
return RedirectResponse(share_url)
|
|
|
|
|
|
@app.get("/frame/face-labels", dependencies=[Depends(require_access_token)])
|
|
def frame_face_labels():
|
|
"""Named-face positions for the manage button's escalated "level 2"
|
|
menu -- who's in the current photo, per Immich's own face
|
|
recognition (no detection/recognition happens here, see
|
|
app/face_labels.py). Response is a flattened, fixed-slot shape
|
|
(name_0/x_0/y_0, name_1/x_1/y_1, ...) rather than a JSON array, so
|
|
the device's hand-rolled parser can read it with the same flat-
|
|
scalar helpers it already has, instead of needing a real array
|
|
parser. Empty (count: 0) if no faces are named, or if anything about
|
|
fetching them fails -- this is a "nice to have" addition to the
|
|
overlay, not worth failing the whole menu over."""
|
|
_touch_last_seen()
|
|
cfg = config.load()
|
|
_require_configured(cfg)
|
|
|
|
client = ImmichClient(cfg.immich_url, cfg.immich_api_key)
|
|
assets = _list_assets(client, cfg)
|
|
|
|
with config.locked():
|
|
cfg = config.load() # re-read: state may have changed since the unlocked read above
|
|
if photo_queue.get_current(cfg, assets):
|
|
config.save(cfg)
|
|
|
|
if not cfg.current_asset_id:
|
|
return {"count": 0}
|
|
|
|
try:
|
|
faces = client.get_asset_faces(cfg.current_asset_id)
|
|
except httpx.HTTPError as e:
|
|
logger.warning("Could not fetch faces for asset %s: %s", cfg.current_asset_id, e)
|
|
return {"count": 0}
|
|
|
|
if not any((face.get("person") or {}).get("name") for face in faces):
|
|
return {"count": 0} # skip the extra preview download in the common no-named-faces case
|
|
|
|
try:
|
|
preview_bytes = client.download_asset_preview(cfg.current_asset_id)
|
|
except httpx.HTTPError as e:
|
|
logger.warning("Could not download asset %s for face-label mapping: %s", cfg.current_asset_id, e)
|
|
return {"count": 0}
|
|
|
|
labels = compute_face_labels(preview_bytes, faces, cfg.smart_crop_faces, cfg.orientation)
|
|
|
|
result: dict[str, object] = {"count": len(labels)}
|
|
for i, label in enumerate(labels):
|
|
result[f"name_{i}"] = label["name"]
|
|
result[f"x_{i}"] = label["x"]
|
|
result[f"y_{i}"] = label["y"]
|
|
return result
|
|
|
|
|
|
@app.get("/api/queue", dependencies=[Depends(require_access_token)])
|
|
def api_queue():
|
|
cfg = config.load()
|
|
_require_configured(cfg)
|
|
|
|
client = ImmichClient(cfg.immich_url, cfg.immich_api_key)
|
|
assets = _list_assets(client, cfg)
|
|
|
|
with config.locked():
|
|
cfg = config.load() # re-read: state may have changed since the unlocked read above
|
|
current_changed = photo_queue.get_current(cfg, assets)
|
|
queue_before = list(cfg.queue)
|
|
photo_queue.sync_queue_length(cfg, assets)
|
|
if current_changed or cfg.queue != queue_before:
|
|
config.save(cfg)
|
|
|
|
def entry(asset_id: str) -> dict:
|
|
return {"id": asset_id, "thumbnail_url": f"/api/photo-thumbnail/{asset_id}"}
|
|
|
|
now = time.time()
|
|
return {
|
|
"current": entry(cfg.current_asset_id) if cfg.current_asset_id else None,
|
|
"upcoming": [entry(asset_id) for asset_id in cfg.queue],
|
|
"device": {
|
|
"last_seen": cfg.last_seen or None,
|
|
"overdue": bool(cfg.last_seen and now - cfg.last_seen > _max_expected_gap_s(cfg) * OVERDUE_FACTOR),
|
|
"firmware_version": cfg.device_firmware_version or None,
|
|
"firmware_available": cfg.firmware_available_version or None,
|
|
"battery": (
|
|
{"percent": cfg.battery_percent, "as_of": cfg.battery_as_of}
|
|
if cfg.battery_percent >= 0
|
|
else None
|
|
),
|
|
"on_battery_since": cfg.battery_history[0][0] if cfg.battery_history else None,
|
|
"battery_estimate_s": _battery_estimate_s(cfg),
|
|
},
|
|
}
|
|
|
|
|
|
@app.get("/api/battery-log", dependencies=[Depends(require_access_token)])
|
|
def api_battery_log():
|
|
cfg = config.load()
|
|
return {"log": cfg.battery_log}
|
|
|
|
|
|
class QueueReorderRequest(BaseModel):
|
|
queue: list[str]
|
|
|
|
|
|
@app.post("/api/queue/reorder", dependencies=[Depends(require_access_token)])
|
|
def api_queue_reorder(body: QueueReorderRequest):
|
|
"""Applies the client's requested order, tolerating drift between the
|
|
browser's last-fetched snapshot and the server's current queue (e.g.
|
|
a top-up/trim landed in between) instead of hard-rejecting: any ID
|
|
the client sent that's no longer actually queued is dropped, and any
|
|
ID the server has that the client didn't know about is appended
|
|
rather than lost."""
|
|
with config.locked():
|
|
cfg = config.load()
|
|
current_set = set(cfg.queue)
|
|
reordered = [asset_id for asset_id in body.queue if asset_id in current_set]
|
|
reordered += [asset_id for asset_id in cfg.queue if asset_id not in set(reordered)]
|
|
cfg.queue = reordered
|
|
config.save(cfg)
|
|
return {"status": "saved"}
|
|
|
|
|
|
class QueuePromoteRequest(BaseModel):
|
|
asset_id: str
|
|
|
|
|
|
@app.post("/api/queue/promote", dependencies=[Depends(require_access_token)])
|
|
def api_queue_promote(body: QueuePromoteRequest):
|
|
"""Moves a single photo to the front of the queue -- "Show next" in
|
|
the web UI. Unlike /api/queue/reorder, this doesn't depend on the
|
|
client supplying a full, exactly-current snapshot of the queue at
|
|
all, so it can't fail due to the queue having shifted server-side
|
|
since the browser's last fetch."""
|
|
with config.locked():
|
|
cfg = config.load()
|
|
if body.asset_id not in cfg.queue:
|
|
raise HTTPException(400, "That photo is no longer in the upcoming queue")
|
|
cfg.queue = [body.asset_id] + [asset_id for asset_id in cfg.queue if asset_id != body.asset_id]
|
|
config.save(cfg)
|
|
return {"status": "saved"}
|
|
|
|
|
|
class QueueRemoveRequest(BaseModel):
|
|
asset_id: str
|
|
|
|
|
|
@app.post("/api/queue/remove", dependencies=[Depends(require_access_token)])
|
|
def api_queue_remove(body: QueueRemoveRequest):
|
|
"""Permanently removes a photo from this frame's rotation -- "Remove"
|
|
in the web UI, on either an upcoming card or the current photo. Does
|
|
NOT touch Immich or the album itself; see photo_queue.remove_from_rotation()."""
|
|
cfg = config.load()
|
|
_require_configured(cfg)
|
|
|
|
client = ImmichClient(cfg.immich_url, cfg.immich_api_key)
|
|
assets = _list_assets(client, cfg)
|
|
|
|
with config.locked():
|
|
cfg = config.load() # re-read: state may have changed since the unlocked read above
|
|
photo_queue.remove_from_rotation(cfg, assets, body.asset_id)
|
|
config.save(cfg)
|
|
return {"status": "removed"}
|
|
|
|
|
|
@app.get("/api/photo-thumbnail/{asset_id}", dependencies=[Depends(require_access_token)])
|
|
def api_photo_thumbnail(asset_id: str):
|
|
cfg = config.load()
|
|
_require_configured(cfg)
|
|
client = ImmichClient(cfg.immich_url, cfg.immich_api_key)
|
|
try:
|
|
content, content_type = client.download_asset_thumbnail(asset_id)
|
|
except httpx.HTTPError as e:
|
|
raise HTTPException(502, f"Could not download thumbnail from Immich: {e}") from e
|
|
return Response(content=content, media_type=content_type)
|