Spaces:

pearlyjam21
/

reachy_mini_multimodal_emotion

Running

App Files Files Community

pearlyjam21 commited on 2 days ago

Commit

d626f8b

verified ·

1 Parent(s): d39d2a1

Reachy Mini multimodal emotion app

Browse files

Files changed (24) hide show

.gitignore +3 -0
README.md +42 -10
index.html +40 -19
pyproject.toml +36 -0
reachy_mini_multimodal_emotion/__init__.py +0 -0
reachy_mini_multimodal_emotion/engine/__init__.py +39 -0
reachy_mini_multimodal_emotion/engine/asr.py +99 -0
reachy_mini_multimodal_emotion/engine/face.py +112 -0
reachy_mini_multimodal_emotion/engine/fusion.py +57 -0
reachy_mini_multimodal_emotion/engine/labels.py +34 -0
reachy_mini_multimodal_emotion/engine/pipeline.py +120 -0
reachy_mini_multimodal_emotion/engine/speech.py +144 -0
reachy_mini_multimodal_emotion/engine/text.py +80 -0
reachy_mini_multimodal_emotion/main.py +335 -0
reachy_mini_multimodal_emotion/reactions.py +59 -0
reachy_mini_multimodal_emotion/static/index.html +75 -0
reachy_mini_multimodal_emotion/static/main.js +145 -0
reachy_mini_multimodal_emotion/static/style.css +140 -0
style.css +411 -28
tests/conftest.py +14 -0
tests/test_branches.py +104 -0
tests/test_fusion.py +65 -0
tests/test_pipeline.py +68 -0
tests/test_policy.py +67 -0

.gitignore ADDED Viewed

	@@ -0,0 +1,3 @@

+__pycache__/
+*.egg-info/
+build/

README.md CHANGED Viewed

@@ -1,10 +1,42 @@
----
-title: Reachy Mini Multimodal Emotion
-emoji: 📉
-colorFrom: pink
-colorTo: green
-sdk: static
-pinned: false
----
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

