271 lines
10 KiB
Python
271 lines
10 KiB
Python
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()
|