Import ASL video pipeline

signspeak/__init__.py

@@ -0,0 +1,2 @@
+"""SignSpeak local ASL-to-speech pipeline package."""
+

signspeak/asl/__init__.py

@@ -0,0 +1,6 @@
+"""ASL video, landmark, and emotion processing helpers."""
+
+from .pipeline import build_intent_input, process_asl_frames, process_asl_video
+
+__all__ = ["build_intent_input", "process_asl_frames", "process_asl_video"]
+

signspeak/asl/asl_detector.py

@@ -0,0 +1,149 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+
+from .landmarks_detector import LandmarksDetector
+
+
+class ASLDetector:
+    def __init__(self, model_dir: str | Path | None = None) -> None:
+        repo_root = Path(__file__).resolve().parents[2]
+        default_model_dir = repo_root / "data" / "models" / "asl"
+        configured_model_dir = model_dir or __import__("os").getenv("ASL_MODEL_DIR")
+        self.model_dir = Path(configured_model_dir) if configured_model_dir else default_model_dir
+        self.model_path = self.model_dir / "model.tflite"
+        self.train_csv_path = self.model_dir / "train.csv"
+        self.labels = self._load_labels()
+
+    def predict_from_frames(self, frames: list[np.ndarray]) -> dict[str, Any]:
+        detector = LandmarksDetector(missing_value=0.0)
+        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,
+            "frames_used": len(frames),
+            "keypoints_shape": list(keypoints.shape),
+            "landmarks_status": landmark_result.status,
+            "landmarks_detector": landmark_result.detector,
+        }
+        if landmark_result.error:
+            base["landmarks_error"] = landmark_result.error
+
+        if not self.model_path.exists():
+            return base
+
+        try:
+            interpreter = self._load_interpreter()
+            interpreter.allocate_tensors()
+            input_details = interpreter.get_input_details()
+            output_details = interpreter.get_output_details()
+
+            input_data = self._prepare_input(keypoints, input_details[0])
+            interpreter.set_tensor(input_details[0]["index"], input_data)
+            interpreter.invoke()
+            output = interpreter.get_tensor(output_details[0]["index"])
+
+            probs = self._softmax_if_needed(np.asarray(output).reshape(-1))
+            top_idx = int(np.argmax(probs))
+            top_prediction = self._label_for_index(top_idx)
+            confidence = float(probs[top_idx])
+
+            base.update(
+                {
+                    "status": "ok",
+                    "gloss_sequence": [top_prediction] if top_prediction else [],
+                    "top_prediction": top_prediction,
+                    "confidence": confidence,
+                }
+            )
+            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
+
+            return Interpreter(model_path=str(self.model_path))
+        except Exception:
+            from tflite_runtime.interpreter import Interpreter
+
+            return Interpreter(model_path=str(self.model_path))
+
+    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)
+        data = np.nan_to_num(keypoints, nan=0.0, posinf=0.0, neginf=0.0).astype(np.float32)
+
+        if shape is None or len(shape) == 0:
+            return data.astype(dtype)
+
+        target_shape = [int(dim) for dim in shape]
+        if target_shape[0] == 1:
+            no_batch_shape = target_shape[1:]
+        else:
+            no_batch_shape = target_shape
+
+        prepared = self._fit_to_shape(data, no_batch_shape)
+        if target_shape[0] == 1:
+            prepared = np.expand_dims(prepared, axis=0)
+        return prepared.astype(dtype)
+
+    def _fit_to_shape(self, data: np.ndarray, target_shape: list[int]) -> np.ndarray:
+        if list(data.shape) == target_shape:
+            return data
+
+        flat = data.reshape(-1)
+        target_size = int(np.prod(target_shape))
+        fitted = np.zeros(target_size, dtype=np.float32)
+        copy_size = min(flat.size, target_size)
+        fitted[:copy_size] = flat[:copy_size]
+        return fitted.reshape(target_shape)
+
+    def _load_labels(self) -> list[str]:
+        if not self.train_csv_path.exists():
+            return []
+        try:
+            import pandas as pd
+
+            df = pd.read_csv(self.train_csv_path)
+            for column in ("sign", "label", "gloss", "target"):
+                if column in df.columns:
+                    values = df[column].dropna().astype(str).unique().tolist()
+                    return sorted(values)
+        except Exception:
+            return []
+        return []
+
+    def _label_for_index(self, index: int) -> str | None:
+        if 0 <= index < len(self.labels):
+            return self.labels[index]
+        return str(index)
+
+    def _softmax_if_needed(self, values: np.ndarray) -> np.ndarray:
+        if values.size == 0:
+            return np.asarray([0.0], dtype=np.float32)
+        if np.all(values >= 0) and np.isclose(np.sum(values), 1.0, atol=1e-3):
+            return values.astype(np.float32)
+        shifted = values - np.max(values)
+        exp = np.exp(shifted)
+        denom = np.sum(exp)
+        if denom <= 0:
+            return np.zeros_like(values, dtype=np.float32)
+        return (exp / denom).astype(np.float32)

