espresso_frame/server/app/routers/common.py

"""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.display_mode == "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,
                        palette_rgb=frame.palette_rgb, display_mode=frame.display_mode)


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 shell_context(request, db: Session, user, active_frame: Frame | None = None,
                  active_nav: str | None = None) -> dict:
    """Template context every app-shell (sidebar) page needs: the user's
    frame list with an online indicator, the active highlights, and the
    session's CSRF token. Import here (not auth) keeps the router
    modules' template plumbing in one place."""
    import time as _time

    from .. import quiet_hours
    from ..auth import current_session, user_frames

    session = current_session(request, db)
    frames = user_frames(db, user)
    now = _time.time()
    for f in frames:
        # Same "not overdue" definition the Device panel uses.
        gap = quiet_hours.max_expected_gap_s(f) * OVERDUE_FACTOR
        f.recently_seen = bool(f.last_seen and now - f.last_seen <= gap)
    return {
        "request": request,
        "user": user,
        "csrf_token": session.csrf_token if session else None,
        "sidebar_frames": frames,
        "active_frame": active_frame,
        "active_nav": active_nav,
    }