diff --git a/src/om1_vlm/anonymizationSys/face_recog_stream/config.py b/src/om1_vlm/anonymizationSys/face_recog_stream/config.py index 0c4ff44..34330c6 100644 --- a/src/om1_vlm/anonymizationSys/face_recog_stream/config.py +++ b/src/om1_vlm/anonymizationSys/face_recog_stream/config.py @@ -120,16 +120,16 @@ # - Used in: face_tracker.FaceTracker(max_recog_attempts=...) # - Hot-tunable: NO -TRACK_BUFFER = 30 +TRACK_BUFFER = 60 # BoTSORT frames to keep a 'lost' track alive (waiting for re-detection). # Higher = better at handling brief occlusions but increases the chance of # track-id mixups when two people swap positions. # - Used in: face_tracker.FaceTracker(track_buffer=...) # - Hot-tunable: NO -# - Frame-count, scales with FPS: 30 frames @ 15fps = 2 seconds of occlusion +# - Frame-count, scales with FPS: 60 frames @ 30fps = 2 seconds of occlusion # tolerance. If you change CAMERA_FPS, scale this accordingly to maintain # the same wall-clock duration. -# - Example: at 15fps, a face occluded for <2s gets the same track_id back +# - Example: at 30fps, a face occluded for <2s gets the same track_id back # via BoTSORT's Kalman prediction. After 2s, occlusion timeout fires and # a new track_id is assigned on re-detection. @@ -220,18 +220,19 @@ # walk-throughs enroll: raise to 1.5s. If robot feels slow: lower # to 0.7-0.8s. -AUTO_ENROLL_MIN_FACE_PX = 26 +AUTO_ENROLL_MIN_FACE_PX = 65 # Minimum bbox short side for an auto-enroll sample. Below this, samples # are dropped — too small to produce reliable embeddings. -# Sized for 640×480 camera (Unitree G1 default). 30 px short side ≈ 6.25% -# of frame height, equivalent to a person ~2-3m from the camera at ~60° field of view. +# Sized for the current 1920×1080 camera resolution (~87% of +# SELFIE_MIN_FACE_PX — see the "slightly looser" note below), equivalent to +# a person ~2-3m from the camera at ~60° field of view. -# At higher camera resolutions, scale up to maintain the same physical -# distance threshold: -# - 640×480 → 30 px -# - 1280×720 → 50 px -# - 1920×1080 → 75 px +# At other camera resolutions, scale to maintain the same physical distance +# threshold: +# - 640×480 → 26 px +# - 1280×720 → 43 px +# - 1920×1080 → 65 px (current) # - Used in: auto_enroller.AutoEnroller(min_face_pixels=...) # - CLI: --auto-enroll-min-face-px @@ -415,24 +416,24 @@ # - cfg key: "selfie_max_samples" # - Hot-tunable: YES -SELFIE_MIN_ENGAGEMENT = 0.0025 +SELFIE_MIN_ENGAGEMENT = 0.0023 # Engagement score floor below which a face isn't considered a selfie # target. score = (bbox_area / frame_area) × frontality, in [0, 1]. -# Calibrated for 640×480 camera + SELFIE_MIN_FACE_PX=30 +# Calibrated for 1920×1080 camera + SELFIE_MIN_FACE_PX=75 # --------------------------------------------------------- # The minimum sensible engagement is the score of a barely-passing face: -# 30 px frontal face → score = (30×30) / (640×480) × 1.0 = 0.00293. +# 75 px frontal face → score = (75×75) / (1920×1080) × 1.0 = 0.00271. -# 0.0025 sits just below this with a small margin to absorb detection +# 0.0023 sits just below this with a small margin to absorb detection # box wobble (1-2 px jitter can shrink bbox area by ~10%). # Behavior at this threshold: -# - 30 px frontal face → 0.00293 ✓ pass -# - 30 px ¾-profile (f=0.85) → 0.00249 ✗ borderline -# - 25 px frontal face → 0.00203 ✗ too small -# - 40 px slight side (f=0.5) → 0.00260 ✓ pass -# - 70 px frontal at 1m → 0.0159 ✓ pass (typical close selfie) +# - 75 px frontal face → 0.00271 ✓ pass +# - 75 px ¾-profile (f=0.85) → 0.00231 ✓ borderline +# - 62 px frontal face → 0.00185 ✗ too small +# - 100 px slight side (f=0.5) → 0.00241 ✓ pass +# - 175 px frontal at 1m → 0.01477 ✓ pass (typical close selfie) # If you change CAMERA_WIDTH/HEIGHT or SELFIE_MIN_FACE_PX, recompute: # min_engagement = (min_face_px)² / (W × H) × 0.85 @@ -453,17 +454,18 @@ # - Example: 0.80 means if two faces have engagement scores 0.05 and 0.045, # ratio = 0.9 > 0.80 → ambiguous, skip frame. -SELFIE_MIN_FACE_PX = 30 +SELFIE_MIN_FACE_PX = 75 # Selfie per-frame quality gate: bbox short side in pixels. -# Sized for 640×480 camera (Unitree G1 default). 30 px short side ≈ 6.25% -# of frame height, equivalent to a person ~2-3m from the camera at ~60° field of view. +# Sized for the current 1920×1080 camera resolution. 75 px short side ≈ +# 6.9% of frame height, equivalent to a person ~2-3m from the camera at +# ~60° field of view. -# At higher camera resolutions (e.g. 1280×720) you can raise this to keep -# the same physical distance threshold: +# At other camera resolutions, scale to keep the same physical distance +# threshold: # - 640×480 → 30 px # - 1280×720 → 50 px -# - 1920×1080 → 75 px +# - 1920×1080 → 75 px (current) # - cfg key: "selfie_min_face_px" # - Hot-tunable: YES @@ -616,27 +618,26 @@ # ============================================================================ CAMERA_DEVICE = "/dev/video0" -CAMERA_WIDTH = 640 -CAMERA_HEIGHT = 480 -# Camera resolution. Unitree G1's onboard camera default is 640×480. - -# All bbox-size thresholds (SELFIE_MIN_FACE_PX=30, AUTO_ENROLL_MIN_FACE_PX=30) -# are sized for this resolution. If you bump to a higher resolution camera, -# scale those thresholds proportionally: -# 640×480 → 30 px (current) -# 1280×720 → 50 px -# 1920×1080 → 75 px -# to keep the same effective "person distance" threshold. -CAMERA_FPS = 15 -# Capture frame rate. The Unitree G1 onboard camera runs at 15 fps. +CAMERA_WIDTH = 1920 +CAMERA_HEIGHT = 1080 + + +# All bbox-size thresholds (SELFIE_MIN_FACE_PX, AUTO_ENROLL_MIN_FACE_PX) are +# sized for this resolution. If you change CAMERA_WIDTH/HEIGHT, scale those +# thresholds proportionally to keep the same effective "person distance" +# threshold: +# 640×480 → 30 px +# 1280×720 → 50 px +# 1920×1080 → 75 px (current) +CAMERA_FPS = 30 # All wall-clock-based parameters (recog_interval, min_unknown_sec, # re_identify_interval, stale_track_sec) are independent of this — they # measure seconds, not frames. # Frame-count parameters that DO scale with fps: -# - TRACK_BUFFER (currently 30 = 2s @ 15fps) -# - gc_stale modulo in run.py main loop (15 = ~1s @ 15fps) +# - TRACK_BUFFER (currently 60 = 2s @ 30fps) +# - gc_stale modulo in run.py main loop (30 = ~1s @ 30fps) LOCAL_RTSP_URL = "rtsp://localhost:8554/top_camera" RAW_LOCAL_RTSP_URL = "rtsp://localhost:8554/top_camera_raw" diff --git a/src/om1_vlm/anonymizationSys/face_recog_stream/face_tracker.py b/src/om1_vlm/anonymizationSys/face_recog_stream/face_tracker.py index ec73c38..860e699 100644 --- a/src/om1_vlm/anonymizationSys/face_recog_stream/face_tracker.py +++ b/src/om1_vlm/anonymizationSys/face_recog_stream/face_tracker.py @@ -28,8 +28,6 @@ # results = [TrackResult(track_id=1, raw_name="wendy", tier="confident", ...), ...] """ -from __future__ import annotations - import logging import time from collections import Counter @@ -47,11 +45,6 @@ log = logging.getLogger(__name__) -# --------------------------------------------------------------------------- -# Public dataclasses -# --------------------------------------------------------------------------- - - @dataclass class TrackResult: """Per-track output each frame. @@ -96,11 +89,6 @@ class TrackResult: is_known: bool = False -# --------------------------------------------------------------------------- -# Internal state -# --------------------------------------------------------------------------- - - @dataclass class _TrackIdentity: """Internal identity state accumulated for one BoTSORT track. @@ -160,11 +148,6 @@ class _TrackIdentity: last_confident_uuid: Optional[str] = None -# --------------------------------------------------------------------------- -# FaceTracker -# --------------------------------------------------------------------------- - - class FaceTracker: """BoTSORT face tracker with tier-aware recognition and multi-frame voting. @@ -281,10 +264,6 @@ def __init__( # Current frame faces (for status queries) self._current_faces: list = [] - # ------------------------------------------------------------------ - # Setup / runtime config - # ------------------------------------------------------------------ - @staticmethod def _init_tracker(track_buffer: int, det_conf: float = 0.5): """Initialize BoTSORT with thresholds aligned to detection confidence.""" @@ -446,10 +425,6 @@ def set_sim_thr(self, thr: float) -> None: """Back-compat shim — prefer ``set_thresholds(sim_thr=thr)``.""" self.sim_thr = float(thr) - # ------------------------------------------------------------------ - # Main per-frame entry point - # ------------------------------------------------------------------ - def update( self, frame: np.ndarray, @@ -485,6 +460,7 @@ def update( self._active_ids = set() results: List[TrackResult] = [] need_recog: List[Tuple[int, np.ndarray, Optional[np.ndarray], float]] = [] + vvad_track_boxes: List[Tuple[int, Tuple[float, float, float, float]]] = [] for track in tracks: x1, y1, x2, y2 = map(int, track[:4]) @@ -498,21 +474,9 @@ def update( self._active_ids.add(track_id) - # Feed the speaking-detector buffer. Pad the (tight) SCRFD box so the - # crop keeps the full mouth/chin context the VVAD model expects; - # clipping the mouth squashes the speaking score. Recognition still - # uses the original tight box below. if self.vvad is not None: - bw, bh = x2 - x1, y2 - y1 - px = int(bw * 0.20) - pyt = int(bh * 0.15) # a little on top - pyb = int(bh * 0.30) # more on bottom (mouth/chin) - cx1, cy1 = max(0, x1 - px), max(0, y1 - pyt) - cx2, cy2 = min(W, x2 + px), min(H, y2 + pyb) - self.vvad.push( - track_id, - frame[cy1:cy2, cx1:cx2], - kps=track_det_map.get(track_id, {}).get("kps"), + vvad_track_boxes.append( + (track_id, (float(x1), float(y1), float(x2), float(y2))) ) # Get or create identity state @@ -581,6 +545,7 @@ def update( self._cleanup_stale() if self.vvad is not None: + self.vvad.push_frame(frame, vvad_track_boxes) self.vvad.evict(self._active_ids) # Build the by-area sorted face list used by /who and unknown capture @@ -588,10 +553,6 @@ def update( return results - # ------------------------------------------------------------------ - # Re-identify / display - # ------------------------------------------------------------------ - def _maybe_reset_for_reidentify(self, ident: _TrackIdentity, now: float) -> None: """Reset votes if it's been a while since last recognition attempt. @@ -808,10 +769,6 @@ def _build_current_faces(self, results: List[TrackResult]) -> list: ) return sorted(faces, key=lambda f: f["area"], reverse=True) - # ------------------------------------------------------------------ - # BoTSORT plumbing - # ------------------------------------------------------------------ - def _run_tracker(self, dets: np.ndarray, frame: np.ndarray) -> np.ndarray: """Run BoTSORT. Returns (M, 7) [x1,y1,x2,y2,track_id,conf,cls].""" if dets is None or len(dets) == 0: @@ -872,10 +829,6 @@ def _match_tracks_to_dets( return result - # ------------------------------------------------------------------ - # Recognition - # ------------------------------------------------------------------ - def _run_recognition_batch( self, frame: np.ndarray, @@ -1208,10 +1161,6 @@ def _set_unknown(self, ident: _TrackIdentity, now: float, *, reason: str) -> Non ident.unknown_since = now log.debug("Track %d: unknown (%s)", ident.track_id, reason) - # ------------------------------------------------------------------ - # Cleanup / introspection - # ------------------------------------------------------------------ - def _cleanup_stale(self) -> None: """Drop identity state for tracks BoTSORT no longer reports.""" stale = [tid for tid in self._identities if tid not in self._active_ids] diff --git a/src/om1_vlm/anonymizationSys/face_recog_stream/mediapipe_mesh.py b/src/om1_vlm/anonymizationSys/face_recog_stream/mediapipe_mesh.py new file mode 100644 index 0000000..d922e71 --- /dev/null +++ b/src/om1_vlm/anonymizationSys/face_recog_stream/mediapipe_mesh.py @@ -0,0 +1,76 @@ +from typing import List, Optional + +import cv2 +import mediapipe as mp +import numpy as np + + +class MediaPipeFaceMesh: + """Wrapper around MediaPipe Face Mesh for detecting face landmarks.""" + + def __init__( + self, + max_num_faces: int = 1, + min_detection_confidence: float = 0.5, + min_tracking_confidence: float = 0.5, + refine_landmarks: bool = True, + ) -> None: + self._mp = mp + self._mesh = mp.solutions.face_mesh.FaceMesh( # type: ignore + max_num_faces=max_num_faces, + refine_landmarks=refine_landmarks, + min_detection_confidence=min_detection_confidence, + min_tracking_confidence=min_tracking_confidence, + ) + + def __call__(self, frame_bgr: np.ndarray) -> Optional[np.ndarray]: + """ + Detect the first face in a BGR frame and return its landmarks. + + Parameters + ---------- + frame_bgr : np.ndarray + Input image in BGR format (as read by OpenCV). + + Returns + ------- + Optional[np.ndarray] + Detected face represented as an array of shape (N, 2) containing the + (x, y) pixel coordinates of the landmarks, or None if no face is detected. + """ + faces = self.all(frame_bgr) + return faces[0] if faces else None + + def all(self, frame_bgr: np.ndarray) -> List[np.ndarray]: + """ + Detect all faces in a BGR frame and return their landmarks. + + Parameters + ---------- + frame_bgr : np.ndarray + Input image in BGR format (as read by OpenCV). + + Returns + ------- + List[np.ndarray] + List of detected faces, each represented as an array of shape (N, 2) + containing the (x, y) pixel coordinates of the landmarks. + """ + h, w = frame_bgr.shape[:2] + rgb = cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB) + rgb.flags.writeable = False + res = self._mesh.process(rgb) + if not res.multi_face_landmarks: + return [] + + out = [] + for face in res.multi_face_landmarks: + pts = np.array( + [[p.x * w, p.y * h] for p in face.landmark], dtype=np.float64 + ) + out.append(pts) + return out + + def close(self) -> None: + """Release resources.""" + self._mesh.close() diff --git a/src/om1_vlm/anonymizationSys/face_recog_stream/mouth_features.py b/src/om1_vlm/anonymizationSys/face_recog_stream/mouth_features.py new file mode 100644 index 0000000..50339f0 --- /dev/null +++ b/src/om1_vlm/anonymizationSys/face_recog_stream/mouth_features.py @@ -0,0 +1,83 @@ +from dataclasses import dataclass + +import numpy as np + +UPPER_INNER_LIP = 13 +LOWER_INNER_LIP = 14 +LEFT_MOUTH_CORNER = 78 +RIGHT_MOUTH_CORNER = 308 + +LEFT_EYE_OUTER = 33 +RIGHT_EYE_OUTER = 263 + +NOSE_TIP = 1 +CHIN = 152 + + +@dataclass(frozen=True) +class LandmarkScheme: + """Which landmark indices to read. Defaults to MediaPipe FaceMesh.""" + + upper_inner_lip: int = UPPER_INNER_LIP + lower_inner_lip: int = LOWER_INNER_LIP + left_mouth_corner: int = LEFT_MOUTH_CORNER + right_mouth_corner: int = RIGHT_MOUTH_CORNER + left_eye: int = LEFT_EYE_OUTER + right_eye: int = RIGHT_EYE_OUTER + + +@dataclass(frozen=True) +class MouthFeature: + """A single frame's scale-normalized mouth measurements.""" + + aperture: float # inner-lip vertical gap / inter-ocular distance + width_ratio: float # mouth width / inter-ocular distance + scale_px: float # inter-ocular distance in pixels (proxy for face size) + valid: bool # False if the face was too small / degenerate to trust + + +def _dist(a: np.ndarray, b: np.ndarray) -> float: + return float(np.linalg.norm(a - b)) + + +def extract_mouth_feature( + landmarks: np.ndarray, + scheme: LandmarkScheme = LandmarkScheme(), + min_scale_px: float = 12.0, +) -> MouthFeature: + """Compute a scale-normalized mouth feature from one frame of landmarks. + + Parameters + ---------- + landmarks: + Array of shape ``(N, 2)`` or ``(N, 3)`` in pixel coordinates. Only the + first two columns (x, y) are used. + scheme: + Landmark index scheme. Defaults to MediaPipe FaceMesh. + min_scale_px: + If the inter-ocular distance is below this many pixels the face is too + small/far to yield a reliable aperture, and we flag the frame invalid. + """ + pts = np.asarray(landmarks, dtype=np.float64)[:, :2] + + eye_l = pts[scheme.left_eye] + eye_r = pts[scheme.right_eye] + scale = _dist(eye_l, eye_r) + + if not np.isfinite(scale) or scale < min_scale_px: + return MouthFeature(0.0, 0.0, scale, valid=False) + + upper = pts[scheme.upper_inner_lip] + lower = pts[scheme.lower_inner_lip] + corner_l = pts[scheme.left_mouth_corner] + corner_r = pts[scheme.right_mouth_corner] + + aperture = _dist(upper, lower) / scale + width = _dist(corner_l, corner_r) / scale + + return MouthFeature( + aperture=aperture, + width_ratio=width, + scale_px=scale, + valid=True, + ) diff --git a/src/om1_vlm/anonymizationSys/face_recog_stream/run.py b/src/om1_vlm/anonymizationSys/face_recog_stream/run.py index 316e21c..8dce155 100644 --- a/src/om1_vlm/anonymizationSys/face_recog_stream/run.py +++ b/src/om1_vlm/anonymizationSys/face_recog_stream/run.py @@ -153,6 +153,7 @@ from .sample_refresh import SampleRefreshManager from .scrfd import TRTSCRFD from .vvad_overlay import draw as draw_speaking +from .vvad_overlay import flag as flag_speaking from .who_tracker import WhoTracker, match_falls_to_faces from .yolo_pose import TRTYOLOPose @@ -263,7 +264,6 @@ def main() -> None: script_dir = os.path.dirname(os.path.abspath(__file__)) engines_dir = os.path.join(script_dir, "..", "engine") - models_dir = os.path.join(script_dir, "..", "model") platform_prefix = get_platform_prefix() scrfd_name = f"{platform_prefix}_scrfd_10g.engine" @@ -274,7 +274,6 @@ def main() -> None: default_arc_engine = os.path.join(engines_dir, arc_name) default_gallery = os.path.join(script_dir, "..", "gallery") default_pose_engine = os.path.join(engines_dir, pose_name) - default_vvad_model = os.path.join(models_dir, "vvad_face.onnx") ap = argparse.ArgumentParser( "Jetson real-time detection + recognition + blur + RTSP + HTTP" @@ -496,7 +495,7 @@ def main() -> None: "--width", type=int, default=config.CAMERA_WIDTH, - help=f"Capture width. Default {config.CAMERA_WIDTH} (Unitree G1 onboard camera).", + help=f"Capture width. Default {config.CAMERA_WIDTH} (pass 640 for the Unitree G1 onboard camera).", ) ap.add_argument( "--height", @@ -553,11 +552,9 @@ def main() -> None: ) ap.add_argument( - "--vvad-model", - nargs="?", - default=None, - const=default_vvad_model, - help="VVAD ONNX model path; bare flag uses /vvad_face.onnx", + "--vvad", + action="store_true", + help="Enable visual speaking detection (MediaPipe FaceMesh + mouth-aperture VAD).", ) ap.add_argument("--vvad-speaking-thr", type=float, default=0.5) ap.add_argument( @@ -996,20 +993,20 @@ def run_job_sync(fn): face_tracker=face_tracker, ) - # --- vision-only speaking-face scorer (VVAD) ------------------------- vvad = None - if getattr(args, "vvad_model", None): + if getattr(args, "vvad", False): try: from .vvad_speaker import VVADScorer vvad = VVADScorer( - args.vvad_model, speaking_thr=float(args.vvad_speaking_thr), buffer_sec=float(args.vvad_buffer_sec), - use_cuda=False, - ) # CPU build on Jetson + fps=float(cap.fps or args.fps), + ) + if face_tracker is not None: face_tracker.vvad = vvad + http_api.vvad = vvad http_api.vvad_shadow = not bool(args.vvad_active) logger.info("VVAD enabled (shadow=%s)", http_api.vvad_shadow) @@ -1100,9 +1097,9 @@ def handle_sigint(_sig, _frame): frame_state.kpss = None if kpss is None else kpss.copy() frame_state.last_ts = time.monotonic() - # ---- Track-based recognition (BoTSORT + low-freq AdaFace) ---- names: List[Optional[str]] = [] known_mask: List[bool] = [] + track_results = [] if face_tracker is not None: with gal_lock: @@ -1227,7 +1224,7 @@ def handle_sigint(_sig, _frame): # Update who tracker WITH tier + fall info who.update_now(names_for_tracking, tiers=tiers, fall_infos=fall_infos) - # AutoEnroller maintenance, every ~1 second (15 frames @ 15fps): + # AutoEnroller maintenance, every ~1 second # 1. try_commit_pending — proactively commits buffers whose # gates have all passed since the last observe() call. # Without this, a track that went status="unknown" with a @@ -1236,7 +1233,10 @@ def handle_sigint(_sig, _frame): # 2. gc_stale — drops buffers for tracks BoTSORT stopped # reporting. # Both ops are cheap (small dict scans). - if auto_enroller is not None and total % 15 == 0: + if ( + auto_enroller is not None + and total % max(1, int(round(cap.fps or args.fps))) == 0 + ): auto_enroller.try_commit_pending() auto_enroller.gc_stale() @@ -1301,7 +1301,18 @@ def handle_sigint(_sig, _frame): draw_names=args.draw_names, ) - frame = draw_speaking(frame) # brief "speaking" badge (no-op when none) + if vvad is not None and track_results: + live_status = [ + ( + tuple(float(v) for v in tr.bbox), + vvad.is_speaking(tr.track_id), + vvad.score_track(tr.track_id), + ) + for tr in track_results + ] + flag_speaking(live_status, ttl=0.5) + + frame = draw_speaking(frame) # Stats dt_ms = (time.perf_counter() - t_start) * 1000.0 diff --git a/src/om1_vlm/anonymizationSys/face_recog_stream/speaking_vad.py b/src/om1_vlm/anonymizationSys/face_recog_stream/speaking_vad.py new file mode 100644 index 0000000..fb54d0a --- /dev/null +++ b/src/om1_vlm/anonymizationSys/face_recog_stream/speaking_vad.py @@ -0,0 +1,132 @@ +from collections import deque +from dataclasses import dataclass, field +from typing import Deque, Optional + +import numpy as np + +from .mouth_features import MouthFeature + + +@dataclass +class DetectorConfig: + """Configuration for :class:`SpeakingDetector`.""" + + fps: float = 30.0 + window_sec: float = 1.0 + + band_low_hz: float = 2.0 + band_high_hz: float = 8.0 + + score_on: float = 0.35 + score_off: float = 0.22 + + min_on_frames: int = 3 + min_off_frames: int = 5 + + max_invalid_frames: int = 5 + + +@dataclass +class DetectorState: + """State of a :class:`SpeakingDetector`.""" + + speaking: bool = False + score: float = 0.0 + _buf: Deque[float] = field(default_factory=deque) + _pending: bool = False + _pending_count: int = 0 + _invalid_count: int = 0 + + +class SpeakingDetector: + """Streaming visual VAD. Feed it one :class:`MouthFeature` per frame.""" + + def __init__(self, config: Optional[DetectorConfig] = None) -> None: + self.cfg = config or DetectorConfig() + self.state = DetectorState() + self._window_n = max(8, int(round(self.cfg.window_sec * self.cfg.fps))) + self.state._buf = deque(maxlen=self._window_n) + + def reset(self) -> None: + """Reset the detector state to its initial values.""" + self.state = DetectorState() + self.state._buf = deque(maxlen=self._window_n) + + def update(self, feature: Optional[MouthFeature]) -> DetectorState: + """Advance one frame. Pass ``None`` or an invalid feature when no + trustworthy face was found this frame. + """ + st = self.state + cfg = self.cfg + + if feature is None or not feature.valid: + st._invalid_count += 1 + if st._invalid_count >= cfg.max_invalid_frames: + st._buf.clear() + self._commit(False) + + st.score = 0.0 + return st + + st._invalid_count = 0 + st._buf.append(feature.aperture) + st.score = self._speaking_score() + self._apply_hysteresis(st.score) + + return st + + def _speaking_score(self) -> float: + buf = self.state._buf + # Need enough samples for a meaningful spectrum. + if len(buf) < self._window_n // 2: + return 0.0 + + x = np.asarray(buf, dtype=np.float64) + x = x - x.mean() + + std = float(x.std()) + if std < 1e-6: + return 0.0 + + freqs = np.fft.rfftfreq(len(x), d=1.0 / self.cfg.fps) + power = np.abs(np.fft.rfft(x)) ** 2 + + total = float(power.sum()) + if total <= 0: + return 0.0 + + band = (freqs >= self.cfg.band_low_hz) & (freqs <= self.cfg.band_high_hz) + band_ratio = float(power[band].sum()) / total # fraction in speech band + + amp_gate = float(np.tanh(std / 0.03)) + + return band_ratio * amp_gate + + def _apply_hysteresis(self, score: float) -> None: + st = self.state + cfg = self.cfg + target = st.speaking + if not st.speaking and score >= cfg.score_on: + target = True + elif st.speaking and score <= cfg.score_off: + target = False + + if target == st.speaking: + st._pending_count = 0 + return + + if target == st._pending: + st._pending_count += 1 + else: + st._pending = target + st._pending_count = 1 + + need = cfg.min_on_frames if target else cfg.min_off_frames + if st._pending_count >= need: + self._commit(target) + + def _commit(self, speaking: bool) -> None: + st = self.state + st.speaking = speaking + st._pending = speaking + st._pending_count = 0 diff --git a/src/om1_vlm/anonymizationSys/face_recog_stream/vvad_speaker.py b/src/om1_vlm/anonymizationSys/face_recog_stream/vvad_speaker.py index e60f090..45723a4 100644 --- a/src/om1_vlm/anonymizationSys/face_recog_stream/vvad_speaker.py +++ b/src/om1_vlm/anonymizationSys/face_recog_stream/vvad_speaker.py @@ -1,441 +1,224 @@ -""" -Vision speaking detector. -""" - -from __future__ import annotations - import logging -import os import threading import time from collections import deque from typing import Dict, List, Optional, Tuple -import cv2 import numpy as np -import onnxruntime as ort -from . import head_pose +from .mediapipe_mesh import MediaPipeFaceMesh +from .mouth_features import LandmarkScheme, MouthFeature, extract_mouth_feature +from .speaking_vad import DetectorConfig, SpeakingDetector log = logging.getLogger(__name__) -# landmark model I/O (don't change unless swapping models) -LM_SIZE = 192 # 2d106det input size -LM_SWAP_RB = True # BGR -> RGB -LM_SCALE = 1.0 # pixel scale (2d106det uses raw [0,255]) -LM_MEAN = 0.0 # mean subtracted -N_LANDMARKS = 106 # 2d106det point count -MOUTH_IDX = list(range(52, 72)) # mouth point block (2d106det only) - -# FaceMesh (468) backend: auto-enabled when model outputs 1404 -FACEMESH_N = 468 # point count -FACEMESH_SIZE = 192 # input size -FACEMESH_DIV255 = True # normalize to [0,1]; set False if points look wrong -MP_UP_INNER, MP_LO_INNER = 13, 14 # upper/lower inner-lip centers (opening) -MP_L_CORNER, MP_R_CORNER = 61, 291 # mouth corners (width) - -# signal / windowing -WINDOW_SEC = 1.6 # decision window (s); LOWER = snappier but jumpier/more false fires -BUFFER_SEC = 6.0 # history kept (s); how far back you can look; leave as-is -SLIDE_STEP_SEC = 0.5 # sliding step for long utterances; leave as-is -WINDOW_PAD_SEC = 0.4 # query-span padding; leave as-is -MIN_SAMPLES = 6 # min frames before scoring; LOWER = scores on shorter windows - -# sensitivity / threshold -MAR_STD_FULL = ( - 0.06 # mouth motion that maps to score 1.0; LOWER = higher scores (more sensitive) -) -MIN_CROSS_HZ = 1.0 # min open/close rhythm (crossings/s); LOWER = looser - -# mouth-shape motion gates (reject head turn / profile) -TURN_WIDTH_CV = ( - 0.06 # mouth-width wobble above this = turning -> suppress; RAISE = looser -) -PROFILE_WIDTH_FRAC = ( - 0.90 # width < this x frontal width = profile -> suppress; LOWER = looser -) -GATE_SUPPRESS = 0.6 # how hard a tripped gate cuts the score; RAISE = looser (0.4 = dampen, not kill) ★MASTER - -# head-pose gates (independent of the mouth) -FRONT_MIN = 0.45 # frontality below this = side/down -> suppress; LOWER = looser (allow more tilt) -YAW_TURN_STD = ( - 8.0 # head yaw swing (deg) above this = turning -> suppress; RAISE = looser -) - -# body-sway gate -SWAY_MAX = 0.25 # face-center drift (in face-widths) above this = swaying -> suppress; RAISE = looser - - class VVADScorer: - """Landmark-MAR speaking scorer (interface-compatible with the CNN one).""" + """ + Per-track visual speaking detector. + + push_frame(), evict(), score_track(), score_window(), resolve_speaking(), + resolve_window(), and the `speaking_thr` attribute. + """ def __init__( self, - engine_path: str, speaking_thr: float = 0.5, - window_sec: float = WINDOW_SEC, - buffer_sec: float = BUFFER_SEC, - min_frames: int = 0, - use_cuda: bool = False, - intra_op_threads: int = 2, + window_sec: float = 1.0, + buffer_sec: float = 6.0, + min_window_samples: int = 4, + fps: float = 15.0, ): - if ort is None: - raise RuntimeError("onnxruntime not installed") self.speaking_thr = float(speaking_thr) self.window_sec = float(window_sec) self.buffer_sec = max(float(buffer_sec), self.window_sec) - # min_frames kept for signature compatibility; mapped to min samples. - self.min_samples = int(min_frames) if min_frames > 0 else MIN_SAMPLES + self.fps = float(fps) + self._min_window_samples = int(min_window_samples) - self._buf: Dict[int, deque] = {} # track_id -> deque[(ts, mar, width)] - self._lock = threading.Lock() - # set VVAD_DEBUG=1 in the environment to log per-track score breakdowns - self.debug = bool(os.environ.get("VVAD_DEBUG")) - - so = ort.SessionOptions() - if intra_op_threads > 0: - so.intra_op_num_threads = int(intra_op_threads) - providers = ( - ["CUDAExecutionProvider", "CPUExecutionProvider"] - if use_cuda - else ["CPUExecutionProvider"] - ) - self._sess = ort.InferenceSession( - engine_path, sess_options=so, providers=providers - ) - self._in_name = self._sess.get_inputs()[0].name - # Auto-detect backend from the model's output size: 468*3=1404 -> FaceMesh. - _osh = self._sess.get_outputs()[0].shape - _onumel = 1 - for _d in _osh: - if isinstance(_d, int) and _d > 0: - _onumel *= _d - self.backend = "facemesh" if _onumel == FACEMESH_N * 3 else "2d106det" - # facemesh models come in NHWC ([1,192,192,3]) or NCHW ([1,3,192,192]). - _ish = self._sess.get_inputs()[0].shape - self._facemesh_nhwc = len(_ish) == 4 and _ish[-1] == 3 log.info( - "VVAD landmark backend = %s (output numel=%s%s)", - self.backend, - _onumel, - ", NHWC" if (self.backend == "facemesh" and self._facemesh_nhwc) else "", - ) - log.info( - "VVAD(landmark-MAR) loaded: %s (input=%s, thr=%.2f, " - "window=%.1fs, buffer=%.1fs, mouth_idx=%d..%d)", - engine_path, - self._in_name, + "VVAD(speaking_detector) landmark backend = MediaPipeFaceMesh, " + "run per-frame (thr=%.2f, window=%.1fs, buffer=%.1fs, fps=%.1f)", self.speaking_thr, self.window_sec, self.buffer_sec, - MOUTH_IDX[0], - MOUTH_IDX[-1], + self.fps, ) - # landmark inference + MAR - def _landmarks(self, crop_bgr: np.ndarray) -> Optional[np.ndarray]: - """Run the landmark model on a face crop -> (N_LANDMARKS, 2) in a - consistent [0,1]-ish space. Ratios below are invariant to the exact - scaling, so we don't map back to image pixels. - """ - if crop_bgr is None or crop_bgr.size == 0: - return None - h, w = crop_bgr.shape[:2] - if h < 8 or w < 8: - return None - if getattr(self, "backend", "2d106det") == "facemesh": - scale = (1.0 / 255.0) if FACEMESH_DIV255 else 1.0 - img = cv2.resize(crop_bgr, (FACEMESH_SIZE, FACEMESH_SIZE)) - img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB).astype(np.float32) * scale - if self._facemesh_nhwc: - blob = img[None, ...] # (1,192,192,3) NHWC - else: - blob = np.transpose(img, (2, 0, 1))[None] # (1,3,192,192) NCHW - out = self._sess.run(None, {self._in_name: blob})[0] - pred = np.asarray(out, dtype=np.float32).reshape(-1, 3) - if pred.shape[0] < FACEMESH_N: - return None - return pred[:FACEMESH_N, :2] # (468,2); ratios are scale-invariant + self._scheme = LandmarkScheme() + self._lock = threading.Lock() + + self._mesh: Optional[MediaPipeFaceMesh] = None + self._mesh_capacity = 0 + + self._live: Dict[int, SpeakingDetector] = {} + self._hist: Dict[int, deque] = {} - blob = cv2.dnn.blobFromImage( - crop_bgr, - LM_SCALE, - (LM_SIZE, LM_SIZE), - (LM_MEAN, LM_MEAN, LM_MEAN), - swapRB=LM_SWAP_RB, - crop=False, + def _new_detector(self, fps: Optional[float] = None) -> SpeakingDetector: + cfg = DetectorConfig( + fps=float(fps) if fps else self.fps, + window_sec=self.window_sec, ) - out = self._sess.run(None, {self._in_name: blob})[0] - pred = np.asarray(out, dtype=np.float32).reshape(-1, 2) - if pred.shape[0] < N_LANDMARKS: - return None - # 2d106det returns coords in ~[-1,1]; map to [0,1] (scale-invariant use). - return (pred[:N_LANDMARKS] + 1.0) * 0.5 + return SpeakingDetector(cfg) - @staticmethod - def _mouth_metrics(lms: np.ndarray) -> Optional[Tuple[float, float]]: - """(MAR, mouth_width) from the mouth landmark block. + def _get_mesh(self, capacity: int) -> MediaPipeFaceMesh: + capacity = max(capacity, 1) + with self._lock: + if self._mesh is None or self._mesh_capacity < capacity: + if self._mesh is not None: + try: + self._mesh.close() + except Exception: + pass + self._mesh = MediaPipeFaceMesh(max_num_faces=capacity) + self._mesh_capacity = capacity + return self._mesh - MAR is rotation-invariant: corners are the horizontal extremes of the - block; rotate the block so the corner line is horizontal, then take - (vertical extent / corner distance). width = corner distance, which we - also use to detect head turning (it shrinks on profile / while turning). - Only MOUTH_IDX must be correct — corners are found dynamically. + @staticmethod + def _match_faces_to_tracks( + faces: List[np.ndarray], + track_boxes: List[Tuple[int, Tuple[float, float, float, float]]], + ) -> Dict[int, np.ndarray]: """ - try: - if lms.shape[0] >= FACEMESH_N: - # FaceMesh: true inner-lip opening (13-14) over corner width - # (61-291). Rotation/scale/translation invariant (Euclidean - # distances). Much more sensitive to small speech motion. - up, lo = lms[MP_UP_INNER], lms[MP_LO_INNER] - lc, rc = lms[MP_L_CORNER], lms[MP_R_CORNER] - width = float(np.hypot(rc[0] - lc[0], rc[1] - lc[1])) - if width < 1e-9: - return None - vert = float(np.hypot(up[0] - lo[0], up[1] - lo[1])) - return vert / width, width - m = lms[MOUTH_IDX] - except (IndexError, TypeError): - return None - if m.shape[0] < 4: - return None - li = int(np.argmin(m[:, 0])) # left corner = min-x point - ri = int(np.argmax(m[:, 0])) # right corner = max-x point - pL, pR = m[li], m[ri] - d = pR - pL - width = float(np.hypot(d[0], d[1])) - if width < 1e-6: - return None - ang = np.arctan2(d[1], d[0]) # rotate corner line to horizontal - c, s = np.cos(-ang), np.sin(-ang) - rot = np.empty_like(m) - rot[:, 0] = m[:, 0] * c - m[:, 1] * s - rot[:, 1] = m[:, 0] * s + m[:, 1] * c - # Openness = gap between the upper-lip and lower-lip point groups - # (mean of the top half vs bottom half of the rotated y's). Averaging - # each lip is far less noisy than the outer top/bottom extremes, and it - # tracks the inner opening that actually moves during speech. - ys = np.sort(rot[:, 1]) - k = max(1, len(ys) // 2) - height = float(ys[-k:].mean() - ys[:k].mean()) - return height / width, width - - @classmethod - def _mar(cls, lms: np.ndarray) -> Optional[float]: - """MAR only (kept for verify_landmarks.py).""" - r = cls._mouth_metrics(lms) - return None if r is None else r[0] + Assign each detected face's landmarks to its nearest track by + center distance — greedy, globally-nearest-pair-first, each track + claims at most one face. + """ + assigned: Dict[int, np.ndarray] = {} + if not faces or not track_boxes: + return assigned + + centers = [ + ((x1 + x2) / 2.0, (y1 + y2) / 2.0) for _tid, (x1, y1, x2, y2) in track_boxes + ] + pairs = [] # (dist_sq, face_idx, track_idx) + for fi, lm in enumerate(faces): + cx, cy = float(lm[:, 0].mean()), float(lm[:, 1].mean()) + for ti, (bcx, bcy) in enumerate(centers): + d = (cx - bcx) ** 2 + (cy - bcy) ** 2 + pairs.append((d, fi, ti)) + pairs.sort(key=lambda p: p[0]) + + used_faces, used_tracks = set(), set() + for _d, fi, ti in pairs: + if fi in used_faces or ti in used_tracks: + continue + used_faces.add(fi) + used_tracks.add(ti) + assigned[track_boxes[ti][0]] = faces[fi] + return assigned # buffer - def push(self, track_id: int, crop_bgr: np.ndarray, kps=None) -> None: - """Frontality from the 5 keypoints FIRST (cheap). If too off-axis - (< FRONT_MIN) the mouth is unreliable and scores quiet anyway, so skip - the expensive landmark inference entirely and store nothing. Otherwise - landmark the crop, compute MAR+width, and append the sample. + def push_frame( + self, + frame_bgr: np.ndarray, + track_boxes: List[Tuple[int, Tuple[float, float, float, float]]], + ) -> None: + """Feed one frame for ALL currently-visible tracks at once. + + Parameters + ---------- + frame_bgr : the full camera frame (BGR), NOT a per-face crop. + track_boxes : list of (track_id, (x1, y1, x2, y2)) for every track + active this frame. """ - front, yaw = 1.0, 0.0 # defaults if we have no kps - cx = cy = sc = 0.0 # face center + size (image px) for sway - if kps is not None and head_pose is not None: - try: - ang = head_pose.head_pose_angles(kps) - if ang is not None: - yaw = float(ang[0]) - front = float(head_pose.frontality_from_angles(*ang)) - except Exception: - pass - - if front < FRONT_MIN: # too off-axis -> skip model, = quiet + if frame_bgr is None or frame_bgr.size == 0: return - if kps is not None: - try: - k = np.asarray(kps, dtype=np.float32).reshape(-1, 2) - if k.shape[0] >= 2: - cx = float(k[:, 0].mean()) - cy = float(k[:, 1].mean()) - sc = float(np.hypot(k[0, 0] - k[1, 0], k[0, 1] - k[1, 1])) - except Exception: - pass + faces: List[np.ndarray] = [] + if track_boxes: + mesh = self._get_mesh(len(track_boxes)) + faces = mesh.all(frame_bgr) + assigned = self._match_faces_to_tracks(faces, track_boxes) - lms = self._landmarks(crop_bgr) # expensive; only when frontal - if lms is None: - return - mm = self._mouth_metrics(lms) - if mm is None: - return - mar, width = mm now = time.time() - with self._lock: - dq = self._buf.get(track_id) - if dq is None: - dq = deque() - self._buf[track_id] = dq - dq.append( - (now, float(mar), float(width), float(front), float(yaw), cx, cy, sc) - ) - cutoff = now - self.buffer_sec - while dq and dq[0][0] < cutoff: - dq.popleft() + for track_id, _bbox in track_boxes: + lm = assigned.get(track_id) + feat: Optional[MouthFeature] = None + if lm is not None: + raw_feat = extract_mouth_feature(lm, scheme=self._scheme) + if raw_feat.valid: + feat = raw_feat + + with self._lock: + det = self._live.get(track_id) + if det is None: + det = self._new_detector() + self._live[track_id] = det + det.update(feat) + + hist = self._hist.get(track_id) + if hist is None: + hist = deque() + self._hist[track_id] = hist + hist.append((now, feat)) + cutoff = now - self.buffer_sec + while hist and hist[0][0] < cutoff: + hist.popleft() def evict(self, alive_track_ids=None) -> None: - """Drop buffered samples for tracks not in alive_track_ids (or all if None).""" + """Drop buffered state for tracks not in alive_track_ids (or all if None).""" with self._lock: if alive_track_ids is None: - self._buf.clear() + self._live.clear() + self._hist.clear() + if self._mesh is not None: + try: + self._mesh.close() + except Exception: + pass + self._mesh = None + self._mesh_capacity = 0 return alive = set(int(t) for t in alive_track_ids) - for tid in [t for t in self._buf if t not in alive]: - del self._buf[tid] + for tid in [t for t in self._live if t not in alive]: + del self._live[tid] + for tid in [t for t in self._hist if t not in alive]: + del self._hist[tid] - def _samples_in( - self, track_id: int, t0: float, t1: float - ) -> List[Tuple[float, ...]]: + def score_track(self, track_id: int) -> Optional[float]: + """Current ("right now") speaking score for this track.""" with self._lock: - dq = self._buf.get(track_id) - if not dq: - return [] - return [s for s in dq if t0 <= s[0] <= t1] - - def _recent(self, track_id: int) -> List[Tuple[float, ...]]: - now = time.time() - return self._samples_in(track_id, now - self.window_sec, now) + det = self._live.get(track_id) + return float(det.state.score) if det is not None else None - def _ref_width(self, track_id: int) -> float: - """This track's frontal mouth width = a high percentile of its buffered - widths (robust to the odd bad-landmark frame). Used to detect profile. - """ + def is_speaking(self, track_id: int) -> bool: + """Current ("right now") hysteresis-debounced speaking state.""" with self._lock: - dq = self._buf.get(track_id) - if not dq: - return 0.0 - ws = sorted(s[2] for s in dq) - return float(ws[int(0.9 * (len(ws) - 1))]) if ws else 0.0 - - # scoring - def _score_samples( - self, - samples: List[Tuple[float, ...]], - ref_width: float = 0.0, - tid: Optional[int] = None, - ) -> Optional[float]: - """MAR time series -> speaking score in [0,1]. - - amplitude = std of the detrended MAR (how much the mouth moves); - rhythm = baseline crossings per second (speech-like open/close); - then head-motion rejection using mouth width: - * turning (width CV high) -> suppress - * profile (width << ref) -> suppress - so a head turn no longer mimics speaking, and a still frontal talker - keeps its score. - """ - if len(samples) < self.min_samples: - return None - ts = np.array([s[0] for s in samples], dtype=np.float64) - v = np.array([s[1] for s in samples], dtype=np.float64) - w = np.array([s[2] for s in samples], dtype=np.float64) - fr = np.array([s[3] if len(s) > 3 else 1.0 for s in samples], dtype=np.float64) - yw = np.array([s[4] if len(s) > 4 else 0.0 for s in samples], dtype=np.float64) - cx = np.array([s[5] if len(s) > 5 else 0.0 for s in samples], dtype=np.float64) - cy = np.array([s[6] if len(s) > 6 else 0.0 for s in samples], dtype=np.float64) - sc = np.array([s[7] if len(s) > 7 else 0.0 for s in samples], dtype=np.float64) - dur = float(ts[-1] - ts[0]) - if dur < 1e-3: - return None - base = np.median(v) - x = v - base - amp = float(np.std(x)) - sign = np.sign(x) - sign[sign == 0] = 1.0 - crossings = int(np.count_nonzero(np.diff(sign) != 0)) - cross_hz = crossings / dur - - mean_w = float(np.mean(w)) - width_cv = float(np.std(w) / mean_w) if mean_w > 1e-6 else 1.0 - width_ratio = mean_w / ref_width if ref_width > 1e-6 else 1.0 - front_med = float(np.median(fr)) - yaw_std = float(np.std(yw)) - med_sc = float(np.median(sc)) - sway = ( - float(np.hypot(np.std(cx), np.std(cy)) / med_sc) if med_sc > 1e-6 else 0.0 - ) - - score = float(np.clip(amp / MAR_STD_FULL, 0.0, 1.0)) - why = "" - if cross_hz < MIN_CROSS_HZ: - score *= 0.25 # motion without speech rhythm (yawn/bite) - why += "norhythm " - if width_cv > TURN_WIDTH_CV: - score *= GATE_SUPPRESS # mouth width swinging => head turning - why += "turning " - if width_ratio < PROFILE_WIDTH_FRAC: - score *= GATE_SUPPRESS # mouth much narrower than frontal => profile - why += "profile " - if front_med < FRONT_MIN: - score *= GATE_SUPPRESS # head turned / side-on => can't judge mouth - why += "lowfront " - if yaw_std > YAW_TURN_STD: - score *= GATE_SUPPRESS # yaw swinging => head is TURNING right now - why += "yawmove " - if sway > SWAY_MAX: - score *= GATE_SUPPRESS # face sliding in frame => body swaying/walking - why += "sway " - if self.debug: - log.info( - "VVAD tid=%s amp=%.3f cross=%.1fHz wcv=%.2f wratio=%.2f " - "front=%.2f yaw_sd=%.1f sway=%.2f -> %.2f %s", - tid, - amp, - cross_hz, - width_cv, - width_ratio, - front_med, - yaw_std, - sway, - score, - why or "ok", - ) - return score - - def score_track(self, track_id: int) -> Optional[float]: - """Most-recent WINDOW_SEC ('who is talking now' / fallback).""" - return self._score_samples( - self._recent(track_id), self._ref_width(track_id), track_id - ) + det = self._live.get(track_id) + return bool(det.state.speaking) if det is not None else False def score_window( self, track_id: int, win_start: float, win_end: float ) -> Optional[float]: - """Score an explicit [start,end] epoch span. If longer than - WINDOW_SEC, slide a WINDOW_SEC window across it and take the MAX. + """Score an explicit [start, end] epoch span by replaying the + buffered features for that span through a fresh detector (so the + result depends only on that window, independent of live state). """ - t0 = float(win_start) - WINDOW_PAD_SEC - t1 = float(win_end) + WINDOW_PAD_SEC - span = self._samples_in(track_id, t0, t1) - if len(span) < self.min_samples: + with self._lock: + hist = self._hist.get(track_id) + span = ( + [(ts, f) for ts, f in hist if win_start <= ts <= win_end] + if hist + else [] + ) + span = [(ts, f) for ts, f in span if f is not None] + if len(span) < self._min_window_samples: return None - ref_w = self._ref_width(track_id) - if (t1 - t0) <= self.window_sec: - return self._score_samples(span, ref_w, track_id) - best: Optional[float] = None - start = t0 - while start <= t1 - self.window_sec + 1e-6: - sub = [s for s in span if start <= s[0] <= start + self.window_sec] - s = self._score_samples(sub, ref_w, track_id) - if s is not None and (best is None or s > best): - best = s - start += SLIDE_STEP_SEC - # also score the trailing window so the very end isn't missed - tail = [s for s in span if s[0] >= t1 - self.window_sec] - s = self._score_samples(tail, ref_w, track_id) - if s is not None and (best is None or s > best): - best = s - return best + + dur = span[-1][0] - span[0][0] + fps = (len(span) - 1) / dur if dur > 1e-3 else self.fps + det = self._new_detector(fps=fps) + state = None + for _, feat in span: + state = det.update(feat) + return float(state.score) if state is not None else None def resolve_speaking( self, candidate_track_ids: List[int] ) -> Tuple[Optional[int], Dict[int, float]]: - """Score each candidate track and return the top speaker (if above threshold) with all scores.""" + """Score each candidate track (current state) and return the top + speaker (if above threshold) with all scores. + """ scores: Dict[int, float] = {} for tid in candidate_track_ids: s = self.score_track(int(tid))