Phase 11

src/visualizer.py

@@ -0,0 +1,184 @@
+"""OpenCV overlay rendering."""
+
+from datetime import datetime
+
+import cv2
+import numpy as np
+
+from .config_loader import DisplaySettings, TriggerLineSettings
+from .models import (Detection, FrameInfo, HeadPoseResult, PixelRegion, TrackedPerson, TrackState,
+                     TurnTrackState, VoicePromptTrackState, PersonStateSnapshot)
+from .turn_rules import is_weak_or_higher
+
+
+class Visualizer:
+    """Render detections, experimental geometry, and runtime status."""
+
+    def __init__(self, config: DisplaySettings, warn_fps_threshold: float, stable_frames: int) -> None:
+        self.config = config
+        self.warn_fps_threshold = warn_fps_threshold
+        self.stable_frames = stable_frames
+
+    def draw(
+        self, frame: np.ndarray, detections: list[Detection], info: FrameInfo, region: PixelRegion | None,
+        trigger: TriggerLineSettings, device: str, paused: bool = False, debug: bool = False,
+        tracked_persons: list[TrackedPerson] | None = None, track_states: dict[int, TrackState] | None = None,
+        tracking_counters: dict[str, int] | None = None, tracker_name: str = "disabled",
+        face_results: dict[int, HeadPoseResult] | None = None, face_enabled: bool = False,
+        turn_states: dict[int, TurnTrackState] | None = None, turn_enabled: bool = False,
+        voice_states: dict[int, VoicePromptTrackState] | None = None, voice_mode: str = "disabled",
+        voice_cooldown: float = 0.0,
+        person_states: dict[int, PersonStateSnapshot] | None = None, person_state_enabled: bool = False,
+    ) -> np.ndarray:
+        output = frame.copy()
+        height, width = output.shape[:2]
+        if region and self.config.show_evaluation_region:
+            cv2.rectangle(output, (region.x1, region.y1), (region.x2, region.y2), (255, 180, 0), 2)
+            cv2.putText(output, "Evaluation Region", (region.x1 + 4, max(20, region.y1 + 20)), cv2.FONT_HERSHEY_SIMPLEX, .55, (255, 180, 0), 2)
+        position = trigger.pixel_position(width, height)
+        if position is not None and self.config.show_trigger_line:
+            points = ((position, 0), (position, height - 1)) if trigger.orientation == "vertical" else ((0, position), (width - 1, position))
+            cv2.line(output, *points, (255, 0, 255), 2)
+        tracked_persons = tracked_persons or []
+        track_states = track_states or {}
+        face_results = face_results or {}
+        turn_states = turn_states or {}
+        voice_states = voice_states or {}
+        person_states = person_states or {}
+        if self.config.show_bounding_boxes and not tracked_persons:
+            for detection in detections:
+                b = detection.bbox
+                color = (0, 220, 0) if detection.inside_evaluation_region else (0, 140, 255)
+                cv2.rectangle(output, (b.x1, b.y1), (b.x2, b.y2), color, 2)
+                label = "person"
+                if self.config.show_confidence:
+                    label += f" {detection.confidence:.2f}"
+                (_, text_h), _ = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, .55, 2)
+                label_y = max(text_h + 4, b.y1 - 5)
+                cv2.putText(output, label, (max(0, b.x1), label_y), cv2.FONT_HERSHEY_SIMPLEX, .55, color, 2)
+        for person in tracked_persons:
+            state = track_states.get(person.track_id)
+            inside = state.current_inside_region if state else person.inside_evaluation_region
+            color = (0, 220, 0) if inside else (0, 140, 255)
+            b = person.bbox
+            if self.config.show_bounding_boxes:
+                cv2.rectangle(output, (b.x1, b.y1), (b.x2, b.y2), color, 2)
+            labels: list[str] = []
+            if self.config.show_track_ids:
+                labels.append(f"ID:{person.track_id}")
+            if self.config.show_confidence:
+                labels.append(f"{person.confidence:.2f}")
+            if self.config.show_movement_direction and state:
+                labels.append(state.movement_direction)
+            if state and state.trigger_crossed:
+                labels.append("CROSSED")
+            if labels:
+                cv2.putText(output, "  ".join(labels), (max(0, b.x1), max(18, b.y1 - 5)), cv2.FONT_HERSHEY_SIMPLEX, .52, color, 2)
+            if self.config.show_trajectories and state:
+                points = list(state.trajectory)[-self.config.trajectory_max_points:]
+                for previous, current in zip(points, points[1:]):
+                    cv2.line(output, (round(previous.foot_x), round(previous.foot_y)),
+                             (round(current.foot_x), round(current.foot_y)), color, 2)
+            face = face_results.get(person.track_id)
+            if face is not None:
+                if self.config.show_face_roi and face.roi is not None:
+                    roi = face.roi
+                    face_color = (255, 255, 0) if face.face_detected else (0, 0, 255)
+                    cv2.rectangle(output, (roi.x1, roi.y1), (roi.x2, roi.y2), face_color, 1)
+                if self.config.show_head_pose:
+                    if face.pose_estimated:
+                        pose_label = f"Yaw:{face.yaw:.1f} Pitch:{face.pitch:.1f} Roll:{face.roll:.1f}"
+                    else:
+                        raw = ""
+                        if all(value is not None for value in (face.pitch, face.yaw, face.roll)):
+                            raw = f" P:{face.pitch:.1f} Y:{face.yaw:.1f} R:{face.roll:.1f}"
+                        pose_label = f"Pose: NG ({face.failure_reason}){raw}"
+                    cv2.putText(output, pose_label, (max(0, b.x1), min(height - 5, b.y1 + 20)),
+                                cv2.FONT_HERSHEY_SIMPLEX, .45, (255, 255, 0), 1)
+                if self.config.show_face_landmarks:
+                    for landmark in face.landmarks:
+                        cv2.circle(output, (round(landmark.x), round(landmark.y)), 1, (255, 255, 0), -1)
+            turn = turn_states.get(person.track_id)
+            if self.config.show_turn_status and turn is not None:
+                if turn.baseline_acquired:
+                    delta = turn.max_yaw_delta_toward_signage
+                    delta_text = f" dYaw:{delta:.1f}" if self.config.show_yaw_delta and delta is not None else ""
+                    turn_label = f"Turn:{turn.turn_level}{delta_text}"
+                else:
+                    turn_label = "Turn:N/A"
+                turn_color = (0, 0, 255) if turn.turn_confirmed else (255, 255, 255)
+                cv2.putText(output, turn_label, (max(0, b.x1), min(height - 5, b.y1 + 40)),
+                            cv2.FONT_HERSHEY_SIMPLEX, .48, turn_color, 2)
+            voice = voice_states.get(person.track_id)
+            if self.config.show_voice_status and voice is not None:
+                status = "control" if voice.pseudo_prompt else ("played" if voice.prompt_played else
+                         ("failed" if voice.prompt_failed else "waiting"))
+                cv2.putText(output, f"Voice:{status}", (max(0, b.x1), min(height - 5, b.y1 + 60)),
+                            cv2.FONT_HERSHEY_SIMPLEX, .48, (255, 180, 80), 2)
+            person_state = person_states.get(person.track_id)
+            if self.config.show_person_state and person_state is not None:
+                short = {"observing_response": "observing", "pseudo_prompted": "pseudo",
+                         "not_evaluable": "not_eval"}.get(person_state.status, person_state.status)
+                cv2.putText(output, f"State:{short}", (max(0, b.x1), min(height - 5, b.y1 + 80)),
+                            cv2.FONT_HERSHEY_SIMPLEX, .48, (180, 255, 180), 2)
+        lines: list[str] = []
+        if self.config.show_fps:
+            lines.append(f"FPS: {info.measured_fps:.1f}")
+        if self.config.show_detection_count:
+            lines.append(f"Raw Detections: {len(detections)}")
+        if self.config.show_tracking_counters and tracking_counters is not None:
+            lines.extend((
+                f"Active Tracks: {tracking_counters.get('active', 0)}",
+                f"Unique Tracks: {tracking_counters.get('unique', 0)}",
+                f"Region Visitors: {tracking_counters.get('visitors', 0)}",
+                f"Trigger Crossings: {tracking_counters.get('crossings', 0)}",
+            ))
+        if self.config.show_face_counters:
+            current_faces = [face_results.get(person.track_id) for person in tracked_persons]
+            lines.append(f"Face: {'enabled' if face_enabled else 'disabled'}")
+            if face_enabled:
+                lines.append(f"Face OK: {sum(bool(item and item.face_detected) for item in current_faces)}/{len(tracked_persons)}")
+                lines.append(f"Pose OK: {sum(bool(item and item.pose_estimated) for item in current_faces)}/{len(tracked_persons)}")
+        if self.config.show_turn_counters:
+            active_turns = [turn_states.get(person.track_id) for person in tracked_persons]
+            lines.append(f"Turn: {'enabled' if turn_enabled else 'disabled'}")
+            if turn_enabled:
+                lines.append(f"Turn Eval: {sum(bool(item and item.baseline_acquired) for item in active_turns)}/{len(tracked_persons)}")
+                lines.append(f"Turn Confirmed: {sum(bool(item and item.turn_confirmed) for item in active_turns)}")
+                lines.append(f"Subtle: {sum(bool(item and item.strongest_level == 'subtle') for item in active_turns)}")
+                lines.append(f"Weak+: {sum(bool(item and is_weak_or_higher(item.strongest_level)) for item in active_turns)}")
+        if self.config.show_voice_counters:
+            voice_values = list(voice_states.values())
+            lines.append(f"Voice: {voice_mode}")
+            if voice_mode != "disabled":
+                lines.append(f"Prompted: {sum(item.prompt_played or item.pseudo_prompt for item in voice_values)}")
+                lines.append(f"Responses: {sum(item.response_detected for item in voice_values)}")
+                lines.append(f"Cooldown: {voice_cooldown:.1f}s")
+        if self.config.show_person_state_counters:
+            counts: dict[str, int] = {}
+            for state in person_states.values(): counts[state.status] = counts.get(state.status, 0) + 1
+            lines.append(f"State: {'enabled' if person_state_enabled else 'disabled'}")
+            if person_state_enabled:
+                labels = ("tracked", "eligible", "prompted", "observing_response", "responded", "excluded")
+                lines.append(" ".join(f"{name[:4]}={counts.get(name, 0)}" for name in labels))
+        if self.config.show_frame_number:
+            lines.append(f"Frame: {info.frame_number}")
+        if self.config.show_timestamp:
+            try:
+                shown_time = datetime.fromisoformat(info.timestamp_iso).strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
+            except ValueError:
+                shown_time = info.timestamp_iso
+            lines.append(f"Time: {shown_time}")
+        lines.append(f"Device: {device.upper()}")
+        lines.append(f"Tracker: {tracker_name}")
+        if debug:
+            lines.append(f"Resolution: {width}x{height}")
+        for index, text in enumerate(lines):
+            y = 25 + index * 24
+            cv2.putText(output, text, (11, y + 1), cv2.FONT_HERSHEY_SIMPLEX, .62, (0, 0, 0), 3)
+            cv2.putText(output, text, (10, y), cv2.FONT_HERSHEY_SIMPLEX, .62, (255, 255, 255), 1)
+        if info.frame_number >= self.stable_frames and 0 < info.measured_fps < self.warn_fps_threshold:
+            cv2.putText(output, "LOW FPS", (width - 130, 30), cv2.FONT_HERSHEY_SIMPLEX, .75, (0, 0, 255), 2)
+        if paused:
+            cv2.putText(output, "PAUSED", (max(10, width // 2 - 70), max(40, height // 2)), cv2.FONT_HERSHEY_SIMPLEX, 1.2, (0, 255, 255), 3)
+        return output