Align ASL inference with upstream model

README.md

@@ -85,6 +85,26 @@ 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.
 
+Good first signs to test because they are in the model vocabulary:
+
+```text
+hello, where, who, why, yes, no, thankyou, please, water, happy, sad
+```
+
+Reference clips/GIFs from the upstream demo list:
+
+```text
+hello    https://media.giphy.com/media/3o7TKNKOfKlIhbD3gY/giphy.gif
+where    https://lifeprint.com/asl101/gifs/w/where.gif
+who      https://lifeprint.com/asl101/gifs/w/who.gif
+why      https://lifeprint.com/asl101/gifs/w/why.gif
+yes      https://media.tenor.com/oYIirlyIih0AAAAC/yes-asl.gif
+no       https://lifeprint.com/asl101/gifs/n/no-2-movement.gif
+thankyou https://lifeprint.com/asl101/gifs/t/thank-you.gif
+please   https://lifeprint.com/asl101/gifs-animated/pleasecloseup.gif
+water    https://lifeprint.com/asl101/gifs/w/water-2.gif
+```
+
 ## GPU dependencies
 
 `flash-attn` is only useful on a CUDA GPU Space with compatible PyTorch/CUDA versions.

app.py

@@ -5,6 +5,7 @@ from pathlib import Path
 import gradio as gr
 
 from signspeak.llm import generate_subtitle_and_instruction
+from signspeak.live_debug import process_live_debug_frame
 from signspeak.pipeline import DEFAULT_INTENT, json_text, run_asl_video
 from signspeak.tts import generate_tts
 
@@ -70,38 +71,6 @@ def build_video_input(label: str) -> gr.Video:
     )
 
 
