espresso_frame/server/app/routers/device.py

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

manage=1 is the one addition: appended by firmware's manage button to
whichever of these three GET/POST requests it was already about to make
(see firmware/main/frame_client.c's fetch_and_display -- it no longer
does its own overlay fetching/compositing, that's all server-side now,
see manage_overlay.py and common.build_manage_content)."""

from __future__ import annotations

import logging
import time

from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import FileResponse, Response
from pydantic import BaseModel
from sqlalchemy import delete, func, select
from sqlalchemy.orm import Session

from .. import grid, mail, quiet_hours
from ..auth import get_server_settings, require_device
from ..db import frame_locked, get_db
from ..firmware import firmware_path
from ..image_pipeline import logical_render_size, render_panel, render_placeholder
from ..models import BatteryLog, Frame, FrameButtonAction, Widget
from ..widgets import WIDGET_TYPES
from .common import (
    BATTERY_HISTORY_MAX,
    BATTERY_LOG_MAX,
    RECHARGE_JUMP_PCT,
    RECHARGE_LOOKBACK,
    build_manage_content,
)

logger = logging.getLogger(__name__)

router = APIRouter()


def _setup_placeholder(frame: Frame, request: Request, manage: dict | None = None,
                       as_png: bool = False) -> bytes:
    """What an unclaimed or widget-less frame displays instead of real
    content -- instructions with a QR, rendered at 200 so the device
    treats it as a perfectly normal image and never error-loops. The
    URLs are built from the request's own base URL: whatever address the
    device reached us at is by definition an address that works on this
    network."""
    base = str(request.base_url).rstrip("/")
    if frame.owner_user_id is None and frame.device_id:
        claim_url = f"{base}/claim?device_id={frame.device_id}"
        return render_placeholder(
            ["This frame isn't claimed yet", "Scan to link it to your account:"],
            qr_url=claim_url,
            orientation=frame.orientation,
            palette_rgb=frame.palette_rgb,
            manage=manage,
            as_png=as_png,
        )
    if frame.owner_user_id is None:
        return render_placeholder(
            ["Almost there!", f"Open {base} to finish setting up this frame."],
            orientation=frame.orientation,
            palette_rgb=frame.palette_rgb,
            manage=manage,
            as_png=as_png,
        )
    return render_placeholder(
        ["Almost there!", "Add a widget for this frame at", base],
        qr_url=base,
        orientation=frame.orientation,
        palette_rgb=frame.palette_rgb,
        manage=manage,
        as_png=as_png,
    )


def _render_widgets(db: Session, frame: Frame, manage: dict | None, is_normal_wake: bool,
                    as_png: bool = False) -> bytes:
    """The widget-system compositor: renders every widget on this frame
    into its own region (see app/grid.py for grid-cell -> pixel math) and
    hands the results to image_pipeline.render_panel for the single
    shared paste/enhance/overlay/quantize/pack pass. Replaces the old
    per-mode RENDERERS dict -- a frame can now show several widgets at
    once instead of exactly one mode owning the whole panel."""
    all_widgets = db.scalars(
        select(Widget).where(Widget.frame_id == frame.id).order_by(Widget.sort_order)
    ).all()
    panel_w, panel_h = logical_render_size(frame.orientation)
    regions = []
    for widget in all_widgets:
        module = WIDGET_TYPES.get(widget.widget_type)
        if module is None:
            continue  # unrecognized widget_type -- shouldn't happen, skip defensively rather than 500
        px, py, pw, ph = grid.cell_to_pixels(
            frame.orientation, panel_w, panel_h, (widget.x, widget.y, widget.w, widget.h)
        )
        img = module.render(db, frame, widget, pw, ph, is_normal_wake=is_normal_wake)
        regions.append(((px, py, pw, ph), img))
    return render_panel(
        regions, orientation=frame.orientation, palette_rgb=frame.palette_rgb,
        color_boost=frame.color_boost, contrast_boost=frame.contrast_boost,
        dither_strength=frame.dither_strength, manage=manage, as_png=as_png,
    )


def _render_frame_content(db: Session, frame: Frame, request: Request | None, manage: dict | None,
                          is_normal_wake: bool, as_png: bool = False) -> bytes:
    """The top-level "what does this frame show right now" entry point.
    An unclaimed frame or one with no widgets yet gets the setup
    placeholder (needs `request` for its QR URLs -- only available on the
    normal-wake path where a real request is on hand, never on an
    advance/back button press); otherwise every widget on it gets
    composited via _render_widgets. Individual widgets that are
    themselves unconfigured show their own small placeholder within
    their own region (see app/widgets/*.py) rather than blanking the
    whole panel -- a partially-set-up multi-widget frame still shows
    whatever IS configured."""
    has_widgets = frame.owner_user_id is not None and (
        db.scalars(select(Widget.id).where(Widget.frame_id == frame.id).limit(1)).first() is not None
    )
    if not has_widgets:
        if request is None:
            return render_placeholder(
                ["Almost there!"], orientation=frame.orientation, palette_rgb=frame.palette_rgb,
                manage=manage, as_png=as_png,
            )
        return _setup_placeholder(frame, request, manage=manage, as_png=as_png)

    return _render_widgets(db, frame, manage, is_normal_wake, as_png=as_png)


def render_frame_preview_png(db: Session, frame: Frame, request: Request) -> bytes:
    """The web UI's live "how it's displaying" thumbnail (see
    routers/api_frames.py's /preview endpoint) -- same compositor
    /frame/image uses, just handed back as a small upright PNG instead of
    packed native-panel bytes. Exported from here (rather than
    duplicated) since this module already owns the full widget-
    compositing pipeline; nothing about the /frame/* paths themselves
    changes."""
    return _render_frame_content(db, frame, request, manage=None, is_normal_wake=True, as_png=True)


def _run_button_actions(db: Session, frame: Frame, button: str) -> None:
    """Executes every (widget, action) binding assigned to this physical
    button, in order -- see models.FrameButtonAction and the button-
    assignment UI (a later phase). Each action runs to completion (its
    own widget_locked span) before the next one starts -- never nested,
    since db.widget_locked's underlying lock isn't reentrant (see its own
    docstring) -- a button assigned several actions would deadlock
    instantly if this looped any other way. One action failing
    unexpectedly doesn't block the others, or the eventual re-render,
    from happening -- the user pressed a physical button and expects
    *something* to happen even if one of several assigned widgets is
    having a bad moment."""
    actions = db.scalars(
        select(FrameButtonAction)
        .where(FrameButtonAction.frame_id == frame.id, FrameButtonAction.button == button)
        .order_by(FrameButtonAction.sort_order)
    ).all()
    for action_row in actions:
        widget = db.get(Widget, action_row.widget_id)
        if widget is None:
            continue
        module = WIDGET_TYPES.get(widget.widget_type)
        action_fn = module.ACTIONS.get(action_row.action) if module else None
        if action_fn is None:
            continue
        try:
            action_fn(db, frame, widget)
        except Exception:
            logger.exception(
                "Button action %r failed for widget %d (frame %d)", action_row.action, widget.id, frame.id
            )


@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


def _manage_flag(request: Request) -> bool:
    return request.query_params.get("manage") == "1"


@router.get("/frame/image")
def frame_image(
    request: Request, frame: Frame = Depends(require_device), db: Session = Depends(get_db)
):
    """Returns the frame's current image -- every widget on the frame
    composited into one panel (see _render_widgets). Each widget's own
    render is idempotent in whatever way makes sense for its type (e.g.
    a photo widget only actually advances once its own refresh interval
    has elapsed, see app/photo_queue.py) -- safe to call as often as the
    device wants, including after an unplanned reboot, without skipping
    ahead. An unclaimed frame or one with no widgets yet gets a rendered
    instruction placeholder (200, not an error) so a fresh device never
    error-loops.

    ?manage=1 (the manage button) composites the manage overlay onto
    whatever this would have returned anyway -- see build_manage_content.
    This is also the "normal wake" that resets any calendar widget's
    browse position back to today (see app/widgets/calendar.py)."""
    manage = build_manage_content(db, frame, request) if _manage_flag(request) else None
    content = _render_frame_content(db, frame, request, manage, is_normal_wake=True)
    return Response(content=content, media_type="application/octet-stream")


@router.post("/frame/advance")
def frame_advance(request: Request, frame: Frame = Depends(require_device), db: Session = Depends(get_db)):
    """Forces an immediate move forward on whatever widget(s) the NEXT
    button is assigned to (see models.FrameButtonAction) -- e.g. the next
    photo for a photo widget, or the next day/week/month for a calendar
    widget -- then re-renders and returns the whole panel. Used by the
    device's next-photo button."""
    _run_button_actions(db, frame, "next")
    manage = build_manage_content(db, frame, request) if _manage_flag(request) else None
    content = _render_frame_content(db, frame, request=None, manage=manage, is_normal_wake=False)
    return Response(content=content, media_type="application/octet-stream")


@router.post("/frame/back")
def frame_back(request: Request, frame: Frame = Depends(require_device), db: Session = Depends(get_db)):
    """The mirror of /frame/advance, for whatever widget(s) the BACK
    button is assigned to. A no-op (still 200, unchanged) for any widget
    with nothing to go back to. Used by the device's back-photo button."""
    _run_button_actions(db, frame, "back")
    manage = build_manage_content(db, frame, request) if _manage_flag(request) else None
    content = _render_frame_content(db, frame, request=None, manage=manage, is_normal_wake=False)
    return Response(content=content, 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()
    should_alert = False
    alert_email = ""
    alert_frame_name = ""
    with frame_locked(db, frame.id) as locked:
        locked.stats_battery_reports += 1
        # See RECHARGE_LOOKBACK: compared against the max of the last few
        # reports, not just the single previous one, so a lone noisy dip
        # can't make the next normal reading look like a recharge.
        recent = locked.battery_history[-RECHARGE_LOOKBACK:]
        recent_max = max((pct for _, pct in recent), default=None)
        if recent_max is not None and body.percent >= recent_max + 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 -- and let a
            # low-battery alert fire again next time it actually gets low.
            locked.battery_history = []
            locked.stats_recharge_cycles += 1
            locked.battery_alert_sent = False
        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)))

        # Once per discharge cycle (see the recharge reset above), not
        # once per report -- a frame idling at 4% would otherwise get an
        # email every wake.
        if (
            locked.battery_alert_threshold_pct >= 0
            and body.percent <= locked.battery_alert_threshold_pct
            and not locked.battery_alert_sent
            and locked.owner is not None
            and locked.owner.email
        ):
            should_alert = True
            alert_email = locked.owner.email
            alert_frame_name = locked.name or f"Frame {locked.id}"

    if should_alert:
        # Network I/O outside the lock, same convention as everywhere
        # else in this file -- then a short re-lock to record that it
        # went out, only on actual success (an SMTP hiccup should let
        # the next report's still-below-threshold reading try again
        # rather than silently giving up for the rest of the cycle).
        settings = get_server_settings(db)
        sent = mail.send_email(
            settings, alert_email, f"{alert_frame_name}: battery low",
            f"{alert_frame_name}'s battery is at {body.percent}%.",
        )
        if sent:
            with frame_locked(db, frame.id) as locked:
                locked.battery_alert_sent = True
    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")