signspeak/asl/emotion_detector.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+import os
+from collections import defaultdict
+from typing import Any
+
+import cv2
+import numpy as np
+
+
+EMOTIONS = ("angry", "disgust", "fear", "happy", "sad", "surprise", "neutral")
+
+
+def detect_emotion_on_frames(frames: list[np.ndarray]) -> dict[str, Any]:
+    if not frames:
+        return _error_result("No frames provided", frames_analyzed=0)
+
+    try:
+        os.environ.setdefault("TF_USE_LEGACY_KERAS", "1")
+        from deepface import DeepFace
+    except Exception as exc:
+        return _error_result(f"DeepFace import failed: {type(exc).__name__}: {exc}", frames_analyzed=0)
+
+    scores_sum: dict[str, float] = defaultdict(float)
+    frames_analyzed = 0
+    errors = []
+
+    for frame_rgb in frames:
+        try:
+            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 {}
+            emotion = analysis.get("emotion", {}) if isinstance(analysis, dict) else {}
+            if not emotion:
+                continue
+
+            total = 0.0
+            normalized = {}
+            for key, value in emotion.items():
+                score = float(value)
+                normalized[key] = score
+                total += score
+            if total > 1.5:
+                normalized = {key: value / 100.0 for key, value in normalized.items()}
+
+            for key in EMOTIONS:
+                scores_sum[key] += float(normalized.get(key, 0.0))
+            frames_analyzed += 1
+        except Exception as exc:
+            errors.append(f"{type(exc).__name__}: {exc}")
+
+    if frames_analyzed == 0:
+        message = errors[0] if errors else "No emotion scores produced"
+        return _error_result(message, frames_analyzed=0)
+
+    averaged = {key: scores_sum[key] / frames_analyzed for key in EMOTIONS}
+    dominant = max(averaged, key=averaged.get)
+    intensity = float(np.clip(averaged[dominant], 0.0, 1.0))
+
+    result = {
+        "status": "ok",
+        "dominant_emotion": dominant,
+        "emotion_scores": averaged,
+        "intensity": intensity,
+        "frames_analyzed": frames_analyzed,
+    }
+    if errors:
+        result["frame_errors"] = errors[:3]
+    return result
+
+
+def _error_result(error: str, frames_analyzed: int) -> dict[str, Any]:
+    return {
+        "status": "emotion_error",
+        "dominant_emotion": "unknown",
+        "emotion_scores": {key: 0.0 for key in EMOTIONS},
+        "intensity": 0.0,
+        "frames_analyzed": frames_analyzed,
+        "error": error,
+    }

signspeak/asl/landmarks_detector.py

