tfaour/espresso_frame
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases 13 Wiki Activity

Redesign phase A: SQLite storage, per-frame data model, device identity

Browse Source 
Replaces the single global config.json (whole-file pydantic model under
one RLock) with SQLite via SQLAlchemy 2.0: users/sessions/frames/links/
pending-claims/battery_log tables (models.py), a per-frame lock registry
(db.frame_locked) succeeding config.locked(), and hand-rolled schema
versioning (migration.py). A pre-database deployment's config.json is
imported verbatim as frame #1 on first boot and left untouched as the
rollback path; the old single firmware.bin slot becomes per-frame
firmware/<id>.bin.

Routes split out of the 900-line main.py into routers/device.py (the
frozen /frame/* protocol) and routers/api.py (web UI, still on the old
single-frame paths for now). Device auth moves to require_device, which
already speaks the full multi-frame protocol: per-frame device tokens
pushed via /frame/config and acknowledged on first use, self-
registration of unknown device ids as unclaimed frames, pending-claim
attachment, and the legacy-token migration window that keeps the
currently-deployed firmware (no id, shared MANAGEMENT_TOKEN) resolving
to frame #1 -- including the one-time binding of its device id when it
first reports one after a future OTA.

Externally identical for existing deployments: same paths, same token
semantics, same response shapes -- verified with a migration fixture,
the legacy-device curl suite, a 20-way concurrent-advance smoke test,
and a mutate-restart-assert persistence check against a fake Immich.

photo_queue.py ports nearly verbatim onto the Frame ORM row (MutableList
JSON columns make its in-place list mutations dirty-track); quiet-hours
math extracted unchanged into quiet_hours.py.
This commit is contained in:
tfaour
2026-07-21 23:21:38 -04:00
parent 6a0072e383
commit 9fbbb8ed2b
16 changed files with 1650 additions and 1013 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/README.md
+8 -5
View File
@@ -223,11 +223,14 @@ algorithm itself -- it just streams the response straight to the panel.
## Notes
- Album/order/refresh-interval/current photo/upcoming queue/etc. are
stored in `./data/config.json` on the host via the compose volume
mount. Immich URL/API key are too if set via the web UI, but
`IMMICH_URL`/`IMMICH_API_KEY` env vars (see Setup above) always take
precedence when present.
- All state (settings, current photo, upcoming queue, battery history,
stats) lives in a SQLite database at `./data/espresso.db` on the host
via the compose volume mount (`DATABASE_URL` env var to override --
any SQLAlchemy URL works, so a future move to Postgres is a config
change). A pre-database deployment's `./data/config.json` is imported
automatically on first boot (it becomes frame #1) and left untouched
afterwards as the rollback path. `IMMICH_URL`/`IMMICH_API_KEY` env
vars (see Setup above) still take precedence when present.
- The upcoming queue is a bounded lookahead, not the whole album --
"Upcoming photos to show" in the config UI (`queue_target_len`, 5-50,
default 20) controls its size and takes effect immediately (the queue
server/app/auth.py
+142
View File
@@ -0,0 +1,142 @@
"""Authentication dependencies.
Phase A scope: browser routes keep the legacy shared-token gate
(MANAGEMENT_TOKEN env var -- empty means open on a trusted LAN, exactly
the old behavior), and device routes move to require_device, which
already implements the full multi-frame resolution: per-frame device
tokens, self-registration by device id, the legacy-token migration
window, and pending-claim attachment. User sessions arrive in Phase B.
"""
from __future__ import annotations
import logging
import os
import time
from fastapi import Depends, HTTPException, Request
from sqlalchemy import select
from sqlalchemy.orm import Session
from .db import get_db
from .migration import new_device_token, new_manage_token
from .models import Frame, PendingClaim, UserFrame
logger = logging.getLogger(__name__)
MANAGEMENT_TOKEN_COOKIE = "mgmt_token"
def management_token() -> str:
"""The legacy shared secret. Env-only, never stored -- same as the old
server, where the env var overrode anything on disk on every load."""
return os.environ.get("MANAGEMENT_TOKEN", "")
def browser_token_valid(request: Request) -> bool:
"""No MANAGEMENT_TOKEN configured means the web UI 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 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)."""
token = management_token()
if not token:
return True
supplied = request.query_params.get("token") or request.cookies.get(MANAGEMENT_TOKEN_COOKIE)
return supplied is not None and supplied == token
def require_access_token(request: Request) -> None:
"""Dependency for the browser-facing /api/* routes (Phase A only --
replaced by real sessions in Phase B). index() handles the
unauthorized case itself with a friendlier HTML prompt."""
if not browser_token_valid(request):
raise HTTPException(401, "Missing or invalid access token")
def _register_frame(db: Session, device_id: str) -> Frame:
"""A device id we've never seen: self-register it as an unclaimed
frame (this fires from ANY /frame/* route -- the wake cycle hits
/frame/image before /frame/config). If a user already submitted a
claim for this id (they beat the device to the server after
provisioning), attach it now."""
frame = Frame(
name=f"Frame {device_id[-6:]}",
device_id=device_id,
device_token=new_device_token(),
manage_token=new_manage_token(),
created_at=time.time(),
)
db.add(frame)
db.flush()
now = time.time()
# Opportunistically prune expired claims while we're here.
for stale in db.scalars(select(PendingClaim).where(PendingClaim.expires_at < now)):
db.delete(stale)
pending = db.get(PendingClaim, device_id)
if pending is not None and pending.expires_at >= now:
frame.owner_user_id = pending.user_id
frame.claimed_at = now
db.add(UserFrame(user_id=pending.user_id, frame_id=frame.id))
db.delete(pending)
logger.info("Frame %s self-registered and attached pending claim by user %d",
device_id, pending.user_id)
else:
logger.info("Frame %s self-registered (unclaimed)", device_id)
return frame
def require_device(request: Request, db: Session = Depends(get_db)) -> Frame:
"""Resolves and authenticates the frame behind a /frame/* request.
New firmware sends ?id=<12-hex-mac>&token=<per-frame device token>.
Deployed legacy firmware sends only ?token=<shared MANAGEMENT_TOKEN>
(or nothing, on an open server) -- those requests resolve to the
unique legacy_token_enabled frame for as long as that migration
window stays open. The first id-bearing request arriving with legacy
credentials while the legacy frame has no device_id yet BINDS that id
to it -- that's the moment the deployed frame comes back up on new
firmware after its OTA, and it must not register as a second frame.
"""
device_id = request.query_params.get("id", "").strip().lower()
token = request.query_params.get("token", "")
legacy = management_token()
legacy_ok = not legacy or token == legacy
if device_id:
frame = db.scalars(select(Frame).where(Frame.device_id == device_id)).first()
if frame is None:
legacy_frame = db.scalars(
select(Frame).where(Frame.legacy_token_enabled == True) # noqa: E712
).first()
if legacy_frame is not None and legacy_frame.device_id is None and legacy_ok:
legacy_frame.device_id = device_id
frame = legacy_frame
logger.info("Bound device id %s to legacy frame #%d", device_id, frame.id)
else:
frame = _register_frame(db, device_id)
else:
token_ok = bool(token) and token == frame.device_token
if token_ok and not frame.device_token_ack:
frame.device_token_ack = True
logger.info("Frame #%d acknowledged its device token", frame.id)
if not token_ok and not (frame.legacy_token_enabled and legacy_ok):
raise HTTPException(401, "Missing or invalid access token")
else:
if not legacy_ok:
raise HTTPException(401, "Missing or invalid access token")
frame = db.scalars(
select(Frame).where(Frame.legacy_token_enabled == True) # noqa: E712
).first()
if frame is None:
# Nothing to resolve a no-id request to. migration.py always
# creates frame #1 at startup, so this only happens if it was
# deleted -- treat like an unknown device.
raise HTTPException(401, "No frame accepts legacy credentials")
frame.last_seen = time.time()
db.commit()
return frame
server/app/config.py
+32 -103
View File
@@ -1,137 +1,85 @@
"""JSON-file-backed config: Immich connection, selected album, and cursor
state (which photo /frame/image serves next)."""
"""LEGACY config.json model -- kept only so migration.py can import an
existing single-frame deployment's state into the database on first
boot. Nothing else should import this module; runtime state lives in
SQLite (see models.py/db.py).
The file at CONFIG_PATH is deliberately never modified or deleted by the
migration: it's the rollback path (redeploying the pre-database server
image picks it right back up).
"""
from __future__ import annotations
import json
import os
from contextlib import contextmanager
from pathlib import Path
from threading import RLock
from typing import Iterator
from pydantic import BaseModel
CONFIG_PATH = Path(os.environ.get("CONFIG_PATH", "/data/config.json"))
# Reentrant so load()/save() can each take it internally for their own I/O
# while a caller also holds it for a whole locked() span (see below).
_lock = RLock()
class FrameStats(BaseModel):
"""Cumulative, lifetime counters -- purely informational, never read
back to drive any behavior, so there's no harm in them being a little
approximate at the edges. Shown in a collapsed "Stats" section in the
web UI (GET /api/stats). Never reset except by deleting config.json."""
first_seen: float = 0.0 # first time this frame ever checked in
device_wakes: int = 0 # wake cycles, counted once each via GET /frame/config
photos_displayed: int = 0 # times the current photo actually changed (any cause)
photos_removed: int = 0 # times a photo was permanently excluded from rotation
battery_reports: int = 0 # POST /frame/battery calls
recharge_cycles: int = 0 # times a battery recharge was detected
ota_updates_applied: int = 0 # times the device's reported firmware version changed
config_saves: int = 0 # POST /api/config calls
first_seen: float = 0.0
device_wakes: int = 0
photos_displayed: int = 0
photos_removed: int = 0
battery_reports: int = 0
recharge_cycles: int = 0
ota_updates_applied: int = 0
config_saves: int = 0
class FrameConfig(BaseModel):
immich_url: str = ""
immich_api_key: str = ""
management_token: str = "" # gates the web UI (see main.py); empty = no gate, open on trusted LAN
management_token: str = ""
album_id: str = ""
order: str = "sequential" # or "shuffle"
order: str = "sequential"
refresh_interval_s: int = 3600
# Quiet hours: no point waking the device overnight just to swap a
# photo nobody's looking at. Times are "HH:MM" interpreted in
# `timezone` below and may wrap past midnight (e.g. start=22:00,
# end=07:00). Purely a server-side decision -- the device is unaware,
# it just gets told a longer refresh_interval_s by GET /frame/config
# while quiet hours are in effect (see main.py's
# _effective_refresh_interval_s).
quiet_hours_enabled: bool = False
quiet_hours_start: str = "22:00"
quiet_hours_end: str = "07:00"
# IANA zone name (e.g. "America/New_York") quiet_hours_start/end are
# interpreted in. Set from the web UI rather than the container's TZ
# environment variable, so it survives container recreation and
# doesn't need a docker-compose.yml edit to change.
timezone: str = "UTC"
smart_crop_faces: bool = True
# How the physical frame is hung: landscape (native), portrait,
# landscape_flipped, portrait_flipped. Purely a server-side render
# decision -- the device always receives native 800x480 bytes.
orientation: str = "landscape"
# Current photo + upcoming queue (see app/photo_queue.py). current_asset_set_at
# is what lets the server decide "has it been long enough to advance" on its
# own clock, independent of how/why the device asked for a photo.
current_asset_id: str = ""
current_asset_set_at: float = 0.0
queue: list[str] = []
queue_cursor: int = 0 # internal bookkeeping for sequential queue top-up; not user-facing
queue_target_len: int = 20 # how many upcoming photos to keep queued/shown in the web UI
history: list[str] = [] # bounded stack of previously-current asset ids, most recent last
excluded_asset_ids: list[str] = [] # permanently removed from this frame's rotation (not deleted from Immich)
queue_cursor: int = 0
queue_target_len: int = 20
history: list[str] = []
excluded_asset_ids: list[str] = []
# Last battery report from the device (POST /frame/battery); -1 = never
# reported / not battery-powered. battery_as_of mirrors the
# current_asset_set_at timestamp pattern.
battery_percent: int = -1
battery_as_of: float = 0.0
# [timestamp, percent] pairs for the CURRENT discharge cycle only --
# reset whenever a report jumps up enough to indicate a recharge (see
# main.py). Feeds the "on battery for" and "estimated remaining"
# numbers in the web UI's Device panel.
battery_history: list = []
# Every report ever received, never reset by a recharge -- the
# permanent record behind the web UI's battery history graph. Capped
# generously (not a real limit at realistic report rates, just a
# safety bound), unlike battery_history above which is deliberately
# scoped to one cycle.
battery_log: list = []
# Device liveness/telemetry: last_seen is touched by every /frame/*
# request; device_firmware_version/device_board_variant come from the
# X-Frame-Version/X-Frame-Board headers the device sends with its
# config poll (CONFIG_FRAME_BOARD_NAME on the firmware side).
last_seen: float = 0.0
device_firmware_version: str = ""
device_board_variant: str = "" # "" until a device has ever checked in
# Version parsed out of the most recently uploaded OTA image
# (POST /api/firmware); "" = none uploaded yet.
device_board_variant: str = ""
firmware_available_version: str = ""
# Gitea-hosted firmware auto-update (see app/gitea_releases.py).
# repo_url empty = feature off, no Gitea calls made at all. Which
# release asset to pull is learned from the device itself
# (device_board_variant below, from its X-Frame-Board header) rather
# than picked by the user -- must match one of the names
# .gitea/workflows/firmware-release-build.yml publishes
# (firmware-<board_variant>.bin).
firmware_update_repo_url: str = "" # e.g. "https://git.example.com/owner/repo"
firmware_auto_update: bool = False # pull+stage a newer release with no button click
# Optional Gitea PAT (read-only access is enough) for a private repo's
# releases; blank is fine for a public repo. GITEA_FIRMWARE_TOKEN env
# var overrides, mirroring MANAGEMENT_TOKEN below -- never exposed to
# the web UI template or any JSON response.
firmware_update_repo_url: str = ""
firmware_auto_update: bool = False
firmware_update_token: str = ""
firmware_update_checked_at: float = 0.0 # throttle bookkeeping, see gitea_releases.UPDATE_CHECK_INTERVAL_S
firmware_gitea_latest_version: str = "" # latest release's version, from its tag name
firmware_update_checked_at: float = 0.0
firmware_gitea_latest_version: str = ""
stats: FrameStats = FrameStats()
def load() -> FrameConfig:
with _lock:
"""Reads the legacy file with the same env-override behavior the old
server applied on every load -- which is exactly how env-configured
IMMICH_URL/IMMICH_API_KEY get baked into the database at migration
time even though they were never written to the file itself."""
if not CONFIG_PATH.exists():
cfg = FrameConfig()
else:
cfg = FrameConfig(**json.loads(CONFIG_PATH.read_text()))
# IMMICH_URL/IMMICH_API_KEY/MANAGEMENT_TOKEN/GITEA_FIRMWARE_TOKEN set in
# the environment (e.g. docker-compose.yml, see
# docker-compose.yml.example) take precedence over whatever's saved in
# CONFIG_PATH, so credentials never need to go through the web UI.
env_url = os.environ.get("IMMICH_URL")
env_key = os.environ.get("IMMICH_API_KEY")
env_token = os.environ.get("MANAGEMENT_TOKEN")
@@ -146,22 +94,3 @@ def load() -> FrameConfig:
cfg.firmware_update_token = env_gitea_token
return cfg
def save(cfg: FrameConfig) -> None:
with _lock:
CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True)
CONFIG_PATH.write_text(cfg.model_dump_json(indent=2))
@contextmanager
def locked() -> Iterator[None]:
"""Serializes an entire load-mutate-save cycle. load()/save() each
only lock their own I/O, which isn't enough by itself: uvicorn
dispatches sync routes to a thread pool, so two concurrent requests
(e.g. the device's own poll landing alongside a web UI edit) can each
load() the same on-disk state and the second save() silently clobber
the first's changes. Route handlers that mutate config should wrap
their whole load/mutate/save span in this."""
with _lock:
yield
server/app/db.py
+88
View File
@@ -0,0 +1,88 @@
"""Engine, sessions, and the per-frame lock that replaces the old
whole-config.json RLock.
Single uvicorn worker (see Dockerfile) -- handlers are sync and run in
the threadpool, so this is ordinary multi-threading in one process: the
same regime the old config.locked() RLock handled, now scoped per frame.
"""
from __future__ import annotations
import os
import threading
from contextlib import contextmanager
from typing import Iterator
from sqlalchemy import create_engine, event
from sqlalchemy.orm import Session, sessionmaker
from .models import Frame
DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:////data/espresso.db")
_is_sqlite = DATABASE_URL.startswith("sqlite")
engine = create_engine(
DATABASE_URL,
connect_args={"check_same_thread": False} if _is_sqlite else {},
)
if _is_sqlite:
@event.listens_for(engine, "connect")
def _sqlite_pragmas(dbapi_connection, _record):
cursor = dbapi_connection.cursor()
cursor.execute("PRAGMA journal_mode=WAL")
cursor.execute("PRAGMA foreign_keys=ON")
cursor.execute("PRAGMA busy_timeout=5000")
cursor.close()
# expire_on_commit=False so a Frame resolved by the require_device
# dependency (which commits its last_seen touch) stays usable in the
# route handler without a re-select per attribute. Freshness inside
# mutation spans is handled explicitly by frame_locked()'s refresh.
SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)
def get_db() -> Iterator[Session]:
"""FastAPI dependency: one session per request (FastAPI caches the
dependency, so require_device and the route handler share it)."""
db = SessionLocal()
try:
yield db
finally:
db.close()
# One lock per frame id, created on demand. Guarded by a module lock so
# two threads can't race to create different Lock objects for the same
# frame (which would defeat the whole point).
_frame_locks: dict[int, threading.Lock] = {}
_frame_locks_guard = threading.Lock()
def _get_lock(frame_id: int) -> threading.Lock:
with _frame_locks_guard:
lock = _frame_locks.get(frame_id)
if lock is None:
lock = threading.Lock()
_frame_locks[frame_id] = lock
return lock
@contextmanager
def frame_locked(db: Session, frame_id: int) -> Iterator[Frame]:
"""Serializes a whole read-modify-write span on one frame -- the
direct successor of the old config.locked(). The refresh() inside the
lock is what makes it correct: without it the session could hold
attribute state read *before* another thread's committed write, and
saving would silently clobber it (the same lost-update race the old
pattern's 're-read inside the lock' comment guarded against)."""
with _get_lock(frame_id):
frame = db.get(Frame, frame_id)
if frame is None:
raise LookupError(f"Frame {frame_id} does not exist")
db.refresh(frame)
yield frame
db.commit()
server/app/firmware.py
+7 -6
View File
@@ -1,7 +1,8 @@
"""Local firmware image storage + esp_app_desc_t parsing. Shared by the
manual upload path (POST /api/firmware) and the Gitea auto-update path
(see gitea_releases.py) -- both end up writing the same firmware.bin slot
that GET /frame/firmware streams to the device."""
"""Per-frame firmware image storage + esp_app_desc_t parsing. Shared by
the manual upload path and the Gitea auto-update path -- both end up
writing the same per-frame slot that GET /frame/firmware streams to the
device. The migration moves the old single /data/firmware.bin into frame
#1's slot."""
from __future__ import annotations
@@ -18,8 +19,8 @@ APP_DESC_MAGIC = 0xABCD5432
EXPECTED_PROJECT_NAME = "espresso_frame"
def firmware_path():
return config.CONFIG_PATH.parent / "firmware.bin"
def firmware_path(frame_id: int):
return config.CONFIG_PATH.parent / "firmware" / f"{frame_id}.bin"
def parse_app_version(data: bytes) -> str:
server/app/main.py
+37 -880
View File
@@ -1,199 +1,36 @@
"""ESPresso Frame server: pulls photos from Immich, pre-processes them for
the panel, and serves the ESP32 a ready-to-display frame."""
"""ESPresso Frame server: pulls photos from Immich, pre-processes them
for the panel, and serves ESP32 frames ready-to-display images.
This module is assembly only -- routes live in app/routers/ (device.py
for the firmware-facing /frame/* protocol, api.py for the web UI's
/api/*), storage in SQLite via models.py/db.py, with migration.py
importing a pre-database config.json deployment on first boot."""
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 import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from PIL import Image
from pydantic import BaseModel
from . import config, gitea_releases, photo_queue
from .face_labels import compute_face_labels
from .firmware import firmware_path, parse_app_version
from .image_pipeline import render_frame
from .immich_client import ImmichClient
from . import migration
from .auth import MANAGEMENT_TOKEN_COOKIE, browser_token_valid, management_token
from .db import SessionLocal
from .quiet_hours import ALL_TIMEZONES
from .routers import api, device
from .routers.common import default_frame, immich_creds
logger = logging.getLogger(__name__)
# Schema + legacy-config import, before the first request is served.
migration.run_migrations()
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 _in_quiet_hours(cfg: config.FrameConfig) -> bool:
"""Whether quiet hours are in effect right now -- separate from
_effective_refresh_interval_s, which only shapes what the *device* is
told to sleep for. This instead gates photo_queue.get_current()'s
time-based advance, since that check runs independent of the device
(also triggered by the web UI's /api/queue, e.g. an open browser tab
polling overnight) and would otherwise happily advance the current
photo mid-quiet-hours on raw elapsed time alone."""
if not cfg.quiet_hours_enabled:
return False
now = datetime.now(_zoneinfo(cfg.timezone))
in_quiet, _ = _quiet_hours_state(now, cfg.quiet_hours_start, cfg.quiet_hours_end)
return in_quiet
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.include_router(device.router)
app.include_router(api.router)
@app.get("/health")
@@ -201,53 +38,33 @@ 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 and board variant (X-Frame-
Version/X-Frame-Board headers -- the latter is how the Gitea
auto-update feature learns which release asset to fetch, instead of
a user picking it in the web UI) 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", "")
reported_board = request.headers.get("X-Frame-Board", "")
with config.locked():
cfg = config.load()
cfg.last_seen = time.time()
if cfg.stats.first_seen == 0:
cfg.stats.first_seen = cfg.last_seen
cfg.stats.device_wakes += 1
if reported_version:
if cfg.device_firmware_version and reported_version != cfg.device_firmware_version:
cfg.stats.ota_updates_applied += 1
cfg.device_firmware_version = reported_version
if reported_board:
cfg.device_board_variant = reported_board
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):
"""The web UI (Phase A: still the single-frame page, bound to the
default frame). Handles the unauthorized case itself with a friendly
token prompt rather than a bare 401, since this is the one route a
human lands on with no token yet."""
if not browser_token_valid(request):
supplied = request.query_params.get("token")
return templates.TemplateResponse(
"token_prompt.html", {"request": request, "wrong": supplied is not None}
)
with SessionLocal() as db:
frame = default_frame(db)
immich_url, _ = immich_creds(frame)
response = templates.TemplateResponse(
"index.html", {"request": request, "cfg": cfg, "timezones": ALL_TIMEZONES}
"index.html",
{
"request": request,
"cfg": frame,
"immich_url": immich_url,
"timezones": ALL_TIMEZONES,
},
)
supplied = request.query_params.get("token")
if cfg.management_token and supplied == cfg.management_token:
if management_token() and supplied == 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.
@@ -255,663 +72,3 @@ def index(request: Request):
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"),
firmware_update_repo_url: str = Form(""),
firmware_auto_update: bool = Form(False),
):
# Immich URL/API key/Gitea token are env-var only (IMMICH_URL/
# IMMICH_API_KEY/GITEA_FIRMWARE_TOKEN, see docker-compose.yml.example)
# -- config.load() already applies them, and this handler doesn't touch
# cfg.immich_url/immich_api_key/firmware_update_token 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
cfg.firmware_update_repo_url = firmware_update_repo_url.strip()
cfg.firmware_auto_update = firmware_auto_update
cfg.stats.config_saves += 1
config.save(cfg)
return {"status": "saved"}
@app.get("/api/stats", dependencies=[Depends(require_access_token)])
def api_stats():
return config.load().stats.model_dump()
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, in_quiet_hours=_in_quiet_hours(cfg)):
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()
cfg.stats.battery_reports += 1
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.stats.recharge_cycles += 1
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"}
@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 _fetch_latest_release(cfg: config.FrameConfig) -> dict | None:
try:
return gitea_releases.fetch_latest_release(cfg.firmware_update_repo_url, cfg.firmware_update_token)
except httpx.HTTPError as e:
raise HTTPException(502, f"Could not reach Gitea at {cfg.firmware_update_repo_url}: {e}") from e
def _apply_gitea_update(cfg: config.FrameConfig) -> str:
"""Downloads the configured Gitea repo's latest release asset for this
frame's board variant and stages it exactly like a manual
POST /api/firmware upload would. The board comes from the device
itself (device_board_variant, learned from its X-Frame-Board header
on GET /frame/config -- see frame_config()), not a user picker, so
there's nothing to fetch until a device has checked in at least
once. Network I/O happens before the lock is taken, matching the
load/mutate/save concurrency pattern used elsewhere (see
config.locked())."""
if not cfg.device_board_variant:
raise HTTPException(400, "No frame has checked in yet -- can't tell which board's build to fetch")
release = _fetch_latest_release(cfg)
if not release:
raise HTTPException(404, "No releases found in the configured Gitea repo")
asset_name = gitea_releases.asset_name_for_board(cfg.device_board_variant)
asset_url = release["assets"].get(asset_name)
if not asset_url:
raise HTTPException(404, f"Latest release has no '{asset_name}' asset")
try:
data = gitea_releases.download_asset(asset_url, cfg.firmware_update_token)
except httpx.HTTPError as e:
raise HTTPException(502, f"Could not download '{asset_name}' from Gitea: {e}") from e
version = parse_app_version(data) # same validation the manual upload path applies
firmware_path().write_bytes(data)
with config.locked():
cfg = config.load()
cfg.firmware_available_version = version
cfg.firmware_gitea_latest_version = version
cfg.firmware_update_checked_at = time.time()
config.save(cfg)
return version
@app.get("/api/firmware/check", dependencies=[Depends(require_access_token)])
def api_firmware_check(force: bool = False):
"""Throttled check of the configured Gitea repo's latest release
(gitea_releases.UPDATE_CHECK_INTERVAL_S) -- cheap, since it only reads
the release's tag name, not its binaries. If firmware_auto_update is
on and a newer version is found, applies it immediately; otherwise
just reports it so the web UI can offer the "Update frame" button.
Applying (auto or manual) needs to know the frame's board, which is
learned from the device's own X-Frame-Board header rather than
picked by the user -- update_available stays false until a device
has checked in at least once, regardless of what Gitea has.
force=true (the "Check now" button) bypasses the throttle and always
hits Gitea -- otherwise a genuinely new release can sit invisible in
the UI for up to the full throttle interval even though it's already
live, since the passive poll won't look again until then."""
cfg = config.load()
if not cfg.firmware_update_repo_url:
return {"enabled": False}
now = time.time()
if force or now - cfg.firmware_update_checked_at >= gitea_releases.UPDATE_CHECK_INTERVAL_S:
# Deliberately not updated on failure (see below) -- checked_at only
# advances on a successful reach, so a Gitea outage gets retried
# every poll instead of waiting out the full throttle interval.
release = _fetch_latest_release(cfg)
with config.locked():
cfg = config.load() # re-read: state may have changed since the unlocked read above
cfg.firmware_update_checked_at = now
if release:
cfg.firmware_gitea_latest_version = release["version"]
config.save(cfg)
cfg = config.load()
update_available = (
bool(cfg.firmware_gitea_latest_version)
and cfg.firmware_gitea_latest_version != cfg.firmware_available_version
and bool(cfg.device_board_variant)
)
if update_available and cfg.firmware_auto_update:
_apply_gitea_update(cfg)
cfg = config.load()
update_available = False
return {
"enabled": True,
"board": cfg.device_board_variant or None,
"latest_version": cfg.firmware_gitea_latest_version or None,
"staged_version": cfg.firmware_available_version or None,
"update_available": update_available,
}
@app.post("/api/firmware/apply-latest", dependencies=[Depends(require_access_token)])
def api_firmware_apply_latest():
"""The "Update frame" button: applies the latest Gitea release right
now, bypassing the check throttle -- this is an explicit user action,
not a background poll."""
cfg = config.load()
if not cfg.firmware_update_repo_url:
raise HTTPException(400, "No Gitea firmware repo configured")
version = _apply_gitea_update(cfg)
return {"status": "saved", "version": version}
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, in_quiet_hours=_in_quiet_hours(cfg)):
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, in_quiet_hours=_in_quiet_hours(cfg)):
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, in_quiet_hours=_in_quiet_hours(cfg))
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)
server/app/migration.py
+146
View File
@@ -0,0 +1,146 @@
"""Schema versioning + one-time import of a legacy config.json deployment.
Hand-rolled on purpose (vs alembic): single worker, single SQLite file,
~30 lines of runner. Each migration is (version, fn(connection)); v1 is
just create_all. DDL stays dialect-neutral so a future move to Postgres
is a DATABASE_URL change, not a rewrite.
Run at import time from main.py, before any request is served.
"""
from __future__ import annotations
import logging
import secrets
import shutil
import time
from sqlalchemy import select, text
from . import config
from .db import SessionLocal, engine
from .models import Base, BatteryLog, Frame
logger = logging.getLogger(__name__)
def _migration_1(conn) -> None:
Base.metadata.create_all(bind=conn)
MIGRATIONS = [
(1, _migration_1),
]
def run_migrations() -> None:
with engine.begin() as conn:
conn.execute(text("CREATE TABLE IF NOT EXISTS schema_version (version INTEGER NOT NULL)"))
row = conn.execute(text("SELECT version FROM schema_version")).fetchone()
current = row[0] if row else 0
for version, fn in MIGRATIONS:
if version > current:
logger.info("Applying schema migration %d", version)
fn(conn)
if row is None:
conn.execute(
text("INSERT INTO schema_version (version) VALUES (:v)"), {"v": version}
)
row = (version,)
else:
conn.execute(text("UPDATE schema_version SET version = :v"), {"v": version})
_ensure_frame_one()
def new_device_token() -> str:
return secrets.token_urlsafe(32)
def new_manage_token() -> str:
return secrets.token_urlsafe(16)
def _ensure_frame_one() -> None:
"""First boot only (frames table empty): create frame #1 -- imported
verbatim from a legacy config.json if one exists, otherwise fresh
defaults. Either way it's the legacy-token frame: the deployed
firmware sends no device id and (at most) the shared MANAGEMENT_TOKEN,
and require_device resolves those requests here. The frames-nonempty
guard makes this idempotent; config.json is left untouched as the
rollback path."""
with SessionLocal() as db:
if db.scalars(select(Frame).limit(1)).first() is not None:
return
cfg = config.load() # all defaults if the file doesn't exist
had_file = config.CONFIG_PATH.exists()
frame = Frame(
name="Frame 1",
device_id=None,
device_token=new_device_token(),
manage_token=new_manage_token(),
legacy_token_enabled=True,
created_at=time.time(),
immich_url=cfg.immich_url,
immich_api_key=cfg.immich_api_key,
album_id=cfg.album_id,
order=cfg.order,
refresh_interval_s=cfg.refresh_interval_s,
quiet_hours_enabled=cfg.quiet_hours_enabled,
quiet_hours_start=cfg.quiet_hours_start,
quiet_hours_end=cfg.quiet_hours_end,
timezone=cfg.timezone,
smart_crop_faces=cfg.smart_crop_faces,
orientation=cfg.orientation,
queue_target_len=cfg.queue_target_len,
current_asset_id=cfg.current_asset_id,
current_asset_set_at=cfg.current_asset_set_at,
queue=list(cfg.queue),
queue_cursor=cfg.queue_cursor,
history=list(cfg.history),
excluded_asset_ids=list(cfg.excluded_asset_ids),
battery_percent=cfg.battery_percent,
battery_as_of=cfg.battery_as_of,
battery_history=[list(pair) for pair in cfg.battery_history],
last_seen=cfg.last_seen,
device_firmware_version=cfg.device_firmware_version,
device_board_variant=cfg.device_board_variant,
firmware_available_version=cfg.firmware_available_version,
firmware_update_repo_url=cfg.firmware_update_repo_url,
firmware_auto_update=cfg.firmware_auto_update,
firmware_update_token=cfg.firmware_update_token,
firmware_update_checked_at=cfg.firmware_update_checked_at,
firmware_gitea_latest_version=cfg.firmware_gitea_latest_version,
stats_first_seen=cfg.stats.first_seen,
stats_device_wakes=cfg.stats.device_wakes,
stats_photos_displayed=cfg.stats.photos_displayed,
stats_photos_removed=cfg.stats.photos_removed,
stats_battery_reports=cfg.stats.battery_reports,
stats_recharge_cycles=cfg.stats.recharge_cycles,
stats_ota_updates_applied=cfg.stats.ota_updates_applied,
stats_config_saves=cfg.stats.config_saves,
)
db.add(frame)
db.flush() # assign frame.id for the battery log rows
for pair in cfg.battery_log:
db.add(BatteryLog(frame_id=frame.id, ts=pair[0], percent=pair[1]))
db.commit()
# The single legacy firmware slot becomes frame #1's per-frame slot.
legacy_bin = config.CONFIG_PATH.parent / "firmware.bin"
if legacy_bin.exists():
per_frame_dir = config.CONFIG_PATH.parent / "firmware"
per_frame_dir.mkdir(parents=True, exist_ok=True)
shutil.copy2(legacy_bin, per_frame_dir / f"{frame.id}.bin")
if had_file:
logger.info(
"Imported legacy config.json as frame #%d (%d battery log entries)",
frame.id,
len(cfg.battery_log),
)
else:
logger.info("Fresh install: created default frame #%d", frame.id)
server/app/models.py
+205
View File
@@ -0,0 +1,205 @@
"""SQLAlchemy models: users, sessions, frames, links, claims, battery log.
One deliberately WIDE `frames` row per frame (settings + state + telemetry
+ stats together): every device request touches exactly one row, so the
per-frame lock in db.frame_locked() keeps the old whole-config-lock
semantics trivially correct, and SQLite doesn't care about row width.
The queue/history/excluded/battery_history columns are MutableList-mapped
JSON: photo_queue.py mutates them in place (pop/append/insert), which a
plain JSON column would silently not persist -- MutableList marks the row
dirty on in-place changes.
The ORM attribute for the photo ordering setting is `order` (matching the
old FrameConfig field name so photo_queue.py ports unchanged) but the
column is named photo_order to stay clear of the SQL keyword.
"""
from __future__ import annotations
import time
from sqlalchemy import JSON, Boolean, Float, ForeignKey, Index, Integer, String
from sqlalchemy.ext.mutable import MutableList
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
# Normalized to lowercase in code before insert/lookup -- portable
# case-insensitive uniqueness without SQLite-only COLLATE NOCASE.
username: Mapped[str] = mapped_column(String, unique=True)
display_name: Mapped[str] = mapped_column(String, default="")
# Pluggable identity: "local" now; an OIDC provider later would set
# provider_subject and leave password_hash NULL.
identity_provider: Mapped[str] = mapped_column(String, default="local")
provider_subject: Mapped[str] = mapped_column(String, default="")
password_hash: Mapped[str | None] = mapped_column(String, nullable=True)
is_admin: Mapped[bool] = mapped_column(Boolean, default=False)
immich_url: Mapped[str] = mapped_column(String, default="")
immich_api_key: Mapped[str] = mapped_column(String, default="")
created_at: Mapped[float] = mapped_column(Float, default=time.time)
__table_args__ = (
Index(
"ix_users_provider_subject",
"identity_provider",
"provider_subject",
unique=True,
sqlite_where=provider_subject != "",
),
)
class UserSession(Base):
__tablename__ = "sessions"
id: Mapped[int] = mapped_column(primary_key=True)
token_hash: Mapped[str] = mapped_column(String, unique=True) # sha256 hex of cookie value
user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"))
csrf_token: Mapped[str] = mapped_column(String)
created_at: Mapped[float] = mapped_column(Float, default=time.time)
expires_at: Mapped[float] = mapped_column(Float, index=True)
user: Mapped[User] = relationship()
class Frame(Base):
__tablename__ = "frames"
id: Mapped[int] = mapped_column(primary_key=True)
# 12 lowercase hex chars of the device's full STA MAC. NULL only for
# the migrated legacy frame until its device first reports an id.
device_id: Mapped[str | None] = mapped_column(String, unique=True, nullable=True)
name: Mapped[str] = mapped_column(String, default="")
# Renderer dispatch seam for future calendar/canva modes -- only
# "photos" is registered today (see routers/device.py RENDERERS).
mode: Mapped[str] = mapped_column(String, default="photos")
# Whose Immich library this frame pulls from; NULL = unclaimed.
owner_user_id: Mapped[int | None] = mapped_column(
ForeignKey("users.id", ondelete="SET NULL"), nullable=True
)
# "Take control" soft lock -- only this user may mutate settings/queue.
controlled_by_user_id: Mapped[int | None] = mapped_column(
ForeignKey("users.id", ondelete="SET NULL"), nullable=True
)
device_token: Mapped[str] = mapped_column(String)
# Device has authenticated with device_token at least once -- stop
# pushing it in /frame/config responses.
device_token_ack: Mapped[bool] = mapped_column(Boolean, default=False)
manage_token: Mapped[str] = mapped_column(String, unique=True)
# Migration window: this frame also accepts the legacy shared
# MANAGEMENT_TOKEN (and no-id requests resolve to it). Only ever the
# migrated frame #1; cleared from /admin once the device is on
# per-frame auth.
legacy_token_enabled: Mapped[bool] = mapped_column(Boolean, default=False)
claimed_at: Mapped[float | None] = mapped_column(Float, nullable=True)
created_at: Mapped[float] = mapped_column(Float, default=time.time)
# Migration staging only: Immich creds imported from the legacy
# config.json/env live here until /setup copies them to admin #1.
# Runtime resolution prefers owner creds, then env, then these (see
# routers/common.py immich_creds()).
immich_url: Mapped[str] = mapped_column(String, default="")
immich_api_key: Mapped[str] = mapped_column(String, default="")
# -- settings (attribute names match the old FrameConfig fields) --
album_id: Mapped[str] = mapped_column(String, default="")
order: Mapped[str] = mapped_column("photo_order", String, default="sequential")
refresh_interval_s: Mapped[int] = mapped_column(Integer, default=3600)
quiet_hours_enabled: Mapped[bool] = mapped_column(Boolean, default=False)
quiet_hours_start: Mapped[str] = mapped_column(String, default="22:00")
quiet_hours_end: Mapped[str] = mapped_column(String, default="07:00")
timezone: Mapped[str] = mapped_column(String, default="UTC")
smart_crop_faces: Mapped[bool] = mapped_column(Boolean, default=True)
orientation: Mapped[str] = mapped_column(String, default="landscape")
queue_target_len: Mapped[int] = mapped_column(Integer, default=20)
# -- state --
current_asset_id: Mapped[str] = mapped_column(String, default="")
current_asset_set_at: Mapped[float] = mapped_column(Float, default=0.0)
queue: Mapped[list] = mapped_column(MutableList.as_mutable(JSON), default=list)
queue_cursor: Mapped[int] = mapped_column(Integer, default=0)
history: Mapped[list] = mapped_column(MutableList.as_mutable(JSON), default=list)
excluded_asset_ids: Mapped[list] = mapped_column(MutableList.as_mutable(JSON), default=list)
# -- telemetry --
battery_percent: Mapped[int] = mapped_column(Integer, default=-1)
battery_as_of: Mapped[float] = mapped_column(Float, default=0.0)
# Current discharge cycle only (reset on recharge detection); the
# permanent record is the battery_log table.
battery_history: Mapped[list] = mapped_column(MutableList.as_mutable(JSON), default=list)
last_seen: Mapped[float] = mapped_column(Float, default=0.0)
device_firmware_version: Mapped[str] = mapped_column(String, default="")
device_board_variant: Mapped[str] = mapped_column(String, default="")
# -- firmware / OTA (per frame; image lives at /data/firmware/<id>.bin) --
firmware_available_version: Mapped[str] = mapped_column(String, default="")
firmware_update_repo_url: Mapped[str] = mapped_column(String, default="")
firmware_auto_update: Mapped[bool] = mapped_column(Boolean, default=False)
firmware_update_token: Mapped[str] = mapped_column(String, default="")
firmware_update_checked_at: Mapped[float] = mapped_column(Float, default=0.0)
firmware_gitea_latest_version: Mapped[str] = mapped_column(String, default="")
# -- stats (flattened from the old nested FrameStats) --
stats_first_seen: Mapped[float] = mapped_column(Float, default=0.0)
stats_device_wakes: Mapped[int] = mapped_column(Integer, default=0)
stats_photos_displayed: Mapped[int] = mapped_column(Integer, default=0)
stats_photos_removed: Mapped[int] = mapped_column(Integer, default=0)
stats_battery_reports: Mapped[int] = mapped_column(Integer, default=0)
stats_recharge_cycles: Mapped[int] = mapped_column(Integer, default=0)
stats_ota_updates_applied: Mapped[int] = mapped_column(Integer, default=0)
stats_config_saves: Mapped[int] = mapped_column(Integer, default=0)
owner: Mapped[User | None] = relationship(foreign_keys=[owner_user_id])
controlled_by: Mapped[User | None] = relationship(foreign_keys=[controlled_by_user_id])
class UserFrame(Base):
"""A user linked to a frame: sees it in their sidebar, may view its
pages, and may take control. Ownership (whose Immich creds the frame
renders from) is frames.owner_user_id, separate from linking."""
__tablename__ = "user_frames"
user_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), primary_key=True
)
frame_id: Mapped[int] = mapped_column(
ForeignKey("frames.id", ondelete="CASCADE"), primary_key=True
)
created_at: Mapped[float] = mapped_column(Float, default=time.time)
class PendingClaim(Base):
"""A claim submitted before the frame's first check-in (the user beat
the device to the server after provisioning). Attached automatically
when a device with this id self-registers; expired rows are pruned
opportunistically."""
__tablename__ = "pending_claims"
device_id: Mapped[str] = mapped_column(String, primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"))
created_at: Mapped[float] = mapped_column(Float, default=time.time)
expires_at: Mapped[float] = mapped_column(Float)
class BatteryLog(Base):
"""Every battery report ever, per frame -- the permanent record behind
the battery history chart (was a 20k-entry JSON array in config.json)."""
__tablename__ = "battery_log"
id: Mapped[int] = mapped_column(primary_key=True)
frame_id: Mapped[int] = mapped_column(ForeignKey("frames.id", ondelete="CASCADE"))
ts: Mapped[float] = mapped_column(Float)
percent: Mapped[int] = mapped_column(Integer)
__table_args__ = (Index("ix_battery_log_frame_ts", "frame_id", "ts"),)
server/app/photo_queue.py
+11 -11
View File
@@ -34,12 +34,12 @@ from __future__ import annotations
import random
import time
from .config import FrameConfig
from .models import Frame
HISTORY_MAX_LEN = 20
def _top_up(cfg: FrameConfig, assets: list[dict]) -> None:
def _top_up(cfg: Frame, assets: list[dict]) -> None:
valid_ids = {a["id"] for a in assets}
excluded_ids = set(cfg.excluded_asset_ids)
cfg.queue = [asset_id for asset_id in cfg.queue if asset_id in valid_ids and asset_id not in excluded_ids]
@@ -85,7 +85,7 @@ def _top_up(cfg: FrameConfig, assets: list[dict]) -> None:
cfg.queue_cursor = (cfg.queue_cursor + i + 1) % n
def advance_forced(cfg: FrameConfig, assets: list[dict]) -> None:
def advance_forced(cfg: Frame, assets: list[dict]) -> None:
"""Unconditionally moves to the next photo, ignoring elapsed time, and
resets the interval clock from now. Used by the explicit next-photo
action (POST /frame/advance) and by get_current() once the refresh
@@ -105,7 +105,7 @@ def advance_forced(cfg: FrameConfig, assets: list[dict]) -> None:
# only asset is already current) -- keep showing what we have.
cfg.current_asset_id = assets[0]["id"]
cfg.current_asset_set_at = time.time()
cfg.stats.photos_displayed += 1
cfg.stats_photos_displayed += 1
# Refill back up to queue_target_len now that current_asset_id has
# changed -- otherwise the queue is left one short until the *next*
# advance, since the pop above consumes one of the items _top_up just
@@ -113,7 +113,7 @@ def advance_forced(cfg: FrameConfig, assets: list[dict]) -> None:
_top_up(cfg, assets)
def back_forced(cfg: FrameConfig, assets: list[dict]) -> bool:
def back_forced(cfg: Frame, assets: list[dict]) -> bool:
"""Unconditionally moves to the previously-current photo, the mirror
image of advance_forced() -- pops the most recent entry off history,
pushes the photo it's replacing onto the front of queue (so pressing
@@ -132,12 +132,12 @@ def back_forced(cfg: FrameConfig, assets: list[dict]) -> bool:
cfg.queue.insert(0, cfg.current_asset_id)
cfg.current_asset_id = previous_id
cfg.current_asset_set_at = time.time()
cfg.stats.photos_displayed += 1
cfg.stats_photos_displayed += 1
return True
return False
def remove_from_rotation(cfg: FrameConfig, assets: list[dict], asset_id: str) -> bool:
def remove_from_rotation(cfg: Frame, assets: list[dict], asset_id: str) -> bool:
"""Permanently excludes asset_id from this frame's rotation (see the
module docstring) -- doesn't touch Immich, just this frame's own
selection. Scrubs it out of queue and history too, so it can't
@@ -149,7 +149,7 @@ def remove_from_rotation(cfg: FrameConfig, assets: list[dict], asset_id: str) ->
changed as a result."""
if asset_id not in cfg.excluded_asset_ids:
cfg.excluded_asset_ids.append(asset_id)
cfg.stats.photos_removed += 1
cfg.stats_photos_removed += 1
cfg.queue = [a for a in cfg.queue if a != asset_id]
cfg.history = [a for a in cfg.history if a != asset_id]
@@ -167,12 +167,12 @@ def remove_from_rotation(cfg: FrameConfig, assets: list[dict], asset_id: str) ->
remaining = [a["id"] for a in assets if a["id"] not in excluded_ids]
cfg.current_asset_id = remaining[0] if remaining else ""
cfg.current_asset_set_at = time.time()
cfg.stats.photos_displayed += 1
cfg.stats_photos_displayed += 1
_top_up(cfg, assets)
return True
def sync_queue_length(cfg: FrameConfig, assets: list[dict]) -> None:
def sync_queue_length(cfg: Frame, assets: list[dict]) -> None:
"""Tops up or trims cfg.queue to match cfg.queue_target_len without
otherwise touching current_asset_id. Used by GET /api/queue so a
change to the "upcoming photos to show" setting takes effect on page
@@ -180,7 +180,7 @@ def sync_queue_length(cfg: FrameConfig, assets: list[dict]) -> None:
_top_up(cfg, assets)
def get_current(cfg: FrameConfig, assets: list[dict], in_quiet_hours: bool = False) -> bool:
def get_current(cfg: Frame, assets: list[dict], in_quiet_hours: bool = False) -> bool:
"""Time-based, idempotent path used by GET /frame/image. Advances only
if the current photo is unset/invalid or refresh_interval_s has
elapsed since it was set. Returns whether it changed anything, so the
server/app/quiet_hours.py
+117
View File
@@ -0,0 +1,117 @@
"""Quiet-hours math, extracted verbatim from the old main.py. Everything
takes the frame-like object duck-typed on quiet_hours_enabled/start/end,
timezone, and refresh_interval_s -- both the old FrameConfig and the
Frame ORM model satisfy it."""
from __future__ import annotations
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo, available_timezones
# 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.
ALL_TIMEZONES = sorted(available_timezones())
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;
the config-save route validates against ALL_TIMEZONES before saving,
so this only matters for state hand-edited or written by an older
version of this code."""
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) -> 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 in_quiet_hours(cfg) -> bool:
"""Whether quiet hours are in effect right now -- separate from
effective_refresh_interval_s, which only shapes what the *device* is
told to sleep for. This instead gates photo_queue.get_current()'s
time-based advance, since that check runs independent of the device
(also triggered by the web UI's queue endpoint, e.g. an open browser
tab polling overnight) and would otherwise happily advance the
current photo mid-quiet-hours on raw elapsed time alone."""
if not cfg.quiet_hours_enabled:
return False
now = datetime.now(_zoneinfo(cfg.timezone))
in_quiet, _ = _quiet_hours_state(now, cfg.quiet_hours_start, cfg.quiet_hours_end)
return in_quiet
def max_expected_gap_s(cfg) -> 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 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
server/app/routers/__init__.py
View File
server/app/routers/api.py
+363
View File
@@ -0,0 +1,363 @@
"""Browser-facing /api/* routes -- Phase A keeps the old single-frame
paths, resolved to the default frame (frame #1), behind the legacy
shared-token gate. Phase B moves auth to sessions; Phase D moves paths
to /api/frames/{id}/... together with the new multi-frame UI."""
from __future__ import annotations
import logging
import time
import httpx
from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile
from fastapi.responses import Response
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.orm import Session
from .. import gitea_releases, photo_queue, quiet_hours
from ..auth import require_access_token
from ..db import frame_locked, get_db
from ..firmware import firmware_path, parse_app_version
from ..models import BatteryLog, Frame
from .common import (
OVERDUE_FACTOR,
battery_estimate_s,
default_frame,
immich_client_for,
immich_creds,
list_assets,
require_configured,
)
logger = logging.getLogger(__name__)
router = APIRouter(dependencies=[Depends(require_access_token)])
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")
@router.get("/api/albums")
def api_albums(db: Session = Depends(get_db)):
frame = default_frame(db)
url, key = immich_creds(frame)
if not url or not key:
raise HTTPException(400, "Immich URL/API key not configured yet")
try:
albums = immich_client_for(frame).list_albums()
except httpx.HTTPError as e:
raise HTTPException(502, f"Could not reach Immich at {url}: {e}") from e
return [{"id": a["id"], "name": a["albumName"], "count": a.get("assetCount", 0)} for a in albums]
@router.post("/api/config")
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"),
firmware_update_repo_url: str = Form(""),
firmware_auto_update: bool = Form(False),
db: Session = Depends(get_db),
):
# Immich creds are per-user (Phase B) / env-fallback -- this handler
# deliberately never touches them, same as the old env-only rule.
frame = default_frame(db)
with frame_locked(db, frame.id) as cfg:
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 quiet_hours.valid_hhmm(quiet_hours_start):
cfg.quiet_hours_start = quiet_hours_start
if quiet_hours.valid_hhmm(quiet_hours_end):
cfg.quiet_hours_end = quiet_hours_end
if timezone in quiet_hours.ALL_TIMEZONES:
cfg.timezone = timezone
cfg.firmware_update_repo_url = firmware_update_repo_url.strip()
cfg.firmware_auto_update = firmware_auto_update
cfg.stats_config_saves += 1
return {"status": "saved"}
@router.get("/api/stats")
def api_stats(db: Session = Depends(get_db)):
frame = default_frame(db)
return {
"first_seen": frame.stats_first_seen,
"device_wakes": frame.stats_device_wakes,
"photos_displayed": frame.stats_photos_displayed,
"photos_removed": frame.stats_photos_removed,
"battery_reports": frame.stats_battery_reports,
"recharge_cycles": frame.stats_recharge_cycles,
"ota_updates_applied": frame.stats_ota_updates_applied,
"config_saves": frame.stats_config_saves,
}
@router.get("/api/queue")
def api_queue(db: Session = Depends(get_db)):
frame = default_frame(db)
require_configured(frame)
client = immich_client_for(frame)
assets = list_assets(client, frame)
with frame_locked(db, frame.id) as cfg:
photo_queue.get_current(cfg, assets, in_quiet_hours=quiet_hours.in_quiet_hours(cfg))
photo_queue.sync_queue_length(cfg, assets)
snapshot = {
"current_asset_id": cfg.current_asset_id,
"queue": list(cfg.queue),
"last_seen": cfg.last_seen,
"overdue_gap": quiet_hours.max_expected_gap_s(cfg) * OVERDUE_FACTOR,
"firmware_version": cfg.device_firmware_version,
"firmware_available": cfg.firmware_available_version,
"battery_percent": cfg.battery_percent,
"battery_as_of": cfg.battery_as_of,
"on_battery_since": cfg.battery_history[0][0] if cfg.battery_history else None,
"battery_estimate_s": battery_estimate_s(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(snapshot["current_asset_id"]) if snapshot["current_asset_id"] else None,
"upcoming": [entry(asset_id) for asset_id in snapshot["queue"]],
"device": {
"last_seen": snapshot["last_seen"] or None,
"overdue": bool(
snapshot["last_seen"] and now - snapshot["last_seen"] > snapshot["overdue_gap"]
),
"firmware_version": snapshot["firmware_version"] or None,
"firmware_available": snapshot["firmware_available"] or None,
"battery": (
{"percent": snapshot["battery_percent"], "as_of": snapshot["battery_as_of"]}
if snapshot["battery_percent"] >= 0
else None
),
"on_battery_since": snapshot["on_battery_since"],
"battery_estimate_s": snapshot["battery_estimate_s"],
},
}
@router.get("/api/battery-log")
def api_battery_log(db: Session = Depends(get_db)):
frame = default_frame(db)
rows = db.execute(
select(BatteryLog.ts, BatteryLog.percent)
.where(BatteryLog.frame_id == frame.id)
.order_by(BatteryLog.ts)
).all()
return {"log": [[ts, percent] for ts, percent in rows]}
class QueueReorderRequest(BaseModel):
queue: list[str]
@router.post("/api/queue/reorder")
def api_queue_reorder(body: QueueReorderRequest, db: Session = Depends(get_db)):
"""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."""
frame = default_frame(db)
with frame_locked(db, frame.id) as cfg:
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
return {"status": "saved"}
class QueuePromoteRequest(BaseModel):
asset_id: str
@router.post("/api/queue/promote")
def api_queue_promote(body: QueuePromoteRequest, db: Session = Depends(get_db)):
"""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."""
frame = default_frame(db)
with frame_locked(db, frame.id) as cfg:
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]
return {"status": "saved"}
class QueueRemoveRequest(BaseModel):
asset_id: str
@router.post("/api/queue/remove")
def api_queue_remove(body: QueueRemoveRequest, db: Session = Depends(get_db)):
"""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()."""
frame = default_frame(db)
require_configured(frame)
client = immich_client_for(frame)
assets = list_assets(client, frame)
with frame_locked(db, frame.id) as cfg:
photo_queue.remove_from_rotation(cfg, assets, body.asset_id)
return {"status": "removed"}
@router.get("/api/photo-thumbnail/{asset_id}")
def api_photo_thumbnail(asset_id: str, db: Session = Depends(get_db)):
frame = default_frame(db)
require_configured(frame)
client = immich_client_for(frame)
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)
@router.post("/api/firmware")
def api_firmware_upload(file: UploadFile = File(...), db: Session = Depends(get_db)):
"""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."""
frame = default_frame(db)
data = file.file.read()
version = parse_app_version(data)
path = firmware_path(frame.id)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(data)
with frame_locked(db, frame.id) as cfg:
cfg.firmware_available_version = version
return {"status": "saved", "version": version, "size": len(data)}
def _fetch_latest_release(frame: Frame) -> dict | None:
try:
return gitea_releases.fetch_latest_release(
frame.firmware_update_repo_url, frame.firmware_update_token
)
except httpx.HTTPError as e:
raise HTTPException(502, f"Could not reach Gitea at {frame.firmware_update_repo_url}: {e}") from e
def _apply_gitea_update(db: Session, frame: Frame) -> str:
"""Downloads the configured Gitea repo's latest release asset for this
frame's board variant and stages it exactly like a manual upload
would. The board comes from the device itself (device_board_variant,
learned from its X-Frame-Board header), not a user picker, so
there's nothing to fetch until the device has checked in at least
once. Network I/O happens before the lock is taken."""
if not frame.device_board_variant:
raise HTTPException(400, "The frame hasn't checked in yet -- can't tell which board's build to fetch")
release = _fetch_latest_release(frame)
if not release:
raise HTTPException(404, "No releases found in the configured Gitea repo")
asset_name = gitea_releases.asset_name_for_board(frame.device_board_variant)
asset_url = release["assets"].get(asset_name)
if not asset_url:
raise HTTPException(404, f"Latest release has no '{asset_name}' asset")
try:
data = gitea_releases.download_asset(asset_url, frame.firmware_update_token)
except httpx.HTTPError as e:
raise HTTPException(502, f"Could not download '{asset_name}' from Gitea: {e}") from e
version = parse_app_version(data) # same validation the manual upload path applies
path = firmware_path(frame.id)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(data)
with frame_locked(db, frame.id) as cfg:
cfg.firmware_available_version = version
cfg.firmware_gitea_latest_version = version
cfg.firmware_update_checked_at = time.time()
return version
@router.get("/api/firmware/check")
def api_firmware_check(force: bool = False, db: Session = Depends(get_db)):
"""Throttled check of the configured Gitea repo's latest release
(gitea_releases.UPDATE_CHECK_INTERVAL_S) -- cheap, since it only reads
the release's tag name, not its binaries. If firmware_auto_update is
on and a newer version is found, applies it immediately; otherwise
just reports it so the web UI can offer the "Update frame" button.
force=true (the "Check now" button) bypasses the throttle and always
hits Gitea -- otherwise a genuinely new release can sit invisible in
the UI for up to the full throttle interval."""
frame = default_frame(db)
if not frame.firmware_update_repo_url:
return {"enabled": False}
now = time.time()
if force or now - frame.firmware_update_checked_at >= gitea_releases.UPDATE_CHECK_INTERVAL_S:
# checked_at only advances on a successful reach, so a Gitea
# outage gets retried every poll instead of waiting out the full
# throttle interval.
release = _fetch_latest_release(frame)
with frame_locked(db, frame.id) as cfg:
cfg.firmware_update_checked_at = now
if release:
cfg.firmware_gitea_latest_version = release["version"]
update_available = (
bool(frame.firmware_gitea_latest_version)
and frame.firmware_gitea_latest_version != frame.firmware_available_version
and bool(frame.device_board_variant)
)
if update_available and frame.firmware_auto_update:
_apply_gitea_update(db, frame)
update_available = False
return {
"enabled": True,
"board": frame.device_board_variant or None,
"latest_version": frame.firmware_gitea_latest_version or None,
"staged_version": frame.firmware_available_version or None,
"update_available": update_available,
}
@router.post("/api/firmware/apply-latest")
def api_firmware_apply_latest(db: Session = Depends(get_db)):
"""The "Update frame" button: applies the latest Gitea release right
now, bypassing the check throttle -- this is an explicit user action,
not a background poll."""
frame = default_frame(db)
if not frame.firmware_update_repo_url:
raise HTTPException(400, "No Gitea firmware repo configured")
version = _apply_gitea_update(db, frame)
return {"status": "saved", "version": version}
server/app/routers/common.py
+120
View File
@@ -0,0 +1,120 @@
"""Helpers shared by the device and browser routers."""
from __future__ import annotations
import io
import logging
import os
import httpx
from fastapi import HTTPException
from PIL import Image
from sqlalchemy import select
from sqlalchemy.orm import Session
from ..image_pipeline import render_frame
from ..immich_client import ImmichClient
from ..models import Frame
logger = logging.getLogger(__name__)
# Battery-history / estimate tuning (see /frame/battery and battery_estimate_s).
BATTERY_HISTORY_MAX = 500 # ~20 days at hourly reports
BATTERY_LOG_MAX = 20000 # ~2 years at hourly reports -- cap on the battery_log table per frame
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 immich_creds(frame: Frame) -> tuple[str, str]:
"""Which Immich this frame renders from. Owner's creds once the frame
is claimed (Phase B+); env vars as the operator-level fallback (the
pre-redesign source of truth); the frame's own staging columns last
(populated by the config.json migration for exactly the case where
the old file held creds but the env no longer does)."""
owner = frame.owner
if owner is not None and owner.immich_url and owner.immich_api_key:
return owner.immich_url, owner.immich_api_key
env_url = os.environ.get("IMMICH_URL", "")
env_key = os.environ.get("IMMICH_API_KEY", "")
if env_url and env_key:
return env_url, env_key
return frame.immich_url, frame.immich_api_key
def immich_client_for(frame: Frame) -> ImmichClient:
url, key = immich_creds(frame)
return ImmichClient(url, key)
def require_configured(frame: Frame) -> None:
url, key = immich_creds(frame)
if not url or not key:
raise HTTPException(400, "Immich URL/API key not configured yet")
if not frame.album_id:
raise HTTPException(400, "No album configured yet")
def list_assets(client: ImmichClient, frame: Frame) -> list[dict]:
try:
assets = client.list_album_assets(frame.album_id)
except httpx.HTTPError as e:
raise HTTPException(502, f"Could not reach Immich: {e}") from e
if not assets:
raise HTTPException(404, "Album has no photos")
return assets
def render_asset(client: ImmichClient, frame: Frame, 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 frame.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=frame.orientation)
def battery_estimate_s(frame: Frame) -> 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 = frame.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)
def default_frame(db: Session) -> Frame:
"""Phase A only: the old single-frame /api/* routes all operate on
"the" frame -- the legacy one if flagged, else the lowest id.
Replaced by explicit /api/frames/{id}/ paths in Phase D."""
frame = db.scalars(
select(Frame).where(Frame.legacy_token_enabled == True) # noqa: E712
).first()
if frame is None:
frame = db.scalars(select(Frame).order_by(Frame.id).limit(1)).first()
if frame is None:
raise HTTPException(404, "No frame exists yet")
return frame
server/app/routers/device.py
+365
View File
@@ -0,0 +1,365 @@
"""Device-facing /frame/* routes. These paths are FROZEN -- they're baked
into deployed firmware -- so multi-frame support changes only how the
calling frame is resolved (see auth.require_device), never the paths or
response key names the deployed flat parser depends on
("refresh_interval_s", "firmware_version")."""
from __future__ import annotations
import logging
import time
from datetime import datetime
import httpx
from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import FileResponse, RedirectResponse, Response
from pydantic import BaseModel
from sqlalchemy import delete, func, select
from sqlalchemy.orm import Session
from .. import photo_queue, quiet_hours
from ..auth import require_device
from ..db import frame_locked, get_db
from ..face_labels import compute_face_labels
from ..firmware import firmware_path
from ..models import BatteryLog, Frame
from .common import (
BATTERY_HISTORY_MAX,
BATTERY_LOG_MAX,
RECHARGE_JUMP_PCT,
immich_client_for,
list_assets,
render_asset,
require_configured,
)
logger = logging.getLogger(__name__)
router = APIRouter()
# Renderer dispatch seam for future frame modes (calendar, canva, ...):
# /frame/image looks up the frame's mode here. Only photos exists today.
def _render_photos_mode(db: Session, frame: Frame) -> bytes:
require_configured(frame)
client = immich_client_for(frame)
assets = list_assets(client, frame)
with frame_locked(db, frame.id) as locked:
photo_queue.get_current(locked, assets, in_quiet_hours=quiet_hours.in_quiet_hours(locked))
asset_id = locked.current_asset_id
return render_asset(client, frame, asset_id)
RENDERERS = {
"photos": _render_photos_mode,
}
@router.get("/frame/config")
def frame_config(request: Request, frame: Frame = Depends(require_device), db: Session = Depends(get_db)):
"""Device-facing settings, polled by the frame alongside its
reachability check. Always returns 200 with current settings -- no
Immich-configured gate, since this doubles as the "is the server up"
signal. Also captures the device's running firmware version and board
variant (X-Frame-Version/X-Frame-Board headers) and advertises the
available OTA image's version, so the device's update check costs
zero extra round trips."""
reported_version = request.headers.get("X-Frame-Version", "")
reported_board = request.headers.get("X-Frame-Board", "")
with frame_locked(db, frame.id) as locked:
if locked.stats_first_seen == 0:
locked.stats_first_seen = time.time()
locked.stats_device_wakes += 1
if reported_version:
if locked.device_firmware_version and reported_version != locked.device_firmware_version:
locked.stats_ota_updates_applied += 1
locked.device_firmware_version = reported_version
if reported_board:
locked.device_board_variant = reported_board
response = {
"refresh_interval_s": quiet_hours.effective_refresh_interval_s(locked),
"firmware_version": locked.firmware_available_version or None,
}
# Per-frame token push: only once the device has introduced itself
# by id (so the response to pure-legacy firmware stays byte-
# compatible with its 256-byte parse buffer), and only until the
# device has authenticated with the token once (device_token_ack).
if locked.device_id is not None and not locked.device_token_ack:
response["device_token"] = locked.device_token
return response
@router.get("/frame/image")
def frame_image(frame: Frame = Depends(require_device), db: Session = Depends(get_db)):
"""Returns the frame's current image. For photos mode: 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."""
renderer = RENDERERS.get(frame.mode, _render_photos_mode)
return Response(content=renderer(db, frame), media_type="application/octet-stream")
@router.post("/frame/advance")
def frame_advance(frame: Frame = Depends(require_device), db: Session = Depends(get_db)):
"""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."""
require_configured(frame)
client = immich_client_for(frame)
assets = list_assets(client, frame)
with frame_locked(db, frame.id) as locked:
photo_queue.advance_forced(locked, assets)
asset_id = locked.current_asset_id
return Response(content=render_asset(client, frame, asset_id), media_type="application/octet-stream")
@router.post("/frame/back")
def frame_back(frame: Frame = Depends(require_device), db: Session = Depends(get_db)):
"""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."""
require_configured(frame)
client = immich_client_for(frame)
assets = list_assets(client, frame)
with frame_locked(db, frame.id) as locked:
photo_queue.back_forced(locked, assets)
asset_id = locked.current_asset_id
return Response(content=render_asset(client, frame, asset_id), media_type="application/octet-stream")
class BatteryReport(BaseModel):
percent: int
@router.post("/frame/battery")
def frame_battery(
body: BatteryReport, frame: Frame = Depends(require_device), db: Session = Depends(get_db)
):
"""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; every report also lands in the
permanent battery_log table behind the history chart."""
if not 0 <= body.percent <= 100:
raise HTTPException(400, "percent must be 0-100")
now = time.time()
with frame_locked(db, frame.id) as locked:
locked.stats_battery_reports += 1
if locked.battery_history and body.percent >= locked.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.
locked.battery_history = []
locked.stats_recharge_cycles += 1
locked.battery_history.append([now, body.percent])
locked.battery_history = locked.battery_history[-BATTERY_HISTORY_MAX:]
locked.battery_percent = body.percent
locked.battery_as_of = now
db.add(BatteryLog(frame_id=locked.id, ts=now, percent=body.percent))
# Safety bound, not a real limit at realistic report rates --
# mirrors the old JSON list's cap.
count = db.scalar(select(func.count()).select_from(BatteryLog).where(BatteryLog.frame_id == locked.id))
if count is not None and count >= BATTERY_LOG_MAX:
cutoff_ids = select(BatteryLog.id).where(BatteryLog.frame_id == locked.id).order_by(
BatteryLog.ts
).limit(count + 1 - BATTERY_LOG_MAX)
db.execute(delete(BatteryLog).where(BatteryLog.id.in_(cutoff_ids)))
return {"status": "saved"}
@router.get("/frame/firmware")
def frame_firmware(frame: Frame = Depends(require_device)):
"""The frame's staged OTA image, streamed to the device
(esp_https_ota). 404 until something has been uploaded/fetched."""
path = firmware_path(frame.id)
if not path.exists():
raise HTTPException(404, "No firmware uploaded")
return FileResponse(path, media_type="application/octet-stream")
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
@router.get("/frame/photo-info")
def frame_photo_info(frame: Frame = Depends(require_device), db: Session = Depends(get_db)):
"""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."""
require_configured(frame)
client = immich_client_for(frame)
assets = list_assets(client, frame)
with frame_locked(db, frame.id) as locked:
photo_queue.get_current(locked, assets, in_quiet_hours=quiet_hours.in_quiet_hours(locked))
asset_id = locked.current_asset_id
if not asset_id:
raise HTTPException(404, "No current photo")
try:
asset = client.get_asset(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": 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),
}
@router.get("/frame/share/{asset_id}")
def frame_share(asset_id: str, frame: Frame = Depends(require_device)):
"""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 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 on THIS frame --
not any arbitrary Immich asset id -- as a second layer even a leaked
token wouldn't bypass."""
require_configured(frame)
if asset_id != frame.current_asset_id and asset_id not in frame.queue:
raise HTTPException(404, "That photo isn't currently showing or queued on this frame")
client = immich_client_for(frame)
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)
@router.get("/frame/face-labels")
def frame_face_labels(frame: Frame = Depends(require_device), db: Session = Depends(get_db)):
"""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, ...) rather than a JSON array, so the device's
hand-rolled parser can read it with the same flat-scalar helpers it
already has. 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."""
require_configured(frame)
client = immich_client_for(frame)
assets = list_assets(client, frame)
with frame_locked(db, frame.id) as locked:
photo_queue.get_current(locked, assets, in_quiet_hours=quiet_hours.in_quiet_hours(locked))
asset_id = locked.current_asset_id
smart_crop = locked.smart_crop_faces
orientation = locked.orientation
if not asset_id:
return {"count": 0}
try:
faces = client.get_asset_faces(asset_id)
except httpx.HTTPError as e:
logger.warning("Could not fetch faces for asset %s: %s", 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(asset_id)
except httpx.HTTPError as e:
logger.warning("Could not download asset %s for face-label mapping: %s", asset_id, e)
return {"count": 0}
labels = compute_face_labels(preview_bytes, faces, smart_crop, 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
server/app/templates/index.html
+2 -2
View File
@@ -5,8 +5,8 @@
{% endblock %}
{% block content %}
{% if cfg.immich_url %}
<div class="info-box">Immich: <code>{{ cfg.immich_url }}</code> (API key configured). Set via
{% if immich_url %}
<div class="info-box">Immich: <code>{{ immich_url }}</code> (API key configured). Set via
<code>IMMICH_URL</code>/<code>IMMICH_API_KEY</code> in <code>docker-compose.yml</code> -- see
<code>docker-compose.yml.example</code>.</div>
{% else %}
server/requirements.txt
+1
View File
@@ -5,3 +5,4 @@ httpx==0.28.1
pillow==12.3.0
python-multipart==0.0.20
jinja2==3.1.5
sqlalchemy==2.0.51