from __future__ import annotations import io import json import sys import tempfile import unittest from contextlib import redirect_stdout from pathlib import Path from unittest import mock import numpy as np from evaluate_floor_state_changes import ( NUM_CLASSES, NUM_HORIZONS, collect_floor_event_candidates, evaluate_event_horizon, evaluate_horizon, load_arrays, main, parse_runs, print_event_timeline_examples, validate_recalls, ) def activity(wearer: int, others: int) -> list[int]: return [int(not wearer and not others), wearer, others] class FloorStateMetricTests(unittest.TestCase): def test_post_boundary_candidate_uses_first_post_transition_row(self) -> None: target = np.asarray( [ activity(0, 1), activity(0, 0), # observed other offset activity(1, 0), # wearer response activity(1, 0), ], dtype=np.float32, ) session_ids = np.zeros(len(target), dtype=np.int32) pre, _ = collect_floor_event_candidates( target, session_ids, 1, event_type="response", decision_timing="pre-boundary", ) post, diagnostics = collect_floor_event_candidates( target, session_ids, 1, event_type="response", decision_timing="post-boundary", ) self.assertEqual([(item.source_row, item.label) for item in pre], [(0, True)]) self.assertEqual([(item.source_row, item.label) for item in post], [(1, True)]) self.assertEqual(diagnostics["num_simultaneous_boundary_outcomes_excluded"], 0) def test_post_boundary_excludes_outcome_that_already_happened_at_anchor(self) -> None: target = np.asarray( [ activity(0, 1), activity(1, 0), # simultaneous other offset and wearer onset activity(1, 0), ], dtype=np.float32, ) session_ids = np.zeros(len(target), dtype=np.int32) pre, _ = collect_floor_event_candidates( target, session_ids, 0, event_type="response", decision_timing="pre-boundary", ) post, diagnostics = collect_floor_event_candidates( target, session_ids, 0, event_type="response", decision_timing="post-boundary", ) self.assertEqual(len(pre), 1) self.assertTrue(pre[0].label) self.assertEqual(post, []) self.assertEqual(diagnostics["num_source_boundaries"], 1) self.assertEqual(diagnostics["num_simultaneous_boundary_outcomes_excluded"], 1) def test_post_boundary_reports_competitor_with_no_observable_bin(self) -> None: target = np.asarray( [ activity(0, 1), activity(0, 0), activity(0, 1), # immediate restart after reactive source activity(0, 1), ], dtype=np.float32, ) candidates, diagnostics = collect_floor_event_candidates( target, np.zeros(len(target), dtype=np.int32), 1, event_type="response", decision_timing="post-boundary", ) self.assertEqual(candidates, []) self.assertEqual(diagnostics["num_zero_window_competing_risk"], 1) def test_response_window_stops_before_next_other_onset(self) -> None: target = np.asarray( [ activity(0, 1), # session 0: other offset anchor activity(0, 0), activity(0, 1), # competing other onset; window ends before here activity(1, 1), # must not count as response to row-1 offset activity(0, 1), # session 1: other offset anchor activity(0, 0), activity(1, 0), # valid response before any competing onset activity(1, 1), # later competitor; response has already won ], dtype=np.float32, ) prediction = np.full( (len(target), NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32 ) prediction[1, 1, 0] = 0.1 prediction[2, 1, 1] = 0.2 prediction[3, 1, 2] = 0.99 # excluded by competing-risk censoring prediction[5, 1, 0] = 0.2 prediction[6, 1, 1] = 0.9 prediction[7, 1, 2] = 0.9 valid = np.ones((len(target), NUM_HORIZONS), dtype=bool) session_ids = np.asarray([0, 0, 0, 0, 1, 1, 1, 1], dtype=np.int32) result = evaluate_event_horizon( target, prediction, valid, session_ids, 2, [0.8], midturn_pause_max_steps=10, ) response = result["other_offset_to_wearer_response_event"] self.assertEqual(response["num_positive"], 1) self.assertEqual(response["num_negative"], 1) self.assertEqual(response["num_competing_risk_censored"], 1) self.assertEqual(response["num_short_resume_hard_negative"], 1) self.assertAlmostEqual(response["average_precision"], 1.0) self.assertAlmostEqual(response["min_effective_window_seconds"], 0.2) self.assertAlmostEqual(response["max_effective_window_seconds"], 0.4) def test_yield_window_stops_before_next_other_offset(self) -> None: target = np.asarray( [ activity(1, 0), # session 0: other onset anchor activity(1, 1), activity(1, 0), # competing other offset; window ends before here activity(0, 0), # must not count as yield to row-1 onset activity(1, 0), # session 1: other onset anchor activity(1, 1), activity(0, 1), # valid yield while other remains active activity(0, 0), # later competitor; yield has already won ], dtype=np.float32, ) prediction = np.full( (len(target), NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32 ) prediction[1, 1, 0] = 0.9 prediction[2, 1, 1] = 0.2 prediction[3, 1, 2] = 0.01 # excluded by competing-risk censoring prediction[5, 1, 0] = 0.8 prediction[6, 1, 1] = 0.1 prediction[7, 1, 2] = 0.1 valid = np.ones((len(target), NUM_HORIZONS), dtype=bool) session_ids = np.asarray([0, 0, 0, 0, 1, 1, 1, 1], dtype=np.int32) result = evaluate_event_horizon( target, prediction, valid, session_ids, 2, [0.8], midturn_pause_max_steps=10, ) yield_metric = result["other_onset_to_wearer_yield_event"] self.assertEqual(yield_metric["num_positive"], 1) self.assertEqual(yield_metric["num_negative"], 1) self.assertEqual(yield_metric["num_competing_risk_censored"], 1) self.assertAlmostEqual(yield_metric["average_precision"], 1.0) point = yield_metric["operating_points"][0] self.assertEqual(point["negative_false_positive_count"], 0) def test_observed_response_before_recording_end_is_not_right_censored(self) -> None: target = np.asarray( [ activity(0, 1), activity(0, 0), activity(1, 0), # observed response, but nominal 0.6-s end is absent ], dtype=np.float32, ) prediction = np.full( (len(target), NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32 ) valid = np.ones((len(target), NUM_HORIZONS), dtype=bool) session_ids = np.zeros(len(target), dtype=np.int32) result = evaluate_event_horizon( target, prediction, valid, session_ids, 2, [0.8], midturn_pause_max_steps=10, ) response = result["other_offset_to_wearer_response_event"] self.assertEqual(response["num_positive"], 1) self.assertEqual(response["num_right_censored"], 0) def test_unresolved_event_at_recording_end_is_right_censored(self) -> None: target = np.asarray( [ activity(0, 1), activity(0, 0), activity(0, 0), ], dtype=np.float32, ) prediction = np.full( (len(target), NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32 ) valid = np.ones((len(target), NUM_HORIZONS), dtype=bool) session_ids = np.zeros(len(target), dtype=np.int32) result = evaluate_event_horizon( target, prediction, valid, session_ids, 2, [0.8], midturn_pause_max_steps=10, ) response = result["other_offset_to_wearer_response_event"] self.assertEqual(response["num_evaluated"], 0) self.assertEqual(response["num_right_censored"], 1) def test_simultaneous_wearer_entry_at_competing_onset_is_excluded(self) -> None: target = np.asarray( [ activity(0, 1), activity(0, 0), activity(1, 1), # wearer and others restart simultaneously activity(1, 1), ], dtype=np.float32, ) prediction = np.full( (len(target), NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32 ) valid = np.ones((len(target), NUM_HORIZONS), dtype=bool) session_ids = np.zeros(len(target), dtype=np.int32) result = evaluate_event_horizon( target, prediction, valid, session_ids, 2, [0.8], midturn_pause_max_steps=10, ) response = result["other_offset_to_wearer_response_event"] self.assertEqual(response["num_positive"], 0) self.assertEqual(response["num_negative"], 1) self.assertEqual(response["num_competing_risk_censored"], 1) self.assertEqual(response["num_short_resume_hard_negative"], 0) def test_timeline_marks_competing_boundary_as_excluded(self) -> None: target = np.asarray( [ activity(0, 1), activity(0, 0), activity(0, 1), activity(1, 1), ], dtype=np.float32, ) prediction = np.full( (len(target), NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32 ) valid = np.ones((len(target), NUM_HORIZONS), dtype=bool) session_ids = np.zeros(len(target), dtype=np.int32) stream = io.StringIO() with redirect_stdout(stream): print_event_timeline_examples( target, prediction, valid, session_ids, ["Session_1"], examples_per_type=1, horizon_steps=3, ) output = stream.getvalue() self.assertIn("status=competing-censored negative", output) self.assertIn("C/excluded", output) def test_fixed_calibration_threshold_is_not_reselected(self) -> None: target = np.asarray( [ activity(0, 1), activity(0, 0), activity(1, 0), activity(1, 0), ], dtype=np.float32, ) prediction = np.full( (len(target), NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32 ) valid = np.ones((len(target), NUM_HORIZONS), dtype=bool) session_ids = np.zeros(len(target), dtype=np.int32) result = evaluate_event_horizon( target, prediction, valid, session_ids, 1, [0.8], midturn_pause_max_steps=10, operating_thresholds={ "response": {0.8: 0.75}, "yield": {0.8: 0.25}, }, threshold_source="calibration_sessions:Session_12", ) point = result["other_offset_to_wearer_response_event"][ "operating_points" ][0] self.assertEqual(point["threshold"], 0.75) self.assertEqual( point["threshold_source"], "calibration_sessions:Session_12" ) self.assertEqual(point["achieved_positive_recall"], 0.0) def test_event_score_uses_full_observable_window(self) -> None: target = np.asarray( [ activity(0, 1), activity(0, 0), activity(1, 0), # response occurs here activity(1, 0), ], dtype=np.float32, ) prediction = np.full( (len(target), NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32 ) prediction[1, 1, 0] = 0.1 prediction[2, 1, 1] = 0.2 prediction[3, 1, 2] = 0.9 # still part of the by-horizon event score valid = np.ones((len(target), NUM_HORIZONS), dtype=bool) session_ids = np.zeros(len(target), dtype=np.int32) result = evaluate_event_horizon( target, prediction, valid, session_ids, 2, [0.8], midturn_pause_max_steps=10, ) response = result["other_offset_to_wearer_response_event"] self.assertAlmostEqual(response["max_effective_window_seconds"], 0.6) self.assertAlmostEqual(response["operating_points"][0]["threshold"], 0.9) def test_main_calibrates_on_disjoint_training_session(self) -> None: target = np.asarray( [ activity(0, 1), activity(0, 0), # response anchor activity(1, 0), # response activity(1, 0), activity(1, 1), # yield anchor activity(0, 1), # yield activity(0, 1), activity(0, 0), # response anchor activity(0, 0), activity(0, 1), # competing onset, no response activity(1, 0), # response anchor and response activity(1, 0), activity(1, 1), # yield anchor activity(1, 1), activity(1, 0), # competing offset, no yield ] + [activity(0, 1)] * 15, dtype=np.float32, ) prediction = np.full( (len(target), NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32 ) valid = np.ones((len(target), NUM_HORIZONS), dtype=bool) with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) target_root = root / "target" run_root = root / "run" (run_root / "pred").mkdir(parents=True) (run_root / "valid").mkdir(parents=True) target_root.mkdir() for session in ("Session_1", "Session_12"): np.save(target_root / f"{session}.npy", target) np.save(run_root / "pred" / f"{session}.npy", prediction) np.save(run_root / "valid" / f"{session}.npy", valid) output = root / "report.json" argv = [ "evaluate_floor_state_changes.py", "--target-root", str(target_root), "--run", f"model={run_root}", "--sessions", "Session_1", "--calibration-sessions", "Session_12", "--matched-recalls", "0.8", "--output", str(output), ] with mock.patch.object(sys, "argv", argv), redirect_stdout(io.StringIO()): main() payload = json.loads(output.read_text()) self.assertEqual( payload["protocol"], "wearer_floor_event_competing_risk_v2" ) self.assertEqual(payload["calibration_sessions"], ["Session_12"]) point = payload["runs"]["model"]["event_horizons"][1][ "other_offset_to_wearer_response_event" ]["operating_points"][0] self.assertEqual( point["threshold_source"], "calibration_sessions:Session_12", ) framewise_point = payload["runs"]["model"]["horizons"][1][ "eot_response_vs_other_hold_proxy" ]["operating_points"][0] self.assertEqual( framewise_point["threshold_source"], "calibration_sessions:Session_12", ) self.assertTrue(output.with_name("report_events.csv").is_file()) def test_main_refuses_official_test_without_calibration(self) -> None: argv = [ "evaluate_floor_state_changes.py", "--target-root", "unused", "--run", "model=unused", "--sessions", "Session_1", ] with mock.patch.object(sys, "argv", argv): with self.assertRaisesRegex(ValueError, "requires --calibration-sessions"): main() def test_main_refuses_test_session_as_official_calibration(self) -> None: argv = [ "evaluate_floor_state_changes.py", "--target-root", "unused", "--run", "model=unused", "--sessions", "Session_1", "--calibration-sessions", "Session_2", ] with mock.patch.object(sys, "argv", argv): with self.assertRaisesRegex(ValueError, "Sessions 4--12"): main() def test_event_level_counts_each_boundary_once(self) -> None: target = np.asarray( [ activity(0, 1), # other offset follows activity(0, 0), activity(1, 0), # wearer responds within 0.4s activity(1, 0), # other entry follows activity(1, 1), activity(0, 1), # wearer yields within 0.4s activity(0, 1), activity(0, 1), activity(0, 0), # short pause, no wearer response activity(0, 1), activity(1, 0), # another other entry follows activity(1, 1), activity(1, 1), # wearer continues ], dtype=np.float32, ) prediction = np.full( (len(target), NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32 ) # Fixed source row 0: response score max(0.2, 0.9). prediction[1, 1, 0] = 0.2 prediction[2, 1, 1] = 0.9 # Fixed source row 7: no-response score max(0.1, 0.2). prediction[8, 1, 0] = 0.1 prediction[9, 1, 1] = 0.2 # Fixed source row 3: yield score max(0.2, 0.9). prediction[4, 1, 0] = 0.8 prediction[5, 1, 1] = 0.1 # Fixed source row 10: continue score max(0.1, 0.2). prediction[11, 1, 0] = 0.9 prediction[12, 1, 1] = 0.8 valid = np.ones((len(target), NUM_HORIZONS), dtype=bool) session_ids = np.zeros(len(target), dtype=np.int32) session_ids[10:] = 1 result = evaluate_event_horizon( target, prediction, valid, session_ids, 1, [0.8], midturn_pause_max_steps=10, ) response = result["other_offset_to_wearer_response_event"] self.assertEqual(response["num_positive"], 1) self.assertEqual(response["num_negative"], 1) self.assertEqual(response["num_short_resume_hard_negative"], 1) self.assertAlmostEqual(response["average_precision"], 1.0) self.assertAlmostEqual( response["operating_points"][0]["negative_false_positive_rate"], 0.0, ) yield_metric = result["other_onset_to_wearer_yield_event"] self.assertEqual(yield_metric["num_positive"], 1) self.assertEqual(yield_metric["num_negative"], 1) self.assertAlmostEqual(yield_metric["average_precision"], 1.0) def test_horizon_one_proxies_and_session_boundary(self) -> None: # t0->t1: EOT-response; t3->t4: HOLD; t4->t5: early entry; # t6->t7: other-entry yield; t8->t9: other-entry continue. target = np.asarray( [ activity(0, 1), activity(1, 0), activity(0, 0), activity(0, 1), activity(0, 1), activity(1, 1), activity(1, 0), activity(0, 1), activity(1, 0), activity(1, 1), ], dtype=np.float32, ) prediction = np.zeros((len(target), NUM_CLASSES, NUM_HORIZONS), dtype=np.float32) prediction[:, 1, :] = 0.5 prediction[1, 1, 0] = 0.9 prediction[4, 1, 0] = 0.1 prediction[5, 1, 0] = 0.8 prediction[7, 1, 0] = 0.1 prediction[9, 1, 0] = 0.9 valid = np.ones((len(target), NUM_HORIZONS), dtype=bool) session_ids = np.zeros(len(target), dtype=np.int32) result = evaluate_horizon( target, prediction, valid, session_ids, 0, [0.8] ) floor = result["eot_response_vs_other_hold_proxy"] self.assertEqual(floor["num_positive"], 2) self.assertEqual(floor["num_negative"], 1) self.assertEqual(floor["num_early_entry_proxy"], 1) self.assertAlmostEqual(floor["average_precision"], 1.0) self.assertAlmostEqual( floor["operating_points"][0]["negative_false_positive_rate"], 0.0 ) yield_metric = result["other_entry_yield_vs_continue_proxy"] self.assertEqual(yield_metric["num_positive"], 1) self.assertEqual(yield_metric["num_negative"], 1) self.assertAlmostEqual(yield_metric["average_precision"], 1.0) # The apparent t4->t5 early entry is removed if t5 begins a new session. session_ids[5:] = 1 bounded = evaluate_horizon( target, prediction, valid, session_ids, 0, [0.8] ) self.assertEqual( bounded["eot_response_vs_other_hold_proxy"]["num_early_entry_proxy"], 0, ) def test_load_arrays_validates_layout(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) target_root = root / "target" run_root = root / "run" (run_root / "pred").mkdir(parents=True) (run_root / "valid").mkdir(parents=True) target_root.mkdir() target = np.asarray([activity(0, 0), activity(1, 0)], dtype=np.float32) np.save(target_root / "Session_1.npy", target) np.save( run_root / "pred/Session_1.npy", np.zeros((2, NUM_CLASSES, NUM_HORIZONS), dtype=np.float32), ) np.save( run_root / "valid/Session_1.npy", np.ones((2, NUM_HORIZONS), dtype=bool), ) loaded = load_arrays(target_root, run_root, ["Session_1"]) self.assertEqual(loaded[0].shape, (2, NUM_CLASSES)) self.assertEqual(loaded[1].shape, (2, NUM_CLASSES, NUM_HORIZONS)) def test_argument_validation(self) -> None: self.assertEqual(parse_runs(["G=results/g"]), [("G", Path("results/g"))]) with self.assertRaises(ValueError): parse_runs(["results/g"]) with self.assertRaises(ValueError): parse_runs(["G=a", "G=b"]) self.assertEqual(validate_recalls([0.9, 0.7]), [0.7, 0.9]) with self.assertRaises(ValueError): validate_recalls([0.0]) if __name__ == "__main__": unittest.main()