"""Deterministic offline stand-ins for the model runtimes. They implement the same small runtime interfaces as the real backends, so the operation code around them (frame IO, tracking, masks, encoding) runs for real. Synthetic inputs only: bright shapes on dark backgrounds stand in for faces and marks. """ from __future__ import annotations import threading from typing import Any import cv2 import numpy as np from app.models import Backend, BackendSpec from app.ops.faces import Detection def fake_backend(runtime: Any, model: str = "fake", experimental: bool = False) -> Backend: return Backend( BackendSpec( id=f"fake-{model}", model=model, license="MIT", commercialUse=True, experimental=experimental, source=None, ), lambda _files, _settings: runtime, ) def bright_blobs(frame: np.ndarray, threshold: int = 200) -> list[tuple[int, int, int, int]]: gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) _, binary = cv2.threshold(gray, threshold, 255, cv2.THRESH_BINARY) contours, _ = cv2.findContours(binary, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) boxes = [cv2.boundingRect(c) for c in contours if cv2.contourArea(c) > 20] return sorted(boxes) class FakeFaceDetector: """Every bright blob is a 'face'; landmarks sit at fixed box fractions.""" def detect(self, frame: np.ndarray) -> list[Detection]: result = [] for x, y, w, h in bright_blobs(frame): marks = [(0.3, 0.4), (0.7, 0.4), (0.5, 0.6), (0.35, 0.8), (0.65, 0.8)] result.append(Detection((x, y, w, h), 0.9, [(x + fx * w, y + fy * h) for fx, fy in marks])) return result class FakeEmbedder: model = "fake-embed" dim = 4 def embed(self, frames: list[np.ndarray]) -> np.ndarray: vectors = np.array([[*frame.reshape(-1, 3).mean(axis=0), 1.0] for frame in frames], dtype=np.float32) return vectors / np.linalg.norm(vectors, axis=1, keepdims=True) class FakeClassifier: model = "fake-classifier" def classify(self, frames: list[np.ndarray]) -> list[dict[str, float]]: return [{"violence": float(frame[..., 2].mean() / 255.0), "weapons": 0.0} for frame in frames] class FakeTranscriber: def transcribe( self, _path: str, language: str | None, _cancel: threading.Event ) -> tuple[str | None, list[dict[str, Any]]]: words = [ {"startMs": 0, "endMs": 400, "text": "synthetic", "confidence": 0.9}, {"startMs": 450, "endMs": 900, "text": "tone", "confidence": 0.8}, ] return language or "en", words class FakeSegmenter: """Segments the bright blob in each sampled frame.""" def segment( self, frames: list[str], prompt_index: int, _box: Any, _point: Any, propagate: bool, cancel: threading.Event, ) -> Any: indices = range(len(frames)) if propagate else [prompt_index] for index in indices: if cancel.is_set(): return image = cv2.imread(frames[index if propagate else 0]) gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) yield index, gray > 200, 0.95 class FakeInpainter: """Classical OpenCV inpainting (Telea) instead of LaMa.""" def inpaint(self, image: np.ndarray, mask: np.ndarray) -> np.ndarray: return cv2.inpaint(image, (mask > 127).astype(np.uint8), 5, cv2.INPAINT_TELEA) class BlockingEmbedder(FakeEmbedder): """Holds the GPU slot until released, to exercise queue overflow.""" def __init__(self) -> None: self.entered = threading.Event() self.release = threading.Event() def embed(self, frames: list[np.ndarray]) -> np.ndarray: self.entered.set() self.release.wait(10) return super().embed(frames) def fake_backends() -> dict[str, Backend]: detector = FakeFaceDetector() return { "transcribe": fake_backend(FakeTranscriber(), "fake-whisper"), "detectFaces": fake_backend(detector, "fake-faces"), "trackFaces": fake_backend(detector, "fake-faces"), "faceLandmarks": fake_backend(detector, "fake-faces"), "embedFrames": fake_backend(FakeEmbedder(), "fake-embed"), "classifyContent": fake_backend(FakeClassifier(), "fake-classifier", experimental=True), "segmentPrompted": fake_backend(FakeSegmenter(), "fake-segment", experimental=True), "inpaintVideo": fake_backend(FakeInpainter(), "fake-inpaint", experimental=True), }