Speed up live ASL debug diagnostics

README.md

@@ -85,6 +85,12 @@ This model recognizes the isolated signs listed in `sign_to_prediction_index_map
 a full sentence or fingerspelling recognizer. Predictions below `ASL_CONFIDENCE_THRESHOLD`
 defaulting to `0.70` are reported as `low_confidence` and are not forwarded as detected glosses.
 
+Live camera debug prioritizes speed over long temporal batching. It starts predicting after
+`LIVE_ASL_MIN_FRAMES=4`, keeps a rolling buffer of `LIVE_ASL_MAX_FRAMES=12`, and runs ASL
+prediction every `LIVE_ASL_PREDICT_EVERY=1` frame. DeepFace emotion is heavier, so it runs every
+`LIVE_EMOTION_EVERY=45` frames by default. The overlay and status panel still show the current top
+candidates even when the accepted gloss is empty because the confidence is below threshold.
+
 Good first signs to test because they are in the model vocabulary:
 
 ```text

signspeak/asl/pipeline.py

@@ -36,6 +36,8 @@ def process_asl_frames(
             "gloss_sequence": asl.get("gloss_sequence", []),
             "top_prediction": asl.get("top_prediction"),
             "confidence": float(asl.get("confidence", 0.0) or 0.0),
+            "confidence_threshold": float(asl.get("confidence_threshold", 0.0) or 0.0),
+            "top_predictions": asl.get("top_predictions", []),
             "frames_used": int(asl.get("frames_used", len(asl_frames)) or 0),
             "keypoints_shape": asl.get("keypoints_shape", []),
             "landmarks_status": asl.get("landmarks_status"),

signspeak/live_debug.py

@@ -15,8 +15,14 @@ class LiveASLSession:
         self.interpreter: Any | None = None
         self.frame_keypoints: list[np.ndarray] = []
         self.latest_prediction = ""
+        self.latest_top_candidate = ""
+        self.latest_top_predictions: list[dict[str, Any]] = []
         self.latest_emotion = "unknown"
-        self.latest_status = "Waiting for 30 frames."
+        self.min_prediction_frames = max(1, int(os.getenv("LIVE_ASL_MIN_FRAMES", "4")))
+        self.max_prediction_frames = max(self.min_prediction_frames, int(os.getenv("LIVE_ASL_MAX_FRAMES", "12")))
+        self.predict_every = max(1, int(os.getenv("LIVE_ASL_PREDICT_EVERY", "1")))
+        self.emotion_every = max(1, int(os.getenv("LIVE_EMOTION_EVERY", "45")))
+        self.latest_status = f"Waiting for live frames: 0/{self.min_prediction_frames}."
         self.frames_seen = 0
         self.mp: Any | None = None
         self.holistic: Any | None = None
@@ -34,15 +40,17 @@ class LiveASLSession:
             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:
+            self.frame_keypoints = self.frame_keypoints[-self.max_prediction_frames :]
+            if self.frames_seen % self.emotion_every == 0:
                 self._update_emotion(frame_rgb)
-            if len(self.frame_keypoints) == 30:
+            if len(self.frame_keypoints) < self.min_prediction_frames:
+                self.latest_status = f"Waiting for live frames: {len(self.frame_keypoints)}/{self.min_prediction_frames}."
+            elif self.frames_seen % self.predict_every == 0:
                 self._predict_latest()
         except Exception as exc:
             self.latest_status = f"Live ASL error: {type(exc).__name__}: {exc}"
 
-        return self._draw(output), self.latest_status
+        return self._draw(output), self._status_text()
 
     def _load_mediapipe(self) -> None:
         try:
@@ -50,6 +58,7 @@ class LiveASLSession:
 
             self.mp = mp
             self.holistic = mp.solutions.holistic.Holistic(
+                model_complexity=0,
                 min_detection_confidence=0.5,
                 min_tracking_confidence=0.5,
             )
@@ -71,13 +80,22 @@ class LiveASLSession:
         top_idx = int(np.argmax(probs))
         label = self.asl._label_for_index(top_idx)
         confidence = float(probs[top_idx])
+        top_predictions = self.asl._top_predictions(probs, limit=3)
+        self.latest_top_predictions = top_predictions
+        self.latest_top_candidate = f"{label} ({confidence:.0%})"
 
         if confidence >= self.asl.confidence_threshold:
             self.latest_prediction = f"{label} ({confidence:.0%})"
-            self.latest_status = f"Accepted: {label} at {confidence:.2f}"
+            self.latest_status = (
+                f"Accepted: {label} at {confidence:.2f} "
+                f"with {len(self.frame_keypoints)} live frames."
+            )
         else:
             self.latest_prediction = ""
-            self.latest_status = f"Low confidence: {label} at {confidence:.2f}"
+            self.latest_status = (
+                f"Top candidate: {label} at {confidence:.2f}; "
+                f"waiting for {self.asl.confidence_threshold:.2f}."
+            )
 
     def _update_emotion(self, frame_rgb: np.ndarray) -> None:
         try:
@@ -104,7 +122,7 @@ class LiveASLSession:
         import cv2
 
         height, width = frame_rgb.shape[:2]
-        cv2.rectangle(frame_rgb, (0, 0), (width, 116), (8, 11, 16), -1)
+        cv2.rectangle(frame_rgb, (0, 0), (width, 146), (8, 11, 16), -1)
         cv2.putText(
             frame_rgb,
             f"Sign: {self.latest_prediction or '-'}",
@@ -117,10 +135,20 @@ class LiveASLSession:
         )
         cv2.putText(
             frame_rgb,
-            f"Emotion: {self.latest_emotion}",
+            f"Top: {self.latest_top_candidate or '-'}",
             (14, 68),
             cv2.FONT_HERSHEY_SIMPLEX,
             0.58,
+            (129, 140, 248),
+            2,
+            cv2.LINE_AA,
+        )
+        cv2.putText(
+            frame_rgb,
+            f"Emotion: {self.latest_emotion}",
+            (14, 98),
+            cv2.FONT_HERSHEY_SIMPLEX,
+            0.58,
             (245, 158, 11),
             2,
             cv2.LINE_AA,
@@ -128,7 +156,7 @@ class LiveASLSession:
         cv2.putText(
             frame_rgb,
             self.latest_status[:96],
-            (14, 100),
+            (14, 128),
             cv2.FONT_HERSHEY_SIMPLEX,
             0.48,
             (248, 250, 252),
@@ -137,6 +165,21 @@ class LiveASLSession:
         )
         return frame_rgb
 
+    def _status_text(self) -> str:
+        top_lines = [
+            f"- {item.get('label')}: {float(item.get('confidence', 0.0) or 0.0):.2f}"
+            for item in self.latest_top_predictions
+        ]
+        top_block = "\n".join(top_lines) if top_lines else "- None yet"
+        return (
+            f"{self.latest_status}\n"
+            f"Accepted sign: {self.latest_prediction or 'None'}\n"
+            f"Top candidates:\n{top_block}\n"
+            f"Frames in rolling buffer: {len(self.frame_keypoints)}/{self.max_prediction_frames}\n"
+            f"Acceptance threshold: {self.asl.confidence_threshold:.2f}\n"
+            f"Emotion: {self.latest_emotion}"
+        )
+
 
 _live_session: LiveASLSession | None = None
 

signspeak/pipeline.py

@@ -134,14 +134,25 @@ def summarize_asl_result(result: dict[str, Any]) -> str:
     top_prediction = asl.get("top_prediction") or "None"
     confidence = float(asl.get("confidence", 0.0) or 0.0)
     threshold = float(asl.get("confidence_threshold", 0.0) or 0.0)
+    top_predictions = _format_top_predictions(asl.get("top_predictions", []))
     override = result.get("intent_input", {}).get("diagnostics", {}).get("manual_gloss_override")
     override_line = "\nOverride: manual glosses applied" if override else ""
     return (
         f"ASL status: {asl.get('status', 'unknown')}\n"
         f"Detected words: {gloss_line}\n"
         f"Top candidate: {top_prediction} ({confidence:.2f}, threshold {threshold:.2f})\n"
+        f"Top candidates: {top_predictions}\n"
         f"Landmarks: {asl.get('landmarks_status', 'unknown')} via {asl.get('landmarks_detector', 'unknown')}\n"
         f"Emotion: {emotion.get('dominant_emotion', 'unknown')} "
         f"({float(emotion.get('intensity', 0.0) or 0.0):.2f})"
         f"{override_line}"
     )
+
+
+def _format_top_predictions(top_predictions: list[dict[str, Any]]) -> str:
+    if not top_predictions:
+        return "None"
+    return ", ".join(
+        f"{item.get('label')} {float(item.get('confidence', 0.0) or 0.0):.2f}"
+        for item in top_predictions[:5]
+    )

tests/test_asl_pipeline.py

@@ -1,4 +1,7 @@
-from signspeak.asl.pipeline import build_intent_input
+import numpy as np
+
+import signspeak.asl.pipeline as asl_pipeline
+from signspeak.asl.pipeline import build_intent_input, process_asl_frames
 from signspeak.pipeline import apply_gloss_override, parse_gloss_override, resolve_video_path, summarize_asl_result
 
 
@@ -25,6 +28,41 @@ def test_build_intent_input_matches_llm_schema():
     assert intent["diagnostics"]["asl_status"] == "ok"
 
 
+def test_process_asl_frames_preserves_detector_diagnostics(monkeypatch):
+    class FakeASLDetector:
+        def predict_from_frames(self, frames):
+            return {
+                "status": "ok",
+                "gloss_sequence": ["talk"],
+                "top_prediction": "talk",
+                "confidence": 0.96,
+                "confidence_threshold": 0.70,
+                "top_predictions": [{"label": "talk", "confidence": 0.96}],
+                "frames_used": len(frames),
+                "keypoints_shape": [8, 543, 3],
+                "landmarks_status": "ok",
+                "landmarks_detector": "holistic",
+            }
+
+    monkeypatch.setattr(asl_pipeline, "ASLDetector", FakeASLDetector)
+    monkeypatch.setattr(
+        asl_pipeline,
+        "detect_emotion_on_frames",
+        lambda frames: {
+            "status": "ok",
+            "dominant_emotion": "neutral",
+            "intensity": 0.29,
+            "emotion_scores": {"neutral": 0.29},
+        },
+    )
+
+    result = process_asl_frames([np.zeros((2, 2, 3), dtype=np.uint8)] * 8)
+
+    assert result["asl"]["confidence_threshold"] == 0.70
+    assert result["asl"]["top_predictions"] == [{"label": "talk", "confidence": 0.96}]
+    assert result["intent_input"]["detected_glosses"] == ["talk"]
+
+
 def test_summarize_asl_result_is_stable_for_missing_fields():
     summary = summarize_asl_result(
         {