backend/vision_session.py

"""Per-session camera ownership and face-result freshness (TTL)."""
from __future__ import annotations

import os
import threading
import time
from typing import Any

PRESENCE_TTL_S = float(os.getenv("CEPHEUS_PRESENCE_TTL", "10") or "10")

_lock = threading.Lock()
_frame_meta: dict[str, dict[str, Any]] = {}
_camera_owners: dict[str, str] = {}
_client_cameras: dict[str, set[str]] = {}
_client_alive: dict[str, float] = {}
_client_selection: dict[str, tuple[str, str]] = {}
_frame_counter = 0


def register_client(client_id: str) -> None:
    with _lock:
        _client_alive[client_id] = time.time()


def touch_client(client_id: str) -> None:
    with _lock:
        _client_alive[client_id] = time.time()


def release_client(client_id: str) -> None:
    with _lock:
        _client_alive.pop(client_id, None)
        _client_selection.pop(client_id, None)
        cams = _client_cameras.pop(client_id, set())
        for cam_id in cams:
            if _camera_owners.get(cam_id) == client_id:
                _camera_owners.pop(cam_id, None)


def camera_count() -> int:
    with _lock:
        return len(_camera_owners)


def claim_camera(client_id: str, cam_id: str, index: Any) -> tuple[bool, str]:
    """One owner per cam_id; one active camera per client."""
    idx_key = str(index)
    sel = (cam_id, idx_key)
    with _lock:
        touch_client(client_id)
        prev = _client_selection.get(client_id)
        if prev == sel:
            return True, "duplicate_ignored"
        owner = _camera_owners.get(cam_id)
        if owner and owner != client_id and owner in _client_alive:
            return False, "camera_owned_by_other_client"
        if prev and prev[0] != cam_id:
            old_cam = prev[0]
            if _camera_owners.get(old_cam) == client_id:
                _camera_owners.pop(old_cam, None)
            _client_cameras.get(client_id, set()).discard(old_cam)
        _camera_owners[cam_id] = client_id
        _client_cameras.setdefault(client_id, set()).add(cam_id)
        _client_selection[client_id] = sel
        return True, "ok"


def record_processed_frame(cam_id: str, had_match: bool) -> int:
    """Bump frame id + TTL only when a frame was actually processed."""
    global _frame_counter
    now = time.time()
    with _lock:
        _frame_counter += 1
        prev = _frame_meta.get(cam_id, {})
        _frame_meta[cam_id] = {
            "last_frame_ts": now,
            "last_match_ts": now if had_match else prev.get("last_match_ts"),
            "frame_id": _frame_counter,
            "presence_expires_at": now + PRESENCE_TTL_S,
        }
        return _frame_counter


def get_cam_meta(cam_id: str) -> dict[str, Any]:
    with _lock:
        return dict(_frame_meta.get(cam_id, {}))


def is_cam_stale(cam_id: str) -> bool:
    with _lock:
        meta = _frame_meta.get(cam_id)
        if not meta:
            return True
        return time.time() > float(meta.get("presence_expires_at", 0))


def presence_summary() -> dict[str, Any]:
    now = time.time()
    with _lock:
        if not _frame_meta:
            return {
                "presence_state": "inactive",
                "presence_is_stale": True,
                "last_frame_age_ms": None,
                "last_presence_update_age_ms": None,
            }
        latest_ts = max(m.get("last_frame_ts", 0) for m in _frame_meta.values())
        match_ts = max((m.get("last_match_ts") or 0) for m in _frame_meta.values())
        age_ms = int((now - latest_ts) * 1000) if latest_ts else None
        match_age_ms = int((now - match_ts) * 1000) if match_ts else None
        stale = age_ms is None or age_ms > int(PRESENCE_TTL_S * 1000)
        return {
            "presence_state": "inactive" if stale else "active",
            "presence_is_stale": stale,
            "last_frame_age_ms": age_ms,
            "last_presence_update_age_ms": match_age_ms,
            "last_match_age_ms": match_age_ms,
        }


def build_face_results_payload(raw_results: dict, strip_fn) -> dict:
    """Return per-camera payload; never resurrect expired recognition."""
    now = time.time()
    out: dict[str, Any] = {}
    for cam_id, faces in (raw_results or {}).items():
        meta = get_cam_meta(cam_id)
        expires = float(meta.get("presence_expires_at", 0))
        last_ts = float(meta.get("last_frame_ts", 0))
        age_ms = int((now - last_ts) * 1000) if last_ts else None
        stale = not meta or now > expires
        if stale:
            out[cam_id] = {
                "live": False,
                "stale": True,
                "reason": "frame_timeout" if meta else "no_frames",
                "tracked_faces": [],
                "faces": [],
                "confidence": None,
                "presence": False,
                "last_frame_ts": last_ts or None,
                "last_match_ts": meta.get("last_match_ts"),
                "presence_expires_at": expires or None,
                "last_frame_age_ms": age_ms,
                "frame_id": meta.get("frame_id"),
            }
        else:
            public = strip_fn({cam_id: faces}).get(cam_id, faces)
            out[cam_id] = {
                "live": True,
                "stale": False,
                "tracked_faces": public,
                "faces": public,
                "presence": bool(public),
                "last_frame_ts": last_ts,
                "last_match_ts": meta.get("last_match_ts"),
                "presence_expires_at": expires,
                "last_frame_age_ms": age_ms,
                "frame_id": meta.get("frame_id"),
            }
    return out