from __future__ import annotations import argparse import json import tempfile import unittest from pathlib import Path from unittest import mock import numpy as np import evaluate_annotated_floor_events as evaluator from evaluate_annotated_floor_events import ( AnnotatedOpportunity, _validate_protocol_args, evaluate_family_horizon, load_annotated_opportunities, score_opportunity, write_event_csv, ) from evaluate_floor_state_changes import NUM_CLASSES, NUM_HORIZONS def reviewed_row( uid: str, *, family: str, verdict: str, anchor_frame: int, subtype: str | None = None, correction: float = 0.0, region_start: float | None = None, region_end: float | None = None, ) -> dict[str, object]: event_type = None if verdict == "POSITIVE": event_type = "WEARER_ENTRY" if family == "entry" else "INT_YIELD" anchor_seconds = anchor_frame / 20.0 return { "candidate_id": uid, "candidate_uid": uid, "session": "Session_4", "candidate_family": family, "review": { "human_status": "REVIEWED", "human_verdict": verdict, "human_event_type": event_type, "human_subtype": subtype, "boundary_correction_seconds": correction, }, "candidate": { "anchor_dense_frame": anchor_frame, "anchor_dense_seconds": anchor_seconds, "region_start_seconds": anchor_seconds if region_start is None else region_start, "region_end_seconds": anchor_seconds if region_end is None else region_end, "temporal_evidence": { "behavior_anchor_frame": anchor_frame, "behavior_anchor_seconds": anchor_seconds, }, }, } class AnnotatedFloorEventTests(unittest.TestCase): def write_reviews(self, rows: list[dict[str, object]]) -> tuple[tempfile.TemporaryDirectory[str], Path]: temporary = tempfile.TemporaryDirectory() path = Path(temporary.name) / "reviews.jsonl" path.write_text("".join(json.dumps(row) + "\n" for row in rows), encoding="utf-8") return temporary, path def test_positive_stays_one_bin_and_boundary_correction_shifts_it(self) -> None: temporary, path = self.write_reviews([ reviewed_row( "entry-positive", family="entry", verdict="POSITIVE", anchor_frame=445, # 22.25 s -> EasyCom 5-Hz row 111 correction=0.2, ) ]) self.addCleanup(temporary.cleanup) opportunities, diagnostics = load_annotated_opportunities([path], ["Session_4"]) self.assertEqual(diagnostics["num_scoring_eligible"], 1) self.assertEqual(opportunities[0].start_row, 112) self.assertEqual(opportunities[0].end_row_exclusive, 113) def test_hard_negative_is_one_region_sample_with_max_score(self) -> None: temporary, path = self.write_reviews([ reviewed_row( "entry-negative", family="entry", verdict="HARD_NEGATIVE", subtype="ENTRY_OTHER_TO_OTHER", anchor_frame=40, region_start=2.0, region_end=2.6, ) ]) self.addCleanup(temporary.cleanup) opportunities, _ = load_annotated_opportunities([path], ["Session_4"]) self.assertEqual((opportunities[0].start_row, opportunities[0].end_row_exclusive), (10, 13)) prediction = np.zeros((20, NUM_CLASSES, NUM_HORIZONS), dtype=np.float32) prediction[10:13, 1, 0] = [0.2, 0.9, 0.4] arrays = {"Session_4": (prediction, np.ones((20, NUM_HORIZONS), dtype=bool))} self.assertAlmostEqual(score_opportunity(opportunities[0], arrays, 0), 0.9) def test_yield_uses_one_minus_wearer_probability(self) -> None: opportunity = AnnotatedOpportunity( uid="yield", session="Session_4", family="int", positive=True, subtype="INT_YIELD", start_row=5, end_row_exclusive=6, anchor_seconds=1.0, region_start_seconds=1.0, region_end_seconds=1.0, ) prediction = np.zeros((10, NUM_CLASSES, NUM_HORIZONS), dtype=np.float32) prediction[5, 1, 2] = 0.15 arrays = {"Session_4": (prediction, np.ones((10, NUM_HORIZONS), dtype=bool))} self.assertAlmostEqual(score_opportunity(opportunity, arrays, 2), 0.85) def test_partial_valid_hard_negative_region_is_omitted(self) -> None: opportunity = AnnotatedOpportunity( "negative", "Session_4", "entry", False, "ENTRY_OTHER_TO_OTHER", 5, 8, 1.0, 1.0, 1.6, ) prediction = np.zeros((10, NUM_CLASSES, NUM_HORIZONS), dtype=np.float32) valid = np.ones((10, NUM_HORIZONS), dtype=bool) valid[6, 0] = False self.assertIsNone( score_opportunity(opportunity, {"Session_4": (prediction, valid)}, 0) ) def test_event_ap_and_false_positive_use_one_row_per_opportunity(self) -> None: opportunities = [ AnnotatedOpportunity("p", "Session_4", "entry", True, "WEARER_ENTRY", 4, 5, 0.8, 0.8, 0.8), AnnotatedOpportunity("n", "Session_4", "entry", False, "ENTRY_OTHER_TO_OTHER", 8, 10, 1.6, 1.6, 2.0), ] prediction = np.zeros((12, NUM_CLASSES, NUM_HORIZONS), dtype=np.float32) prediction[4, 1, 0] = 0.9 prediction[8:10, 1, 0] = [0.1, 0.2] arrays = {"Session_4": (prediction, np.ones((12, NUM_HORIZONS), dtype=bool))} metric, rows = evaluate_family_horizon(opportunities, arrays, "entry", 0, [0.8]) self.assertEqual(len(rows), 2) self.assertEqual(metric["num_positive"], 1) self.assertEqual(metric["num_hard_negative"], 1) self.assertAlmostEqual(metric["average_precision"], 1.0) self.assertAlmostEqual(metric["operating_points"][0]["negative_false_positive_rate"], 0.0) def test_official_test_requires_disjoint_calibration(self) -> None: args = argparse.Namespace( sessions=["Session_1"], calibration_reviews=None, calibration_sessions=None, calibration_run=None, allow_in_sample_thresholds=False, ) with self.assertRaisesRegex(ValueError, "Official Sessions"): _validate_protocol_args(args, [("model", Path("result"))]) def test_calibration_must_be_training_side_and_disjoint(self) -> None: args = argparse.Namespace( sessions=["Session_1"], calibration_reviews=[Path("calibration.jsonl")], calibration_sessions=["Session_1"], calibration_run=None, allow_in_sample_thresholds=False, ) with self.assertRaisesRegex(ValueError, "Sessions 4--12"): _validate_protocol_args(args, [("model", Path("result"))]) def test_event_csv_keeps_one_scored_opportunity_row(self) -> None: with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "events.csv" row = { "run": "model", "uid": "entry-positive", "session": "Session_4", "family": "entry", "positive": True, "subtype": "WEARER_ENTRY", "horizon_index": 1, "nominal_horizon_seconds": 0.2, "score": 0.9, "anchor_seconds": 10.0, "region_start_seconds": 10.0, "region_end_seconds": 10.0, "target_start_row": 50, "target_end_row_exclusive": 51, "source_start_row": 49, "source_end_row_exclusive": 50, } write_event_csv(path, [row]) text = path.read_text(encoding="utf-8") self.assertIn("entry-positive", text) self.assertEqual(len(text.splitlines()), 2) def test_main_writes_summary_metric_and_event_outputs(self) -> None: rows = [ reviewed_row("ep", family="entry", verdict="POSITIVE", anchor_frame=40), reviewed_row( "en", family="entry", verdict="HARD_NEGATIVE", subtype="ENTRY_OTHER_TO_OTHER", anchor_frame=80, region_start=4.0, region_end=4.4, ), reviewed_row("ip", family="int", verdict="POSITIVE", anchor_frame=120), reviewed_row( "in", family="int", verdict="HARD_NEGATIVE", subtype="INT_OTHER_BACKCHANNEL", anchor_frame=160, region_start=8.0, region_end=8.4, ), ] temporary, reviews = self.write_reviews(rows) self.addCleanup(temporary.cleanup) root = Path(temporary.name) run_dir = root / "run" (run_dir / "pred").mkdir(parents=True) (run_dir / "valid").mkdir() prediction = np.full((50, NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32) np.save(run_dir / "pred" / "Session_4.npy", prediction) np.save( run_dir / "valid" / "Session_4.npy", np.ones((50, NUM_HORIZONS), dtype=bool), ) output = root / "summary.json" args = argparse.Namespace( reviews=[reviews], run=[f"model={run_dir}"], sessions=["Session_4"], calibration_reviews=None, calibration_run=None, calibration_sessions=None, matched_recalls=[0.8], allow_in_sample_thresholds=False, output=output, csv_output=None, event_csv_output=None, print_event_examples=0, ) with mock.patch.object(evaluator, "parse_args", return_value=args): evaluator.main() self.assertTrue(output.is_file()) self.assertTrue((root / "summary_metrics.csv").is_file()) self.assertTrue((root / "summary_events.csv").is_file()) payload = json.loads(output.read_text(encoding="utf-8")) self.assertEqual(payload["protocol"], "human-adjudicated-fixed-horizon-floor-events-v1") self.assertEqual(len(payload["runs"][0]["horizons"]), 20) if __name__ == "__main__": unittest.main()