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