Add experimental WLASL I3D backend

.gitignore

@@ -2,4 +2,5 @@ __pycache__/
 *.py[cod]
 .pytest_cache/
 data/examples/*.mp4
+data/models/wlasl_i3d/
 external/

README.md

@@ -102,6 +102,24 @@ 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.
 
+An experimental WLASL2000 I3D backend is also available for broader vocabulary coverage. It uses
+`raghuhasan/asl2000-i3d` from Hugging Face, downloads the I3D architecture helper if needed, and
+falls back to the TFLite detector when `ASL_DETECTOR_BACKEND=auto` cannot initialize it.
+
+```text
+ASL_DETECTOR_BACKEND=tflite     # default lightweight backend
+ASL_DETECTOR_BACKEND=wlasl_i3d  # WLASL2000 I3D only
+ASL_DETECTOR_BACKEND=auto       # try WLASL2000, then fallback to TFLite
+WLASL_I3D_CONFIDENCE_THRESHOLD=0.20
+WLASL_I3D_SEQUENCE_WINDOW=64
+WLASL_I3D_SEQUENCE_STRIDE=32
+WLASL_I3D_FRAME_SIZE=224
+```
+
+The WLASL backend is heavier and more experimental. The model card reports 2,000 classes with
+32.48% top-1, 57.31% top-5, and 66.31% top-10 accuracy, so the UI exposes top candidates and
+segment diagnostics instead of hiding uncertainty.
+
 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/pipeline.py

@@ -9,6 +9,7 @@ import numpy as np
 from .asl_detector import ASLDetector
 from .emotion_detector import detect_emotion_on_frames
 from .video_utils import sample_video_frames_for_emotion, sample_video_frames_sequential
+from .wlasl_i3d_detector import WLASLI3DDetector
 
 
 def process_asl_video(video_path: str | Path) -> dict[str, Any]:
@@ -30,7 +31,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_sequence_from_frames(asl_frames)
+    asl = predict_asl_sequence(asl_frames)
     emotion = detect_emotion_on_frames(emotion_frames)
     intent_input = build_intent_input(asl, emotion)
 
@@ -49,6 +50,9 @@ def process_asl_frames(
             "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"),
+            "model_repo_id": asl.get("model_repo_id"),
+            "label_count": int(asl.get("label_count", 0) or 0),
+            "fallback_from": asl.get("fallback_from"),
             "keypoints_shape": asl.get("keypoints_shape", []),
             "landmarks_status": asl.get("landmarks_status"),
             "landmarks_detector": asl.get("landmarks_detector"),
@@ -59,6 +63,24 @@ def process_asl_frames(
     }
 
 
+def predict_asl_sequence(asl_frames: list[np.ndarray]) -> dict[str, Any]:
+    backend = os.getenv("ASL_DETECTOR_BACKEND", "tflite").strip().lower()
+    if backend in ("wlasl", "wlasl_i3d", "i3d"):
+        return WLASLI3DDetector().predict_sequence_from_frames(asl_frames)
+    if backend == "auto":
+        wlasl = WLASLI3DDetector().predict_sequence_from_frames(asl_frames)
+        if wlasl.get("status") not in ("wlasl_i3d_unavailable", "wlasl_i3d_error"):
+            return wlasl
+        tflite = ASLDetector().predict_sequence_from_frames(asl_frames)
+        tflite["fallback_from"] = {
+            "backend": "wlasl_i3d",
+            "status": wlasl.get("status"),
+            "error": wlasl.get("error"),
+        }
+        return tflite
+    return ASLDetector().predict_sequence_from_frames(asl_frames)
+
+
 def build_intent_input(asl: dict[str, Any], emotion: dict[str, Any]) -> dict[str, Any]:
     glosses = asl.get("gloss_sequence", []) or []
     dominant_emotion = emotion.get("dominant_emotion", "unknown")
@@ -99,6 +121,9 @@ def build_intent_input(asl: dict[str, Any], emotion: dict[str, Any]) -> dict[str
             "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"),
+            "model_repo_id": asl.get("model_repo_id"),
+            "label_count": int(asl.get("label_count", 0) or 0),
+            "fallback_from": asl.get("fallback_from"),
             "landmarks_status": asl.get("landmarks_status"),
             "landmarks_detector": asl.get("landmarks_detector"),
         },
@@ -112,6 +137,8 @@ def build_intent_input(asl: dict[str, Any], emotion: dict[str, Any]) -> dict[str
             "sign_confidence": confidence,
             "confidence_threshold": threshold,
             "accepted": bool(glosses),
+            "recognition_mode": asl.get("recognition_mode"),
+            "fallback_from": asl.get("fallback_from"),
         },
     }
 

signspeak/asl/wlasl_i3d_detector.py

@@ -0,0 +1,263 @@
+from __future__ import annotations
+
+import json
+import importlib.util
+import os
+import urllib.request
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+
+
+WLASL_MODEL_REPO_ID = "raghuhasan/asl2000-i3d"
+WLASL_LABELS_URL = "https://raw.githubusercontent.com/dxli94/WLASL/master/start_kit/WLASL_v0.3.json"
+PYTORCH_I3D_URL = "https://raw.githubusercontent.com/piergiaj/pytorch-i3d/master/pytorch_i3d.py"
+
+
+class WLASLI3DDetector:
+    """Experimental WLASL2000 video classifier backed by a Hugging Face I3D checkpoint."""
+
+    def __init__(self, cache_dir: str | Path | None = None) -> None:
+        repo_root = Path(__file__).resolve().parents[2]
+        self.cache_dir = Path(cache_dir or os.getenv("WLASL_I3D_CACHE_DIR") or repo_root / "data" / "models" / "wlasl_i3d")
+        self.repo_id = os.getenv("WLASL_I3D_REPO_ID", WLASL_MODEL_REPO_ID)
+        self.confidence_threshold = float(os.getenv("WLASL_I3D_CONFIDENCE_THRESHOLD", "0.20"))
+        self.sequence_window = max(8, int(os.getenv("WLASL_I3D_SEQUENCE_WINDOW", "64")))
+        self.sequence_stride = max(1, int(os.getenv("WLASL_I3D_SEQUENCE_STRIDE", "32")))
+        self.frame_size = max(64, int(os.getenv("WLASL_I3D_FRAME_SIZE", "224")))
+        self.labels = self._load_labels()
+
+    def predict_sequence_from_frames(self, frames: list[np.ndarray]) -> dict[str, Any]:
+        base = {
+            "status": "wlasl_i3d_unavailable",
+            "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,
+            "recognition_mode": "wlasl_i3d_sliding_window",
+            "model_repo_id": self.repo_id,
+            "label_count": len(self.labels),
+        }
+        if not frames:
+            base["error"] = "No frames supplied to WLASL I3D detector."
+            return base
+
+        try:
+            torch = self._load_torch()
+            model = self._load_model(torch)
+            segments = []
+            for window_index, start, end, window_frames in self._iter_frame_windows(frames):
+                segments.append(self._predict_window(torch, model, window_frames, window_index, start, end))
+
+            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": "wlasl_i3d_error",
+                    "error": f"{type(exc).__name__}: {exc}",
+                }
+            )
+            return base
+
+    def _load_torch(self) -> Any:
+        try:
+            import torch
+
+            return torch
+        except Exception as exc:
+            raise RuntimeError("PyTorch is required for WLASL I3D. Install torch to enable this backend.") from exc
+
+    def _load_model(self, torch: Any) -> Any:
+        InceptionI3d = self._load_inception_i3d()
+
+        weights_path = self._download_weights()
+        model = InceptionI3d(400, in_channels=3)
+        model.replace_logits(2000)
+        state_dict = torch.load(weights_path, map_location="cpu")
+        model.load_state_dict(state_dict)
+        model.eval()
+        return model
+
+    def _load_inception_i3d(self) -> Any:
+        try:
+            from pytorch_i3d import InceptionI3d
+
+            return InceptionI3d
+        except Exception:
+            pass
+
+        module_path = self.cache_dir / "pytorch_i3d.py"
+        if not module_path.exists():
+            self.cache_dir.mkdir(parents=True, exist_ok=True)
+            try:
+                with urllib.request.urlopen(PYTORCH_I3D_URL, timeout=20) as response:
+                    module_path.write_bytes(response.read())
+            except Exception as exc:
+                raise RuntimeError("Could not download pytorch_i3d architecture for WLASL I3D.") from exc
+
+        spec = importlib.util.spec_from_file_location("signspeak_cached_pytorch_i3d", module_path)
+        if spec is None or spec.loader is None:
+            raise RuntimeError(f"Could not import cached pytorch_i3d module from {module_path}.")
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        return module.InceptionI3d
+
+    def _download_weights(self) -> str:
+        try:
+            from huggingface_hub import hf_hub_download
+
+            return hf_hub_download(self.repo_id, "pytorch_model.bin")
+        except Exception as exc:
+            raise RuntimeError(f"Could not download WLASL I3D weights from {self.repo_id}.") from exc
+
+    def _predict_window(
+        self,
+        torch: Any,
+        model: Any,
+        frames: list[np.ndarray],
+        window_index: int,
+        start_frame: int,
+        end_frame: int,
+    ) -> dict[str, Any]:
+        tensor = self._frames_to_tensor(torch, frames)
+        with torch.no_grad():
+            logits = model(tensor)
+            if logits.ndim == 3:
+                logits = logits.mean(dim=2)
+            probs = torch.softmax(logits, dim=1)[0].detach().cpu().numpy()
+        top_idx = int(np.argmax(probs))
+        confidence = float(probs[top_idx])
+        label = self._label_for_index(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=10),
+        }
+
+    def _frames_to_tensor(self, torch: Any, frames: list[np.ndarray]) -> Any:
+        cv2 = self._load_cv2()
+        processed = []
+        for frame in frames:
+            resized = cv2.resize(frame, (self.frame_size, self.frame_size), interpolation=cv2.INTER_AREA)
+            normalized = resized.astype(np.float32) / 127.5 - 1.0
+            processed.append(normalized)
+        array = np.stack(processed, axis=0)
+        array = np.transpose(array, (3, 0, 1, 2))
+        return torch.from_numpy(array).unsqueeze(0).float()
+
+    def _iter_frame_windows(self, frames: list[np.ndarray]):
+        frame_count = len(frames)
+        if frame_count <= self.sequence_window:
+            yield 0, 0, frame_count, self._fit_window(frames)
+            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, frames[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, frames[last_start:frame_count]
+
+    def _fit_window(self, frames: list[np.ndarray]) -> list[np.ndarray]:
+        if len(frames) >= self.sequence_window:
+            return frames
+        return frames + [frames[-1]] * (self.sequence_window - len(frames))
+
+    def _load_labels(self) -> list[str]:
+        labels_path = self.cache_dir / "wlasl_2000_labels.json"
+        if labels_path.exists():
+            try:
+                labels = json.loads(labels_path.read_text(encoding="utf-8"))
+                if isinstance(labels, list):
+                    return [str(label) for label in labels]
+            except Exception:
+                pass
+
+        try:
+            self.cache_dir.mkdir(parents=True, exist_ok=True)
+            with urllib.request.urlopen(WLASL_LABELS_URL, timeout=20) as response:
+                data = json.loads(response.read().decode("utf-8"))
+            labels = [str(item["gloss"]) for item in data[:2000] if "gloss" in item]
+            labels_path.write_text(json.dumps(labels, ensure_ascii=False, indent=2), encoding="utf-8")
+            return labels
+        except Exception:
+            return []
+
+    def _label_for_index(self, index: int) -> str:
+        if 0 <= index < len(self.labels):
+            return self.labels[index]
+        return str(index)
+
+    def _top_predictions(self, probs: np.ndarray, limit: int = 10) -> list[dict[str, Any]]:
+        top_indices = np.argsort(probs)[::-1][:limit]
+        return [
+            {
+                "label": self._label_for_index(int(index)),
+                "confidence": float(probs[int(index)]),
+            }
+            for index in top_indices
+        ]
+
+    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 = 10) -> 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 "")
+                confidence = float(item.get("confidence", 0.0) or 0.0)
+                if label:
+                    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 _load_cv2(self):
+        try:
+            import cv2
+
+            return cv2
+        except Exception as exc:
+            raise RuntimeError("OpenCV is required for WLASL I3D frame preprocessing.") from exc

tests/test_asl_pipeline.py

@@ -108,6 +108,43 @@ def test_process_asl_frames_preserves_detector_diagnostics(monkeypatch):
     assert result["intent_input"]["candidate_gloss_sequence"][0]["gloss"] == "talk"
 
 
+def test_predict_asl_sequence_auto_falls_back_to_tflite(monkeypatch):
+    class FakeWLASLDetector:
+        def predict_sequence_from_frames(self, frames):
+            return {
+                "status": "wlasl_i3d_error",
+                "error": "missing torch",
+                "gloss_sequence": [],
+            }
+
+    class FakeASLDetector:
+        def predict_sequence_from_frames(self, frames):
+            return {
+                "status": "ok",
+                "gloss_sequence": ["hello"],
+                "top_prediction": "hello",
+                "confidence": 0.91,
+                "confidence_threshold": 0.70,
+                "top_predictions": [{"label": "hello", "confidence": 0.91}],
+                "segment_predictions": [],
+                "frames_used": len(frames),
+                "windows_used": 1,
+                "sequence_window": 30,
+                "sequence_stride": 15,
+                "recognition_mode": "sliding_window_sequence",
+            }
+
+    monkeypatch.setenv("ASL_DETECTOR_BACKEND", "auto")
+    monkeypatch.setattr(asl_pipeline, "WLASLI3DDetector", FakeWLASLDetector)
+    monkeypatch.setattr(asl_pipeline, "ASLDetector", FakeASLDetector)
+
+    result = asl_pipeline.predict_asl_sequence([np.zeros((2, 2, 3), dtype=np.uint8)] * 30)
+
+    assert result["gloss_sequence"] == ["hello"]
+    assert result["fallback_from"]["backend"] == "wlasl_i3d"
+    assert result["fallback_from"]["error"] == "missing torch"
+
+
 def test_summarize_asl_result_is_stable_for_missing_fields():
     summary = summarize_asl_result(
         {

tests/test_wlasl_i3d_detector.py

@@ -0,0 +1,28 @@
+import json
+
+import numpy as np
+
+from signspeak.asl.wlasl_i3d_detector import WLASLI3DDetector
+
+
+def test_wlasl_detector_loads_cached_labels(tmp_path):
+    cache_dir = tmp_path / "wlasl_i3d"
+    cache_dir.mkdir()
+    (cache_dir / "wlasl_2000_labels.json").write_text(json.dumps(["book", "hello"]), encoding="utf-8")
+
+    detector = WLASLI3DDetector(cache_dir=cache_dir)
+
+    assert detector.labels == ["book", "hello"]
+    assert detector._label_for_index(1) == "hello"
+    assert detector._label_for_index(99) == "99"
+
+
+def test_wlasl_detector_reports_missing_runtime(monkeypatch, tmp_path):
+    (tmp_path / "wlasl_2000_labels.json").write_text(json.dumps(["book"]), encoding="utf-8")
+    detector = WLASLI3DDetector(cache_dir=tmp_path)
+    monkeypatch.setattr(detector, "_load_torch", lambda: (_ for _ in ()).throw(RuntimeError("missing torch")))
+
+    result = detector.predict_sequence_from_frames([np.zeros((2, 2, 3), dtype=np.uint8)])
+
+    assert result["status"] == "wlasl_i3d_error"
+    assert "missing torch" in result["error"]