@@ -0,0 +1,192 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+import numpy as np
+
+
+FACE_POINTS = 468
+HAND_POINTS = 21
+POSE_POINTS = 33
+TOTAL_POINTS = FACE_POINTS + HAND_POINTS + POSE_POINTS + HAND_POINTS
+
+
+@dataclass
+class LandmarkDetectionResult:
+    keypoints: np.ndarray
+    detector: str
+    status: str
+    error: str | None = None
+
+
+class LandmarksDetector:
+    """MediaPipe detector with Holistic first and component fallback."""
+
+    def __init__(self, missing_value: float = 0.0) -> None:
+        self.missing_value = missing_value
+        self.mp: Any | None = None
+        self.mode = "unavailable"
+        self.error: str | None = None
+        self._holistic = None
+        self._hands = None
+        self._face_mesh = None
+        self._pose = None
+        self._load_mediapipe()
+
+    def close(self) -> None:
+        for detector in (self._holistic, self._hands, self._face_mesh, self._pose):
+            if detector is not None:
+                detector.close()
+
+    def detect_sequence(self, frames: list[np.ndarray]) -> LandmarkDetectionResult:
+        if self.mp is None:
+            return LandmarkDetectionResult(
+                keypoints=self._empty_sequence(len(frames)),
+                detector=self.mode,
+                status="landmarks_unavailable",
+                error=self.error,
+            )
+
+        output = []
+        try:
+            for frame in frames:
+                output.append(self._detect_frame(frame))
+            return LandmarkDetectionResult(
+                keypoints=np.asarray(output, dtype=np.float32),
+                detector=self.mode,
+                status="ok",
+            )
+        except Exception as exc:
+            return LandmarkDetectionResult(
+                keypoints=self._empty_sequence(len(frames)),
+                detector=self.mode,
+                status="landmarks_error",
+                error=f"{type(exc).__name__}: {exc}",
+            )
+
+    def _load_mediapipe(self) -> None:
+        try:
+            import mediapipe as mp
+
+            self.mp = mp
+        except Exception as exc:
+            self.error = f"MediaPipe import failed: {type(exc).__name__}: {exc}"
+            return
+
+        try:
+            self._holistic = self.mp.solutions.holistic.Holistic(
+                static_image_mode=False,
+                model_complexity=1,
+                smooth_landmarks=True,
+                refine_face_landmarks=False,
+                min_detection_confidence=0.5,
+                min_tracking_confidence=0.5,
+            )
+            self.mode = "holistic"
+            return
+        except Exception as exc:
+            self.error = f"Holistic unavailable, using fallback if possible: {type(exc).__name__}: {exc}"
+
+        self._init_component_fallback()
+
+    def _init_component_fallback(self) -> bool:
+        if self.mp is None:
+            return False
+        if self._hands is not None and self._face_mesh is not None and self._pose is not None:
+            self.mode = "components"
+            return True
+
+        try:
+            self._hands = self.mp.solutions.hands.Hands(
+                static_image_mode=False,
+                max_num_hands=2,
+                min_detection_confidence=0.5,
+                min_tracking_confidence=0.5,
+            )
+            self._face_mesh = self.mp.solutions.face_mesh.FaceMesh(
+                static_image_mode=False,
+                max_num_faces=1,
+                refine_landmarks=False,
+                min_detection_confidence=0.5,
+                min_tracking_confidence=0.5,
+            )
+            self._pose = self.mp.solutions.pose.Pose(
+                static_image_mode=False,
+                model_complexity=1,
+                smooth_landmarks=True,
+                min_detection_confidence=0.5,
+                min_tracking_confidence=0.5,
+            )
+            self.mode = "components"
+            return True
+        except Exception as exc:
+            self.mode = "unavailable"
+            self.error = f"MediaPipe component fallback failed: {type(exc).__name__}: {exc}"
+            return False
+
+    def _detect_frame(self, frame_rgb: np.ndarray) -> np.ndarray:
+        if self.mode == "holistic":
+            try:
+                results = self._holistic.process(frame_rgb)
+                return self._from_holistic(results)
+            except Exception as exc:
+                self.error = f"Holistic process failed, switched to components: {type(exc).__name__}: {exc}"
+                if self._holistic is not None:
+                    self._holistic.close()
+                    self._holistic = None
+                if self._init_component_fallback():
+                    return self._from_components(frame_rgb)
+                raise
+        if self.mode == "components":
+            return self._from_components(frame_rgb)
+        return self._empty_frame()
+
+    def _from_holistic(self, results: Any) -> np.ndarray:
+        face = self._landmark_array(getattr(results, "face_landmarks", None), FACE_POINTS)
+        left = self._landmark_array(getattr(results, "left_hand_landmarks", None), HAND_POINTS)
+        pose = self._landmark_array(getattr(results, "pose_landmarks", None), POSE_POINTS)
+        right = self._landmark_array(getattr(results, "right_hand_landmarks", None), HAND_POINTS)
+        return np.vstack([face, left, pose, right]).astype(np.float32)
+
+    def _from_components(self, frame_rgb: np.ndarray) -> np.ndarray:
+        face_results = self._face_mesh.process(frame_rgb) if self._face_mesh else None
+        pose_results = self._pose.process(frame_rgb) if self._pose else None
+        hands_results = self._hands.process(frame_rgb) if self._hands else None
+
+        face_landmarks = None
+        if getattr(face_results, "multi_face_landmarks", None):
+            face_landmarks = face_results.multi_face_landmarks[0]
+
+        left_landmarks = None
+        right_landmarks = None
+        if getattr(hands_results, "multi_hand_landmarks", None):
+            handedness = getattr(hands_results, "multi_handedness", []) or []
+            for hand_lms, hand_info in zip(hands_results.multi_hand_landmarks, handedness):
+                label = hand_info.classification[0].label.lower()
+                if label == "left":
+                    left_landmarks = hand_lms
+                elif label == "right":
+                    right_landmarks = hand_lms
+
+        face = self._landmark_array(face_landmarks, FACE_POINTS)
+        left = self._landmark_array(left_landmarks, HAND_POINTS)
+        pose = self._landmark_array(getattr(pose_results, "pose_landmarks", None), POSE_POINTS)
+        right = self._landmark_array(right_landmarks, HAND_POINTS)
+        return np.vstack([face, left, pose, right]).astype(np.float32)
+
+    def _landmark_array(self, landmark_list: Any, expected_points: int) -> np.ndarray:
+        empty = np.full((expected_points, 3), self.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
+
+    def _empty_frame(self) -> np.ndarray:
+        return np.full((TOTAL_POINTS, 3), self.missing_value, dtype=np.float32)
+
+    def _empty_sequence(self, frame_count: int) -> np.ndarray:
+        return np.full((frame_count, TOTAL_POINTS, 3), self.missing_value, dtype=np.float32)

signspeak/asl/pipeline.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+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
+
+
+def process_asl_video(video_path: str | Path) -> dict[str, Any]:
+    path = Path(video_path)
+    asl_frames = sample_video_frames(path, target_frames=30)
+    emotion_frames = sample_video_frames_for_emotion(path, target_frames=12)
+
+    return process_asl_frames(asl_frames, emotion_frames, source=str(path))
+
+
+def process_asl_frames(
+    asl_frames: list[np.ndarray],
+    emotion_frames: list[np.ndarray] | None = None,
+    *,
+    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)
+    emotion = detect_emotion_on_frames(emotion_frames)
+    intent_input = build_intent_input(asl, emotion)
+
+    return {
+        "source": source,
+        "asl": {
+            "status": asl.get("status", "unknown"),
+            "gloss_sequence": asl.get("gloss_sequence", []),
+            "top_prediction": asl.get("top_prediction"),
+            "confidence": float(asl.get("confidence", 0.0) or 0.0),
+            "frames_used": int(asl.get("frames_used", len(asl_frames)) or 0),
+            "keypoints_shape": asl.get("keypoints_shape", []),
+            "landmarks_status": asl.get("landmarks_status"),
+            "landmarks_detector": asl.get("landmarks_detector"),
+            **_optional_error_fields(asl, ("error", "landmarks_error")),
+        },
+        "emotion": emotion,
+        "intent_input": intent_input,
+    }
+
+
+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")
+    return {
+        "detected_glosses": glosses,
+        "detected_facial_expression": dominant_emotion,
+        "emotion_profile": {
+            "dominant": dominant_emotion,
+            "confidence": float(emotion.get("intensity", 0.0) or 0.0),
+            "scores": emotion.get("emotion_scores", {}),
+        },
+        "communication_intent": "derived_from_asl_video",
+        "sign_confidence": float(asl.get("confidence", 0.0) or 0.0),
+        "pipeline_stage": "asl_video_to_llama_cpp_intent",
+        "diagnostics": {
+            "asl_status": asl.get("status", "unknown"),
+            "emotion_status": emotion.get("status", "unknown"),
+            "frames_used": int(asl.get("frames_used", 0) or 0),
+        },
+    }
+
+
+def _optional_error_fields(data: dict[str, Any], keys: tuple[str, ...]) -> dict[str, Any]:
+    return {key: data[key] for key in keys if key in data and data[key]}

signspeak/asl/video_utils.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+
+def sample_video_frames(video_path: str | Path, target_frames: int = 30) -> list:
+    """Sample RGB frames uniformly from a video."""
+    return _sample_video_frames(video_path, target_frames)
+
+
+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)
+
+
+def _sample_video_frames(video_path: str | Path, target_frames: int) -> list:
+    cv2 = _load_cv2()
+    path = Path(video_path)
+    if target_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:
+        frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT) or 0)
+        if frame_count <= 0:
+            return _sample_unknown_length(cap, target_frames)
+
+        if frame_count <= target_frames:
+            indices = list(range(frame_count))
+        else:
+            step = (frame_count - 1) / float(target_frames - 1) if target_frames > 1 else 0
+            indices = [round(i * step) for i in range(target_frames)]
+
+        frames = []
+        for idx in indices:
+            cap.set(cv2.CAP_PROP_POS_FRAMES, idx)
+            ok, frame_bgr = cap.read()
+            if ok and frame_bgr is not None:
+                frames.append(cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB))
+
+        if not frames:
+            raise ValueError(f"No readable frames found in: {path}")
+        return frames
+    finally:
+        cap.release()
+
+
+def _sample_unknown_length(cap, target_frames: int) -> list:
+    cv2 = _load_cv2()
+    frames = []
+    while True:
+        ok, frame_bgr = cap.read()
+        if not ok or frame_bgr is None:
+            break
+        frames.append(cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB))
+
+    if not frames:
+        return []
+    if len(frames) <= target_frames:
+        return frames
+
+    step = (len(frames) - 1) / float(target_frames - 1) if target_frames > 1 else 0
+    return [frames[round(i * step)] for i in range(target_frames)]
+
+
+def _load_cv2():
+    try:
+        import cv2
+
+        return cv2
+    except Exception as exc:
+        raise RuntimeError(
+            "OpenCV is required for video sampling. Install opencv-python-headless "
+            "or opencv-contrib-python to enable this brick."
+        ) from exc