You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
73 lines
3.5 KiB
73 lines
3.5 KiB
from unittest.mock import Mock
|
|
|
|
from src.config_loader import VoicePromptSettings
|
|
from src.models import BoundingBox, FrameInfo, TrackEvent, TrackedPerson, TurnFrameResult
|
|
from src.voice_prompt_controller import VoicePromptController
|
|
|
|
|
|
def cfg(mode: str = "prompt", **kw: object) -> VoicePromptSettings:
|
|
values = dict(enabled=True, mode=mode, trigger_strategy="trigger_line_crossing", require_confirmed_track=True,
|
|
require_inside_region=True, skip_if_already_turned=True, one_prompt_per_track=True,
|
|
global_cooldown_sec=3., minimum_track_age_sec=0., response_window_sec=3.,
|
|
fixed_position_axis="x", fixed_position_value=640., fixed_position_direction="decreasing")
|
|
values.update(kw)
|
|
return VoicePromptSettings(**values)
|
|
|
|
|
|
def info(frame: int, elapsed: float) -> FrameInfo:
|
|
return FrameInfo(frame, f"t{frame}", elapsed, 100, 100, 30.)
|
|
|
|
|
|
def person(track_id: int = 1) -> TrackedPerson:
|
|
return TrackedPerson(track_id, BoundingBox(1, 1, 20, 40), .9, 0, "person", 10, 20, 10, 40, True, True, 10, 0)
|
|
|
|
|
|
def crossing(track_id: int = 1) -> TrackEvent:
|
|
return TrackEvent("trigger_crossed", track_id, 1, "t", 1., 10, 40)
|
|
|
|
|
|
def turn(frame: int, elapsed: float, confirmed: bool = True) -> TurnFrameResult:
|
|
return TurnFrameResult(1, frame, f"t{frame}", elapsed, True, True, True, 0., 25., 25., 25., "positive",
|
|
True, confirmed, "weak", .4, "")
|
|
|
|
|
|
def test_prompt_success_duplicate_and_response() -> None:
|
|
audio = Mock(audio_file="a.wav")
|
|
audio.play.return_value = True
|
|
controller = VoicePromptController(cfg(), audio)
|
|
decisions, events = controller.update([person()], [crossing()], [], info(1, 1.))
|
|
assert decisions[0].prompt_played
|
|
assert {e.event_type for e in events} >= {"prompt_played", "response_window_started"}
|
|
controller.update([person()], [crossing()], [], info(2, 1.1))
|
|
assert audio.play.call_count == 1
|
|
_, events = controller.update([person()], [], [turn(3, 1.8)], info(3, 1.8))
|
|
response = next(e for e in events if e.event_type == "response_detected")
|
|
assert response.reaction_time_sec == .8
|
|
|
|
|
|
def test_prompt_failure_control_and_expiration() -> None:
|
|
failed = Mock(audio_file="missing.wav")
|
|
failed.play.return_value = False
|
|
controller = VoicePromptController(cfg(), failed)
|
|
_, events = controller.update([person()], [crossing()], [], info(1, 1.))
|
|
assert "prompt_failed" in {event.event_type for event in events}
|
|
control_audio = Mock(audio_file="a.wav")
|
|
control = VoicePromptController(cfg("control", response_window_sec=1.), control_audio)
|
|
decisions, events = control.update([person()], [crossing()], [], info(1, 1.))
|
|
assert decisions[0].pseudo_prompt and not decisions[0].prompt_played
|
|
control_audio.play.assert_not_called()
|
|
_, events = control.update([person()], [], [], info(2, 2.1))
|
|
assert "response_window_expired" in {event.event_type for event in events}
|
|
|
|
|
|
def test_cooldown_and_finalize_all() -> None:
|
|
audio = Mock(audio_file="a.wav")
|
|
audio.play.return_value = True
|
|
controller = VoicePromptController(cfg(), audio)
|
|
controller.update([person(1)], [crossing(1)], [], info(1, 1.))
|
|
decisions, _ = controller.update([person(2)], [crossing(2)], [], info(2, 2.))
|
|
assert decisions[0].reason == "global_cooldown"
|
|
states, events = controller.finalize_all(info(3, 5.))
|
|
assert len(states) == 2 and all(state.finalized for state in states)
|
|
assert all(event.event_type == "voice_track_finalized" for event in events)
|
|
|
|
|