tests/test_policy.py

"""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,)