wx44wx/icassp2027-experiment
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add annotated fixed-horizon floor evaluation

Browse Source
This commit is contained in:
Xin Wang
2026-09-15 02:27:48 +08:00
parent 3a127a3f9f
commit 2ab56f48d7
7 changed files with 1099 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

270
test_evaluate_annotated_floor_events.py Normal file
View File

@@ -0,0 +1,270 @@
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()