Spaces:
Sleeping
Sleeping
| """Frame pipeline + threaded camera manager. | |
| `InterpreterPipeline` is the stateless-per-instance glue: frame in, annotated | |
| frame + optional recognition out. It is shared by the CLI runner and the web app. | |
| `CameraInterpreter` owns a camera, runs the pipeline on a background thread, and | |
| keeps the latest annotated JPEG + transcript so a web server can stream them | |
| without blocking. It is a process-wide singleton (one camera, many viewers). | |
| """ | |
| from __future__ import annotations | |
| import threading | |
| import time | |
| from collections import deque | |
| from dataclasses import dataclass | |
| from typing import Deque, List, Optional | |
| import numpy as np | |
| from .depth import DepthEstimator | |
| from .landmarks import HolisticTracker, LandmarkFrame | |
| from .recognizer import Recognition, SignRecognizer | |
| from .skeleton import draw_skeleton | |
| try: | |
| import cv2 | |
| except ImportError as exc: # pragma: no cover | |
| raise ImportError("opencv-python is required (pip install -r requirements.txt)") from exc | |
| class FrameResult: | |
| annotated: np.ndarray | |
| landmarks: LandmarkFrame | |
| recognition: Optional[Recognition] | |
| depth_source: str | |
| class InterpreterPipeline: | |
| """Holistic detection + depth refinement + skeleton draw + recognition.""" | |
| def __init__( | |
| self, | |
| *, | |
| templates_path: str = "signs/lsf_signs.json", | |
| model_path: Optional[str] = None, | |
| enable_monocular_depth: bool = False, | |
| model_complexity: int = 1, | |
| refine_face_landmarks: bool = True, | |
| ) -> None: | |
| self.tracker = HolisticTracker( | |
| model_complexity=model_complexity, | |
| refine_face_landmarks=refine_face_landmarks, | |
| ) | |
| self.depth = DepthEstimator(enable_monocular=enable_monocular_depth) | |
| self.recognizer = SignRecognizer( | |
| templates_path=templates_path, model_path=model_path | |
| ) | |
| def probe_depth(self, capture) -> None: | |
| self.depth.probe(capture) | |
| def process( | |
| self, | |
| bgr_frame: np.ndarray, | |
| external_depth: Optional[np.ndarray] = None, | |
| *, | |
| draw: bool = True, | |
| ) -> FrameResult: | |
| rgb = cv2.cvtColor(bgr_frame, cv2.COLOR_BGR2RGB) | |
| lf = self.tracker.process(rgb) | |
| depth_info = self.depth.depth_for_frame(rgb, external_depth) | |
| lf = self.depth.refine_landmarks(lf, depth_info) | |
| # Server-side drawing is only used by the CLI; the web app draws the | |
| # skeleton in the browser from returned landmarks, so it skips this. | |
| annotated = draw_skeleton(bgr_frame, lf, self.tracker) if draw else bgr_frame | |
| recognition = self.recognizer.update(lf) | |
| return FrameResult(annotated, lf, recognition, depth_info.source) | |
| def close(self) -> None: | |
| self.tracker.close() | |
| class CameraInterpreter: | |
| """Threaded singleton that captures, processes, and buffers results.""" | |
| _instance: Optional["CameraInterpreter"] = None | |
| _lock = threading.Lock() | |
| def __init__( | |
| self, | |
| camera_index: int = 0, | |
| *, | |
| templates_path: str = "signs/lsf_signs.json", | |
| model_path: Optional[str] = None, | |
| enable_monocular_depth: bool = False, | |
| max_transcript: int = 200, | |
| ) -> None: | |
| self.camera_index = camera_index | |
| self._pipeline = InterpreterPipeline( | |
| templates_path=templates_path, | |
| model_path=model_path, | |
| enable_monocular_depth=enable_monocular_depth, | |
| ) | |
| self._cap: Optional[cv2.VideoCapture] = None | |
| self._thread: Optional[threading.Thread] = None | |
| self._running = False | |
| self._frame_lock = threading.Lock() | |
| self._latest_jpeg: Optional[bytes] = None | |
| self._transcript: Deque[dict] = deque(maxlen=max_transcript) | |
| self._transcript_seq = 0 | |
| self._depth_source = "mediapipe" | |
| self._fps = 0.0 | |
| # -- singleton access ----------------------------------------------------- | |
| def instance(cls, **kwargs) -> "CameraInterpreter": | |
| with cls._lock: | |
| if cls._instance is None: | |
| cls._instance = cls(**kwargs) | |
| cls._instance.start() | |
| return cls._instance | |
| # -- lifecycle ------------------------------------------------------------ | |
| def start(self) -> None: | |
| if self._running: | |
| return | |
| self._cap = cv2.VideoCapture(self.camera_index) | |
| if not self._cap.isOpened(): | |
| raise RuntimeError(f"Could not open camera index {self.camera_index}.") | |
| self._pipeline.probe_depth(self._cap) | |
| self._running = True | |
| self._thread = threading.Thread(target=self._loop, daemon=True) | |
| self._thread.start() | |
| def stop(self) -> None: | |
| self._running = False | |
| if self._thread is not None: | |
| self._thread.join(timeout=2.0) | |
| if self._cap is not None: | |
| self._cap.release() | |
| self._pipeline.close() | |
| # -- worker loop ---------------------------------------------------------- | |
| def _loop(self) -> None: | |
| last = time.time() | |
| while self._running: | |
| ok, frame = self._cap.read() | |
| if not ok: | |
| time.sleep(0.01) | |
| continue | |
| frame = cv2.flip(frame, 1) # mirror, like a selfie view | |
| result = self._pipeline.process(frame) | |
| self._depth_source = result.depth_source | |
| if result.recognition is not None: | |
| self._append_transcript(result.recognition) | |
| self._overlay_hud(result.annotated) | |
| ok, buf = cv2.imencode(".jpg", result.annotated, [int(cv2.IMWRITE_JPEG_QUALITY), 80]) | |
| if ok: | |
| with self._frame_lock: | |
| self._latest_jpeg = buf.tobytes() | |
| now = time.time() | |
| self._fps = 0.9 * self._fps + 0.1 * (1.0 / max(now - last, 1e-3)) | |
| last = now | |
| def _overlay_hud(self, frame: np.ndarray) -> None: | |
| rec = self._pipeline.recognizer | |
| status = "REC " + (rec._record_label or "") if rec.is_recording else "" | |
| text = f"depth:{self._depth_source} fps:{self._fps:0.0f} {status}" | |
| cv2.putText( | |
| frame, text, (10, 24), cv2.FONT_HERSHEY_SIMPLEX, 0.6, | |
| (0, 0, 0), 4, cv2.LINE_AA, | |
| ) | |
| cv2.putText( | |
| frame, text, (10, 24), cv2.FONT_HERSHEY_SIMPLEX, 0.6, | |
| (0, 255, 128), 1, cv2.LINE_AA, | |
| ) | |
| # -- shared state accessors ---------------------------------------------- | |
| def _append_transcript(self, rec: Recognition) -> None: | |
| self._transcript_seq += 1 | |
| with self._frame_lock: | |
| self._transcript.append( | |
| { | |
| "seq": self._transcript_seq, | |
| "gloss": rec.gloss, | |
| "confidence": round(rec.confidence, 3), | |
| "source": rec.source, | |
| "t": rec.timestamp, | |
| } | |
| ) | |
| def latest_jpeg(self) -> Optional[bytes]: | |
| with self._frame_lock: | |
| return self._latest_jpeg | |
| def transcript(self, since: int = 0) -> List[dict]: | |
| with self._frame_lock: | |
| return [e for e in self._transcript if e["seq"] > since] | |
| def clear_transcript(self) -> None: | |
| with self._frame_lock: | |
| self._transcript.clear() | |
| # -- recording controls (proxied to recognizer) -------------------------- | |
| def start_recording(self, label: str) -> None: | |
| self._pipeline.recognizer.start_recording(label) | |
| def stop_recording(self) -> Optional[str]: | |
| return self._pipeline.recognizer.stop_recording() | |
| def cancel_recording(self) -> None: | |
| self._pipeline.recognizer.cancel_recording() | |
| def template_counts(self) -> dict: | |
| return self._pipeline.recognizer.template_counts() | |
| def delete_gloss(self, gloss: str) -> bool: | |
| return self._pipeline.recognizer.delete_gloss(gloss) | |
| def status(self) -> dict: | |
| rec = self._pipeline.recognizer | |
| return { | |
| "depth_source": self._depth_source, | |
| "depth_reason": self._pipeline.depth.reason, | |
| "fps": round(self._fps, 1), | |
| "recording": rec.is_recording, | |
| "vocabulary": rec.vocabulary, | |
| } | |