-def render_live_frame_debug(frame):
-    if frame is None:
-        return None, "Waiting for camera frame."
-
-    import cv2
-
-    output = frame.copy()
-    height, width = output.shape[:2]
-    cv2.rectangle(output, (0, 0), (width, 72), (8, 11, 16), -1)
-    cv2.putText(
-        output,
-        "LIVE CAMERA DEBUG",
-        (14, 28),
-        cv2.FONT_HERSHEY_SIMPLEX,
-        0.62,
-        (45, 212, 191),
-        2,
-        cv2.LINE_AA,
-    )
-    cv2.putText(
-        output,
-        "Use Analyze ASL for 30-frame TFLite gloss inference",
-        (14, 58),
-        cv2.FONT_HERSHEY_SIMPLEX,
-        0.46,
-        (248, 250, 252),
-        1,
-        cv2.LINE_AA,
-    )
-    return output, "Live frame received. Sequence inference runs from the recorded/uploaded clip."
-
-
 with gr.Blocks(title="SignSpeak Local") as demo:
     gr.HTML(
         """
@@ -306,7 +275,7 @@ with gr.Blocks(title="SignSpeak Local") as demo:
     )
 
     live_camera_input.stream(
-        fn=render_live_frame_debug,
+        fn=process_live_debug_frame,
         inputs=[live_camera_input],
         outputs=[live_camera_output, live_camera_status],
     )

signspeak/asl/asl_detector.py

@@ -23,7 +23,7 @@ class ASLDetector:
         self.labels = self._load_labels()
 
     def predict_from_frames(self, frames: list[np.ndarray]) -> dict[str, Any]:
-        detector = LandmarksDetector(missing_value=0.0)
+        detector = LandmarksDetector(missing_value=np.nan)
         try:
             landmark_result = detector.detect_sequence(frames)
         finally:
@@ -48,12 +48,25 @@ class ASLDetector:
 
         try:
             interpreter = self._load_interpreter()
-            output = self._predict(interpreter, keypoints)
-
-            probs = self._softmax_if_needed(np.asarray(output).reshape(-1))
-            top_idx = int(np.argmax(probs))
+            window_results = []
+            for window_index, window_keypoints in enumerate(self._keypoint_windows(keypoints)):
+                output = self._predict(interpreter, window_keypoints)
+                probs = self._softmax_if_needed(np.asarray(output).reshape(-1))
+                top_idx = int(np.argmax(probs))
+                window_results.append(
+                    {
+                        "window_index": window_index,
+                        "probs": probs,
+                        "top_idx": top_idx,
+                        "confidence": float(probs[top_idx]),
+                    }
+                )
+
+            best = max(window_results, key=lambda item: item["confidence"])
+            probs = best["probs"]
+            top_idx = int(best["top_idx"])
             top_prediction = self._label_for_index(top_idx)
-            confidence = float(probs[top_idx])
+            confidence = float(best["confidence"])
             accepted = confidence >= self.confidence_threshold
 
             base.update(
@@ -64,6 +77,8 @@ class ASLDetector:
                     "confidence": confidence,
                     "confidence_threshold": self.confidence_threshold,
                     "top_predictions": self._top_predictions(probs),
+                    "windows_analyzed": len(window_results),
+                    "best_window_index": int(best["window_index"]),
                 }
             )
             return base
@@ -91,7 +106,7 @@ class ASLDetector:
         if "serving_default" in signatures:
             prediction_fn = interpreter.get_signature_runner("serving_default")
             input_name = self._signature_input_name(signatures["serving_default"])
-            input_data = np.nan_to_num(keypoints, nan=0.0, posinf=0.0, neginf=0.0).astype(np.float32)
+            input_data = keypoints.astype(np.float32)
             prediction = prediction_fn(**{input_name: input_data})
             output_name = self._signature_output_name(prediction)
             return np.asarray(prediction[output_name])
@@ -114,6 +129,24 @@ class ASLDetector:
             return "outputs"
         return next(iter(prediction))
 
+    def _keypoint_windows(self, keypoints: np.ndarray, window_size: int = 30, max_windows: int = 5) -> list[np.ndarray]:
+        frame_count = int(keypoints.shape[0])
+        if frame_count <= window_size:
+            return [keypoints.astype(np.float32)]
+
+        max_start = frame_count - window_size
+        if max_windows <= 1:
+            starts = [max_start // 2]
+        else:
+            starts = np.linspace(0, max_start, num=min(max_windows, max_start + 1), dtype=int).tolist()
+
+        unique_starts = []
+        for start in starts:
+            if start not in unique_starts:
+                unique_starts.append(int(start))
+
+        return [keypoints[start : start + window_size].astype(np.float32) for start in unique_starts]
+
     def _prepare_input(self, keypoints: np.ndarray, input_detail: dict[str, Any]) -> np.ndarray:
         shape = input_detail.get("shape")
         dtype = input_detail.get("dtype", np.float32)

signspeak/asl/pipeline.py

@@ -12,7 +12,7 @@ from .video_utils import sample_video_frames, sample_video_frames_for_emotion
 
 def process_asl_video(video_path: str | Path) -> dict[str, Any]:
     path = Path(video_path)
-    asl_frames = sample_video_frames(path, target_frames=30)
+    asl_frames = sample_video_frames(path, target_frames=90)
     emotion_frames = sample_video_frames_for_emotion(path, target_frames=12)
 
     return process_asl_frames(asl_frames, emotion_frames, source=str(path))

signspeak/live_debug.py

@@ -0,0 +1,137 @@
+from __future__ import annotations
+
+import os
+from typing import Any
+
+import numpy as np
+
+from .asl.asl_detector import ASLDetector
+from .asl.landmarks_detector import LandmarksDetector
+
+
+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] = []
+        self.latest_prediction = ""
+        self.latest_emotion = "unknown"
+        self.latest_status = "Waiting for 30 frames."
+        self.frames_seen = 0
+
+    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."
+            return self._draw(output), self.latest_status
+
+        try:
+            keypoints = self.detector._detect_frame(frame_rgb)
+            self.frame_keypoints.append(keypoints)
+            self.frame_keypoints = self.frame_keypoints[-30:]
+            if self.frames_seen % 15 == 0:
+                self._update_emotion(frame_rgb)
+            if len(self.frame_keypoints) == 30:
+                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
+
+    def _predict_latest(self) -> None:
+        if not self.asl.model_path.exists():
+            self.latest_prediction = ""
+            self.latest_status = "ASL model missing."
+            return
+
+        if self.interpreter is None:
+            self.interpreter = self.asl._load_interpreter()
+
+        keypoints = np.asarray(self.frame_keypoints, dtype=np.float32)
+        output = self.asl._predict(self.interpreter, keypoints)
+        probs = self.asl._softmax_if_needed(np.asarray(output).reshape(-1))
+        top_idx = int(np.argmax(probs))
+        label = self.asl._label_for_index(top_idx)
+        confidence = float(probs[top_idx])
+
+        if confidence >= self.asl.confidence_threshold:
+            self.latest_prediction = f"{label} ({confidence:.0%})"
+            self.latest_status = f"Accepted: {label} at {confidence:.2f}"
+        else:
+            self.latest_prediction = ""
+            self.latest_status = f"Low confidence: {label} at {confidence:.2f}"
+
+    def _update_emotion(self, frame_rgb: np.ndarray) -> None:
+        try:
+            os.environ.setdefault("TF_USE_LEGACY_KERAS", "1")
+            import cv2
+            from deepface import DeepFace
+
+            frame_bgr = cv2.cvtColor(frame_rgb, cv2.COLOR_RGB2BGR)
+            analysis = DeepFace.analyze(
+                img_path=frame_bgr,
+                actions=["emotion"],
+                enforce_detection=False,
+                silent=True,
+            )
+            if isinstance(analysis, list):
+                analysis = analysis[0] if analysis else {}
+            dominant = analysis.get("dominant_emotion") if isinstance(analysis, dict) else None
+            if dominant:
+                self.latest_emotion = str(dominant)
+        except Exception:
+            self.latest_emotion = "unavailable"
+
+    def _draw(self, frame_rgb: np.ndarray) -> np.ndarray:
+        import cv2
+
+        height, width = frame_rgb.shape[:2]
+        cv2.rectangle(frame_rgb, (0, 0), (width, 116), (8, 11, 16), -1)
+        cv2.putText(
+            frame_rgb,
+            f"Sign: {self.latest_prediction or '-'}",
+            (14, 36),
+            cv2.FONT_HERSHEY_SIMPLEX,
+            0.86,
+            (45, 212, 191),
+            2,
+            cv2.LINE_AA,
+        )
+        cv2.putText(
+            frame_rgb,
+            f"Emotion: {self.latest_emotion}",
+            (14, 68),
+            cv2.FONT_HERSHEY_SIMPLEX,
+            0.58,
+            (245, 158, 11),
+            2,
+            cv2.LINE_AA,
+        )
+        cv2.putText(
+            frame_rgb,
+            self.latest_status[:96],
+            (14, 100),
+            cv2.FONT_HERSHEY_SIMPLEX,
+            0.48,
+            (248, 250, 252),
+            1,
+            cv2.LINE_AA,
+        )
+        return frame_rgb
+
+
+_live_session: LiveASLSession | None = None
+
+
+def process_live_debug_frame(frame: np.ndarray | None) -> tuple[np.ndarray | None, str]:
+    global _live_session
+
+    if frame is None:
+        return None, "Waiting for camera frame."
+
+    if _live_session is None:
+        _live_session = LiveASLSession()
+
+    return _live_session.process_frame(frame)

signspeak/llm.py

@@ -76,24 +76,6 @@ def normalize_llm_output(parsed: dict[str, Any]) -> dict[str, str]:
 def deterministic_speech_from_intent(intent: dict[str, Any]) -> dict[str, str]:
     glosses = [str(gloss).upper() for gloss in intent.get("detected_glosses", []) if str(gloss).strip()]
     emotion = str(intent.get("detected_facial_expression") or intent.get("emotion_profile", {}).get("dominant") or "neutral")
-    phrase_key = " ".join(glosses)
-
-    phrase_map = {
-        "I LOVE YOU": "I love you.",
-        "LOVE YOU": "I love you.",
-        "I HAPPY SEE YOU": "I am happy to see you.",
-        "I SEE YOU": "I see you.",
-        "WHERE": "Where?",
-        "WHO": "Who?",
-        "WHY": "Why?",
-        "WHAT": "What?",
-        "HOW": "How?",
-        "THANK YOU": "Thank you.",
-        "THANKYOU": "Thank you.",
-        "HELLO": "Hello.",
-        "YES": "Yes.",
-        "NO": "No.",
-    }
 
     if not glosses:
         return {
@@ -101,9 +83,14 @@ def deterministic_speech_from_intent(intent: dict[str, Any]) -> dict[str, str]:
             "voice_instruction": "Speak calmly and clearly, indicating that no sign was detected.",
         }
 
-    subtitle = phrase_map.get(phrase_key)
-    if subtitle is None:
-        subtitle = " ".join(gloss.lower() for gloss in glosses).capitalize() + "."
+    words = [normalize_gloss_token(gloss) for gloss in glosses]
+    subtitle = " ".join(word for word in words if word).strip()
+    if not subtitle:
+        subtitle = " ".join(gloss.lower() for gloss in glosses).strip()
+    punctuation = "?" if len(words) == 1 and words[0].lower() in {"where", "who", "why", "what", "how"} else "."
+    subtitle = subtitle[:1].upper() + subtitle[1:]
+    if subtitle[-1:] not in ".?!":
+        subtitle += punctuation
 
     tone = "clearly and naturally"
     if emotion in ("happy", "joy"):
@@ -119,6 +106,21 @@ def deterministic_speech_from_intent(intent: dict[str, Any]) -> dict[str, str]:
     }
 
 
+def normalize_gloss_token(gloss: str) -> str:
+    token = gloss.strip().lower().replace("_", " ").replace("-", " ")
+    compact_map = {
+        "thankyou": "thank you",
+        "callonphone": "call on phone",
+        "frenchfries": "french fries",
+        "glasswindow": "glass window",
+        "haveto": "have to",
+        "hesheit": "he she it",
+        "minemy": "mine my",
+        "weus": "we us",
+    }
+    return compact_map.get(token, token)
+
+
 def enforce_intent_consistency(intent: dict[str, Any], normalized: dict[str, Any]) -> dict[str, Any]:
     glosses = [str(gloss).upper() for gloss in intent.get("detected_glosses", []) if str(gloss).strip()]
     phrase_key = " ".join(glosses)

tests/test_asl_detector.py

@@ -28,6 +28,7 @@ class FakeSignatureInterpreter:
 
         def predict(**kwargs):
             assert kwargs["inputs"].shape == (543, 3)
+            assert np.isnan(kwargs["inputs"][0][0])
             return {"outputs": np.asarray([[0.1, 0.8, 0.1]], dtype=np.float32)}
 
         return predict
@@ -36,6 +37,7 @@ class FakeSignatureInterpreter:
 def test_predict_uses_tflite_signature_runner(tmp_path):
     detector = ASLDetector(model_dir=tmp_path)
     keypoints = np.zeros((543, 3), dtype=np.float32)
+    keypoints[0][0] = np.nan
 
     output = detector._predict(FakeSignatureInterpreter(), keypoints)
 
@@ -83,3 +85,13 @@ def test_low_confidence_prediction_is_not_accepted(monkeypatch, tmp_path):
     assert result["status"] == "low_confidence"
     assert result["top_prediction"] == "where"
     assert result["gloss_sequence"] == []
+
+
+def test_keypoint_windows_cover_long_sequences(tmp_path):
+    detector = ASLDetector(model_dir=tmp_path)
+    keypoints = np.zeros((90, 543, 3), dtype=np.float32)
+
+    windows = detector._keypoint_windows(keypoints, window_size=30, max_windows=5)
+
+    assert len(windows) == 5
+    assert all(window.shape == (30, 543, 3) for window in windows)