silver_pose/v1/tests/test_view_model.py

from types import SimpleNamespace

import pytest

from v1.fall_state import FallState
from v1.video_source import SourceStatus
from v1.view_model import (
    DraftValidationError,
    SettingsDraft,
    StatusColor,
    build_monitor_view,
    config_version,
)


def _keypoint(x, y, confidence=0.9):
    return SimpleNamespace(x=float(x), y=float(y), confidence=float(confidence))


def _person(track_id, state, keypoints=None, box=(10.0, 10.0, 40.0, 120.0)):
    points = keypoints or [_keypoint(index, index) for index in range(17)]
    pose = SimpleNamespace(box_xyxy=box, box_confidence=0.9, keypoints=tuple(points))
    tracked = SimpleNamespace(track_id=track_id, detected_at_monotonic=1.0, pose=pose)
    return SimpleNamespace(tracked_pose=tracked, pose_evidence=None, state=state)


def _analysis(status=SourceStatus.CONNECTED, people=(), events=(), has_image=True):
    image = object() if has_image else None
    packet = SimpleNamespace(image=image, timestamp_monotonic=1.0, status=status, error=None)
    return SimpleNamespace(packet=packet, people=tuple(people), events=tuple(events))


def test_state_maps_to_semantic_color_and_red_is_confirmed_only():
    view = build_monitor_view(
        _analysis(
            people=[
                _person("P-0001", FallState.NORMAL),
                _person("P-0002", FallState.SUSPECT),
                _person("P-0003", FallState.CONFIRMED),
            ]
        )
    )
    colors = {overlay.track_id: overlay.color for overlay in view.people}
    assert colors["P-0001"] is StatusColor.SUCCESS
    assert colors["P-0002"] is StatusColor.CAUTION
    assert colors["P-0003"] is StatusColor.CRITICAL
    critical = [o for o in view.people if o.color is StatusColor.CRITICAL]
    assert [o.state for o in critical] == [FallState.CONFIRMED]
    assert view.highest_state is FallState.CONFIRMED


def test_disconnected_frame_shows_reconnect_text_and_no_fall():
    view = build_monitor_view(_analysis(status=SourceStatus.RETRYING, people=(), has_image=False))
    assert view.connected is False
    assert view.status_text == "正在重连…"
    assert view.status_color is StatusColor.OFFLINE
    assert view.people == ()
    assert view.highest_state is FallState.NORMAL


def test_skeleton_overlay_drops_low_confidence_keypoints_and_labels_person():
    points = [_keypoint(index, index, 0.9) for index in range(17)]
    points[9] = _keypoint(9, 9, 0.1)  # left wrist below threshold
    view = build_monitor_view(
        _analysis(people=[_person("P-0007", FallState.SUSPECT, keypoints=points)]),
        keypoint_min_confidence=0.4,
    )
    overlay = view.people[0]
    assert overlay.keypoints[9] is None
    assert overlay.keypoints[7] is not None
    # Edges touching keypoint 9 (5-7-9 arm) must not draw the 7->9 segment.
    assert all(
        not (a is overlay.keypoints[7] and b is None) for a, b in overlay.skeleton_segments
    )
    assert overlay.label == "P-0007 · SUSPECT"
    assert overlay.box_xyxy == (10.0, 10.0, 40.0, 120.0)


def test_confirmed_event_becomes_a_badge():
    event = SimpleNamespace(event_id="FALL-000001", track_id="P-0003", latency_seconds=1.8)
    view = build_monitor_view(_analysis(events=[event]))
    assert len(view.events) == 1
    assert view.events[0].event_id == "FALL-000001"
    assert view.events[0].latency_seconds == 1.8


def test_person_diagnostic_line_shows_decision_fields():
    from v1.view_model import person_diagnostic_line

    pose_evidence = SimpleNamespace(
        accepted=True, horizontal_pose=False, rapid_vertical_change=False,
        horizontal_angle_degrees=72.0, reason="accepted",
    )
    evidence = SimpleNamespace(is_fall_candidate=False, is_recovery_candidate=False)
    person = SimpleNamespace(
        tracked_pose=SimpleNamespace(track_id="P-0001"),
        state=FallState.NORMAL, pose_evidence=pose_evidence, evidence=evidence,
    )

    assert person_diagnostic_line(person) == "P-0001 NORMAL acc=1 horiz=0 ang=72deg rapid=0 cand=0"


def test_person_diagnostic_line_shows_reject_reason():
    from v1.view_model import person_diagnostic_line

    pose_evidence = SimpleNamespace(accepted=False, reason="required_joint_low_confidence")
    person = SimpleNamespace(
        tracked_pose=SimpleNamespace(track_id="P-0002"),
        state=FallState.NORMAL, pose_evidence=pose_evidence, evidence=None,
    )

    assert person_diagnostic_line(person) == "P-0002 NORMAL acc=0 required_joint_low_confidence"


def test_build_monitor_view_includes_a_diagnostic_per_person():
    view = build_monitor_view(_analysis(people=[_person("P-0001", FallState.NORMAL)]))

    assert len(view.diagnostics) == 1
    assert view.diagnostics[0].startswith("P-0001 NORMAL")


def _initial():
    return {
        "keypoint_confidence_threshold": 0.4,
        "suspect_window_seconds": 0.5,
        "confirm_window_seconds": 1.8,
        "recovery_window_seconds": 2.0,
        "cooldown_seconds": 10.0,
        "model_confidence_threshold": 0.25,
    }


def test_editing_draft_does_not_change_running_or_saved_config():
    draft = SettingsDraft(_initial())
    draft.edit("confirm_window_seconds", 2.5)
    assert draft.is_dirty is True
    assert draft.draft_values["confirm_window_seconds"] == 2.5
    assert draft.saved_values["confirm_window_seconds"] == 1.8
    assert draft.running_values["confirm_window_seconds"] == 1.8
    assert draft.has_pending_for_next_start is False


def test_save_marks_pending_but_running_only_changes_on_start():
    draft = SettingsDraft(_initial())
    before_version = draft.running_version
    draft.edit("confirm_window_seconds", 2.5)
    message = draft.save()
    assert "下次开始监控" in message
    assert draft.is_dirty is False
    assert draft.has_pending_for_next_start is True
    assert draft.running_values["confirm_window_seconds"] == 1.8
    assert draft.running_version == before_version

    new_version = draft.start_monitoring()
    assert draft.running_values["confirm_window_seconds"] == 2.5
    assert draft.has_pending_for_next_start is False
    assert new_version != before_version


def test_discard_reverts_draft_to_saved():
    draft = SettingsDraft(_initial())
    draft.edit("cooldown_seconds", 20.0)
    draft.discard()
    assert draft.is_dirty is False
    assert draft.draft_values["cooldown_seconds"] == 10.0


def test_out_of_range_draft_value_is_rejected():
    draft = SettingsDraft(_initial())
    with pytest.raises(DraftValidationError):
        draft.edit("confirm_window_seconds", 5.0)
    with pytest.raises(DraftValidationError):
        draft.edit("keypoint_confidence_threshold", 1.5)
    with pytest.raises(DraftValidationError):
        draft.edit("unknown_field", 1.0)


def test_config_version_is_stable_and_sensitive_to_values():
    values = _initial()
    assert config_version(values) == config_version(dict(values))
    changed = dict(values)
    changed["cooldown_seconds"] = 12.0
    assert config_version(values) != config_version(changed)