File size: 5,603 Bytes
c794b6b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 | """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
|