silver_pose/v1/tests/test_fall_state.py

import pytest

from v1.fall_state import Evidence, FallState, FallStateMachine


CONFIG_VERSION = "cfg-test-20260721"


def test_confirmed_event_is_emitted_once_after_persistent_evidence():
    machine = FallStateMachine(
        confirm_window_seconds=1.8,
        recovery_window_seconds=2.0,
        config_version=CONFIG_VERSION,
    )

    assert machine.update("P-0007", Evidence(True, True), now=0.0) == []
    events = machine.update("P-0007", Evidence(True, True), now=1.8)

    assert len(events) == 1
    assert events[0].track_id == "P-0007"
    assert events[0].latency_seconds == 1.8
    assert events[0].config_version == CONFIG_VERSION
    assert machine.update("P-0007", Evidence(True, True), now=2.0) == []
    assert machine.state_of("P-0007") is FallState.CONFIRMED


def test_brief_low_posture_returns_to_normal_without_event():
    machine = FallStateMachine(
        confirm_window_seconds=1.0,
        recovery_window_seconds=2.0,
        config_version=CONFIG_VERSION,
    )

    machine.update("P-0007", Evidence(True, True), now=0.0)
    events = machine.update("P-0007", Evidence(True, False), now=0.3)

    assert events == []
    assert machine.state_of("P-0007") is FallState.NORMAL


def test_rejected_pose_resets_suspect_and_cannot_shorten_confirmation_window():
    machine = FallStateMachine(
        confirm_window_seconds=1.0,
        recovery_window_seconds=2.0,
        config_version=CONFIG_VERSION,
    )

    machine.update("P-0007", Evidence(True, True), now=0.0)
    machine.update("P-0007", Evidence(False, False), now=0.9)
    events = machine.update("P-0007", Evidence(True, True), now=1.1)

    assert events == []
    assert machine.state_of("P-0007") is FallState.SUSPECT


def test_recovery_must_persist_before_new_event_is_allowed():
    machine = FallStateMachine(
        confirm_window_seconds=1.0,
        recovery_window_seconds=2.0,
        config_version=CONFIG_VERSION,
    )
    machine.update("P-0007", Evidence(True, True), now=0.0)
    first_event = machine.update("P-0007", Evidence(True, True), now=1.0)
    assert len(first_event) == 1

    machine.update("P-0007", Evidence(True, False, True), now=1.1)
    machine.update("P-0007", Evidence(True, False, True), now=3.1)
    assert machine.state_of("P-0007") is FallState.NORMAL

    machine.update("P-0007", Evidence(True, True), now=4.0)
    second_event = machine.update("P-0007", Evidence(True, True), now=5.0)
    assert len(second_event) == 1
    assert first_event[0].event_id != second_event[0].event_id


def test_confirmation_window_must_remain_within_customer_target():
    with pytest.raises(ValueError, match="between 1 and 3"):
        FallStateMachine(
            confirm_window_seconds=0.9,
            recovery_window_seconds=2.0,
            config_version=CONFIG_VERSION,
        )


def test_session_id_makes_event_ids_unique_across_runs():
    run_a = FallStateMachine(
        confirm_window_seconds=1.0, recovery_window_seconds=2.0,
        config_version="cfg", session_id="run-a",
    )
    run_b = FallStateMachine(
        confirm_window_seconds=1.0, recovery_window_seconds=2.0,
        config_version="cfg", session_id="run-b",
    )
    run_a.update("P-0001", Evidence(True, True), now=0.0)
    event_a = run_a.update("P-0001", Evidence(True, True), now=1.0)
    run_b.update("P-0001", Evidence(True, True), now=0.0)
    event_b = run_b.update("P-0001", Evidence(True, True), now=1.0)

    assert event_a[0].event_id != event_b[0].event_id
    assert "run-a" in event_a[0].event_id
    assert "run-b" in event_b[0].event_id


def test_each_track_has_an_independent_confirmation_window():
    machine = FallStateMachine(
        confirm_window_seconds=1.0,
        recovery_window_seconds=2.0,
        config_version=CONFIG_VERSION,
    )

    machine.update("P-0001", Evidence(True, True), now=0.0)
    machine.update("P-0002", Evidence(True, True), now=0.6)
    first_events = machine.update("P-0001", Evidence(True, True), now=1.0)
    second_events = machine.update("P-0002", Evidence(True, True), now=1.6)

    assert first_events[0].track_id == "P-0001"
    assert second_events[0].track_id == "P-0002"
    assert first_events[0].event_id != second_events[0].event_id


def test_cooldown_delays_recovery_after_confirmation():
    machine = FallStateMachine(
        confirm_window_seconds=1.0,
        recovery_window_seconds=2.0,
        config_version=CONFIG_VERSION,
        cooldown_seconds=3.0,
    )
    machine.update("P-0007", Evidence(True, True), now=0.0)
    machine.update("P-0007", Evidence(True, True), now=1.0)

    machine.update("P-0007", Evidence(True, False, True), now=1.1)
    assert machine.state_of("P-0007") is FallState.CONFIRMED

    machine.update("P-0007", Evidence(True, False, True), now=4.0)
    assert machine.state_of("P-0007") is FallState.RECOVERING