+---
+title: Reachy Mini Multimodal Emotion
+emoji: 🙂
+colorFrom: indigo
+colorTo: pink
+sdk: static
+pinned: false
+short_description: Face + voice + Chinese words emotion, empathetic reactions
+tags:
+ - reachy_mini
+ - reachy_mini_python_app
+models:
+ - pearlyjam21/multimodal-emotion-zh
+ - csukuangfj/sherpa-onnx-sense-voice-zh-en-ja-ko-yue-2024-07-17
+---
+# Reachy Mini Multimodal Emotion
+This app lets Reachy Mini sense emotion from three sources and react with an empathetic move from
+`pollen-robotics/reachy-mini-emotions-library`:
+- **Face**: the camera runs at about 5 fps; YuNet finds the face and a FER+ MobileNetV3 classifies it.
+- **Voice**: the microphone feeds a 2 s window every 0.2 s to a compact ECAPA-TDNN student.
+- **Words**: silero VAD splits speech into utterances, SenseVoice-Small transcribes them, and a Traditional
+  Chinese BERT classifies the text.
+The three results are combined by calibrated late fusion (weighted log-linear pooling). A modality with no
+evidence is left out rather than counted as neutral: no face, silence, English speech (the text model is
+Chinese-only), or a transcript that is too old. Everything runs on the Reachy Mini Wireless CPU with ONNX
+Runtime, and no PyTorch is needed.
+Models download from [`pearlyjam21/multimodal-emotion-zh`](https://huggingface.co/pearlyjam21/multimodal-emotion-zh)
+on first start (about 360 MB). Its model card lists the metrics, licenses and known limitations.
+**Dashboard:** `http://reachy-mini.local:8042` shows fused and per-modality probabilities, the transcript,
+fusion-weight sliders, modality on/off switches and test reactions.
+**Simulation:** set `MM_EMOTION_WAV=/path/to/16k.wav` to loop a file instead of the microphone. Set
+`MM_EMOTION_LOCAL_DIR` to load the models from a local folder instead of the Hub.
+The output is a soft social cue, not a measurement. On natural conversation the individual models are close
+to chance; see the model card.

index.html CHANGED Viewed

@@ -1,19 +1,40 @@
-<!doctype html>
-<html>
-	<head>
-		<meta charset="utf-8" />
-		<meta name="viewport" content="width=device-width" />
-		<title>My static Space</title>
-		<link rel="stylesheet" href="style.css" />
-	</head>
-	<body>
-		<div class="card">
-			<h1>Welcome to your static Space!</h1>
-			<p>You can modify this app directly by editing <i>index.html</i> in the Files and versions tab.</p>
-			<p>
-				Also don't forget to check the
-				<a href="https://huggingface.co/docs/hub/spaces" target="_blank">Spaces documentation</a>.
-			</p>
-		</div>
-	</body>
-</html>

+<!doctype html>
+<html>
+<head>
+	<meta charset="utf-8" />
+	<meta name="viewport" content="width=device-width" />
+	<title> Reachy Mini Multimodal Emotion </title>
+	<link rel="stylesheet" href="style.css" />
+</head>
+<body>
+	<div class="hero">
+		<div class="hero-content">
+			<div class="app-icon">🙂🎙️💬</div>
+			<h1> Reachy Mini Multimodal Emotion </h1>
+			<p class="tagline">Reachy Mini reads your face, hears your voice and understands your (Chinese) words, then responds with empathy.</p>
+		</div>
+	</div>
+	<div class="container">
+		<div class="main-card">
+			<div class="app-preview">
+				<div class="preview-image">
+					<div class="camera-feed">😊 😢 😠 😲 😨 🤢</div>
+				</div>
+			</div>
+		</div>
+	</div>
+	<div class="footer">
+		<p>
+			🤖 Reachy Mini Multimodal Emotion •
+			<a href="https://github.com/pollen-robotics" target="_blank">Pollen Robotics</a> •
+			<a href="https://huggingface.co/spaces/pollen-robotics/reachy-mini-landing-page#apps" target="_blank">Browse More
+				Apps</a>
+		</p>
+	</div>
+</body>
+</html>

pyproject.toml ADDED Viewed

	@@ -0,0 +1,36 @@

+[build-system]
+requires = ["setuptools>=61.0"]
+build-backend = "setuptools.build_meta"
+[project]
+name = "reachy_mini_multimodal_emotion"
+version = "0.1.0"
+description = "Reachy Mini senses emotion from face, voice and (Chinese) words, and responds empathetically"
+readme = "README.md"
+requires-python = ">=3.10"
+dependencies = [
+    "reachy-mini",
+    "numpy",
+    "onnxruntime>=1.17",
+    "pillow",
+    "tokenizers>=0.19",
+    "sherpa-onnx>=1.12",
+    "chinese-converter>=1.0",
+    "huggingface_hub",
+]
+keywords = ["reachy-mini-app"]
+[project.entry-points."reachy_mini_apps"]
+reachy_mini_multimodal_emotion = "reachy_mini_multimodal_emotion.main:ReachyMiniMultimodalEmotion"
+[tool.setuptools]
+package-dir = { "" = "." }
+include-package-data = true
+[tool.setuptools.packages.find]
+where = ["."]
+include = ["reachy_mini_multimodal_emotion*"]  # keep build/ and tests/ out of the package
+[tool.setuptools.package-data]
+reachy_mini_multimodal_emotion = ["**/*"] # Also include all non-.py files

reachy_mini_multimodal_emotion/__init__.py ADDED Viewed

File without changes

reachy_mini_multimodal_emotion/engine/__init__.py ADDED Viewed

	@@ -0,0 +1,39 @@

+"""Multimodal (face + voice + Chinese text) emotion engine. No torch, no Reachy dependency.
+    from reachy_mini_multimodal_emotion.engine import load_config, download_models
+    paths = download_models()            # model repo + SenseVoice, cached by huggingface_hub
+"""
+from __future__ import annotations
+import json
+import os
+from pathlib import Path
+from .fusion import BranchResult, FusedResult, fuse
+from .labels import EMOTIONS
+MODEL_REPO = os.environ.get("MM_EMOTION_MODEL_REPO", "pearlyjam21/multimodal-emotion-zh")
+ASR_REPO = "csukuangfj/sherpa-onnx-sense-voice-zh-en-ja-ko-yue-2024-07-17"
+__all__ = ["EMOTIONS", "BranchResult", "FusedResult", "fuse", "download_models", "load_config"]
+def download_models(repo_id: str = MODEL_REPO, with_asr: bool = True, local_dir: str | None = None) -> dict:
+    """Fetch (or reuse cached) weights. Returns {"root": Path, "asr": Path | None}.
+    Set MM_EMOTION_LOCAL_DIR to a folder with the model repo's layout to run fully offline.
+    """
+    from huggingface_hub import snapshot_download
+    local = local_dir or os.environ.get("MM_EMOTION_LOCAL_DIR")
+    root = Path(local) if local else Path(snapshot_download(repo_id, allow_patterns=["config.json", "face/*",
+                                                                                      "speech/*", "text/*", "asr/*"]))
+    asr = None
+    if with_asr:
+        asr = Path(snapshot_download(ASR_REPO, allow_patterns=["model.int8.onnx", "tokens.txt"]))
+    return {"root": root, "asr": asr}
+def load_config(root: str | Path) -> dict:
+    return json.loads((Path(root) / "config.json").read_text(encoding="utf-8"))

reachy_mini_multimodal_emotion/engine/asr.py ADDED Viewed

	@@ -0,0 +1,99 @@

+"""Utterance segmentation (silero VAD) and transcription (SenseVoice-Small int8) via sherpa-onnx.
+SenseVoice also emits its own emotion tag; it is shown on the dashboard for comparison only and
+is not part of the fusion (it has not been evaluated in this project).
+"""
+from __future__ import annotations
+import queue
+import threading
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Callable
+import numpy as np
+SAMPLE_RATE = 16000
+VAD_WINDOW = 512  # silero VAD frame size at 16 kHz
+@dataclass
+class Utterance:
+    text: str
+    lang: str | None        # "zh", "en", "yue", ... (SenseVoice language tag)
+    asr_emotion: str | None  # SenseVoice's own tag, informational
+    seconds: float
+def _tag(value: str) -> str | None:
+    value = value.strip().strip("<|>").lower()
+    return value or None
+class Transcriber:
+    """Feed mono 16 kHz audio with `push`; finished utterances go to `on_utterance` from a worker."""
+    def __init__(self, sensevoice_model: str | Path, tokens: str | Path, vad_model: str | Path,
+                 on_utterance: Callable[[Utterance], None], threads: int = 2) -> None:
+        import sherpa_onnx
+        self.recognizer = sherpa_onnx.OfflineRecognizer.from_sense_voice(
+            model=str(sensevoice_model), tokens=str(tokens), num_threads=threads, use_itn=True)
+        cfg = sherpa_onnx.VadModelConfig()
+        cfg.silero_vad.model = str(vad_model)
+        cfg.silero_vad.threshold = 0.5
+        cfg.silero_vad.min_silence_duration = 0.5
+        cfg.silero_vad.min_speech_duration = 0.4
+        cfg.silero_vad.max_speech_duration = 10.0
+        cfg.silero_vad.window_size = VAD_WINDOW
+        cfg.sample_rate = SAMPLE_RATE
+        cfg.num_threads = 1
+        self.vad = sherpa_onnx.VoiceActivityDetector(cfg, buffer_size_in_seconds=30)
+        self.on_utterance = on_utterance
+        self._pending = np.zeros(0, dtype=np.float32)
+        self._segments: "queue.Queue[np.ndarray]" = queue.Queue(maxsize=2)
+        self._stop = threading.Event()
+        self._worker = threading.Thread(target=self._run, daemon=True)
+        self._worker.start()
+    @property
+    def speaking(self) -> bool:
+        return bool(self.vad.is_speech_detected())
+    def push(self, chunk: np.ndarray) -> None:
+        self._pending = np.concatenate([self._pending, np.asarray(chunk, dtype=np.float32).reshape(-1)])
+        while len(self._pending) >= VAD_WINDOW:
+            self.vad.accept_waveform(self._pending[:VAD_WINDOW])
+            self._pending = self._pending[VAD_WINDOW:]
+        while not self.vad.empty():
+            segment = np.array(self.vad.front.samples, dtype=np.float32)
+            self.vad.pop()
+            try:
+                self._segments.put_nowait(segment)
+            except queue.Full:  # ASR is behind; drop rather than build latency
+                pass
+    def reset(self) -> None:
+        """Forget partial speech (e.g. audio recorded while the robot moved)."""
+        self.vad.reset()
+        self._pending = np.zeros(0, dtype=np.float32)
+    def transcribe(self, samples: np.ndarray) -> Utterance:
+        stream = self.recognizer.create_stream()
+        stream.accept_waveform(SAMPLE_RATE, samples)
+        self.recognizer.decode_stream(stream)
+        r = stream.result
+        return Utterance(r.text, _tag(r.lang), _tag(r.emotion), len(samples) / SAMPLE_RATE)
+    def _run(self) -> None:
+        while not self._stop.is_set():
+            try:
+                samples = self._segments.get(timeout=0.2)
+            except queue.Empty:
+                continue
+            self.on_utterance(self.transcribe(samples))
+    def close(self) -> None:
+        self._stop.set()
+        self._worker.join(timeout=2.0)

reachy_mini_multimodal_emotion/engine/face.py ADDED Viewed

	@@ -0,0 +1,112 @@

+"""Facial-expression branch: YuNet face detection + newEmotion FER+ MobileNetV3 (fp32 ONNX).
+Preprocessing reproduces newEmotion's `ferplus-gray-rgb-imagenet-v1` contract without torch:
+BGR crop -> cv2 grayscale -> replicated to RGB -> PIL bilinear resize to 112x112 (stretch)
+-> /255 -> ImageNet mean/std. tests/test_face.py checks it against torchvision.
+The int8 export is not used: its saved FER+ macro-F1 is 0.337 versus 0.754 for fp32.
+"""
+from __future__ import annotations
+import time
+from pathlib import Path
+import numpy as np
+from .fusion import OK, BranchResult
+from .labels import softmax, to_canonical
+IMAGE_MEAN = np.array([0.485, 0.456, 0.406], dtype=np.float32)
+IMAGE_STD = np.array([0.229, 0.224, 0.225], dtype=np.float32)
+DETECT_WIDTH = 320     # detect on a downscaled frame; crop from the full-resolution one
+EMA_ALPHA = 0.5
+STALE_S = 1.0          # a face result older than this is treated as absent
+def crop_face(frame_bgr: np.ndarray, box, padding: float) -> np.ndarray:
+    """Same semantics as newEmotion emotion_model.webcam.crop_face (symmetric padding, clamped)."""
+    height, width = frame_bgr.shape[:2]
+    x, y, bw, bh = (int(v) for v in box)
+    pad_x, pad_y = int(bw * padding), int(bh * padding)
+    left, top = max(0, x - pad_x), max(0, y - pad_y)
+    right, bottom = min(width, x + bw + pad_x), min(height, y + bh + pad_y)
+    if right <= left or bottom <= top:
+        raise ValueError("face box falls outside the frame")
+    return frame_bgr[top:bottom, left:right]
+def preprocess_face(face_bgr: np.ndarray, size: int = 112) -> np.ndarray:
+    """BGR face crop -> [1, 3, size, size] float32, matching the training transform."""
+    import cv2
+    from PIL import Image
+    gray = cv2.cvtColor(face_bgr, cv2.COLOR_BGR2GRAY)
+    rgb = Image.fromarray(np.repeat(gray[..., None], 3, axis=2), mode="RGB")
+    x = np.asarray(rgb.resize((size, size), Image.Resampling.BILINEAR), dtype=np.float32) / 255.0
+    x = (x - IMAGE_MEAN) / IMAGE_STD
+    return np.ascontiguousarray(x.transpose(2, 0, 1)[None])
+class FaceBranch:
+    def __init__(self, fer_model: str | Path, yunet_model: str | Path, temperature: float = 1.0,
+                 padding: float = 0.10, score_threshold: float = 0.7, min_face_px: int = 48,
+                 threads: int = 1) -> None:
+        import cv2
+        import onnxruntime as ort
+        opts = ort.SessionOptions()
+        opts.intra_op_num_threads = threads
+        self.session = ort.InferenceSession(str(fer_model), opts, providers=["CPUExecutionProvider"])
+        self.input_name = self.session.get_inputs()[0].name
+        self.detector = cv2.FaceDetectorYN.create(str(yunet_model), "", (DETECT_WIDTH, 240),
+                                                  score_threshold, 0.3, 50)
+        self.temperature, self.padding, self.min_face_px = temperature, padding, min_face_px
+        self._probs: np.ndarray | None = None
+        self._last_seen = float("-inf")
+        self.latency_ms = 0.0
+        self.box: list[int] | None = None
+    def reset(self) -> None:
+        self._probs, self._last_seen, self.box = None, float("-inf"), None
+    def detect(self, frame_bgr: np.ndarray) -> tuple[np.ndarray, float] | None:
+        """Largest face as (x, y, w, h) in full-frame pixels plus its detector score."""
+        import cv2
+        h, w = frame_bgr.shape[:2]
+        scale = DETECT_WIDTH / w
+        small = cv2.resize(frame_bgr, (DETECT_WIDTH, int(round(h * scale))), interpolation=cv2.INTER_AREA)
+        self.detector.setInputSize((small.shape[1], small.shape[0]))
+        _, faces = self.detector.detect(small)
+        if faces is None or not len(faces):
+            return None
+        best = max(faces, key=lambda f: f[2] * f[3])
+        return best[:4] / scale, float(best[-1])
+    def logits(self, face_bgr: np.ndarray) -> np.ndarray:
+        return self.session.run(None, {self.input_name: preprocess_face(face_bgr)})[0][0]
+    def step(self, frame_bgr: np.ndarray | None, now: float) -> BranchResult:
+        if frame_bgr is None:
+            return self.current(now, "no_camera")
+        t0 = time.perf_counter()
+        found = self.detect(frame_bgr)
+        if found is None or min(found[0][2:]) < self.min_face_px:
+            self.box = None
+            return self.current(now, "no_face")
+        box, score = found
+        probs = softmax(to_canonical("face", self.logits(crop_face(frame_bgr, box, self.padding))),
+                        self.temperature)
+        self.latency_ms = (time.perf_counter() - t0) * 1000.0
+        self.box = [int(v) for v in box]
+        self._probs = probs if self._probs is None else EMA_ALPHA * probs + (1 - EMA_ALPHA) * self._probs
+        self._last_seen = now
+        return BranchResult(self._probs.copy(), OK, {"score": score, "box": self.box,
+                                                     "latency_ms": self.latency_ms})
+    def current(self, now: float, missing_status: str = "no_face") -> BranchResult:
+        """Latest smoothed result while fresh; otherwise report the face as absent."""
+        if self._probs is not None and now - self._last_seen <= STALE_S:
+            return BranchResult(self._probs.copy(), OK, {"box": self.box, "latency_ms": self.latency_ms})
+        self._probs = None
+        return BranchResult(status=missing_status)

reachy_mini_multimodal_emotion/engine/fusion.py ADDED Viewed

	@@ -0,0 +1,57 @@

+"""Calibrated late fusion: weighted log-linear pooling over the branches that are present.
+A branch that has no evidence (no face, silence, no transcript, unsupported language, stale
+result) is left out and the remaining weights are renormalised. Absence is never encoded as an
+observed "neutral".
+"""
+from __future__ import annotations
+from dataclasses import dataclass, field
+import numpy as np
+from .labels import EMOTIONS
+OK = "ok"
+@dataclass
+class BranchResult:
+    """Calibrated probabilities in EMOTIONS order, or None with a status explaining why."""
+    probs: np.ndarray | None = None
+    status: str = "waiting"
+    detail: dict = field(default_factory=dict)
+    @property
+    def present(self) -> bool:
+        return self.status == OK and self.probs is not None
+@dataclass
+class FusedResult:
+    probs: np.ndarray | None
+    used: dict[str, float]  # branch -> normalised weight actually applied
+    @property
+    def top(self) -> str | None:
+        return None if self.probs is None else EMOTIONS[int(self.probs.argmax())]
+    @property
+    def confidence(self) -> float:
+        return 0.0 if self.probs is None else float(self.probs.max())
+def fuse(results: dict[str, BranchResult], weights: dict[str, float], eps: float = 1e-6) -> FusedResult:
+    present = {b: r.probs for b, r in results.items()
+               if r.present and r.probs is not None and weights.get(b, 0.0) > 0.0}
+    if not present:
+        return FusedResult(None, {})
+    total = sum(weights[b] for b in present)
+    used = {b: weights[b] / total for b in present}
+    log_p = np.zeros(len(EMOTIONS))
+    for b, w in used.items():
+        log_p += w * np.log(np.clip(present[b], eps, 1.0))
+    p = np.exp(log_p - log_p.max())
+    return FusedResult(p / p.sum(), used)

reachy_mini_multimodal_emotion/engine/labels.py ADDED Viewed

	@@ -0,0 +1,34 @@

+"""One label order for fusion, and the index maps from each branch's own training order.
+Each branch keeps the taxonomy it was trained with; logits are reordered into EMOTIONS before
+calibration and fusion. EMOTIONS[i] == BRANCH_LABELS[b][TO_CANONICAL[b][i]].
+"""
+from __future__ import annotations
+import numpy as np
+EMOTIONS = ["neutral", "happy", "sad", "angry", "surprise", "fear", "disgust"]
+BRANCH_LABELS = {
+    # SER ECAPA student (speech/v3/distill_config_v3.yaml)
+    "speech": ["neutral", "happy", "sad", "angry", "surprise", "fear", "disgust"],
+    # newEmotion FER+ MobileNetV3 (model_fp32_metadata.json)
+    "face": ["angry", "disgust", "fear", "happy", "sad", "surprise", "neutral"],
+    # Traditional Chinese BERT (config.json id2label)
+    "text": ["happy", "sad", "angry", "fear", "disgust", "neutral", "surprise"],
+}
+TO_CANONICAL = {b: np.array([labels.index(e) for e in EMOTIONS]) for b, labels in BRANCH_LABELS.items()}
+def to_canonical(branch: str, logits: np.ndarray) -> np.ndarray:
+    """Reorder a branch's logits (last axis) into EMOTIONS order."""
+    return np.asarray(logits, dtype=np.float64)[..., TO_CANONICAL[branch]]
+def softmax(x: np.ndarray, temperature: float = 1.0) -> np.ndarray:
+    z = np.asarray(x, dtype=np.float64) / temperature
+    z = z - z.max(axis=-1, keepdims=True)
+    e = np.exp(z)
+    return e / e.sum(axis=-1, keepdims=True)

reachy_mini_multimodal_emotion/engine/pipeline.py ADDED Viewed

	@@ -0,0 +1,120 @@

+"""Wires the branches together from the model repo's config.json.
+Usage (robot or PC):
+    pipe = MultimodalPipeline.from_pretrained()
+    pipe.push_audio(mono_16k_chunk)        # as audio arrives
+    pipe.push_frame(bgr_frame)             # a few times per second
+    snap = pipe.step()                     # every ~0.2 s -> fused + per-branch results
+"""
+from __future__ import annotations
+import threading
+import time
+from collections import deque
+from dataclasses import dataclass, field
+from pathlib import Path
+import numpy as np
+from . import download_models, load_config
+from .asr import Transcriber, Utterance
+from .face import FaceBranch
+from .fusion import BranchResult, FusedResult, fuse
+from .speech import SpeechBranch
+from .text import TextBranch
+@dataclass
+class Snapshot:
+    fused: FusedResult
+    branches: dict[str, BranchResult]
+    vad: np.ndarray                      # speech valence/arousal/dominance (0-1)
+    transcript: str = ""
+    asr_emotion: str | None = None
+    asr_lang: str | None = None
+    utterances: list = field(default_factory=list)
+class MultimodalPipeline:
+    def __init__(self, root: str | Path, asr_dir: str | Path | None, use_face: bool = True,
+                 use_text: bool = True) -> None:
+        root = Path(root)
+        cfg = load_config(root)
+        files, fusion = cfg["files"], cfg["fusion"]
+        self.temperatures, self.weights = dict(fusion["temperatures"]), dict(fusion["weights"])
+        self.speech = SpeechBranch(root / files["speech"], self.temperatures["speech"])
+        self.face = FaceBranch(root / files["face"], root / files["face_detector"], self.temperatures["face"],
+                               padding=cfg["face"]["padding"]) if use_face else None
+        self.text = None
+        self.transcriber = None
+        if use_text and asr_dir is not None:
+            self.text = TextBranch(root / files["text"], root / files["text_tokenizer"], self.temperatures["text"])
+            asr_dir = Path(asr_dir)
+            self.transcriber = Transcriber(asr_dir / "model.int8.onnx", asr_dir / "tokens.txt",
+                                           root / files["vad"], self._on_utterance)
+        self.lock = threading.Lock()
+        self.utterances: deque = deque(maxlen=8)
+        self._last_utt: Utterance | None = None
+        self._face_result = BranchResult(status="no_camera" if use_face else "disabled")
+        self.enabled = {"speech": True, "face": use_face, "text": self.text is not None}
+    @classmethod
+    def from_pretrained(cls, repo_id: str | None = None, use_face: bool = True, use_text: bool = True):
+        paths = download_models(repo_id, with_asr=use_text) if repo_id else download_models(with_asr=use_text)
+        return cls(paths["root"], paths["asr"], use_face, use_text)
+    # ------------------------------------------------------------------ inputs
+    def push_audio(self, chunk: np.ndarray) -> None:
+        self.speech.ring.push(chunk)
+        if self.transcriber is not None:
+            self.transcriber.push(chunk)
+    def push_frame(self, frame_bgr: np.ndarray | None, now: float | None = None) -> BranchResult:
+        if self.face is None:
+            return self._face_result
+        result = self.face.step(frame_bgr, time.monotonic() if now is None else now)
+        with self.lock:
+            self._face_result = result
+        return result
+    def _on_utterance(self, utt: Utterance) -> None:
+        assert self.text is not None
+        result = self.text.classify(utt.text, utt.lang, time.monotonic())
+        with self.lock:
+            self._last_utt = utt
+            self.utterances.appendleft({
+                "time": time.strftime("%H:%M:%S"), "text": self.text.transcript, "lang": utt.lang,
+                "asr_emotion": utt.asr_emotion, "text_status": result.status,
+                "text_top": None if result.probs is None else int(result.probs.argmax()),
+            })
+    def reset_audio(self) -> None:
+        """Forget audio context (call after the robot moved or made sound)."""
+        self.speech.reset()
+        if self.transcriber is not None:
+            self.transcriber.reset()
+    # ------------------------------------------------------------------ output
+    def step(self, now: float | None = None) -> Snapshot:
+        now = time.monotonic() if now is None else now
+        branches = {"speech": self.speech.step() if self.enabled["speech"] else BranchResult(status="disabled")}
+        with self.lock:
+            face = self._face_result
+            utt = self._last_utt
+            utterances = list(self.utterances)
+        if self.face is not None and self.enabled["face"]:
+            face = self.face.current(now) if face.present else face
+        branches["face"] = face if self.enabled["face"] else BranchResult(status="disabled")
+        branches["text"] = (self.text.current(now) if self.text is not None and self.enabled["text"]
+                            else BranchResult(status="disabled"))
+        return Snapshot(
+            fused=fuse(branches, self.weights), branches=branches, vad=self.speech.vad.copy(),
+            transcript=self.text.transcript if self.text is not None else "",
+            asr_emotion=utt.asr_emotion if utt else None, asr_lang=utt.lang if utt else None,
+            utterances=utterances,
+        )
+    def close(self) -> None:
+        if self.transcriber is not None:
+            self.transcriber.close()

reachy_mini_multimodal_emotion/engine/speech.py ADDED Viewed

	@@ -0,0 +1,144 @@

+"""Torch-free acoustic emotion branch (compact ECAPA-TDNN student, SER v3).
+The log-mel front end is the NumPy port verified against torchaudio in SER_Reachy_merli_v2
+(MelSpectrogram n_fft=512, win=400 Hann, hop=160, 80 HTK mel bands 20-8000 Hz, power 2,
+log clamp 1e-5, per-utterance mean/std normalisation). tests/test_speech.py re-checks it.
+"""
+from __future__ import annotations
+import threading
+import time
+from pathlib import Path
+import numpy as np
+from .fusion import OK, BranchResult
+from .labels import softmax, to_canonical
+SAMPLE_RATE = 16000
+WINDOW_SEC = 2.0
+RMS_THRESHOLD = 0.005
+EMA_ALPHA = 0.35
+def _hz_to_mel(f):
+    return 2595.0 * np.log10(1.0 + f / 700.0)
+def _mel_to_hz(m):
+    return 700.0 * (10.0 ** (m / 2595.0) - 1.0)
+class LogMel:
+    """NumPy port of utils.audio.LogMelExtractor from the SER training repo."""
+    def __init__(self, sample_rate=SAMPLE_RATE, n_fft=512, win_length=400, hop_length=160,
+                 n_mels=80, f_min=20.0, f_max=8000.0) -> None:
+        self.n_fft, self.hop = n_fft, hop_length
+        # Periodic Hann window, zero-padded to n_fft and centred (as torch.stft does)
+        win = 0.5 - 0.5 * np.cos(2 * np.pi * np.arange(win_length) / win_length)
+        left = (n_fft - win_length) // 2
+        self.window = np.zeros(n_fft)
+        self.window[left : left + win_length] = win
+        # HTK mel filterbank without normalisation (torchaudio.functional.melscale_fbanks)
+        all_freqs = np.linspace(0, sample_rate // 2, n_fft // 2 + 1)
+        f_pts = _mel_to_hz(np.linspace(_hz_to_mel(f_min), _hz_to_mel(f_max), n_mels + 2))
+        f_diff = f_pts[1:] - f_pts[:-1]
+        slopes = f_pts[None, :] - all_freqs[:, None]
+        down = -slopes[:, :-2] / f_diff[:-1]
+        up = slopes[:, 2:] / f_diff[1:]
+        self.fbank = np.maximum(0.0, np.minimum(down, up))  # [n_freqs, n_mels]
+    def __call__(self, wave: np.ndarray) -> np.ndarray:
+        """wave: [T] float -> normalised log-mel [80, frames] float32."""
+        x = np.pad(np.asarray(wave, dtype=np.float64), self.n_fft // 2, mode="reflect")
+        n_frames = 1 + (len(x) - self.n_fft) // self.hop
+        idx = np.arange(self.n_fft)[None, :] + self.hop * np.arange(n_frames)[:, None]
+        spec = np.abs(np.fft.rfft(x[idx] * self.window, axis=1)) ** 2  # [frames, n_freqs]
+        log_mel = np.log(np.maximum(spec @ self.fbank, 1e-5)).T  # [n_mels, frames]
+        mean = log_mel.mean(axis=1, keepdims=True)
+        std = log_mel.std(axis=1, ddof=1, keepdims=True) + 1e-5
+        return ((log_mel - mean) / std).astype(np.float32)
+class RingBuffer:
+    """Thread-safe buffer holding the latest window of mono samples."""
+    def __init__(self, n: int = int(WINDOW_SEC * SAMPLE_RATE)) -> None:
+        self._buf = np.zeros(n, dtype=np.float32)
+        self._filled = 0
+        self._lock = threading.Lock()
+    def push(self, x: np.ndarray) -> None:
+        x = np.asarray(x, dtype=np.float32).reshape(-1)[-len(self._buf):]
+        with self._lock:
+            self._buf = np.roll(self._buf, -len(x))
+            self._buf[-len(x):] = x
+            self._filled = min(len(self._buf), self._filled + len(x))
+    def clear(self) -> None:
+        with self._lock:
+            self._buf[:] = 0.0
+            self._filled = 0
+    def snapshot(self) -> tuple[np.ndarray, bool]:
+        """Returns (window, full) where full means a whole window of fresh audio is present."""
+        with self._lock:
+            return self._buf.copy(), self._filled >= len(self._buf)
+class SpeechBranch:
+    """Energy gate + log-mel + ONNX + EMA smoothing over calibrated probabilities.
+    Also exposes the student's valence/arousal/dominance head (0-1 each) for continuous cues.
+    """
+    def __init__(self, model_path: str | Path, temperature: float = 1.0, threads: int = 2) -> None:
+        import onnxruntime as ort
+        opts = ort.SessionOptions()
+        opts.intra_op_num_threads = threads  # leave CPU for the robot daemon on the CM4
+        self.session = ort.InferenceSession(str(model_path), opts, providers=["CPUExecutionProvider"])
+        self.temperature = temperature
+        self.ring = RingBuffer()
+        self.mel = LogMel()
+        self.lock = threading.Lock()
+        self._probs: np.ndarray | None = None
+        self.vad = np.full(3, 0.5)
+        self.rms = 0.0
+        self.latency_ms = 0.0
+    def reset(self) -> None:
+        """Drop buffered audio and smoothing history (after the robot moved or spoke)."""
+        self.ring.clear()
+        with self.lock:
+            self._probs = None
+            self.vad = np.full(3, 0.5)
+    def logits(self, wave: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
+        """Raw branch-order emotion logits [7] and VAD [3] for one mono 16 kHz window."""
+        logits, vad = self.session.run(None, {"log_mel": self.mel(wave)[None]})
+        return logits[0], vad[0]
+    def step(self) -> BranchResult:
+        wave, full = self.ring.snapshot()
+        rms = float(np.sqrt(np.mean(wave**2)))
+        self.rms = rms
+        if not full:
+            return BranchResult(status="buffering")
+        if rms < RMS_THRESHOLD:  # silence is missing evidence, not "neutral"
+            with self.lock:
+                self._probs = None
+            return BranchResult(status="silence", detail={"rms": rms})
+        t0 = time.perf_counter()
+        logits, vad = self.logits(wave)
+        self.latency_ms = (time.perf_counter() - t0) * 1000.0
+        probs = softmax(to_canonical("speech", logits), self.temperature)
+        with self.lock:
+            if self._probs is None:
+                self._probs, self.vad = probs, vad
+            else:
+                self._probs = EMA_ALPHA * probs + (1 - EMA_ALPHA) * self._probs
+                self.vad = EMA_ALPHA * vad + (1 - EMA_ALPHA) * self.vad
+            return BranchResult(self._probs.copy(), OK, {"rms": rms, "latency_ms": self.latency_ms})

reachy_mini_multimodal_emotion/engine/text.py ADDED Viewed

	@@ -0,0 +1,80 @@

+"""Traditional Chinese text-emotion branch (bert-base-chinese fine-tune, dynamic int8 ONNX).
+Uses the `tokenizers` fast tokenizer (no transformers/torch). ASR output in Simplified Chinese is
+converted to Traditional with chinese-converter, the converter used for the training fear data.
+The model is Chinese-only: English or other transcripts are reported as unsupported, not scored.
+"""
+from __future__ import annotations
+import re
+import threading
+import time
+from pathlib import Path
+import numpy as np
+from .fusion import OK, BranchResult
+from .labels import softmax, to_canonical
+MAX_LENGTH = 128
+TTL_S = 8.0          # how long an utterance's text emotion stays in the fusion
+CHINESE_LANGS = {"zh", "yue"}
+_HAN = re.compile(r"[㐀-鿿]")
+class TextBranch:
+    def __init__(self, model_path: str | Path, tokenizer_path: str | Path, temperature: float = 1.0,
+                 threads: int = 2) -> None:
+        import onnxruntime as ort
+        from tokenizers import Tokenizer
+        opts = ort.SessionOptions()
+        opts.intra_op_num_threads = threads
+        self.session = ort.InferenceSession(str(model_path), opts, providers=["CPUExecutionProvider"])
+        self.tokenizer = Tokenizer.from_file(str(tokenizer_path))
+        self.tokenizer.enable_truncation(MAX_LENGTH)
+        self.temperature = temperature
+        self.lock = threading.Lock()
+        self._result = BranchResult(status="waiting")
+        self._at = float("-inf")
+        self.transcript = ""
+        self.latency_ms = 0.0
+    def logits(self, text: str) -> np.ndarray:
+        enc = self.tokenizer.encode(text)
+        feeds = {
+            "input_ids": np.array([enc.ids], dtype=np.int64),
+            "attention_mask": np.array([enc.attention_mask], dtype=np.int64),
+            "token_type_ids": np.array([enc.type_ids], dtype=np.int64),
+        }
+        return self.session.run(None, feeds)[0][0]
+    def classify(self, text: str, lang: str | None, now: float) -> BranchResult:
+        """Score one finished utterance and keep it as the live text result for TTL_S seconds."""
+        text = text.strip()
+        if not text:
+            result = BranchResult(status="no_transcript")
+        elif (lang is not None and lang not in CHINESE_LANGS) or not _HAN.search(text):
+            result = BranchResult(status="unsupported_language", detail={"lang": lang})
+        else:
+            import chinese_converter
+            text = chinese_converter.to_traditional(text)
+            t0 = time.perf_counter()
+            probs = softmax(to_canonical("text", self.logits(text)), self.temperature)
+            self.latency_ms = (time.perf_counter() - t0) * 1000.0
+            result = BranchResult(probs, OK, {"latency_ms": self.latency_ms})
+        with self.lock:
+            self.transcript, self._result, self._at = text, result, now
+        return result
+    def current(self, now: float) -> BranchResult:
+        with self.lock:
+            if self._result.present and now - self._at > TTL_S:
+                self._result = BranchResult(status="stale")
+            return self._result
+    def reset(self) -> None:
+        with self.lock:
+            self._result, self._at = BranchResult(status="waiting"), float("-inf")

reachy_mini_multimodal_emotion/main.py ADDED Viewed

	@@ -0,0 +1,335 @@

+"""Reachy Mini multimodal emotion: face + voice + (Chinese) words -> fused emotion -> empathetic move.
+Pipeline
+  camera (BGR) ~5 fps -> YuNet face -> FER+ MobileNetV3 ------------------------\
+  microphone 16 kHz -> 2 s window every 0.2 s -> log-mel -> ECAPA student --------> late fusion
+                    -> silero VAD utterances -> SenseVoice ASR -> zh BERT -----/      |
+                                                                          reaction policy -> move
+Branches without evidence (no face, silence, English transcript, stale text) are left out of the
+fusion instead of being counted as "neutral". While a move plays, and briefly after, microphone
+input is ignored so the robot never reacts to its own motors or speaker.
+Models download from the Hugging Face Hub on first start (about 360 MB with ASR). For simulation
+without a microphone set MM_EMOTION_WAV=/path/to/16k.wav to loop a file instead.
+"""
+from __future__ import annotations
+import logging
+import os
+import queue
+import threading
+import time
+import wave
+from collections import deque
+from pathlib import Path
+import numpy as np
+from pydantic import BaseModel
+from reachy_mini import ReachyMini, ReachyMiniApp
+from reachy_mini.reachy_mini import INIT_ANTENNAS_JOINT_POSITIONS, INIT_HEAD_POSE
+from reachy_mini.utils import create_head_pose
+from reachy_mini_multimodal_emotion.engine.labels import EMOTIONS
+from reachy_mini_multimodal_emotion.engine.speech import SAMPLE_RATE
+from reachy_mini_multimodal_emotion.reactions import DEFAULT_THRESHOLDS, ReactionPolicy, pick_move
+HOP_SEC = 0.20            # fusion + speech scoring rate
+FACE_PERIOD_S = 0.20      # ~5 fps face analysis
+POST_MOVE_GUARD_S = 0.8   # keep ignoring audio after a move (speaker tail, settling motors)
+CUE_PERIOD_S = 0.05       # subtle antenna/head cues at 20 Hz
+REST_ANTENNAS = np.array(INIT_ANTENNAS_JOINT_POSITIONS)
+logger = logging.getLogger("reachy_mini_multimodal_emotion")
+class MicSource:
+    """Robot microphone: one chunk per call (a chunk is always ready on the robot), stereo -> mono."""
+    name = "robot microphone"
+    def __init__(self, reachy_mini: ReachyMini) -> None:
+        self.media = reachy_mini.media
+        self.media.start_recording()
+    def read(self) -> np.ndarray | None:
+        sample = self.media.get_audio_sample()
+        if sample is None or not len(sample):
+            return None
+        return (sample.mean(axis=1) if sample.ndim == 2 else sample).astype(np.float32)
+    def close(self) -> None:
+        self.media.stop_recording()
+class WavSource:
+    """Loops a 16 kHz 16-bit wav file in real time (simulation has no microphone)."""
+    def __init__(self, path: str) -> None:
+        with wave.open(path, "rb") as w:
+            if w.getframerate() != SAMPLE_RATE or w.getsampwidth() != 2:
+                raise ValueError(f"{path}: need 16 kHz 16-bit PCM, got {w.getframerate()} Hz")
+            pcm = np.frombuffer(w.readframes(w.getnframes()), dtype=np.int16).reshape(-1, w.getnchannels())
+        self.audio = pcm.mean(axis=1).astype(np.float32) / 32768.0
+        self.name = f"wav file {Path(path).name}"
+        self.pos, self.t_last = 0, time.monotonic()
+    def read(self) -> np.ndarray | None:
+        now = time.monotonic()
+        n = int((now - self.t_last) * SAMPLE_RATE)
+        if n <= 0:
+            return None
+        self.t_last += n / SAMPLE_RATE
+        chunk = np.take(self.audio, range(self.pos, self.pos + n), mode="wrap")
+        self.pos = (self.pos + n) % len(self.audio)
+        return chunk
+    def close(self) -> None:
+        pass
+class Settings(BaseModel):
+    reactions_enabled: bool = True
+    sensitivity: float = 0.5   # 0 = only very confident reactions, 1 = react readily
+    move_sound: bool = True
+    use_face: bool = True
+    use_text: bool = True
+    weight_speech: float = 0.35
+    weight_face: float = 0.65
+    weight_text: float = 0.35
+class TestReaction(BaseModel):
+    emotion: str
+def _round(p) -> dict[str, float] | None:
+    return None if p is None else {e: round(float(v), 4) for e, v in zip(EMOTIONS, p)}
+class ReachyMiniMultimodalEmotion(ReachyMiniApp):
+    custom_app_url: str | None = "http://0.0.0.0:8042"
+    request_media_backend: str | None = None
+    def run(self, reachy_mini: ReachyMini, stop_event: threading.Event) -> None:
+        policy = ReactionPolicy()
+        settings = Settings()
+        wav_path = os.environ.get("MM_EMOTION_WAV")
+        source = WavSource(wav_path) if wav_path else MicSource(reachy_mini)
+        pipe = None
+        status = {"models": "downloading / loading models", "moves": "loading"}
+        moves = None
+        reaction_queue: "queue.Queue[tuple[str, str]]" = queue.Queue(maxsize=1)
+        log: deque = deque(maxlen=12)
+        shared: dict = {"moving": False, "guard_until": 0.0, "snap": None}
+        lock = threading.Lock()
+        def apply_settings() -> None:
+            offset = (0.5 - settings.sensitivity) * 0.4
+            policy.thresholds = {k: float(np.clip(v + offset, 0.3, 0.95)) for k, v in DEFAULT_THRESHOLDS.items()}
+            if pipe is not None:
+                pipe.weights.update(speech=settings.weight_speech, face=settings.weight_face,
+                                    text=settings.weight_text)
+                pipe.enabled["face"] = settings.use_face and pipe.face is not None
+                pipe.enabled["text"] = settings.use_text and pipe.text is not None
+        # ---------------------------------------------------------------- web dashboard API
+        if self.settings_app is not None:
+            @self.settings_app.get("/state")
+            def get_state():
+                with lock:
+                    snap, moving = shared["snap"], shared["moving"]
+                    entries = list(log)
+                out = {"status": status, "moving": moving, "source": source.name, "log": entries,
+                       "settings": settings.model_dump(), "thresholds": policy.thresholds}
+                if snap is None:
+                    return out
+                out.update({
+                    "top": snap.fused.top, "confidence": round(snap.fused.confidence, 4),
+                    "fused": _round(snap.fused.probs), "used": {b: round(w, 3) for b, w in snap.fused.used.items()},
+                    "branches": {b: {"status": r.status, "probs": _round(r.probs),
+                                     "latency_ms": round(float(r.detail.get("latency_ms", 0.0)), 1)}
+                                 for b, r in snap.branches.items()},
+                    "vad": [round(float(v), 4) for v in snap.vad],
+                    "transcript": snap.transcript, "asr_lang": snap.asr_lang, "asr_emotion": snap.asr_emotion,
+                    "utterances": [{**u, "text_top": None if u["text_top"] is None else EMOTIONS[u["text_top"]]}
+                                   for u in snap.utterances],
+                })
+                return out
+            @self.settings_app.post("/settings")
+            def update_settings(new: Settings):
+                nonlocal settings
+                settings = new
+                apply_settings()
+                return settings.model_dump()
+            @self.settings_app.post("/test_reaction")
+            def test_reaction(req: TestReaction):
+                move = pick_move(req.emotion)
+                if move is None:
+                    return {"queued": False, "reason": f"no move for '{req.emotion}'"}
+                try:
+                    reaction_queue.put_nowait((req.emotion, move))
+                    return {"queued": True, "move": move}
+                except queue.Full:
+                    return {"queued": False, "reason": "a reaction is already queued"}
+        # ---------------------------------------------------------------- background loading
+        def load_pipeline() -> None:
+            nonlocal pipe
+            from reachy_mini_multimodal_emotion.engine.pipeline import MultimodalPipeline
+            try:
+                pipe = MultimodalPipeline.from_pretrained()
+                apply_settings()
+                status["models"] = "ready"
+                logger.info("Models ready (face=%s, text=%s)", pipe.face is not None, pipe.text is not None)
+            except Exception as e:  # e.g. no network on first start; keep the app alive and say why
+                logger.exception("Could not load models")
+                status["models"] = f"failed: {e}"
+        def load_moves() -> None:
+            nonlocal moves
+            from reachy_mini.motion.recorded_move import RecordedMoves
+            while not stop_event.is_set():
+                try:
+                    moves = RecordedMoves("pollen-robotics/reachy-mini-emotions-library")
+                    status["moves"] = "ready"
+                    return
+                except Exception as e:
+                    status["moves"] = "unavailable (using antenna gestures)"
+                    logger.warning("Emotion library unavailable, retrying in 30 s: %s", e)
+                    stop_event.wait(30.0)
+        threading.Thread(target=load_pipeline, daemon=True).start()
+        threading.Thread(target=load_moves, daemon=True).start()
+        def fallback_gesture(emotion: str) -> None:
+            up = emotion in ("happy", "surprise")
+            for a in ((30, -30), (0, 0)) * (2 if up else 1):
+                offset = np.deg2rad(a if up else (-a[0], -a[1]))
+                reachy_mini.goto_target(antennas=REST_ANTENNAS + offset, duration=0.4)
+        def reaction_worker() -> None:
+            while not stop_event.is_set():
+                try:
+                    emotion, move_name = reaction_queue.get(timeout=0.2)
+                except queue.Empty:
+                    continue
+                with lock:
+                    shared["moving"] = True
+                    snap = shared["snap"]
+                    log.appendleft({"time": time.strftime("%H:%M:%S"), "emotion": emotion, "move": move_name,
+                                    "used": sorted(snap.fused.used) if snap is not None else []})
+                try:
+                    if moves is not None:
+                        reachy_mini.play_move(moves.get(move_name), initial_goto_duration=0.6,
+                                              sound=settings.move_sound)
+                    else:
+                        fallback_gesture(emotion)
+                    reachy_mini.goto_target(head=INIT_HEAD_POSE, antennas=REST_ANTENNAS, duration=0.6)
+                except Exception:
+                    logger.exception("Reaction %s failed", move_name)
+                finally:
+                    with lock:
+                        shared["moving"] = False
+                        shared["guard_until"] = time.monotonic() + POST_MOVE_GUARD_S
+        def face_worker() -> None:
+            """Camera analysis runs on its own thread so detection never delays audio handling."""
+            next_t = time.monotonic()
+            while not stop_event.is_set():
+                next_t += FACE_PERIOD_S
+                p = pipe
+                if p is not None and p.enabled["face"]:
+                    try:
+                        p.push_frame(reachy_mini.media.get_frame())
+                    except Exception:
+                        logger.exception("Face step failed")
+                stop_event.wait(max(0.0, next_t - time.monotonic()))
+                next_t = max(next_t, time.monotonic())
+        # Wake up silently: the daemon leaves the robot limp in its sleep pose between apps
+        reachy_mini.enable_motors()
+        reachy_mini.goto_target(head=INIT_HEAD_POSE, antennas=REST_ANTENNAS, duration=2.0)
+        workers = [threading.Thread(target=reaction_worker, daemon=True),
+                   threading.Thread(target=face_worker, daemon=True)]
+        for w in workers:
+            w.start()
+        logger.info("Listening on %s", source.name)
+        # ---------------------------------------------------------------- main loop
+        next_hop = next_cue = time.monotonic()
+        cue = np.zeros(3)  # smoothed [antenna_deg, head_roll_deg, head_pitch_deg]
+        was_blocked = False
+        try:
+            while not stop_event.is_set():
+                now = time.monotonic()
+                with lock:
+                    blocked = shared["moving"] or now < shared["guard_until"]
+                chunk = source.read()
+                p = pipe
+                if blocked:
+                    was_blocked = True  # drop our own motor/speaker sound
+                elif p is not None:
+                    if was_blocked:
+                        p.reset_audio()
+                        policy.reset()
+                        cue[:] = 0.0  # the worker already returned the robot to neutral
+                        was_blocked = False
+                    if chunk is not None:
+                        p.push_audio(chunk)
+                if p is not None and now >= next_hop:
+                    next_hop = max(next_hop + HOP_SEC, now)
+                    if not blocked:
+                        snap = p.step(now)
+                        with lock:
+                            shared["snap"] = snap
+                        probs = snap.fused.probs
+                        idle = probs is None
+                        emotion = policy.update(probs if probs is not None else np.zeros(len(EMOTIONS)), idle, now)
+                        if emotion and settings.reactions_enabled and (move := pick_move(emotion)):
+                            try:
+                                reaction_queue.put_nowait((emotion, move))
+                            except queue.Full:
+                                pass
+                if not blocked and now >= next_cue:
+                    # Antennas follow vocal arousal, head tilt follows vocal valence, plus slow breathing.
+                    next_cue = now + CUE_PERIOD_S
+                    with lock:
+                        snap = shared["snap"]
+                    vad = snap.vad if snap is not None and snap.branches["speech"].present else None
+                    target = np.array([
+                        40.0 * (float(vad[1]) - 0.5) if vad is not None else 0.0,
+                        12.0 * (float(vad[0]) - 0.5) if vad is not None else 0.0,
+                        1.5 * np.sin(2 * np.pi * 0.25 * now),
+                    ])
+                    cue += 0.1 * (target - cue)
+                    reachy_mini.set_target(
+                        head=create_head_pose(roll=cue[1], pitch=cue[2], degrees=True),
+                        antennas=REST_ANTENNAS + np.deg2rad([cue[0], -cue[0]]),
+                    )
+                time.sleep(0.005)
+        finally:
+            source.close()
+            if pipe is not None:
+                pipe.close()
+            for w in workers:
+                w.join(timeout=2.0)
+if __name__ == "__main__":
+    app = ReachyMiniMultimodalEmotion()
+    try:
+        app.wrapped_run()
+    except KeyboardInterrupt:
+        app.stop()

reachy_mini_multimodal_emotion/reactions.py ADDED Viewed

	@@ -0,0 +1,59 @@

+"""When to react (decision policy) and how to react (empathetic move mapping)."""
+from __future__ import annotations
+import random
+from dataclasses import dataclass, field
+from reachy_mini_multimodal_emotion.engine.labels import EMOTIONS as EMOTIONS_7
+# Empathetic responses from pollen-robotics/reachy-mini-emotions-library (neutral -> no move)
+EMPATHY_MOVES: dict[str, list[str]] = {
+    "happy": ["cheerful1", "enthusiastic1", "laughing1"],   # join in
+    "sad": ["understanding1", "understanding2", "loving1"],  # comfort
+    "angry": ["calming1", "serenity1"],                      # calm down
+    "surprise": ["surprised1", "amazed1", "curious1"],       # share the surprise
+    "fear": ["calming1", "welcoming1", "helpful1"],          # reassure
+    "disgust": ["thoughtful1", "uncomfortable1"],            # acknowledge
+}
+# Calibrated on held-out validation speakers (ESD 0005/0009 + 9 CREMA-D actors, 18,820 smoothed
+# 2 s windows) for roughly 75-80% precision per emotion. Sad is precise on acted data (98% at 0.70)
+# but over-fires on calm, quiet real voices, so it keeps a strict threshold on purpose.
+DEFAULT_THRESHOLDS = {"happy": 0.55, "sad": 0.70, "angry": 0.45, "surprise": 0.45, "fear": 0.50, "disgust": 0.65}
+@dataclass
+class ReactionPolicy:
+    """Fires once an emotion has been confidently dominant for `hold_s`, then waits `cooldown_s`."""
+    thresholds: dict[str, float] = field(default_factory=lambda: dict(DEFAULT_THRESHOLDS))
+    hold_s: float = 1.5
+    cooldown_s: float = 6.0
+    _candidate: str | None = None
+    _since: float = 0.0
+    _last_fire: float = float("-inf")
+    def reset(self) -> None:
+        self._candidate = None
+    def update(self, probs, idle: bool, now: float) -> str | None:
+        """Feed one smoothed prediction; returns the emotion to react to, or None."""
+        top = EMOTIONS_7[int(probs.argmax())]
+        confident = top in self.thresholds and float(probs.max()) >= self.thresholds[top]
+        if idle or not confident:
+            self._candidate = None
+            return None
+        if top != self._candidate:
+            self._candidate, self._since = top, now
+            return None
+        if now - self._since >= self.hold_s and now - self._last_fire >= self.cooldown_s:
+            self._last_fire = now
+            self._candidate = None
+            return top
+        return None
+def pick_move(emotion: str, rng: random.Random | None = None) -> str | None:
+    options = EMPATHY_MOVES.get(emotion)
+    return (rng or random).choice(options) if options else None

reachy_mini_multimodal_emotion/static/index.html ADDED Viewed

	@@ -0,0 +1,75 @@

+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <title>Multimodal Emotion</title>
+    <meta name="viewport" content="width=device-width, initial-scale=1, viewport-fit=cover">
+    <link rel="stylesheet" href="/static/style.css">
+</head>
+<body>
+    <header>
+        <h1>Multimodal Emotion</h1>
+        <span id="status" class="pill idle">connecting</span>
+    </header>
+    <main>
+        <section class="card hero">
+            <div class="label">Fused emotion</div>
+            <div id="top" class="top-emotion">-</div>
+            <div id="confidence" class="muted">waiting for a face or speech</div>
+            <div class="meta"><span id="used">-</span><span id="source">-</span><span id="models">-</span></div>
+        </section>
+        <section class="card">
+            <h2>Fused probabilities</h2>
+            <div id="bars"></div>
+        </section>
+        <section class="card">
+            <h2>Modalities</h2>
+            <div id="branches" class="branches"></div>
+            <p class="small muted">A modality without evidence (no face, silence, English speech for the Chinese text
+                model) is left out of the fusion; it never counts as "neutral".</p>
+        </section>
+        <section class="card">
+            <h2>What I heard</h2>
+            <div id="transcript" class="transcript muted">-</div>
+            <div id="asr" class="small muted"></div>
+            <div id="utterances"></div>
+        </section>
+        <section class="card">
+            <h2>Controls</h2>
+            <label class="switch-row"><span>Reactions</span><input type="checkbox" id="reactions_enabled" checked></label>
+            <label class="switch-row"><span>Move sounds</span><input type="checkbox" id="move_sound" checked></label>
+            <label class="switch-row"><span>Use face (camera)</span><input type="checkbox" id="use_face" checked></label>
+            <label class="switch-row"><span>Use words (speech-to-text)</span><input type="checkbox" id="use_text" checked></label>
+            <label class="slider-row"><span>Sensitivity <b id="sensitivity-val">0.50</b></span>
+                <input type="range" id="sensitivity" min="0" max="1" step="0.05" value="0.5"></label>
+            <h3>Fusion weights</h3>
+            <label class="slider-row"><span>Voice <b id="weight_speech-val">0.35</b></span>
+                <input type="range" id="weight_speech" min="0" max="1" step="0.05" value="0.35"></label>
+            <label class="slider-row"><span>Face <b id="weight_face-val">0.65</b></span>
+                <input type="range" id="weight_face" min="0" max="1" step="0.05" value="0.65"></label>
+            <label class="slider-row"><span>Words <b id="weight_text-val">0.35</b></span>
+                <input type="range" id="weight_text" min="0" max="1" step="0.05" value="0.35"></label>
+            <p class="small muted">Voice/face weights were tuned on acted CREMA-D video; the words weight is an
+                untested default. Weights are renormalised over the modalities present.</p>
+            <h3>Try a reaction</h3>
+            <div id="tests" class="tests"></div>
+            <div id="test-msg" class="small muted"></div>
+        </section>
+        <section class="card">
+            <h2>Recent reactions</h2>
+            <ul id="log" class="log"><li class="muted">none yet</li></ul>
+            <div id="moves" class="small muted"></div>
+        </section>
+    </main>
+    <script src="/static/main.js"></script>
+</body>
+</html>

reachy_mini_multimodal_emotion/static/main.js ADDED Viewed

	@@ -0,0 +1,145 @@

+const EMOTIONS = ["neutral", "happy", "sad", "angry", "surprise", "fear", "disgust"];
+const REACTABLE = ["happy", "sad", "angry", "surprise", "fear", "disgust"];
+const BRANCHES = { speech: "voice", face: "face", text: "words" };
+const STATUS_TEXT = {
+    ok: "live", silence: "silence", buffering: "listening…", no_face: "no face", no_camera: "no camera",
+    no_transcript: "no words", unsupported_language: "not Chinese", stale: "old", waiting: "waiting", disabled: "off",
+};
+const SWITCHES = ["reactions_enabled", "move_sound", "use_face", "use_text"];
+const SLIDERS = ["sensitivity", "weight_speech", "weight_face", "weight_text"];
+const color = (e) => `var(--${e})`;
+const $ = (id) => document.getElementById(id);
+const pct = (p) => `${(p * 100).toFixed(0)}%`;
+function el(tag, cls, text) {
+    const node = document.createElement(tag);
+    if (cls) node.className = cls;
+    if (text !== undefined) node.textContent = text;
+    return node;
+}
+// Static parts
+$("bars").innerHTML = EMOTIONS.map((e) => `
+  <div class="bar-row">
+    <span class="name">${e}</span>
+    <div class="track"><div class="fill" id="bar-${e}" style="background:${color(e)}"></div></div>
+    <span class="value" id="val-${e}">-</span>
+  </div>`).join("");
+$("branches").innerHTML = Object.entries(BRANCHES).map(([b, name]) => `
+  <div class="branch" id="br-${b}">
+    <div class="head"><span class="name">${name}</span><span class="status-chip" id="br-${b}-status">-</span></div>
+    <div class="top" id="br-${b}-top">-</div>
+    <div class="weight" id="br-${b}-weight">-</div>
+    <div class="mini-bars">${EMOTIONS.map((e) => `<div class="track" title="${e}"><div class="fill" id="br-${b}-${e}" style="background:${color(e)}"></div></div>`).join("")}</div>
+  </div>`).join("");
+$("tests").innerHTML = REACTABLE.map((e) => `<button data-emotion="${e}">${e}</button>`).join("");
+let settingsLoaded = false;
+function renderBranch(b, r, weight) {
+    const status = r ? r.status : "waiting";
+    const chip = $(`br-${b}-status`);
+    chip.textContent = STATUS_TEXT[status] || status;
+    chip.className = `status-chip ${status === "ok" ? "ok" : ""}`;
+    $(`br-${b}`).className = `branch ${status === "disabled" ? "off" : ""}`;
+    const probs = r && r.probs;
+    let top = null;
+    for (const e of EMOTIONS) {
+        const p = probs ? probs[e] : 0;
+        $(`br-${b}-${e}`).style.width = `${(p * 100).toFixed(1)}%`;
+        if (probs && (top === null || p > probs[top])) top = e;
+    }
+    $(`br-${b}-top`).textContent = top ? `${top} ${pct(probs[top])}` : "-";
+    $(`br-${b}-top`).style.color = top ? color(top) : "var(--muted)";
+    const lat = r && r.latency_ms ? ` · ${r.latency_ms} ms` : "";
+    $(`br-${b}-weight`).textContent = weight !== undefined ? `fusion weight ${pct(weight)}${lat}` : `not in fusion${lat}`;
+}
+function render(s) {
+    const statusPill = $("status");
+    const ready = s.status.models === "ready";
+    if (!ready) { statusPill.textContent = s.status.models.startsWith("failed") ? "model error" : "loading models"; statusPill.className = `pill ${s.status.models.startsWith("failed") ? "error" : "idle"}`; }
+    else if (s.moving) { statusPill.textContent = "reacting"; statusPill.className = "pill moving"; }
+    else if (!s.top) { statusPill.textContent = "watching / listening"; statusPill.className = "pill idle"; }
+    else { statusPill.textContent = "sensing"; statusPill.className = "pill listening"; }
+    $("models").textContent = `models: ${s.status.models}`;
+    $("source").textContent = s.source;
+    $("moves").textContent = `Emotion moves: ${s.status.moves}`;
+    const top = s.top || null;
+    $("top").textContent = top || "-";
+    $("top").style.color = top ? color(top) : "var(--muted)";
+    $("confidence").textContent = top ? `${pct(s.confidence)} confident` : "waiting for a face or speech";
+    const used = s.used ? Object.keys(s.used).map((b) => BRANCHES[b]) : [];
+    $("used").textContent = used.length ? `from ${used.join(" + ")}` : "no modality present";
+    for (const e of EMOTIONS) {
+        const p = s.fused ? s.fused[e] : 0;
+        $(`bar-${e}`).style.width = `${(p * 100).toFixed(1)}%`;
+        $(`val-${e}`).textContent = s.fused ? pct(p) : "-";
+    }
+    for (const b of Object.keys(BRANCHES)) renderBranch(b, s.branches && s.branches[b], s.used && s.used[b]);
+    $("transcript").textContent = s.transcript || "-";
+    $("asr").textContent = s.asr_lang ? `language: ${s.asr_lang} · SenseVoice's own tag: ${s.asr_emotion || "-"} (not fused)` : "";
+    $("utterances").replaceChildren(...(s.utterances || []).slice(0, 5).map((u) => {
+        const row = el("div", "utt");
+        row.append(el("span", "muted", u.time), el("span", "", u.text || "…"));
+        const tag = el("span", "", u.text_top || STATUS_TEXT[u.text_status] || u.text_status);
+        if (u.text_top) tag.style.color = color(u.text_top); else tag.className = "muted";
+        row.append(tag);
+        return row;
+    }));
+    $("log").replaceChildren(...(s.log.length ? s.log : [null]).map((r) => {
+        if (!r) return el("li", "muted", "none yet");
+        const li = el("li");
+        const emo = el("span", "emo", r.emotion);
+        if (EMOTIONS.includes(r.emotion)) emo.style.color = color(r.emotion);
+        li.append(el("span", "time", r.time), emo, el("span", "muted", `→ ${r.move}`));
+        if (r.used && r.used.length) li.append(el("span", "muted", `(${r.used.map((b) => BRANCHES[b]).join("+")})`));
+        return li;
+    }));
+    if (!settingsLoaded) {
+        for (const k of SWITCHES) $(k).checked = s.settings[k];
+        for (const k of SLIDERS) { $(k).value = s.settings[k]; $(`${k}-val`).textContent = Number(s.settings[k]).toFixed(2); }
+        settingsLoaded = true;
+    }
+}
+async function poll() {
+    try {
+        render(await (await fetch("/state")).json());
+    } catch (err) {
+        $("status").textContent = "disconnected";
+        $("status").className = "pill error";
+    } finally {
+        setTimeout(poll, 250);
+    }
+}
+async function pushSettings() {
+    const settings = {};
+    for (const k of SWITCHES) settings[k] = $(k).checked;
+    for (const k of SLIDERS) settings[k] = Number($(k).value);
+    await fetch("/settings", { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(settings) });
+}
+for (const k of SWITCHES) $(k).addEventListener("change", pushSettings);
+for (const k of SLIDERS) {
+    $(k).addEventListener("change", pushSettings);
+    $(k).addEventListener("input", () => { $(`${k}-val`).textContent = Number($(k).value).toFixed(2); });
+}
+$("tests").addEventListener("click", async (ev) => {
+    const emotion = ev.target.dataset.emotion;
+    if (!emotion) return;
+    const r = await (await fetch("/test_reaction", {
+        method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ emotion }),
+    })).json();
+    $("test-msg").textContent = r.queued ? `Playing "${r.move}"` : `Not played: ${r.reason}`;
+});
+poll();

reachy_mini_multimodal_emotion/static/style.css ADDED Viewed

	@@ -0,0 +1,140 @@

+:root {
+    --bg: #0f1216;
+    --card: #181d24;
+    --line: #2a313b;
+    --ink: #e8ecf1;
+    --muted: #93a0ae;
+    --accent: #ff9900;
+    --neutral: #b8c0c8;
+    --happy: #4ade80;
+    --sad: #60a5fa;
+    --angry: #f43f5e;
+    --surprise: #e879f9;
+    --fear: #facc15;
+    --disgust: #fb923c;
+}
+* { box-sizing: border-box; }
+body {
+    margin: 0;
+    background: var(--bg);
+    color: var(--ink);
+    font-family: system-ui, -apple-system, "Segoe UI", Roboto, sans-serif;
+    padding: env(safe-area-inset-top) 16px 24px;
+}
+header {
+    display: flex;
+    align-items: center;
+    justify-content: space-between;
+    max-width: 720px;
+    margin: 16px auto 12px;
+}
+h1 { font-size: 1.3rem; margin: 0; }
+h2 { font-size: 0.95rem; margin: 0 0 12px; color: var(--muted); font-weight: 600; }
+h3 { font-size: 0.9rem; margin: 18px 0 8px; color: var(--muted); }
+main { max-width: 720px; margin: 0 auto; display: grid; gap: 12px; }
+.card {
+    background: var(--card);
+    border: 1px solid var(--line);
+    border-radius: 14px;
+    padding: 16px;
+}
+.muted { color: var(--muted); }
+.small { font-size: 0.8rem; }
+.label { color: var(--muted); font-size: 0.8rem; text-transform: uppercase; letter-spacing: 0.06em; }
+.hero { text-align: center; }
+.top-emotion { font-size: 2.6rem; font-weight: 800; margin: 6px 0 2px; text-transform: capitalize; }
+.meta { display: flex; justify-content: center; gap: 14px; flex-wrap: wrap; margin-top: 10px; color: var(--muted); font-size: 0.8rem; }
+.pill { padding: 6px 12px; border-radius: 999px; font-size: 0.8rem; font-weight: 600; border: 1px solid var(--line); }
+.pill.listening { color: var(--happy); border-color: var(--happy); }
+.pill.idle { color: var(--muted); }
+.pill.moving { color: var(--accent); border-color: var(--accent); }
+.pill.error { color: var(--angry); border-color: var(--angry); }
+.bar-row { display: grid; grid-template-columns: 76px 1fr 48px; align-items: center; gap: 10px; margin: 8px 0; }
+.bar-row .name { text-transform: capitalize; font-size: 0.9rem; }
+.bar-row .value { text-align: right; font-variant-numeric: tabular-nums; font-size: 0.85rem; color: var(--muted); }
+.track { height: 10px; background: #232a33; border-radius: 6px; overflow: hidden; }
+.fill { height: 100%; width: 0; border-radius: 6px; transition: width 0.25s ease; }
+.fill.dom { background: var(--sad); }
+.circumplex { position: relative; padding: 22px 64px; margin: 0 auto 12px; max-width: 380px; }
+.plane {
+    position: relative;
+    aspect-ratio: 1;
+    max-width: 240px;
+    margin: 0 auto;
+    border: 1px solid var(--line);
+    border-radius: 8px;
+    background:
+        linear-gradient(var(--line), var(--line)) center / 1px 100% no-repeat,
+        linear-gradient(var(--line), var(--line)) center / 100% 1px no-repeat;
+}
+.axis { position: absolute; font-size: 0.75rem; color: var(--muted); }
+.axis.top { top: 0; left: 50%; transform: translateX(-50%); }
+.axis.bottom { bottom: 0; left: 50%; transform: translateX(-50%); }
+.axis.left { left: 0; top: 50%; transform: translateY(-50%); }
+.axis.right { right: 0; top: 50%; transform: translateY(-50%); }
+.dot {
+    position: absolute;
+    width: 16px;
+    height: 16px;
+    margin: -8px 0 0 -8px;
+    border-radius: 50%;
+    background: var(--ink);
+    box-shadow: 0 0 0 6px rgba(255, 255, 255, 0.12), 0 0 18px 4px rgba(255, 153, 0, 0.45);
+    left: 50%;
+    top: 50%;
+    transition: left 0.3s ease, top 0.3s ease;
+}
+.row { display: flex; justify-content: space-between; margin: 6px 0; }
+.switch-row, .slider-row { display: flex; align-items: center; justify-content: space-between; gap: 12px; min-height: 44px; }
+.slider-row { flex-wrap: wrap; }
+.slider-row input { flex: 1 1 100%; accent-color: var(--accent); min-height: 32px; }
+input[type="checkbox"] { width: 24px; height: 24px; accent-color: var(--accent); }
+.tests { display: grid; grid-template-columns: repeat(3, 1fr); gap: 8px; }
+.tests button {
+    min-height: 44px;
+    border-radius: 10px;
+    border: 1px solid var(--line);
+    background: #20262e;
+    color: var(--ink);
+    font-size: 0.9rem;
+    text-transform: capitalize;
+    cursor: pointer;
+}
+.tests button:active { transform: scale(0.97); }
+.log { list-style: none; margin: 0; padding: 0; }
+.log li { display: flex; gap: 10px; padding: 8px 0; border-bottom: 1px solid var(--line); font-size: 0.9rem; }
+.log li:last-child { border-bottom: none; }
+.log .time { color: var(--muted); font-variant-numeric: tabular-nums; }
+.log .emo { text-transform: capitalize; font-weight: 600; }
+/* multimodal additions */
+.branches { display: grid; grid-template-columns: repeat(auto-fit, minmax(190px, 1fr)); gap: 10px; }
+.branch { border: 1px solid var(--line); border-radius: 10px; padding: 10px 12px; }
+.branch.off { opacity: 0.55; }
+.branch .head { display: flex; justify-content: space-between; align-items: baseline; gap: 8px; }
+.branch .name { font-weight: 700; text-transform: capitalize; }
+.branch .top { font-weight: 700; text-transform: capitalize; margin: 4px 0; }
+.branch .weight { font-size: 0.75rem; color: var(--muted); }
+.mini-bars { display: grid; gap: 3px; margin-top: 6px; }
+.mini-bars .track { height: 6px; }
+.status-chip { font-size: 0.72rem; padding: 2px 8px; border-radius: 999px; border: 1px solid var(--line); color: var(--muted); }
+.status-chip.ok { color: var(--happy); border-color: var(--happy); }
+.transcript { font-size: 1.05rem; min-height: 1.6em; }
+.utt { display: grid; grid-template-columns: 64px 1fr auto; gap: 8px; padding: 6px 0; border-top: 1px solid var(--line); font-size: 0.85rem; }
+.utt:first-child { border-top: 0; }
+.log li .muted { margin-left: 6px; }

style.css CHANGED Viewed

@@ -1,28 +1,411 @@
-body {
-	padding: 2rem;
-	font-family: -apple-system, BlinkMacSystemFont, "Arial", sans-serif;
-}
-h1 {
-	font-size: 16px;
-	margin-top: 0;
-}
-p {
-	color: rgb(107, 114, 128);
-	font-size: 15px;
-	margin-bottom: 10px;
-	margin-top: 5px;
-}
-.card {
-	max-width: 620px;
-	margin: 0 auto;
-	padding: 16px;
-	border: 1px solid lightgray;
-	border-radius: 16px;
-}
-.card p:last-child {
-	margin-bottom: 0;
-}

+* {
+	margin: 0;
+	padding: 0;
+	box-sizing: border-box;
+}
+body {
+	font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, "Helvetica Neue", Arial, sans-serif;
+	line-height: 1.6;
+	color: #333;
+	background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
+	min-height: 100vh;
+}
+.hero {
+	background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
+	color: white;
+	padding: 4rem 2rem;
+	text-align: center;
+}
+.hero-content {
+	max-width: 800px;
+	margin: 0 auto;
+}
+.app-icon {
+	font-size: 4rem;
+	margin-bottom: 1rem;
+	display: inline-block;
+}
+.hero h1 {
+	font-size: 3rem;
+	font-weight: 700;
+	margin-bottom: 1rem;
+	background: linear-gradient(45deg, #fff, #f0f9ff);
+	background-clip: text;
+	-webkit-background-clip: text;
+	-webkit-text-fill-color: transparent;
+}
+.tagline {
+	font-size: 1.25rem;
+	opacity: 0.9;
+	max-width: 600px;
+	margin: 0 auto;
+}
+.container {
+	max-width: 1200px;
+	margin: 0 auto;
+	padding: 0 2rem;
+	position: relative;
+	z-index: 2;
+}
+.main-card {
+	background: white;
+	border-radius: 20px;
+	box-shadow: 0 20px 40px rgba(0, 0, 0, 0.1);
+	margin-top: -2rem;
+	overflow: hidden;
+	margin-bottom: 3rem;
+}
+.app-preview {
+	background: linear-gradient(135deg, #1e3a8a, #3b82f6);
+	padding: 3rem;
+	color: white;
+	text-align: center;
+	position: relative;
+}
+.preview-image {
+	background: #000;
+	border-radius: 15px;
+	padding: 2rem;
+	max-width: 500px;
+	margin: 0 auto;
+	position: relative;
+	overflow: hidden;
+}
+.camera-feed {
+	font-size: 4rem;
+	margin-bottom: 1rem;
+	opacity: 0.7;
+}
+.detection-overlay {
+	position: absolute;
+	top: 50%;
+	left: 50%;
+	transform: translate(-50%, -50%);
+	width: 100%;
+}
+.bbox {
+	background: rgba(34, 197, 94, 0.9);
+	color: white;
+	padding: 0.5rem 1rem;
+	border-radius: 8px;
+	font-size: 0.9rem;
+	font-weight: 600;
+	margin: 0.5rem;
+	display: inline-block;
+	border: 2px solid #22c55e;
+}
+.app-details {
+	padding: 3rem;
+}
+.app-details h2 {
+	font-size: 2rem;
+	color: #1e293b;
+	margin-bottom: 2rem;
+	text-align: center;
+}
+.template-info {
+	display: grid;
+	grid-template-columns: repeat(auto-fit, minmax(300px, 1fr));
+	gap: 2rem;
+	margin-bottom: 3rem;
+}
+.info-box {
+	background: #f0f9ff;
+	border: 2px solid #e0f2fe;
+	border-radius: 12px;
+	padding: 2rem;
+}
+.info-box h3 {
+	color: #0c4a6e;
+	margin-bottom: 1rem;
+	font-size: 1.2rem;
+}
+.info-box p {
+	color: #0369a1;
+	line-height: 1.6;
+}
+.how-to-use {
+	background: #fefce8;
+	border: 2px solid #fde047;
+	border-radius: 12px;
+	padding: 2rem;
+	margin-top: 3rem;
+}
+.how-to-use h3 {
+	color: #a16207;
+	margin-bottom: 1.5rem;
+	font-size: 1.3rem;
+	text-align: center;
+}
+.steps {
+	display: flex;
+	flex-direction: column;
+	gap: 1.5rem;
+}
+.step {
+	display: flex;
+	align-items: flex-start;
+	gap: 1rem;
+}
+.step-number {
+	background: #eab308;
+	color: white;
+	width: 2rem;
+	height: 2rem;
+	border-radius: 50%;
+	display: flex;
+	align-items: center;
+	justify-content: center;
+	font-weight: bold;
+	flex-shrink: 0;
+}
+.step h4 {
+	color: #a16207;
+	margin-bottom: 0.5rem;
+	font-size: 1.1rem;
+}
+.step p {
+	color: #ca8a04;
+}
+.download-card {
+	background: white;
+	border-radius: 20px;
+	box-shadow: 0 20px 40px rgba(0, 0, 0, 0.1);
+	padding: 3rem;
+	text-align: center;
+}
+.download-card h2 {
+	font-size: 2rem;
+	color: #1e293b;
+	margin-bottom: 1rem;
+}
+.download-card>p {
+	color: #64748b;
+	font-size: 1.1rem;
+	margin-bottom: 2rem;
+}
+.dashboard-config {
+	margin-bottom: 2rem;
+	text-align: left;
+	max-width: 400px;
+	margin-left: auto;
+	margin-right: auto;
+}
+.dashboard-config label {
+	display: block;
+	color: #374151;
+	font-weight: 600;
+	margin-bottom: 0.5rem;
+}
+.dashboard-config input {
+	width: 100%;
+	padding: 0.75rem 1rem;
+	border: 2px solid #e5e7eb;
+	border-radius: 8px;
+	font-size: 0.95rem;
+	transition: border-color 0.2s;
+}
+.dashboard-config input:focus {
+	outline: none;
+	border-color: #667eea;
+}
+.install-btn {
+	background: linear-gradient(135deg, #667eea, #764ba2);
+	color: white;
+	border: none;
+	padding: 1.25rem 3rem;
+	border-radius: 16px;
+	font-size: 1.2rem;
+	font-weight: 700;
+	cursor: pointer;
+	transition: all 0.3s ease;
+	display: inline-flex;
+	align-items: center;
+	gap: 0.75rem;
+	margin-bottom: 2rem;
+	box-shadow: 0 8px 25px rgba(102, 126, 234, 0.3);
+}
+.install-btn:hover:not(:disabled) {
+	transform: translateY(-3px);
+	box-shadow: 0 15px 35px rgba(102, 126, 234, 0.4);
+}
+.install-btn:disabled {
+	opacity: 0.7;
+	cursor: not-allowed;
+	transform: none;
+}
+.manual-option {
+	background: #f8fafc;
+	border-radius: 12px;
+	padding: 2rem;
+	margin-top: 2rem;
+}
+.manual-option h3 {
+	color: #1e293b;
+	margin-bottom: 1rem;
+	font-size: 1.2rem;
+}
+.manual-option>p {
+	color: #64748b;
+	margin-bottom: 1rem;
+}
+.btn-icon {
+	font-size: 1.1rem;
+}
+.install-status {
+	padding: 1rem;
+	border-radius: 8px;
+	font-size: 0.9rem;
+	text-align: center;
+	display: none;
+	margin-top: 1rem;
+}
+.install-status.success {
+	background: #dcfce7;
+	color: #166534;
+	border: 1px solid #bbf7d0;
+}
+.install-status.error {
+	background: #fef2f2;
+	color: #dc2626;
+	border: 1px solid #fecaca;
+}
+.install-status.loading {
+	background: #dbeafe;
+	color: #1d4ed8;
+	border: 1px solid #bfdbfe;
+}
+.install-status.info {
+	background: #e0f2fe;
+	color: #0369a1;
+	border: 1px solid #7dd3fc;
+}
+.manual-install {
+	background: #1f2937;
+	border-radius: 8px;
+	padding: 1rem;
+	margin-bottom: 1rem;
+	display: flex;
+	align-items: center;
+	gap: 1rem;
+}
+.manual-install code {
+	color: #10b981;
+	font-family: 'SF Mono', 'Monaco', 'Inconsolata', 'Roboto Mono', monospace;
+	font-size: 0.85rem;
+	flex: 1;
+	overflow-x: auto;
+}
+.copy-btn {
+	background: #374151;
+	color: white;
+	border: none;
+	padding: 0.5rem 1rem;
+	border-radius: 6px;
+	font-size: 0.8rem;
+	cursor: pointer;
+	transition: background-color 0.2s;
+}
+.copy-btn:hover {
+	background: #4b5563;
+}
+.manual-steps {
+	color: #6b7280;
+	font-size: 0.9rem;
+	line-height: 1.8;
+}
+.footer {
+	text-align: center;
+	padding: 2rem;
+	color: white;
+	opacity: 0.8;
+}
+.footer a {
+	color: white;
+	text-decoration: none;
+	font-weight: 600;
+}
+.footer a:hover {
+	text-decoration: underline;
+}
+/* Responsive Design */
+@media (max-width: 768px) {
+	.hero {
+		padding: 2rem 1rem;
+	}
+	.hero h1 {
+		font-size: 2rem;
+	}
+	.container {
+		padding: 0 1rem;
+	}
+	.app-details,
+	.download-card {
+		padding: 2rem;
+	}
+	.features-grid {
+		grid-template-columns: 1fr;
+	}
+	.download-options {
+		grid-template-columns: 1fr;
+	}
+}

tests/conftest.py ADDED Viewed

	@@ -0,0 +1,14 @@

+"""Model-dependent tests use MM_EMOTION_LOCAL_DIR (a folder with the model repo layout) and skip without it."""
+import os
+from pathlib import Path
+import pytest
+@pytest.fixture(scope="session")
+def model_root() -> Path:
+    root = os.environ.get("MM_EMOTION_LOCAL_DIR")
+    if not root or not (Path(root) / "config.json").exists():
+        pytest.skip("set MM_EMOTION_LOCAL_DIR to the model repo folder")
+    return Path(root)

tests/test_branches.py ADDED Viewed

	@@ -0,0 +1,104 @@

+"""Each torch-free branch must reproduce its training-time pipeline."""
+import os
+import sys
+import numpy as np
+import pytest
+from reachy_mini_multimodal_emotion.engine import load_config
+from reachy_mini_multimodal_emotion.engine.face import FaceBranch, crop_face, preprocess_face
+from reachy_mini_multimodal_emotion.engine.speech import LogMel
+# ------------------------------------------------------------------ speech
+@pytest.mark.parametrize("seconds", [0.5, 2.0, 3.3])
+def test_numpy_log_mel_matches_torchaudio(seconds):
+    torch = pytest.importorskip("torch")
+    T = pytest.importorskip("torchaudio.transforms")
+    rng = np.random.default_rng(0)
+    t = np.arange(int(seconds * 16000)) / 16000
+    wave = (0.1 * np.sin(2 * np.pi * 180 * t) + 0.02 * rng.standard_normal(len(t))).astype(np.float32)
+    mel = T.MelSpectrogram(sample_rate=16000, n_fft=512, win_length=400, hop_length=160, n_mels=80,
+                           f_min=20.0, f_max=8000.0, center=True, power=2.0)(torch.from_numpy(wave))
+    log_mel = torch.log(torch.clamp(mel, min=1e-5))
+    ref = ((log_mel - log_mel.mean(-1, keepdim=True)) / (log_mel.std(-1, keepdim=True) + 1e-5)).numpy()
+    ours = LogMel()(wave)
+    assert ours.shape == ref.shape and np.abs(ours - ref).max() < 1e-3
+# ------------------------------------------------------------------ face
+def test_face_preprocess_matches_torchvision_transform():
+    pytest.importorskip("torchvision")
+    import cv2
+    from PIL import Image
+    from torchvision import transforms
+    rng = np.random.default_rng(1)
+    face = rng.integers(0, 256, (97, 83, 3), dtype=np.uint8)
+    gray = cv2.cvtColor(face, cv2.COLOR_BGR2GRAY)
+    tf = transforms.Compose([transforms.Resize((112, 112)), transforms.ToTensor(),
+                             transforms.Normalize((0.485, 0.456, 0.406), (0.229, 0.224, 0.225))])
+    ref = tf(Image.fromarray(np.repeat(gray[..., None], 3, axis=2), mode="RGB")).unsqueeze(0).numpy()
+    np.testing.assert_allclose(preprocess_face(face), ref, atol=1e-5)
+def test_crop_face_pads_and_clamps():
+    frame = np.zeros((100, 200, 3), np.uint8)
+    assert crop_face(frame, (10, 10, 50, 50), 0.10).shape == (60, 60, 3)
+    assert crop_face(frame, (0, 0, 50, 50), 0.10).shape == (55, 55, 3)
+NEW_EMOTION = os.environ.get("NEW_EMOTION_REPO", "D:/newEmotion")
+def test_face_onnx_matches_training_checkpoint(model_root):
+    torch = pytest.importorskip("torch")
+    ckpt = os.path.join(NEW_EMOTION, "runs", "student_direct", "best.pt")
+    if not os.path.exists(ckpt):
+        pytest.skip("newEmotion checkpoint not available")
+    sys.path.insert(0, NEW_EMOTION)
+    from emotion_model.models import create_model
+    payload = torch.load(ckpt, map_location="cpu", weights_only=False)
+    model = create_model("mobilenet_v3_large", num_classes=7, pretrained=False)
+    model.load_state_dict(payload["model_state"])
+    model.eval()
+    cfg = load_config(model_root)
+    branch = FaceBranch(model_root / cfg["files"]["face"], model_root / cfg["files"]["face_detector"])
+    face = np.random.default_rng(2).integers(0, 256, (120, 100, 3), dtype=np.uint8)
+    with torch.no_grad():
+        ref = model(torch.from_numpy(preprocess_face(face))).numpy()[0]
+    np.testing.assert_allclose(branch.logits(face), ref, atol=1e-3)
+# ------------------------------------------------------------------ text
+def test_fast_tokenizer_matches_transformers(model_root):
+    transformers = pytest.importorskip("transformers")
+    from tokenizers import Tokenizer
+    src = os.environ.get("TER_MODEL", "D:/TER/TextEmotionDetection-model-2026-09-23/models/best_model")
+    if not os.path.exists(src):
+        pytest.skip("original text model not available")
+    ref = transformers.AutoTokenizer.from_pretrained(src)
+    ours = Tokenizer.from_file(str(model_root / "text" / "tokenizer.json"))
+    ours.enable_truncation(128)
+    for text in ["我今天終於完成專案了，超開心！", "他說：「我不是不生氣」", "OK啦 123 abc", "很" * 300]:
+        enc = ours.encode(text)
+        r = ref(text, truncation=True, max_length=128)
+        assert enc.ids == r["input_ids"] and enc.type_ids == r["token_type_ids"]
+def test_text_branch_statuses(model_root):
+    pytest.importorskip("chinese_converter")
+    from reachy_mini_multimodal_emotion.engine.text import TTL_S, TextBranch
+    cfg = load_config(model_root)
+    tb = TextBranch(model_root / cfg["files"]["text"], model_root / cfg["files"]["text_tokenizer"])
+    assert tb.classify("", "zh", 0.0).status == "no_transcript"
+    assert tb.classify("I am so happy today", "en", 0.0).status == "unsupported_language"
+    r = tb.classify("我今天终于完成项目了，超开心！", "zh", 10.0)  # Simplified input is converted
+    assert r.present and tb.transcript.startswith("我今天終於")
+    assert int(r.probs.argmax()) == 1  # happy
+    assert tb.current(10.0 + TTL_S - 0.1).present
+    assert tb.current(10.0 + TTL_S + 0.1).status == "stale"

tests/test_fusion.py ADDED Viewed

	@@ -0,0 +1,65 @@

+"""Label maps and late fusion behaviour, including missing modalities."""
+import numpy as np
+from reachy_mini_multimodal_emotion.engine.fusion import OK, BranchResult, fuse
+from reachy_mini_multimodal_emotion.engine.labels import BRANCH_LABELS, EMOTIONS, softmax, to_canonical
+def onehot_logits(branch, emotion, high=5.0):
+    z = np.zeros(7)
+    z[BRANCH_LABELS[branch].index(emotion)] = high
+    return z
+def test_each_branch_reorders_into_canonical_labels():
+    for branch in BRANCH_LABELS:
+        for emotion in EMOTIONS:
+            assert EMOTIONS[int(to_canonical(branch, onehot_logits(branch, emotion)).argmax())] == emotion
+def test_face_reorder_matches_ser_handoff_index_list():
+    # SER_models/README.md: reorder face logits into speech order with [6, 3, 4, 0, 5, 2, 1]
+    z = np.arange(7.0)
+    assert to_canonical("face", z).tolist() == z[[6, 3, 4, 0, 5, 2, 1]].tolist()
+def probs_for(emotion, p):
+    x = np.full(7, (1 - p) / 6)
+    x[EMOTIONS.index(emotion)] = p
+    return x
+W = {"speech": 0.35, "face": 0.65, "text": 0.35}
+def test_missing_branches_are_dropped_not_counted_as_neutral():
+    only_speech = fuse({"speech": BranchResult(probs_for("angry", 0.6), OK),
+                        "face": BranchResult(status="no_face"),
+                        "text": BranchResult(status="unsupported_language")}, W)
+    assert only_speech.used == {"speech": 1.0}
+    np.testing.assert_allclose(only_speech.probs, probs_for("angry", 0.6))
+def test_nothing_present_gives_no_prediction():
+    fused = fuse({"speech": BranchResult(status="silence"), "face": BranchResult(status="no_face")}, W)
+    assert fused.probs is None and fused.top is None and fused.used == {}
+def test_log_linear_pooling_weights_and_normalisation():
+    a, b = probs_for("happy", 0.7), probs_for("sad", 0.7)
+    fused = fuse({"speech": BranchResult(a, OK), "face": BranchResult(b, OK)}, W)
+    expected = np.exp(0.35 * np.log(a) + 0.65 * np.log(b))
+    np.testing.assert_allclose(fused.probs, expected / expected.sum())
+    assert fused.top == "sad" and abs(sum(fused.used.values()) - 1) < 1e-12
+def test_agreeing_branches_raise_confidence():
+    p = probs_for("surprise", 0.5)
+    fused = fuse({b: BranchResult(p, OK) for b in W}, W)
+    assert fused.top == "surprise" and np.isclose(fused.confidence, 0.5)  # identical inputs: unchanged
+def test_temperature_flattens():
+    z = np.array([3.0, 0, 0, 0, 0, 0, 0])
+    assert softmax(z, 6.46).max() < softmax(z, 1.0).max()

tests/test_pipeline.py ADDED Viewed

	@@ -0,0 +1,68 @@

+"""End to end through MultimodalPipeline with the real models (skipped without MM_EMOTION_LOCAL_DIR)."""
+import os
+import time
+import wave
+from pathlib import Path
+import numpy as np
+import pytest
+from reachy_mini_multimodal_emotion.engine.speech import SAMPLE_RATE
+@pytest.fixture(scope="module")
+def pipe(model_root):
+    from reachy_mini_multimodal_emotion.engine.pipeline import MultimodalPipeline
+    asr = os.environ.get("MM_EMOTION_ASR_DIR")
+    p = MultimodalPipeline(model_root, asr if asr and Path(asr).exists() else None)
+    yield p
+    p.close()
+def voiced(seconds=2.0):
+    t = np.arange(int(SAMPLE_RATE * seconds)) / SAMPLE_RATE
+    return (0.08 * np.sign(np.sin(2 * np.pi * 150 * t)) * (0.6 + 0.4 * np.sin(2 * np.pi * 3 * t))).astype(np.float32)
+def test_silence_and_no_face_give_no_emotion(pipe):
+    pipe.reset_audio()
+    pipe.push_audio(np.zeros(SAMPLE_RATE * 2, np.float32))
+    pipe.push_frame(np.full((480, 640, 3), 127, np.uint8))
+    snap = pipe.step()
+    assert snap.branches["speech"].status == "silence"
+    assert snap.branches["face"].status == "no_face"
+    assert snap.fused.probs is None
+def test_voice_only_fusion_uses_speech_alone(pipe):
+    pipe.reset_audio()
+    pipe.push_audio(voiced())
+    pipe.push_frame(None)
+    snap = pipe.step()
+    assert snap.branches["speech"].present and snap.fused.used == {"speech": 1.0}
+    np.testing.assert_allclose(snap.fused.probs, snap.branches["speech"].probs, atol=1e-9)
+def test_asr_utterance_reaches_text_branch(pipe):
+    if pipe.transcriber is None:
+        pytest.skip("set MM_EMOTION_ASR_DIR to the SenseVoice folder")
+    clip = Path(os.environ["MM_EMOTION_ASR_DIR"]) / "test_wavs" / "zh.wav"
+    if not clip.exists():
+        pytest.skip("SenseVoice zh.wav test clip not available")
+    with wave.open(str(clip)) as w:
+        audio = np.frombuffer(w.readframes(w.getnframes()), np.int16).astype(np.float32) / 32768
+    audio = np.concatenate([audio, np.zeros(SAMPLE_RATE, np.float32)])  # trailing silence ends the utterance
+    time.sleep(1.0)  # let the worker finish segments left over from earlier tests
+    pipe.reset_audio()
+    pipe.utterances.clear()
+    for i in range(0, len(audio), 1024):
+        pipe.push_audio(audio[i:i + 1024])
+    deadline = time.monotonic() + 20
+    while not any(u["lang"] == "zh" for u in pipe.utterances) and time.monotonic() < deadline:
+        time.sleep(0.1)
+    zh = [u for u in pipe.utterances if u["lang"] == "zh"]
+    assert zh, f"no Mandarin utterance transcribed: {list(pipe.utterances)}"
+    assert "時間" in zh[0]["text"]  # converted to Traditional
+    assert pipe.step().branches["text"].present

tests/test_policy.py ADDED Viewed

	@@ -0,0 +1,67 @@

+"""Reaction policy and SER engine behaviour."""
+import random
+import numpy as np
+from reachy_mini_multimodal_emotion.engine.labels import EMOTIONS
+from reachy_mini_multimodal_emotion.reactions import EMPATHY_MOVES, ReactionPolicy, pick_move
+def probs_for(emotion, p):
+    x = np.full(7, (1 - p) / 6)
+    x[EMOTIONS.index(emotion)] = p
+    return x
+def test_policy_needs_sustained_confidence_then_cools_down():
+    pol = ReactionPolicy(hold_s=1.5, cooldown_s=6.0)
+    happy = probs_for("happy", 0.8)
+    assert pol.update(happy, False, 0.0) is None          # first sighting starts the hold timer
+    assert pol.update(happy, False, 1.0) is None          # not held long enough
+    assert pol.update(happy, False, 1.6) == "happy"       # fires after 1.5 s
+    assert pol.update(happy, False, 2.0) is None          # restarts hold
+    assert pol.update(happy, False, 4.0) is None          # held, but inside the 6 s cooldown
+    assert pol.update(happy, False, 7.7) == "happy"       # cooldown over
+def test_policy_ignores_neutral_idle_low_confidence_and_resets_on_switch():
+    pol = ReactionPolicy(hold_s=1.0, cooldown_s=0.0)
+    assert pol.update(probs_for("neutral", 0.9), False, 0.0) is None
+    assert pol.update(probs_for("neutral", 0.9), False, 5.0) is None      # neutral never reacts
+    assert pol.update(probs_for("angry", 0.4), False, 6.0) is None        # below threshold
+    pol.update(probs_for("angry", 0.8), False, 7.0)
+    assert pol.update(probs_for("angry", 0.8), True, 8.5) is None         # idle interrupts
+    pol.update(probs_for("angry", 0.8), False, 9.0)
+    pol.update(probs_for("fear", 0.8), False, 9.5)                        # switch restarts hold
+    assert pol.update(probs_for("fear", 0.8), False, 10.0) is None
+    assert pol.update(probs_for("fear", 0.8), False, 10.6) == "fear"
+def test_every_reactable_emotion_has_moves():
+    rng = random.Random(0)
+    for emo in EMOTIONS[1:]:
+        assert pick_move(emo, rng) in EMPATHY_MOVES[emo]
+    assert pick_move("neutral") is None
+def test_mic_source_returns_with_continuous_audio():
+    """Regression from SER_Reachy_merli_v2: a chunk is always ready, so read() must not drain 'until None'."""
+    import threading
+    from types import SimpleNamespace
+    from reachy_mini_multimodal_emotion.main import MicSource
+    class EndlessMic:
+        def start_recording(self): pass
+        def stop_recording(self): pass
+        def get_audio_sample(self): return np.full((256, 2), 0.1, np.float32)
+    src = MicSource(SimpleNamespace(media=EndlessMic()))
+    out = []
+    t = threading.Thread(target=lambda: out.append(src.read()), daemon=True)
+    t.start(); t.join(timeout=1.0)
+    assert not t.is_alive() and out[0].shape == (256,)