espresso_frame/server/app/main.py

"""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
from datetime import datetime

import httpx
from fastapi import Depends, FastAPI, HTTPException, Form, Request
from fastapi.responses import 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 = 50

MANAGEMENT_TOKEN_COOKIE = "mgmt_token"


def _token_valid(request: Request, cfg: config.FrameConfig) -> bool:
    """No management_token configured (MANAGEMENT_TOKEN env var, see
    docker-compose.yml.example) means the management page stays open on a
    trusted LAN, matching this project's existing default. Once one's
    set, a request is authorized by either a ?token= query param (what
    the manage-menu QR code embeds) or the cookie index() sets after a
    valid query-param hit (so the page's own fetch()/<img> calls, which
    carry no query string, stay authorized for the rest of the 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_management_token(request: Request) -> None:
    """Dependency for the /api/* routes behind the management page. index()
    below handles the unauthorized case itself (a friendlier HTML prompt,
    not a bare 401) since that's the one route an unauthorized visitor is
    actually meant to land on."""
    if not _token_valid(request, config.load()):
        raise HTTPException(401, "Missing or invalid management token")


@app.get("/health")
def health() -> dict:
    return {"status": "ok"}


@app.get("/frame/config")
def frame_config():
    """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."""
    cfg = config.load()
    return {"refresh_interval_s": cfg.refresh_interval_s}


@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_management_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_management_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),
):
    # 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.
    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.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))
    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)


@app.get("/frame/image")
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."""
    cfg = config.load()
    _require_configured(cfg)

    client = ImmichClient(cfg.immich_url, cfg.immich_api_key)
    assets = _list_assets(client, cfg)

    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")
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."""
    cfg = config.load()
    _require_configured(cfg)

    client = ImmichClient(cfg.immich_url, cfg.immich_api_key)
    assets = _list_assets(client, cfg)

    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")


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")
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."""
    cfg = config.load()
    _require_configured(cfg)

    client = ImmichClient(cfg.immich_url, cfg.immich_api_key)
    assets = _list_assets(client, cfg)

    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}")
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 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.
    Scoped to the photo currently showing or queued -- not any arbitrary
    Immich asset id -- since this is otherwise an unauthenticated
    endpoint (see server/README.md)."""
    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")
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."""
    cfg = config.load()
    _require_configured(cfg)

    client = ImmichClient(cfg.immich_url, cfg.immich_api_key)
    assets = _list_assets(client, cfg)

    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)

    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_management_token)])
def api_queue():
    cfg = config.load()
    _require_configured(cfg)

    client = ImmichClient(cfg.immich_url, cfg.immich_api_key)
    assets = _list_assets(client, cfg)

    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}"}

    return {
        "current": entry(cfg.current_asset_id) if cfg.current_asset_id else None,
        "upcoming": [entry(asset_id) for asset_id in cfg.queue],
    }


class QueueReorderRequest(BaseModel):
    queue: list[str]


@app.post("/api/queue/reorder", dependencies=[Depends(require_management_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."""
    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_management_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."""
    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"}


@app.get("/api/photo-thumbnail/{asset_id}", dependencies=[Depends(require_management_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)