| import atexit |
| import json |
| import os |
| import threading |
| import time |
| from pathlib import Path |
| from typing import Dict, List, Optional, Tuple |
| import cv2 |
| import numpy as np |
|
|
| from detector import ObjectDetector |
| from filters import AVAILABLE_FILTERS, process_frame |
|
|
| DEFAULT_CAMERAS = [ |
| { |
| "id": "laptop_cam", |
| "name": "Integrated Webcam / Browser Stream HD", |
| "type": "local", |
| "source": "0", |
| "enabled": True, |
| "is_default": True |
| } |
| ] |
|
|
| def apply_targeted_zoom( |
| frame: np.ndarray, |
| zoom: float, |
| center_x: float = 0.5, |
| center_y: float = 0.5 |
| ) -> Tuple[np.ndarray, Tuple[int, int, int, int]]: |
| """ |
| Crops around normalized coordinates (center_x, center_y) and rescales to original dimensions. |
| Returns (zoomed_frame, (crop_x1, crop_y1, crop_w, crop_h)). |
| """ |
| if frame is None: |
| return frame, (0, 0, 1280, 720) |
| |
| h, w = frame.shape[:2] |
| if zoom <= 1.01: |
| return frame, (0, 0, w, h) |
|
|
| new_h = int(h / zoom) |
| new_w = int(w / zoom) |
|
|
| cx_px = int(np.clip(center_x, 0.0, 1.0) * w) |
| cy_px = int(np.clip(center_y, 0.0, 1.0) * h) |
|
|
| x1 = max(0, min(w - new_w, cx_px - new_w // 2)) |
| y1 = max(0, min(h - new_h, cy_px - new_h // 2)) |
| x2 = x1 + new_w |
| y2 = y1 + new_h |
|
|
| cropped = frame[y1:y2, x1:x2] |
| zoomed = cv2.resize(cropped, (w, h), interpolation=cv2.INTER_LINEAR) |
| return zoomed, (x1, y1, new_w, new_h) |
|
|
| class CameraManager: |
| def __init__(self, config_path: Path): |
| self.config_path = config_path |
| self.lock = threading.RLock() |
| |
| |
| self.cameras: List[dict] = [] |
| self.active_camera_id: str = "laptop_cam" |
| self.current_filter: str = "normal" |
| |
| |
| self.zoom_level: float = 1.0 |
| self.zoom_center_x: float = 0.5 |
| self.zoom_center_y: float = 0.5 |
| self.target_object_id: Optional[str] = None |
| self.auto_tracking_mode: str = "off" |
|
|
| self.ai_detection_enabled: bool = True |
| self.show_hud_overlay: bool = True |
| |
| |
| self.detector = ObjectDetector() |
| self.current_detections: List[dict] = [] |
| self.current_telemetry: dict = dict(self.detector.latest_telemetry) |
| |
| self._load_registry() |
|
|
| |
| self.cap: Optional[cv2.VideoCapture] = None |
| self.thread: Optional[threading.Thread] = None |
| self.ai_thread: Optional[threading.Thread] = None |
| self.running: bool = False |
| |
| |
| self.ai_frame_buffer: Optional[np.ndarray] = None |
| self.ai_event = threading.Event() |
| self.external_frame_event = threading.Event() |
| |
| |
| self.latest_raw_frame: Optional[np.ndarray] = None |
| self.latest_jpeg: Optional[bytes] = None |
| self.external_frame: Optional[np.ndarray] = None |
| self.external_frame_ts: float = 0.0 |
| self.actual_width: int = 1280 |
| self.actual_height: int = 720 |
| self.fps: float = 0.0 |
| self.started_at: Optional[float] = None |
| self.client_count: int = 0 |
| |
| self.new_frame_event = threading.Event() |
| atexit.register(self.stop) |
| if os.environ.get("PORT") == "7860": |
| self.ai_detection_enabled = True |
| self.show_hud_overlay = True |
| self.start(skip_local_probe=True) |
|
|
| def push_external_frame(self, jpeg_bytes: bytes) -> bool: |
| """Ingests a live JPEG frame from Browser Webcam (getUserMedia) or local station relay.""" |
| if not jpeg_bytes: |
| return False |
| arr = np.frombuffer(jpeg_bytes, dtype=np.uint8) |
| frame = cv2.imdecode(arr, cv2.IMREAD_COLOR) |
| if frame is None: |
| return False |
| with self.lock: |
| self.external_frame = frame |
| self.external_frame_ts = time.time() |
| self.actual_height, self.actual_width = frame.shape[:2] |
| self.external_frame_event.set() |
| if not self.running: |
| self.start(skip_local_probe=True) |
| return True |
|
|
| def _load_registry(self): |
| if self.config_path.exists(): |
| try: |
| data = json.loads(self.config_path.read_text()) |
| if isinstance(data, list) and len(data) > 0: |
| existing_ids = {c.get("id") for c in data} |
| for def_cam in DEFAULT_CAMERAS: |
| if def_cam["id"] not in existing_ids: |
| data.append(dict(def_cam)) |
| self.cameras = data |
| self.active_camera_id = self.cameras[0]["id"] |
| self._save_registry() |
| return |
| except Exception: |
| pass |
| self.cameras = list(DEFAULT_CAMERAS) |
| self.active_camera_id = "laptop_cam" |
| self._save_registry() |
|
|
| def _save_registry(self): |
| self.config_path.parent.mkdir(parents=True, exist_ok=True) |
| self.config_path.write_text(json.dumps(self.cameras, indent=2)) |
|
|
| def get_active_camera(self) -> dict: |
| with self.lock: |
| for cam in self.cameras: |
| if cam["id"] == self.active_camera_id: |
| return cam |
| return self.cameras[0] if self.cameras else DEFAULT_CAMERAS[0] |
|
|
| def get_cameras(self) -> List[dict]: |
| with self.lock: |
| return [ |
| {**c, "is_active": (c["id"] == self.active_camera_id)} |
| for c in self.cameras |
| ] |
|
|
| def add_camera(self, name: str, source: str, cam_type: str = "network") -> Tuple[bool, str, dict]: |
| with self.lock: |
| cam_id = f"cam_{int(time.time())}" |
| new_cam = { |
| "id": cam_id, |
| "name": name.strip(), |
| "type": cam_type, |
| "source": source.strip(), |
| "enabled": True, |
| "is_default": False |
| } |
| self.cameras.append(new_cam) |
| self._save_registry() |
| return True, "Camera added successfully", new_cam |
|
|
| def remove_camera(self, cam_id: str) -> Tuple[bool, str]: |
| with self.lock: |
| if cam_id == "laptop_cam": |
| return False, "Cannot delete the built-in system camera" |
| |
| orig_len = len(self.cameras) |
| self.cameras = [c for c in self.cameras if c["id"] != cam_id] |
| if len(self.cameras) == orig_len: |
| return False, "Camera ID not found" |
|
|
| self._save_registry() |
| |
| if self.active_camera_id == cam_id: |
| self.active_camera_id = "laptop_cam" |
| fallback_restart = True |
| else: |
| fallback_restart = False |
|
|
| if fallback_restart and self.running: |
| self.switch_camera("laptop_cam") |
|
|
| return True, "Camera removed successfully" |
|
|
| def set_filter(self, filter_name: str) -> Tuple[bool, str]: |
| if filter_name not in AVAILABLE_FILTERS: |
| return False, f"Unknown filter. Available: {list(AVAILABLE_FILTERS.keys())}" |
| with self.lock: |
| self.current_filter = filter_name |
| return True, f"Filter updated to: {AVAILABLE_FILTERS[filter_name]}" |
|
|
| def set_zoom( |
| self, |
| zoom: float, |
| center_x: Optional[float] = None, |
| center_y: Optional[float] = None |
| ) -> Tuple[bool, str, float]: |
| clamped_zoom = max(1.0, min(4.0, round(float(zoom), 2))) |
| with self.lock: |
| self.zoom_level = clamped_zoom |
| if center_x is not None: |
| self.zoom_center_x = float(np.clip(center_x, 0.0, 1.0)) |
| if center_y is not None: |
| self.zoom_center_y = float(np.clip(center_y, 0.0, 1.0)) |
| if clamped_zoom <= 1.01 and center_x is None: |
| self.zoom_center_x = 0.5 |
| self.zoom_center_y = 0.5 |
| self.target_object_id = None |
| return True, f"Zoom set to {clamped_zoom}x", clamped_zoom |
|
|
| def set_zoom_focus_point(self, cx: float, cy: float, zoom_level: Optional[float] = None) -> Tuple[bool, str]: |
| with self.lock: |
| self.zoom_center_x = float(np.clip(cx, 0.0, 1.0)) |
| self.zoom_center_y = float(np.clip(cy, 0.0, 1.0)) |
| if zoom_level is not None: |
| self.zoom_level = max(1.0, min(4.0, round(float(zoom_level), 2))) |
| elif self.zoom_level <= 1.1: |
| self.zoom_level = 2.0 |
| return True, f"Focused on point ({self.zoom_center_x:.2f}, {self.zoom_center_y:.2f}) at {self.zoom_level}x" |
|
|
| def set_focus_target_object(self, object_id: Optional[str], zoom_level: Optional[float] = None) -> Tuple[bool, str]: |
| """Locks digital zoom onto a specific tracked object ID or body part.""" |
| with self.lock: |
| self.target_object_id = object_id |
| if object_id is None: |
| self.auto_tracking_mode = "off" |
| return True, "Target lock cleared" |
|
|
| self.ai_detection_enabled = True |
| self.auto_tracking_mode = "selected" |
| for d in self.current_detections: |
| if d.get("id") == object_id: |
| self.zoom_center_x = d["center"][0] |
| self.zoom_center_y = d["center"][1] |
| if zoom_level is not None: |
| self.zoom_level = max(1.0, min(4.0, float(zoom_level))) |
| elif self.zoom_level <= 1.1: |
| self.zoom_level = 2.2 |
| return True, f"Locked target onto {d['label'].upper()} ({object_id})" |
|
|
| return True, f"Target ID {object_id} queued for tracking" |
|
|
| def set_auto_tracking_mode(self, mode: str) -> Tuple[bool, str]: |
| valid_modes = ["off", "face", "person", "hand", "torso", "pelvic", "feet", "selected"] |
| clean_mode = mode.lower().strip() |
| if clean_mode not in valid_modes: |
| return False, f"Invalid mode. Choose from: {valid_modes}" |
|
|
| with self.lock: |
| self.auto_tracking_mode = clean_mode |
| if clean_mode == "off": |
| self.target_object_id = None |
| else: |
| self.ai_detection_enabled = True |
| if self.zoom_level <= 1.1: |
| self.zoom_level = 2.0 |
|
|
| return True, f"Auto-Tracking set to: {clean_mode.upper()}" |
|
|
| def set_hud_overlay(self, enabled: bool) -> Tuple[bool, str, bool]: |
| with self.lock: |
| self.show_hud_overlay = bool(enabled) |
| state_str = "visible" if self.show_hud_overlay else "hidden (clean view)" |
| return True, f"AI HUD Overlay {state_str}", self.show_hud_overlay |
|
|
| def set_detection(self, enabled: bool) -> Tuple[bool, str, bool]: |
| with self.lock: |
| self.ai_detection_enabled = bool(enabled) |
| state_str = "enabled" if self.ai_detection_enabled else "disabled" |
| if not self.ai_detection_enabled: |
| self.current_detections = [] |
| self.auto_tracking_mode = "off" |
| self.target_object_id = None |
| return True, f"AI Vision & Body-Parts Detection {state_str}", self.ai_detection_enabled |
|
|
| def switch_camera(self, cam_id: str) -> Tuple[bool, str]: |
| with self.lock: |
| found = any(c["id"] == cam_id for c in self.cameras) |
| if not found: |
| return False, "Camera ID not found in registry" |
| self.active_camera_id = cam_id |
|
|
| if self.running: |
| self.stop() |
| return self.start() |
| return True, f"Switched active camera to {cam_id}" |
|
|
| def _make_standby_frame(self, cam_info: dict, status_msg: str = "READY FOR BROWSER WEBCAM OR NETWORK STREAM...") -> np.ndarray: |
| frame = np.zeros((720, 1280, 3), dtype=np.uint8) |
| frame[:] = (18, 20, 28) |
| cv2.rectangle(frame, (40, 40), (1240, 680), (45, 55, 75), 2) |
| cv2.putText(frame, f"CAMERA VIEW: {cam_info.get('name', 'Primary View').upper()}", (80, 140), cv2.FONT_HERSHEY_SIMPLEX, 0.9, (0, 215, 255), 2, cv2.LINE_AA) |
| cv2.putText(frame, f"SOURCE: {cam_info.get('source', 'N/A')}", (80, 200), cv2.FONT_HERSHEY_SIMPLEX, 0.65, (180, 190, 210), 1, cv2.LINE_AA) |
| cv2.putText(frame, f"STATUS: {status_msg}", (80, 260), cv2.FONT_HERSHEY_SIMPLEX, 0.65, (80, 220, 120), 2, cv2.LINE_AA) |
| ts = time.strftime("%Y-%m-%d %H:%M:%S") |
| cv2.putText(frame, f"HEARTBEAT: {ts} | CLICK 'BROWSER WEBCAM' TO STREAM", (80, 620), cv2.FONT_HERSHEY_SIMPLEX, 0.55, (120, 130, 150), 1, cv2.LINE_AA) |
| return frame |
|
|
| def _try_open_capture(self, cam_info: dict) -> Optional[cv2.VideoCapture]: |
| source = cam_info["source"] |
| is_local = (cam_info["type"] == "local") |
| try: |
| if is_local: |
| dev_index = int(source) if source.isdigit() else 0 |
| if not os.path.exists(f"/dev/video{dev_index}"): |
| return None |
| cap = cv2.VideoCapture(dev_index, cv2.CAP_V4L2) |
| if not cap.isOpened(): |
| cap = cv2.VideoCapture(dev_index) |
| if not cap.isOpened(): |
| return None |
| cap.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter_fourcc(*'MJPG')) |
| cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1280) |
| cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 720) |
| cap.set(cv2.CAP_PROP_FPS, 30) |
| return cap |
| else: |
| os.environ["OPENCV_FFMPEG_CAPTURE_OPTIONS"] = "timeout;2500000" |
| cap = cv2.VideoCapture(source) |
| if not cap.isOpened(): |
| return None |
| return cap |
| except Exception: |
| return None |
|
|
| def start(self, skip_local_probe: bool = False) -> Tuple[bool, str]: |
| with self.lock: |
| if self.running: |
| return True, "Camera already running" |
|
|
| cam_info = self.get_active_camera() |
| cap = None if skip_local_probe else self._try_open_capture(cam_info) |
| frame = None |
| if cap is not None: |
| ret, f = cap.read() |
| if ret and f is not None: |
| frame = f |
| else: |
| cap.release() |
| cap = None |
|
|
| if frame is None and self.external_frame is not None: |
| frame = self.external_frame.copy() |
|
|
| if frame is None: |
| frame = self._make_standby_frame(cam_info) |
|
|
| h, w = frame.shape[:2] |
| self.actual_width = w |
| self.actual_height = h |
| self.cap = cap |
| self.running = True |
| self.started_at = time.time() |
|
|
| self.thread = threading.Thread(target=self._capture_loop, daemon=True) |
| self.thread.start() |
|
|
| self.ai_thread = threading.Thread(target=self._ai_worker_loop, daemon=True) |
| self.ai_thread.start() |
|
|
| mode_str = "LIVE" if cap is not None else "STANDBY (Browser/Network Ready)" |
| return True, f"Camera '{cam_info['name']}' started [{mode_str}] ({w}x{h})" |
|
|
| def stop(self) -> Tuple[bool, str]: |
| with self.lock: |
| if not self.running: |
| return True, "Camera already in standby" |
|
|
| self.running = False |
| self.new_frame_event.set() |
| self.ai_event.set() |
| self.external_frame_event.set() |
|
|
| if self.thread: |
| self.thread.join(timeout=2.0) |
| self.thread = None |
|
|
| if self.ai_thread: |
| self.ai_thread.join(timeout=2.0) |
| self.ai_thread = None |
|
|
| with self.lock: |
| if self.cap: |
| self.cap.release() |
| self.cap = None |
| self.latest_jpeg = None |
| self.latest_raw_frame = None |
| self.ai_frame_buffer = None |
| self.fps = 0.0 |
| self.started_at = None |
|
|
| return True, "Camera put into standby, sensor released" |
|
|
| def _ai_worker_loop(self): |
| """Dedicated background AI thread for 80-class detection and 17-keypoint body-parts tracking.""" |
| import urllib.request |
| hf_infer_url = os.environ.get("HF_SPACE_INFER_URL", "").strip() |
| hf_token = os.environ.get("HF_TOKEN", "").strip() |
|
|
| while self.running: |
| if not self.ai_detection_enabled: |
| time.sleep(0.05) |
| continue |
|
|
| self.ai_event.wait(timeout=0.15) |
| self.ai_event.clear() |
|
|
| frame_to_process = None |
| with self.lock: |
| if self.ai_frame_buffer is not None: |
| frame_to_process = self.ai_frame_buffer |
| self.ai_frame_buffer = None |
|
|
| if frame_to_process is not None: |
| try: |
| dets, telem = None, None |
| if hf_infer_url: |
| try: |
| ok_enc, buf = cv2.imencode(".jpg", frame_to_process, [int(cv2.IMWRITE_JPEG_QUALITY), 80]) |
| if ok_enc: |
| headers = {"Content-Type": "application/octet-stream"} |
| if hf_token: |
| headers["Authorization"] = f"Bearer {hf_token}" |
| req = urllib.request.Request(hf_infer_url, data=buf.tobytes(), headers=headers, method="POST") |
| with urllib.request.urlopen(req, timeout=1.5) as resp: |
| payload = json.loads(resp.read().decode("utf-8")) |
| dets = payload.get("detections", []) |
| telem = payload.get("telemetry", self.current_telemetry) |
| self.detector.latest_keypoints = payload.get("keypoints", []) |
| except Exception: |
| dets, telem = None, None |
|
|
| if dets is None: |
| dets, telem = self.detector.detect(frame_to_process) |
|
|
| with self.lock: |
| self.current_detections = dets |
| self.current_telemetry = telem |
| except Exception: |
| pass |
|
|
| def _capture_loop(self): |
| """High-speed 30+ FPS video capture loop with smooth PTZ digital glide tracking.""" |
| frame_counter = 0 |
| fps_timer = time.time() |
| last_reconnect_attempt = 0.0 |
|
|
| while self.running: |
| frame = None |
| if self.cap is not None and self.cap.isOpened(): |
| ret, f = self.cap.read() |
| if ret and f is not None: |
| frame = f |
| else: |
| self.cap.release() |
| self.cap = None |
|
|
| if frame is None: |
| now = time.time() |
| with self.lock: |
| if self.external_frame is not None and (now - self.external_frame_ts < 8.0): |
| frame = self.external_frame |
| if frame is not None: |
| |
| self.external_frame_event.wait(timeout=0.025) |
| self.external_frame_event.clear() |
| with self.lock: |
| if self.external_frame is not None: |
| frame = self.external_frame.copy() |
|
|
| if frame is None: |
| cam_info = self.get_active_camera() |
| if now - last_reconnect_attempt >= 3.0: |
| last_reconnect_attempt = now |
| new_cap = self._try_open_capture(cam_info) |
| if new_cap is not None: |
| ret, f = new_cap.read() |
| if ret and f is not None: |
| self.cap = new_cap |
| frame = f |
| else: |
| new_cap.release() |
|
|
| if frame is None: |
| frame = self._make_standby_frame(cam_info) |
| self.external_frame_event.wait(timeout=0.08) |
| self.external_frame_event.clear() |
|
|
| with self.lock: |
| active_filter = self.current_filter |
| zoom = self.zoom_level |
| ai_enabled = self.ai_detection_enabled |
| show_hud = self.show_hud_overlay |
| cached_dets = list(self.current_detections) |
| cached_telem = dict(self.current_telemetry) |
| tracking_mode = self.auto_tracking_mode |
| target_id = self.target_object_id |
|
|
| |
| if ai_enabled and tracking_mode != "off": |
| matched_target = None |
| if tracking_mode == "face": |
| matched_target = next((d for d in cached_dets if d.get("base_label", d["label"]) == "head / face"), None) |
| elif tracking_mode == "person": |
| matched_target = next((d for d in cached_dets if d.get("base_label", d["label"]) == "person"), None) |
| elif tracking_mode == "hand": |
| matched_target = next((d for d in cached_dets if "hand" in d.get("base_label", d["label"])), None) |
| elif tracking_mode == "torso": |
| matched_target = next((d for d in cached_dets if "torso" in d.get("base_label", d["label"]) or "chest" in d.get("base_label", d["label"])), None) |
| elif tracking_mode == "pelvic": |
| matched_target = next((d for d in cached_dets if "pelvic" in d.get("base_label", d["label"]) or "hips" in d.get("base_label", d["label"])), None) |
| elif tracking_mode == "feet": |
| matched_target = next((d for d in cached_dets if "foot" in d.get("base_label", d["label"])), None) |
| elif tracking_mode == "selected" and target_id: |
| matched_target = next((d for d in cached_dets if d.get("id") == target_id), None) |
|
|
| if matched_target: |
| target_cx, target_cy = matched_target["center"] |
| alpha = 0.18 |
| with self.lock: |
| self.zoom_center_x = round(self.zoom_center_x * (1 - alpha) + target_cx * alpha, 3) |
| self.zoom_center_y = round(self.zoom_center_y * (1 - alpha) + target_cy * alpha, 3) |
|
|
| with self.lock: |
| center_x = self.zoom_center_x |
| center_y = self.zoom_center_y |
|
|
| |
| if ai_enabled: |
| with self.lock: |
| if self.ai_frame_buffer is None: |
| self.ai_frame_buffer = frame |
| self.ai_event.set() |
|
|
| |
| zoomed_frame, crop_meta = apply_targeted_zoom(frame, zoom, center_x, center_y) |
|
|
| |
| clean_processed = process_frame(zoomed_frame, active_filter) |
|
|
| |
| stream_frame = clean_processed |
| if ai_enabled and show_hud: |
| stream_frame = self.detector.render_overlay( |
| clean_processed, |
| cached_dets, |
| cached_telem, |
| crop_meta=crop_meta, |
| active_target_id=target_id |
| ) |
|
|
| |
| ret_enc, jpeg = cv2.imencode('.jpg', stream_frame, [int(cv2.IMWRITE_JPEG_QUALITY), 85]) |
| if ret_enc: |
| jpeg_bytes = jpeg.tobytes() |
| with self.lock: |
| self.latest_raw_frame = clean_processed |
| self.latest_jpeg = jpeg_bytes |
|
|
| self.new_frame_event.set() |
| self.new_frame_event.clear() |
|
|
| frame_counter += 1 |
| now = time.time() |
| if now - fps_timer >= 1.0: |
| self.fps = round(frame_counter / (now - fps_timer), 1) |
| frame_counter = 0 |
| fps_timer = now |
|
|
| with self.lock: |
| if self.cap: |
| self.cap.release() |
| self.cap = None |
|
|
| def get_latest_jpeg(self) -> Optional[bytes]: |
| with self.lock: |
| return self.latest_jpeg |
|
|
| def get_clean_jpeg(self) -> Optional[bytes]: |
| """Returns a clean JPEG encoded directly from latest_raw_frame without any AI HUD overlays.""" |
| with self.lock: |
| raw = self.latest_raw_frame.copy() if self.latest_raw_frame is not None else None |
| if raw is None: |
| return self.get_latest_jpeg() |
| ok, buf = cv2.imencode('.jpg', raw, [int(cv2.IMWRITE_JPEG_QUALITY), 92]) |
| return buf.tobytes() if ok else self.get_latest_jpeg() |
|
|
| def _build_scene_intelligence(self) -> dict: |
| """Builds a real-time natural-language scene title, description, and pose/object tracking breakdown.""" |
| dets = list(self.current_detections) |
| telem = dict(self.current_telemetry) |
| filt_label = AVAILABLE_FILTERS.get(self.current_filter, "Normal HD") |
|
|
| people = [d for d in dets if d.get("base_label", d["label"]) == "person"] |
| body_parts = [d for d in dets if d.get("base_label", d["label"]) in ( |
| "head / face", "chest / torso", "pelvic / hips", "left hand", "right hand", |
| "left arm", "right arm", "left leg", "right leg", "left foot", "right foot" |
| )] |
| objects = [d for d in dets if d not in people and d not in body_parts] |
|
|
| |
| if len(people) > 1: |
| title = f"👥 Multi-Person Scene ({len(people)} Subjects) — {filt_label}" |
| elif len(people) == 1 or body_parts: |
| focus_str = f"Tracking {self.auto_tracking_mode.upper()}" if self.auto_tracking_mode != "off" else f"{self.zoom_level:.1f}x View" |
| title = f"👤 Live Subject Tracking ({focus_str}) — {filt_label}" |
| elif objects: |
| top_objs = ", ".join(sorted({o.get('base_label', o['label']).title() for o in objects[:2]})) |
| title = f"🎥 Environment Feed: {top_objs} — {filt_label}" |
| else: |
| title = f"🎥 Ambient Surveillance Stream — {filt_label}" |
|
|
| |
| parts_list = [p.get("base_label", p["label"]) for p in body_parts] |
| unique_parts = list(dict.fromkeys(parts_list)) |
| obj_list = list(dict.fromkeys(o.get("base_label", o["label"]) for o in objects)) |
|
|
| sentences = [] |
| if people or unique_parts: |
| p_count = max(1, len(people)) |
| subj_word = "1 person is" if p_count == 1 else f"{p_count} people are" |
| if unique_parts: |
| sentences.append(f"{subj_word} active in the scene with tracked {', '.join(unique_parts[:5])}.") |
| else: |
| sentences.append(f"{subj_word} currently visible in the camera frame.") |
| else: |
| sentences.append("No human subjects currently detected in the foreground.") |
|
|
| if obj_list: |
| sentences.append(f"Surrounding objects identified: {', '.join(obj_list[:5])}.") |
|
|
| if self.zoom_level > 1.05 or self.auto_tracking_mode != "off": |
| track_desc = f"locked onto {self.auto_tracking_mode.upper()}" if self.auto_tracking_mode != "off" else f"centered at ({self.zoom_center_x:.2f}, {self.zoom_center_y:.2f})" |
| sentences.append(f"Digital PTZ lens is at {self.zoom_level:.1f}x magnification ({track_desc}).") |
|
|
| kpts_count = int(telem.get("keypoints_count", len(self.detector.latest_keypoints) * 17)) |
| provider_str = "T4 CUDA GPU" if telem.get("gpu_active") else telem.get("provider", "ONNX Runtime") |
|
|
| return { |
| "title": title, |
| "description": " ".join(sentences), |
| "tracking_details": { |
| "status": "ACTIVE TRACKING" if self.ai_detection_enabled else "AI PAUSED", |
| "provider": provider_str, |
| "total_targets": len(dets), |
| "inference_ms": telem.get("inference_ms", 0.0), |
| "visible_body_parts": [p["label"] for p in body_parts], |
| "category_breakdown": { |
| "People": len(people), |
| "Body Parts": len(body_parts), |
| "Objects": len(objects), |
| "Keypoints": kpts_count |
| } |
| } |
| } |
|
|
| def get_status(self) -> dict: |
| uptime = int(time.time() - self.started_at) if self.running and self.started_at else 0 |
| cam_info = self.get_active_camera() |
| return { |
| "running": self.running, |
| "active_camera_id": self.active_camera_id, |
| "active_camera_name": cam_info["name"], |
| "active_camera_type": cam_info["type"], |
| "active_filter": self.current_filter, |
| "active_filter_label": AVAILABLE_FILTERS.get(self.current_filter, "Normal"), |
| "zoom_level": self.zoom_level, |
| "zoom_center_x": self.zoom_center_x, |
| "zoom_center_y": self.zoom_center_y, |
| "target_object_id": self.target_object_id, |
| "auto_tracking_mode": self.auto_tracking_mode, |
| "ai_detection_enabled": self.ai_detection_enabled, |
| "show_hud_overlay": self.show_hud_overlay, |
| "detections": self.current_detections, |
| "telemetry": self.current_telemetry, |
| "scene": self._build_scene_intelligence(), |
| "width": self.actual_width, |
| "height": self.actual_height, |
| "fps": self.fps, |
| "uptime_seconds": uptime, |
| "client_count": self.client_count, |
| "available_filters": AVAILABLE_FILTERS, |
| "cameras": self.get_cameras(), |
| } |
|
|
| def generate_mjpeg(self): |
| self.client_count += 1 |
| try: |
| while self.running: |
| self.new_frame_event.wait(timeout=0.2) |
| jpeg = self.get_latest_jpeg() |
| if jpeg: |
| yield ( |
| b"--frame\r\n" |
| b"Content-Type: image/jpeg\r\n" |
| b"Content-Length: " + str(len(jpeg)).encode() + b"\r\n\r\n" |
| + jpeg + b"\r\n" |
| ) |
| else: |
| time.sleep(0.02) |
| finally: |
| self.client_count = max(0, self.client_count - 1) |
|
|
