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