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.
56 lines
3.0 KiB
56 lines
3.0 KiB
from src.config_loader import PersonStateSettings
|
|
from src.models import (BoundingBox, FrameInfo, HeadPoseResult, TrackEvent, TrackedPerson, TurnFrameResult,
|
|
VoicePromptDecision, VoicePromptEvent, VoicePromptTrackState)
|
|
from src.person_state_manager import PersonStateManager
|
|
|
|
|
|
CFG = PersonStateSettings(True, 0., False, False, False, True, True)
|
|
|
|
|
|
def info(frame: int, elapsed: float) -> FrameInfo:
|
|
return FrameInfo(frame, f"t{frame}", elapsed, 100, 100, 30.)
|
|
|
|
|
|
def person() -> TrackedPerson:
|
|
return TrackedPerson(1, BoundingBox(1, 1, 20, 40), .9, 0, "person", 10, 20, 10, 40, True, True, 10, 0)
|
|
|
|
|
|
def face() -> HeadPoseResult:
|
|
return HeadPoseResult(1, 1, "t1", 1., True, True, 0., 10., 0., 6, None, .9)
|
|
|
|
|
|
def turn() -> TurnFrameResult:
|
|
return TurnFrameResult(1, 1, "t1", 1., True, True, True, 0., 25., 25., 25., "positive", True, True, "weak", .4)
|
|
|
|
|
|
def test_integrates_track_face_turn_voice_and_response() -> None:
|
|
manager = PersonStateManager(CFG, "exp")
|
|
track_events = [TrackEvent("region_entered", 1, 1, "t1", 1., 10, 40),
|
|
TrackEvent("trigger_crossed", 1, 1, "t1", 1., 10, 40, "left_to_right")]
|
|
decision = VoicePromptDecision(1, 1, "t1", 1., "prompt", True, True, True, False, "prompt_played", "a.wav")
|
|
voice_events = [VoicePromptEvent("response_window_started", 1, 1, "t1", 1., "prompt", True),
|
|
VoicePromptEvent("response_detected", 1, 1, "t1", 1.5, "prompt", True,
|
|
turn_confirmed=True, turn_level="weak", reaction_time_sec=.5)]
|
|
voice_state = VoicePromptTrackState(1, "prompt", eligible=True, prompt_played=True,
|
|
response_window_started=True, response_detected=True,
|
|
reaction_time_sec=.5, response_turn_level="weak")
|
|
snapshots, events = manager.update([person()], track_events, [face()], [turn()], None,
|
|
[decision], voice_events, {1: voice_state}, info(1, 1.5))
|
|
state = snapshots[0]
|
|
assert state.status == "responded" and state.trigger_crossed
|
|
assert state.face_detected and state.pose_estimated and state.turn_confirmed
|
|
assert state.prompt_played and state.response_detected and state.reaction_time_sec == .5
|
|
assert any(event.event_type == "status_changed" for event in events)
|
|
summary, _ = manager.finalize_track(1, info(2, 2.), voice_state=voice_state)
|
|
assert summary and summary.trigger_crossed and summary.response_detected
|
|
|
|
|
|
def test_control_no_response_and_finalize_all() -> None:
|
|
manager = PersonStateManager(CFG, "exp")
|
|
decision = VoicePromptDecision(1, 1, "t1", 1., "control", True, True, False, True, "pseudo")
|
|
events = [VoicePromptEvent("response_window_expired", 1, 2, "t2", 2., "control", pseudo_prompt=True)]
|
|
snapshots, _ = manager.update([person()], [], [], [], None, [decision], events, None, info(2, 2.))
|
|
assert snapshots[0].status == "no_response" and snapshots[0].pseudo_prompt
|
|
summaries, _ = manager.finalize_all(info(3, 3.))
|
|
assert len(summaries) == 1
|
|
|
|
|