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() # Camera configuration & registry self.cameras: List[dict] = [] self.active_camera_id: str = "laptop_cam" self.current_filter: str = "normal" # Targeted Zoom & Object Tracking Parameters 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" # "off", "face", "person", "hand", "torso", "pelvic", "feet", "selected" self.ai_detection_enabled: bool = True self.show_hud_overlay: bool = True # High-Speed Multi-Model AI Engine (YOLOv8 + YOLOv8-Pose) self.detector = ObjectDetector() self.current_detections: List[dict] = [] self.current_telemetry: dict = dict(self.detector.latest_telemetry) self._load_registry() # Capture internals self.cap: Optional[cv2.VideoCapture] = None self.thread: Optional[threading.Thread] = None self.ai_thread: Optional[threading.Thread] = None self.running: bool = False # AI Async Worker Queuing self.ai_frame_buffer: Optional[np.ndarray] = None self.ai_event = threading.Event() self.external_frame_event = threading.Event() # Frame buffers 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: # Wait briefly for next pushed frame or process at ~30 FPS without blocking 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 # 1. Smooth Auto-Tracking Glide Calculation 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 # 2. Feed Full-Frame to AI Worker Queue (if worker ready) if ai_enabled: with self.lock: if self.ai_frame_buffer is None: self.ai_frame_buffer = frame self.ai_event.set() # 3. Targeted Digital Zoom & Smart Framing zoomed_frame, crop_meta = apply_targeted_zoom(frame, zoom, center_x, center_y) # 4. Real-Time Visual Filter (Clean Frame for Recording & Snapshots) clean_processed = process_frame(zoomed_frame, active_filter) # 5. Render HUD, Skeletal Wireframe, and Target Reticle if show_hud_overlay is True 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 ) # 6. Encode to JPEG 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] # 1. Dynamic Scene Title 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}" # 2. Natural-Language Scene Description 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)