Add sliding-window ASL phrase prototype

README.md

@@ -17,9 +17,10 @@ MVP Gradio pour tester la pipeline ASL video/camera -> intent JSON -> llama.cpp
 
 ```text
 Video upload / camera capture
--> ASL frame sampling
+-> ASL sequential frame sampling
 -> MediaPipe landmarks, if installed
--> TFLite ASL classifier, if data/models/asl/model.tflite is present
+-> TFLite ASL classifier over sliding temporal windows, if data/models/asl/model.tflite is present
+-> gloss sequence aggregation
 -> DeepFace emotion aggregation, if installed
 -> llama.cpp subtitle + voice instruction
 -> Qwen3-TTS audio
@@ -85,6 +86,22 @@ 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.
 
+Uploaded videos use a phrase-prototype mode: frames are read in temporal order, landmarks are
+extracted once, then the ASL model runs over sliding windows. Accepted window predictions are
+collapsed into an ordered gloss sequence before llama.cpp rewrites them as natural speech. Tune it
+with:
+
+```text
+ASL_UPLOAD_TARGET_FPS=12
+ASL_UPLOAD_MAX_FRAMES=240
+ASL_SEQUENCE_WINDOW=30
+ASL_SEQUENCE_STRIDE=15
+ASL_CONFIDENCE_THRESHOLD=0.70
+```
+
+This is still not full continuous ASL translation, but it lets recorded phrase clips become
+`hello where water`-style gloss sequences instead of one global class.
+
 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

signspeak/asl/asl_detector.py

@@ -20,6 +20,8 @@ class ASLDetector:
         self.train_csv_path = self.model_dir / "train.csv"
         self.sign_map_path = self.model_dir / "sign_to_prediction_index_map.json"
         self.confidence_threshold = float(os.getenv("ASL_CONFIDENCE_THRESHOLD", "0.70"))
+        self.sequence_window = max(1, int(os.getenv("ASL_SEQUENCE_WINDOW", "30")))
+        self.sequence_stride = max(1, int(os.getenv("ASL_SEQUENCE_STRIDE", "15")))
         self.labels = self._load_labels()
 
     def predict_from_frames(self, frames: list[np.ndarray]) -> dict[str, Any]:
@@ -75,6 +77,70 @@ class ASLDetector:
             )
             return base
 
+    def predict_sequence_from_frames(self, frames: list[np.ndarray]) -> dict[str, Any]:
+        detector = LandmarksDetector(missing_value=np.nan)
+        try:
+            landmark_result = detector.detect_sequence(frames)
+        finally:
+            detector.close()
+
+        keypoints = landmark_result.keypoints
+        base = {
+            "status": "model_missing",
+            "gloss_sequence": [],
+            "top_prediction": None,
+            "confidence": 0.0,
+            "confidence_threshold": self.confidence_threshold,
+            "top_predictions": [],
+            "segment_predictions": [],
+            "frames_used": len(frames),
+            "windows_used": 0,
+            "sequence_window": self.sequence_window,
+            "sequence_stride": self.sequence_stride,
+            "keypoints_shape": list(keypoints.shape),
+            "landmarks_status": landmark_result.status,
+            "landmarks_detector": landmark_result.detector,
+            "recognition_mode": "sliding_window_sequence",
+        }
+        if landmark_result.error:
+            base["landmarks_error"] = landmark_result.error
+
+        if not self.model_path.exists():
+            return base
+
+        try:
+            interpreter = self._load_interpreter()
+            segments = []
+            for window_index, start, end, window_keypoints in self._iter_keypoint_windows(keypoints):
+                prediction = self._predict_keypoint_window(interpreter, window_keypoints, window_index, start, end)
+                segments.append(prediction)
+
+            accepted_glosses = self._collapse_segments(segments)
+            best_segment = max(segments, key=lambda item: float(item.get("confidence", 0.0) or 0.0), default=None)
+            confidence = float(best_segment.get("confidence", 0.0) or 0.0) if best_segment else 0.0
+            top_prediction = best_segment.get("label") if best_segment else None
+
+            base.update(
+                {
+                    "status": "ok" if accepted_glosses else "low_confidence",
+                    "gloss_sequence": accepted_glosses,
+                    "top_prediction": top_prediction,
+                    "confidence": confidence,
+                    "top_predictions": self._merge_top_predictions(segments),
+                    "segment_predictions": segments,
+                    "windows_used": len(segments),
+                }
+            )
+            return base
+        except Exception as exc:
+            base.update(
+                {
+                    "status": "inference_error",
+                    "error": f"{type(exc).__name__}: {exc}",
+                }
+            )
+            return base
+
     def _load_interpreter(self) -> Any:
         try:
             from tensorflow.lite.python.interpreter import Interpreter
@@ -104,6 +170,73 @@ class ASLDetector:
         interpreter.invoke()
         return np.asarray(interpreter.get_tensor(output_details[0]["index"]))
 
+    def _iter_keypoint_windows(self, keypoints: np.ndarray):
+        frame_count = int(keypoints.shape[0])
+        if frame_count <= self.sequence_window:
+            yield 0, 0, frame_count, keypoints
+            return
+
+        window_index = 0
+        last_start = frame_count - self.sequence_window
+        for start in range(0, last_start + 1, self.sequence_stride):
+            end = start + self.sequence_window
+            yield window_index, start, end, keypoints[start:end]
+            window_index += 1
+
+        covered_end = (window_index - 1) * self.sequence_stride + self.sequence_window if window_index else 0
+        if covered_end < frame_count:
+            yield window_index, last_start, frame_count, keypoints[last_start:frame_count]
+
+    def _predict_keypoint_window(
+        self,
+        interpreter: Any,
+        keypoints: np.ndarray,
+        window_index: int,
+        start_frame: int,
+        end_frame: int,
+    ) -> dict[str, Any]:
+        output = self._predict(interpreter, keypoints.astype(np.float32))
+        probs = self._softmax_if_needed(np.asarray(output).reshape(-1))
+        top_idx = int(np.argmax(probs))
+        label = self._label_for_index(top_idx)
+        confidence = float(probs[top_idx])
+        return {
+            "window_index": window_index,
+            "start_frame": start_frame,
+            "end_frame": end_frame,
+            "label": label,
+            "confidence": confidence,
+            "accepted": bool(label) and confidence >= self.confidence_threshold,
+            "top_predictions": self._top_predictions(probs, limit=5),
+        }
+
+    def _collapse_segments(self, segments: list[dict[str, Any]]) -> list[str]:
+        glosses = []
+        for segment in segments:
+            if not segment.get("accepted"):
+                continue
+            label = segment.get("label")
+            if not label:
+                continue
+            if glosses and glosses[-1] == label:
+                continue
+            glosses.append(str(label))
+        return glosses
+
+    def _merge_top_predictions(self, segments: list[dict[str, Any]], limit: int = 5) -> list[dict[str, Any]]:
+        best_by_label: dict[str, float] = {}
+        for segment in segments:
+            for item in segment.get("top_predictions", []):
+                label = str(item.get("label") or "")
+                if not label:
+                    continue
+                confidence = float(item.get("confidence", 0.0) or 0.0)
+                best_by_label[label] = max(best_by_label.get(label, 0.0), confidence)
+        return [
+            {"label": label, "confidence": confidence}
+            for label, confidence in sorted(best_by_label.items(), key=lambda item: item[1], reverse=True)[:limit]
+        ]
+
     def _signature_input_name(self, signature: dict[str, Any]) -> str:
         inputs = signature.get("inputs") or ["inputs"]
         return inputs[0]

signspeak/asl/pipeline.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import os
 from pathlib import Path
 from typing import Any
 
@@ -7,12 +8,16 @@ import numpy as np
 
 from .asl_detector import ASLDetector
 from .emotion_detector import detect_emotion_on_frames
-from .video_utils import sample_video_frames, sample_video_frames_for_emotion
+from .video_utils import sample_video_frames_for_emotion, sample_video_frames_sequential
 
 
 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_sequential(
+        path,
+        target_fps=float(os.getenv("ASL_UPLOAD_TARGET_FPS", "12")),
+        max_frames=int(os.getenv("ASL_UPLOAD_MAX_FRAMES", "240")),
+    )
     emotion_frames = sample_video_frames_for_emotion(path, target_frames=12)
 
     return process_asl_frames(asl_frames, emotion_frames, source=str(path))
@@ -25,7 +30,7 @@ def process_asl_frames(
     source: str = "frames",
 ) -> dict[str, Any]:
     emotion_frames = emotion_frames if emotion_frames is not None else asl_frames[:12]
-    asl = ASLDetector().predict_from_frames(asl_frames)
+    asl = ASLDetector().predict_sequence_from_frames(asl_frames)
     emotion = detect_emotion_on_frames(emotion_frames)
     intent_input = build_intent_input(asl, emotion)
 
@@ -38,7 +43,12 @@ def process_asl_frames(
             "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", []),
+            "segment_predictions": asl.get("segment_predictions", []),
             "frames_used": int(asl.get("frames_used", len(asl_frames)) or 0),
+            "windows_used": int(asl.get("windows_used", 0) or 0),
+            "sequence_window": int(asl.get("sequence_window", 0) or 0),
+            "sequence_stride": int(asl.get("sequence_stride", 0) or 0),
+            "recognition_mode": asl.get("recognition_mode"),
             "keypoints_shape": asl.get("keypoints_shape", []),
             "landmarks_status": asl.get("landmarks_status"),
             "landmarks_detector": asl.get("landmarks_detector"),
@@ -56,8 +66,19 @@ def build_intent_input(asl: dict[str, Any], emotion: dict[str, Any]) -> dict[str
     threshold = float(asl.get("confidence_threshold", 0.0) or 0.0)
     top_prediction = asl.get("top_prediction")
     top_predictions = asl.get("top_predictions", [])
+    segments = asl.get("segment_predictions", [])
     return {
         "detected_glosses": glosses,
+        "candidate_gloss_sequence": [
+            {
+                "gloss": segment.get("label"),
+                "confidence": float(segment.get("confidence", 0.0) or 0.0),
+                "accepted": bool(segment.get("accepted")),
+                "start_frame": int(segment.get("start_frame", 0) or 0),
+                "end_frame": int(segment.get("end_frame", 0) or 0),
+            }
+            for segment in segments
+        ],
         "detected_facial_expression": dominant_emotion,
         "emotion_profile": {
             "dominant": dominant_emotion,
@@ -73,6 +94,11 @@ def build_intent_input(asl: dict[str, Any], emotion: dict[str, Any]) -> dict[str
             "confidence": confidence,
             "confidence_threshold": threshold,
             "top_predictions": top_predictions,
+            "segment_predictions": segments,
+            "windows_used": int(asl.get("windows_used", 0) or 0),
+            "sequence_window": int(asl.get("sequence_window", 0) or 0),
+            "sequence_stride": int(asl.get("sequence_stride", 0) or 0),
+            "recognition_mode": asl.get("recognition_mode"),
             "landmarks_status": asl.get("landmarks_status"),
             "landmarks_detector": asl.get("landmarks_detector"),
         },
@@ -81,6 +107,7 @@ def build_intent_input(asl: dict[str, Any], emotion: dict[str, Any]) -> dict[str
             "asl_status": asl.get("status", "unknown"),
             "emotion_status": emotion.get("status", "unknown"),
             "frames_used": int(asl.get("frames_used", 0) or 0),
+            "windows_used": int(asl.get("windows_used", 0) or 0),
             "top_prediction": top_prediction,
             "sign_confidence": confidence,
             "confidence_threshold": threshold,

signspeak/asl/video_utils.py

@@ -8,6 +8,47 @@ def sample_video_frames(video_path: str | Path, target_frames: int = 30) -> list
     return _sample_video_frames(video_path, target_frames)
 
 
+def sample_video_frames_sequential(
+    video_path: str | Path,
+    *,
+    target_fps: float = 12.0,
+    max_frames: int = 240,
+) -> list:
+    """Read RGB frames in temporal order for sliding-window sign recognition."""
+    cv2 = _load_cv2()
+    path = Path(video_path)
+    if max_frames <= 0:
+        return []
+    if not path.exists():
+        raise FileNotFoundError(f"Video not found: {path}")
+
+    cap = cv2.VideoCapture(str(path))
+    if not cap.isOpened():
+        raise ValueError(f"Could not open video: {path}")
+
+    try:
+        source_fps = float(cap.get(cv2.CAP_PROP_FPS) or 0.0)
+        stride = 1
+        if source_fps > 0 and target_fps > 0:
+            stride = max(1, round(source_fps / target_fps))
+
+        frames = []
+        frame_idx = 0
+        while len(frames) < max_frames:
+            ok, frame_bgr = cap.read()
+            if not ok or frame_bgr is None:
+                break
+            if frame_idx % stride == 0:
+                frames.append(cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB))
+            frame_idx += 1
+
+        if not frames:
+            raise ValueError(f"No readable frames found in: {path}")
+        return frames
+    finally:
+        cap.release()
+
+
 def sample_video_frames_for_emotion(video_path: str | Path, target_frames: int = 12) -> list:
     """Sample fewer RGB frames for temporal emotion aggregation."""
     return _sample_video_frames(video_path, target_frames)

signspeak/debug_video.py

@@ -42,9 +42,10 @@ def create_debug_overlay_video(video_path: str | Path, result: dict[str, Any]) -
     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)
+    windows_used = int(asl.get("windows_used", 0) or 0)
     status_text = (
         f"ASL {asl.get('status', 'unknown')} | top {top_prediction} "
-        f"{confidence:.2f}/{threshold:.2f} | EMOTION {emotion.get('status', 'unknown')}"
+        f"{confidence:.2f}/{threshold:.2f} | windows {windows_used} | EMOTION {emotion.get('status', 'unknown')}"
     )
 
     try:

signspeak/pipeline.py

@@ -135,11 +135,14 @@ def summarize_asl_result(result: dict[str, Any]) -> str:
     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", []))
+    windows_used = int(asl.get("windows_used", 0) or 0)
+    recognition_mode = asl.get("recognition_mode") or "single_clip"
     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"Recognition: {recognition_mode}, windows {windows_used}\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"

tests/test_asl_detector.py

@@ -85,3 +85,52 @@ 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_sequence_prediction_collapses_accepted_windows(monkeypatch, tmp_path):
+    model_dir = tmp_path / "asl"
+    model_dir.mkdir()
+    (model_dir / "model.tflite").write_bytes(b"demo")
+    (model_dir / "sign_to_prediction_index_map.json").write_text(
+        json.dumps({"hello": 0, "water": 1}),
+        encoding="utf-8",
+    )
+
+    class SequenceLandmarkResult:
+        keypoints = np.zeros((60, 543, 3), dtype=np.float32)
+        status = "ok"
+        detector = "fake"
+        error = None
+
+    class SequenceLandmarksDetector:
+        def __init__(self, missing_value=0.0):
+            pass
+
+        def detect_sequence(self, frames):
+            return SequenceLandmarkResult()
+
+        def close(self):
+            pass
+
+    calls = iter(
+        [
+            np.asarray([[0.9, 0.1]], dtype=np.float32),
+            np.asarray([[0.86, 0.14]], dtype=np.float32),
+            np.asarray([[0.1, 0.9]], dtype=np.float32),
+        ]
+    )
+    monkeypatch.setattr(asl_detector_module, "LandmarksDetector", SequenceLandmarksDetector)
+    monkeypatch.setenv("ASL_CONFIDENCE_THRESHOLD", "0.70")
+    monkeypatch.setenv("ASL_SEQUENCE_WINDOW", "30")
+    monkeypatch.setenv("ASL_SEQUENCE_STRIDE", "15")
+    monkeypatch.setattr(ASLDetector, "_load_interpreter", lambda self: object())
+    monkeypatch.setattr(ASLDetector, "_predict", lambda self, interpreter, keypoints: next(calls))
+
+    result = ASLDetector(model_dir=model_dir).predict_sequence_from_frames(
+        [np.zeros((2, 2, 3), dtype=np.uint8)] * 60
+    )
+
+    assert result["status"] == "ok"
+    assert result["gloss_sequence"] == ["hello", "water"]
+    assert result["windows_used"] == 3
+    assert result["recognition_mode"] == "sliding_window_sequence"

tests/test_asl_pipeline.py

@@ -59,7 +59,7 @@ def test_build_intent_input_exposes_rejected_top_prediction():
 
 def test_process_asl_frames_preserves_detector_diagnostics(monkeypatch):
     class FakeASLDetector:
-        def predict_from_frames(self, frames):
+        def predict_sequence_from_frames(self, frames):
             return {
                 "status": "ok",
                 "gloss_sequence": ["talk"],
@@ -67,7 +67,21 @@ def test_process_asl_frames_preserves_detector_diagnostics(monkeypatch):
                 "confidence": 0.96,
                 "confidence_threshold": 0.70,
                 "top_predictions": [{"label": "talk", "confidence": 0.96}],
+                "segment_predictions": [
+                    {
+                        "window_index": 0,
+                        "start_frame": 0,
+                        "end_frame": 8,
+                        "label": "talk",
+                        "confidence": 0.96,
+                        "accepted": True,
+                    }
+                ],
                 "frames_used": len(frames),
+                "windows_used": 1,
+                "sequence_window": 30,
+                "sequence_stride": 15,
+                "recognition_mode": "sliding_window_sequence",
                 "keypoints_shape": [8, 543, 3],
                 "landmarks_status": "ok",
                 "landmarks_detector": "holistic",
@@ -91,6 +105,7 @@ def test_process_asl_frames_preserves_detector_diagnostics(monkeypatch):
     assert result["asl"]["top_predictions"] == [{"label": "talk", "confidence": 0.96}]
     assert result["intent_input"]["detected_glosses"] == ["talk"]
     assert result["intent_input"]["sign_detection"]["top_prediction"] == "talk"
+    assert result["intent_input"]["candidate_gloss_sequence"][0]["gloss"] == "talk"
 
 
 def test_summarize_asl_result_is_stable_for_missing_fields():