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 time
from datetime import datetime, timedelta
from urllib.parse import urlparse

import httpx
from fastapi import HTTPException
from PIL import Image
from sqlalchemy import select
from sqlalchemy.orm import Session

from .. import calendar_feed, quiet_hours
from ..db import frame_locked
from ..image_pipeline import render_frame
from ..immich_client import ImmichClient
from ..models import Frame, User, UserFrame

logger = logging.getLogger(__name__)

FRAME_MODES = ("photos", "calendar")

# 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 recent baseline = battery was recharged
# How many of the most recent reports make up that baseline. A lone noisy
# reading (ADC/regulator glitch -- see firmware/main/battery.c) can still
# dip or spike a single report; comparing against just the one immediately
# previous report meant that a normal reading right after a noisy dip
# looked like a 5%+ jump and falsely registered as a recharge. Comparing
# against the max of the last few reports instead means an actual recharge
# still needs to clear all of them, while a single stray low one doesn't
# get to set the bar.
RECHARGE_LOOKBACK = 3
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 fetch_source_and_faces(client: ImmichClient, frame: Frame, asset_id: str) -> tuple[Image.Image, list[dict] | None]:
    """The shared first half of rendering: download the Immich preview
    and (only if display_mode needs it) its detected faces. Used by both
    render_asset (device-facing) and the web UI's rendered-preview
    endpoint (routers/api_frames.py) so they can't drift apart."""
    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)

    return Image.open(io.BytesIO(jpeg_bytes)), faces


def render_asset(client: ImmichClient, frame: Frame, asset_id: str, manage: dict | None = None) -> bytes:
    source, faces = fetch_source_and_faces(client, frame, asset_id)
    return render_frame(source, faces=faces, orientation=frame.orientation,
                        palette_rgb=frame.palette_rgb, display_mode=frame.display_mode,
                        color_boost=frame.color_boost, contrast_boost=frame.contrast_boost,
                        dither_strength=frame.dither_strength, manage=manage)


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


def valid_http_url(url: str) -> bool:
    """http(s)-only URL check -- generalized from what was api_frames.py's
    frame-specific _valid_repo_url, now shared by two call sites (the
    Gitea firmware repo URL, and a user's personal calendar ICS URL)."""
    parsed = urlparse(url)
    return parsed.scheme in ("http", "https") and bool(parsed.netloc)


# --- Location/date-taken text for the manage overlay (see build_manage_content) ---

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


def _manage_content_asset_id(frame: Frame) -> str | None:
    """Whether frame.current_asset_id refers to a photo actually visible
    right now, for whichever mode is active -- always true in photos
    mode; only true in calendar mode when the agenda view's photo inlay
    is on (otherwise current_asset_id could be stale, left over from
    whenever photos mode last ran, and showing its location/date/share
    info on a manage overlay over a view with no visible photo at all
    would be actively misleading, not just unhelpful)."""
    relevant = frame.mode != "calendar" or (frame.calendar_view == "agenda" and frame.calendar_photo_inlay)
    return frame.current_asset_id if relevant and frame.current_asset_id else None


def build_manage_content(db: Session, frame: Frame, request) -> dict:
    """Gathers everything manage_overlay.compose() needs -- what used to
    be two separate device-facing endpoints (/frame/photo-info,
    /frame/face-labels, both removed -- see the module docstring in
    manage_overlay.py) are now just internal calls made here, once,
    server-side, since compositing itself also moved server-side.
    management_url and battery_percent always apply; location/date/
    share-URL/face-labels only when there's a real current photo (see
    _manage_content_asset_id) -- absent otherwise, which
    manage_overlay.compose() already treats as "skip that region",
    exactly the graceful-degradation behavior the old firmware-fetched
    version had."""
    base = str(request.base_url).rstrip("/")
    content: dict = {
        "management_url": f"{base}/m/{frame.manage_token}",
        "battery_percent": frame.battery_percent,
    }

    asset_id = _manage_content_asset_id(frame)
    if not asset_id:
        return content

    client = immich_client_for(frame)
    try:
        asset = client.get_asset(asset_id)
        faces = client.get_asset_faces(asset_id)
    except httpx.HTTPError as e:
        logger.warning("Could not fetch manage-overlay photo info for asset %s: %s", asset_id, e)
        return content

    exif = asset.get("exifInfo") or {}
    content["location_lines"] = _format_location(exif)
    content["taken_at"] = _format_taken_at(exif)
    content["share_url"] = f"{base}/frame/share/{asset_id}"

    if any((face.get("person") or {}).get("name") for face in faces):
        try:
            preview_bytes = client.download_asset_preview(asset_id)
            from ..face_labels import compute_face_labels

            content["face_labels"] = compute_face_labels(preview_bytes, faces, frame.display_mode, frame.orientation)
        except httpx.HTTPError as e:
            logger.warning("Could not download asset %s for face-label mapping: %s", asset_id, e)

    return content


def calendar_sources_for_frame(db: Session, frame: Frame) -> list[tuple[str, str]]:
    """Every user linked to this frame with BOTH a calendar URL set AND
    explicit per-frame opt-in (UserFrame.calendar_included) -- the exact
    set calendar_feed.merge_events needs. [(display_name-or-username,
    ics_url), ...]."""
    rows = db.execute(
        select(User)
        .join(UserFrame, UserFrame.user_id == User.id)
        .where(UserFrame.frame_id == frame.id, UserFrame.calendar_included == True,  # noqa: E712
               User.calendar_ics_url != "")
    ).scalars().all()
    return [(u.display_name or u.username, u.calendar_ics_url) for u in rows]


def get_or_refresh_calendar_events(db: Session, frame: Frame) -> tuple[list[dict], str]:
    """Frame-level throttled merge-fetch (calendar_feed.CHECK_INTERVAL_S)
    -- same shape as the Gitea release-check throttle in api_frames.py's
    api_firmware_check. One shared cache for the whole merged result
    (every included user's events together), not per-user -- ICS feeds
    are small and this refetches at most every ~20 minutes regardless of
    how many are included, so per-user cache columns would add
    bookkeeping for a marginal benefit."""
    now = time.time()
    if frame.calendar_cached_events is not None and now - frame.calendar_checked_at < calendar_feed.CHECK_INTERVAL_S:
        return frame.calendar_cached_events, frame.calendar_fetch_summary

    sources = calendar_sources_for_frame(db, frame)
    today = quiet_hours.local_date(frame)
    events, summary = calendar_feed.merge_events(
        sources,
        today - timedelta(days=calendar_feed.EXPAND_WINDOW_PAST_DAYS),
        today + timedelta(days=calendar_feed.EXPAND_WINDOW_FUTURE_DAYS),
    )
    with frame_locked(db, frame.id) as locked:
        locked.calendar_cached_events = events
        locked.calendar_fetch_summary = summary
        locked.calendar_checked_at = now
    return events, summary