Draw MediaPipe landmarks in live debug

signspeak/asl/mediapipe_utils.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from typing import Any
+
+import numpy as np
+
+
+FACE_POINTS = 468
+HAND_POINTS = 21
+POSE_POINTS = 33
+
+
+def extract_keypoints_from_holistic(results: Any, missing_value: float = np.nan) -> np.ndarray:
+    face = _landmark_array(getattr(results, "face_landmarks", None), FACE_POINTS, missing_value)
+    left = _landmark_array(getattr(results, "left_hand_landmarks", None), HAND_POINTS, missing_value)
+    pose = _landmark_array(getattr(results, "pose_landmarks", None), POSE_POINTS, missing_value)
+    right = _landmark_array(getattr(results, "right_hand_landmarks", None), HAND_POINTS, missing_value)
+    return np.vstack([face, left, pose, right]).astype(np.float32)
+
+
+def draw_holistic_landmarks(mp: Any, image: np.ndarray, results: Any) -> None:
+    drawing = mp.solutions.drawing_utils
+    styles = mp.solutions.drawing_styles
+    holistic = mp.solutions.holistic
+
+    drawing.draw_landmarks(
+        image,
+        getattr(results, "face_landmarks", None),
+        holistic.FACEMESH_CONTOURS,
+        landmark_drawing_spec=None,
+        connection_drawing_spec=styles.get_default_face_mesh_contours_style(),
+    )
+    drawing.draw_landmarks(
+        image,
+        getattr(results, "pose_landmarks", None),
+        holistic.POSE_CONNECTIONS,
+        landmark_drawing_spec=styles.get_default_pose_landmarks_style(),
+    )
+    drawing.draw_landmarks(
+        image,
+        getattr(results, "left_hand_landmarks", None),
+        holistic.HAND_CONNECTIONS,
+        landmark_drawing_spec=styles.get_default_hand_landmarks_style(),
+        connection_drawing_spec=styles.get_default_hand_connections_style(),
+    )
+    drawing.draw_landmarks(
+        image,
+        getattr(results, "right_hand_landmarks", None),
+        holistic.HAND_CONNECTIONS,
+        landmark_drawing_spec=styles.get_default_hand_landmarks_style(),
+        connection_drawing_spec=styles.get_default_hand_connections_style(),
+    )
+
+
+def _landmark_array(landmark_list: Any, expected_points: int, missing_value: float) -> np.ndarray:
+    empty = np.full((expected_points, 3), missing_value, dtype=np.float32)
+    if landmark_list is None or not getattr(landmark_list, "landmark", None):
+        return empty
+
+    points = landmark_list.landmark[:expected_points]
+    for idx, point in enumerate(points):
+        empty[idx] = [point.x, point.y, point.z]
+    return empty
+

signspeak/debug_video.py

@@ -5,9 +5,12 @@ import time
 from pathlib import Path
 from typing import Any
 
+from .asl.mediapipe_utils import draw_holistic_landmarks
+
 
 def create_debug_overlay_video(video_path: str | Path, result: dict[str, Any]) -> str:
     cv2 = _load_cv2()
+    mp, holistic = _load_holistic()
     path = Path(video_path)
     cap = cv2.VideoCapture(str(path))
     if not cap.isOpened():
@@ -49,10 +52,16 @@ def create_debug_overlay_video(video_path: str | Path, result: dict[str, Any]) -
             ok, frame = cap.read()
             if not ok or frame is None:
                 break
+            if mp is not None and holistic is not None:
+                frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+                results = holistic.process(frame_rgb)
+                draw_holistic_landmarks(mp, frame, results)
             _draw_overlay(cv2, frame, gloss_text, emotion_text, status_text)
             writer.write(frame)
     finally:
         cap.release()
+        if holistic is not None:
+            holistic.close()
         writer.release()
 
     return str(output_path)
@@ -94,3 +103,18 @@ def _load_cv2():
         return cv2
     except Exception as exc:
         raise RuntimeError("OpenCV is required for debug overlay video generation.") from exc
+
+
+def _load_holistic():
+    try:
+        import mediapipe as mp
+
+        return (
+            mp,
+            mp.solutions.holistic.Holistic(
+                min_detection_confidence=0.5,
+                min_tracking_confidence=0.5,
+            ),
+        )
+    except Exception:
+        return None, None

signspeak/live_debug.py

@@ -6,12 +6,11 @@ from typing import Any
 import numpy as np
 
 from .asl.asl_detector import ASLDetector
-from .asl.landmarks_detector import LandmarksDetector
+from .asl.mediapipe_utils import draw_holistic_landmarks, extract_keypoints_from_holistic
 
 
 class LiveASLSession:
     def __init__(self) -> None:
-        self.detector = LandmarksDetector(missing_value=np.nan)
         self.asl = ASLDetector()
         self.interpreter: Any | None = None
         self.frame_keypoints: list[np.ndarray] = []
@@ -19,16 +18,21 @@ class LiveASLSession:
         self.latest_emotion = "unknown"
         self.latest_status = "Waiting for 30 frames."
         self.frames_seen = 0
+        self.mp: Any | None = None
+        self.holistic: Any | None = None
+        self._load_mediapipe()
 
     def process_frame(self, frame_rgb: np.ndarray) -> tuple[np.ndarray, str]:
         output = frame_rgb.copy()
         self.frames_seen += 1
-        if self.detector.mp is None:
-            self.latest_status = self.detector.error or "MediaPipe unavailable."
+        if self.mp is None or self.holistic is None:
+            self.latest_status = "MediaPipe unavailable."
             return self._draw(output), self.latest_status
 
         try:
-            keypoints = self.detector._detect_frame(frame_rgb)
+            results = self.holistic.process(frame_rgb)
+            draw_holistic_landmarks(self.mp, output, results)
+            keypoints = extract_keypoints_from_holistic(results, missing_value=np.nan)
             self.frame_keypoints.append(keypoints)
             self.frame_keypoints = self.frame_keypoints[-30:]
             if self.frames_seen % 15 == 0:
@@ -40,6 +44,18 @@ class LiveASLSession:
 
         return self._draw(output), self.latest_status
 
+    def _load_mediapipe(self) -> None:
+        try:
+            import mediapipe as mp
+
+            self.mp = mp
+            self.holistic = mp.solutions.holistic.Holistic(
+                min_detection_confidence=0.5,
+                min_tracking_confidence=0.5,
+            )
+        except Exception as exc:
+            self.latest_status = f"MediaPipe unavailable: {type(exc).__name__}: {exc}"
+
     def _predict_latest(self) -> None:
         if not self.asl.model_path.exists():
             self.latest_prediction = ""