diff --git a/PROJECT_CONTEXT.md b/PROJECT_CONTEXT.md index acac4e1..402b062 100644 --- a/PROJECT_CONTEXT.md +++ b/PROJECT_CONTEXT.md @@ -51,6 +51,11 @@ result. pools the 0.6--2.0-s wearer trajectory over equal-width 0.6-s causal decision regions. It preserves one score per annotated opportunity and never treats the longer annotation evidence interval as a scoring window. +- An opportunity-conditioned dense-output evaluator aligns ENTRY positives and + core negatives at non-wearer releases, aligns INT positives and hold-floor + negatives at competing non-wearer onsets, and applies strict competing-risk + censoring. It uses only the original wearer probabilities, excludes wearer + backchannels from headline ENTRY FPR, and records unanchored/ambiguous cases. ### Measured screening result diff --git a/docs/README.md b/docs/README.md index 3e7d1e7..cf3535c 100644 --- a/docs/README.md +++ b/docs/README.md @@ -34,6 +34,7 @@ tests. | [`easycom_competing_risk_floor_evaluation.md`](easycom_competing_risk_floor_evaluation.md) | Running and auditing the competing-risk entry/yield evaluator with training-side threshold calibration | | [`easycom_floor_event_diagnostics.md`](easycom_floor_event_diagnostics.md) | Separating boundary observability, event-objective mismatch and addressee information with remote commands | | [`easycom_event_review_tool.md`](easycom_event_review_tool.md) | Reviewing automatically mined EasyCom floor-event candidates against local video and transcripts | +| [`easycom_opportunity_event_evaluation.md`](easycom_opportunity_event_evaluation.md) | Current opportunity-conditioned ENTRY/YIELD evaluation from EgoSpeak-compatible wearer forecasts | | [`easycom_egospeak_event_proxy.md`](easycom_egospeak_event_proxy.md) | Scoring the EgoSpeak wearer trajectory over equal-width causal decision regions as an activity-derived ENTRY/INT proxy | | [`easycom_annotated_floor_evaluation.md`](easycom_annotated_floor_evaluation.md) | Reproducing the legacy asymmetric point-positive/region-negative fixed-horizon diagnostic | diff --git a/docs/easycom_egospeak_event_proxy.md b/docs/easycom_egospeak_event_proxy.md index 666a4d6..cf7053e 100644 --- a/docs/easycom_egospeak_event_proxy.md +++ b/docs/easycom_egospeak_event_proxy.md @@ -1,5 +1,10 @@ # EgoSpeak dense-trajectory event proxy +> **Status:** boundary-centered diagnostic retained for reproducibility. The +> manuscript's current main behavioral protocol is the matched-opportunity +> evaluator in +> [`easycom_opportunity_event_evaluation.md`](easycom_opportunity_event_evaluation.md). + `evaluate_egospeak_event_proxy.py` provides a deliberately limited bridge between EgoSpeak-compatible dense future-activity predictions and reviewed wearer ENTRY/INT opportunities. It does **not** claim that wearer voice diff --git a/docs/easycom_opportunity_event_evaluation.md b/docs/easycom_opportunity_event_evaluation.md new file mode 100644 index 0000000..d5b8d3d --- /dev/null +++ b/docs/easycom_opportunity_event_evaluation.md @@ -0,0 +1,59 @@ +# EasyCom opportunity-conditioned ENTRY/YIELD evaluation + +[`evaluate_egospeak_opportunity_events.py`](../evaluate_egospeak_opportunity_events.py) +uses LLM-assisted or human-reviewed semantic labels to define matched +conversational opportunities while scoring every model only from its original +EgoSpeak-compatible wearer probabilities. It does not train a semantic event +head. + +## ENTRY + +A positive must have a reviewer-selected preceding non-wearer release. Its +opportunity starts at that release and resolves at the reviewed wearer entry. +Core hard negatives start at the same type of non-wearer release and resolve +when the same speaker resumes (`ENTRY_OTHER_MIDTURN_PAUSE`) or a different +non-wearer takes the next floor (`ENTRY_OTHER_TO_OTHER`). The score is the +maximum projected wearer probability up to the competing resolution, capped at +2.0 seconds. + +`ENTRY_WEARER_BACKCHANNEL` is excluded from headline ENTRY FPR. Predicting +wearer activity for a real backchannel is correct under EgoSpeak's dense target; +ENTRY-versus-backchannel remains a separate semantic-selectivity diagnostic. + +## YIELD + +An opportunity starts at a non-wearer onset during wearer speech. The score is +the maximum `1 - p_wearer` over 0.2--0.6 seconds, censored strictly at that +non-wearer's offset. Reviewed `INT_YIELD` items are positive; other +backchannels, failed takeovers, and clear wearer-hold cases are hard negatives. + +Current INT reviews do not select one competing onset. The loader therefore +uses the latest supplied other interval that crosses the wearer offset within +the annotation tolerance. It records how many intervals were eligible so +multi-competitor cases can be audited before the protocol is frozen. + +## Provisional EgoSpeak run + +Threshold-free AP on the official test sessions can be run without fitting a +test threshold: + +```bash +python3 evaluate_egospeak_opportunity_events.py \ + --reviews reports/easycom_entry_int_sessions1_4_v16/easycom_eot_int_provisional_llm_reviews.jsonl \ + --run egospeak=results/egospeak_miniroada_floor_eval \ + --sessions Session_1 Session_2 Session_3 \ + --allow-provisional-llm-reviews \ + --print-event-examples 10 \ + --output results/easycom_egospeak_opportunity_events/sessions1_3_provisional_summary.json +``` + +The output JSON records label coverage and every exclusion reason. The metrics +CSV contains threshold-free ENTRY/YIELD AP. The event CSV contains anchors, +causal source cutoffs, censor times, horizons, and wearer probabilities for +timeline auditing. + +Do not request matched-recall operating points on Sessions 1--3 without +disjoint Sessions 4--12 calibration reviews and out-of-fold predictions. +Before manuscript reporting, freeze the 2.0-second ENTRY cap, 0.6-second YIELD +cap, positive anchor-selection policy, and multiple-competitor handling using +training-side data only. diff --git a/evaluate_egospeak_opportunity_events.py b/evaluate_egospeak_opportunity_events.py new file mode 100755 index 0000000..788b9c7 --- /dev/null +++ b/evaluate_egospeak_opportunity_events.py @@ -0,0 +1,785 @@ +#!/usr/bin/env python3 +"""Evaluate opportunity-conditioned ENTRY/YIELD from dense wearer forecasts. + +Semantic review labels define which conversational opportunities end in a +wearer floor action. The evaluated score remains the model's original dense +wearer-activity probability; this script does not train or assume a semantic +event head. + +ENTRY opportunities are anchored at a reviewed preceding non-wearer release. +Positive opportunities resolve at the wearer entry; core hard negatives +resolve when the same or a different non-wearer speaks next. The score is the +maximum wearer probability over available 0.2--2.0 s forecasts up to that +competing resolution. + +YIELD opportunities are anchored at a non-wearer onset during wearer speech. +The score is the maximum future wearer-inactivity probability over 0.2--0.6 s, +censored at that non-wearer's offset. Positive and hard-negative examples +therefore begin at the same kind of observed conversational opportunity. + +Prediction arrays use the repository contract +``[target_time, class, future_horizon]``. All scores are read diagonally from +one causal source row. No input after the labeled outcome is used. +""" + +from __future__ import annotations + +import argparse +from collections import Counter +import csv +from dataclasses import dataclass +import json +import math +from pathlib import Path +from typing import Any, Iterable, Literal + +import numpy as np + +from evaluate_annotated_floor_events import ( + PROVISIONAL_LLM_SOURCE, + _read_jsonl, + _review_payload, + load_session_predictions, +) +from evaluate_floor_state_changes import ( + CLASS_NAMES, + NUM_HORIZONS, + OFFICIAL_TEST_SESSIONS, + OFFICIAL_TRAIN_SESSIONS, + STEP_SECONDS, + WEARER_INDEX, + _operating_points, + _safe_ap, + parse_runs, + validate_recalls, +) + + +Family = Literal["entry", "int"] +FAMILIES: tuple[Family, ...] = ("entry", "int") +ENTRY_CORE_HARD_NEGATIVES = frozenset( + {"ENTRY_OTHER_MIDTURN_PAUSE", "ENTRY_OTHER_TO_OTHER"} +) +ENTRY_DIAGNOSTIC_HARD_NEGATIVES = frozenset({"ENTRY_WEARER_BACKCHANNEL"}) +INT_HARD_NEGATIVES = frozenset( + { + "INT_OTHER_BACKCHANNEL", + "INT_NON_TAKEOVER_OVERLAP", + "INT_WEARER_HOLDS_FLOOR", + } +) +ENTRY_MINED_SOURCES = { + "ENTRY_OTHER_MIDTURN_PAUSE": "SAME_SPEAKER_PAUSE", + "ENTRY_OTHER_TO_OTHER": "OTHER_TO_OTHER", +} +INT_MINED_SOURCES = frozenset({"OTHER_BACKCHANNEL", "NON_TAKEOVER_OVERLAP"}) + + +@dataclass(frozen=True) +class OpportunityEvent: + uid: str + session: str + family: Family + positive: bool + subtype: str + candidate_source: str + opportunity_anchor_seconds: float + resolution_seconds: float + censor_seconds: float + behavior_anchor_seconds: float | None + anchor_basis: str + competing_intervals: int = 0 + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--reviews", type=Path, action="append", required=True) + parser.add_argument( + "--run", + action="append", + required=True, + metavar="NAME=RESULT_DIR", + help="Inference directory containing pred/ and valid/. Repeat per method.", + ) + parser.add_argument("--sessions", nargs="+", default=list(OFFICIAL_TEST_SESSIONS)) + parser.add_argument("--entry-max-seconds", type=float, default=2.0) + parser.add_argument("--yield-max-seconds", type=float, default=0.6) + parser.add_argument("--calibration-reviews", type=Path, action="append") + parser.add_argument("--calibration-run", action="append", metavar="NAME=RESULT_DIR") + parser.add_argument("--calibration-sessions", nargs="+") + parser.add_argument("--matched-recalls", nargs="*", type=float, default=()) + parser.add_argument("--allow-in-sample-thresholds", action="store_true") + parser.add_argument( + "--output", + type=Path, + default=Path("results/easycom_egospeak_opportunity_events/summary.json"), + ) + parser.add_argument("--csv-output", type=Path) + parser.add_argument("--event-csv-output", type=Path) + parser.add_argument("--print-event-examples", type=int, default=0, metavar="N") + parser.add_argument("--allow-provisional-llm-reviews", action="store_true") + return parser.parse_args() + + +def _finite_seconds(value: Any, *, field: str, uid: str) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError(f"{uid}: {field} must be numeric") + result = float(value) + if not math.isfinite(result) or result < 0: + raise ValueError(f"{uid}: {field} must be finite and non-negative") + return result + + +def _optional_seconds(value: Any, *, field: str, uid: str) -> float | None: + if value is None: + return None + return _finite_seconds(value, field=field, uid=uid) + + +def _behavior_anchor( + candidate: dict[str, Any], + review: dict[str, Any], + *, + uid: str, +) -> float: + evidence = candidate.get("temporal_evidence") + evidence = evidence if isinstance(evidence, dict) else {} + raw = evidence.get( + "behavior_anchor_seconds", + evidence.get( + "wearer_offset_seconds", + candidate.get("anchor_dense_seconds", candidate.get("event_time")), + ), + ) + result = _finite_seconds(raw, field="behavior_anchor_seconds", uid=uid) + correction = float(review.get("boundary_correction_seconds") or 0.0) + corrected = result + correction + if corrected < 0: + raise ValueError(f"{uid}: corrected behavior anchor precedes session") + return corrected + + +def _candidate_region( + candidate: dict[str, Any], *, uid: str +) -> tuple[float, float]: + anchor = candidate.get("anchor_dense_seconds", candidate.get("event_time")) + start = _finite_seconds( + candidate.get("region_start_seconds", anchor), + field="region_start_seconds", + uid=uid, + ) + end = _finite_seconds( + candidate.get("region_end_seconds", anchor), + field="region_end_seconds", + uid=uid, + ) + if end <= start: + raise ValueError(f"{uid}: opportunity region must have positive duration") + return start, end + + +def _int_competing_interval( + candidate: dict[str, Any], + *, + behavior_anchor_seconds: float, + uid: str, +) -> tuple[float, float, int] | None: + evidence = candidate.get("temporal_evidence") + evidence = evidence if isinstance(evidence, dict) else {} + raw_intervals = evidence.get("nearby_other_intervals") + if not isinstance(raw_intervals, list): + return None + eligible: list[tuple[float, float]] = [] + for index, interval in enumerate(raw_intervals): + if not isinstance(interval, dict): + continue + if interval.get("crosses_offset_with_tolerance") is not True: + continue + onset = _optional_seconds( + interval.get("onset_seconds"), + field=f"nearby_other_intervals[{index}].onset_seconds", + uid=uid, + ) + offset = _optional_seconds( + interval.get("offset_seconds"), + field=f"nearby_other_intervals[{index}].offset_seconds", + uid=uid, + ) + if onset is None or offset is None: + continue + if onset < behavior_anchor_seconds and offset > onset: + eligible.append((onset, offset)) + if not eligible: + return None + # A semantic INT review does not currently select one competing onset. + # The latest interval crossing the wearer offset is the deterministic, + # most proximal causal candidate. Multiplicity is surfaced for audit. + onset, offset = max(eligible, key=lambda item: (item[0], item[1])) + return onset, offset, len(eligible) + + +def load_opportunity_events( + paths: Iterable[Path], + sessions: Iterable[str], + *, + allow_provisional_llm: bool = False, +) -> tuple[list[OpportunityEvent], dict[str, Any]]: + selected_sessions = set(sessions) + events: list[OpportunityEvent] = [] + seen: set[str] = set() + skipped: Counter[str] = Counter() + provisional = 0 + + for row_number, row in enumerate(_read_jsonl(paths), 1): + uid = str(row.get("candidate_id") or row.get("candidate_uid") or f"row-{row_number}") + if uid in seen: + raise ValueError(f"Duplicate reviewed candidate: {uid}") + seen.add(uid) + session = str(row.get("session", "")) + if session not in selected_sessions: + continue + review, candidate = _review_payload( + row, uid, allow_provisional_llm=allow_provisional_llm + ) + review_source = row.get("review_source", review.get("review_source")) + if review_source == PROVISIONAL_LLM_SOURCE: + provisional += 1 + verdict = review.get("human_verdict") + if verdict == "EXCLUDE": + skipped["review_exclude"] += 1 + continue + if verdict not in {"POSITIVE", "HARD_NEGATIVE"}: + raise ValueError(f"{uid}: invalid review verdict {verdict!r}") + family_value = row.get("candidate_family") + if family_value not in FAMILIES: + raise ValueError(f"{uid}: invalid candidate_family {family_value!r}") + family: Family = family_value + positive = verdict == "POSITIVE" + expected_event_type = "WEARER_ENTRY" if family == "entry" else "INT_YIELD" + if positive and review.get("human_event_type") != expected_event_type: + raise ValueError(f"{uid}: positive {family} must be {expected_event_type}") + subtype_value = ( + "WEARER_ENTRY" if family == "entry" and positive + else "INT_YIELD" if family == "int" and positive + else review.get("human_subtype") + ) + if not isinstance(subtype_value, str): + raise ValueError(f"{uid}: missing subtype") + source = str(candidate.get("candidate_source", "")) + behavior = _behavior_anchor(candidate, review, uid=uid) + + if family == "entry": + if positive: + selected_release = _optional_seconds( + review.get("selected_opportunity_seconds"), + field="selected_opportunity_seconds", + uid=uid, + ) + if selected_release is None: + skipped["entry_positive_without_selected_release"] += 1 + continue + anchor = selected_release + resolution = behavior + censor = behavior + basis = "reviewed_preceding_other_release" + elif subtype_value in ENTRY_CORE_HARD_NEGATIVES: + if source != ENTRY_MINED_SOURCES[subtype_value]: + skipped["entry_hn_without_matching_mined_release"] += 1 + continue + anchor, resolution = _candidate_region(candidate, uid=uid) + censor = resolution + basis = "mined_nonwearer_release" + elif subtype_value in ENTRY_DIAGNOSTIC_HARD_NEGATIVES: + skipped["entry_backchannel_separate_diagnostic"] += 1 + continue + else: + skipped["entry_unsupported_hard_negative"] += 1 + continue + competing_intervals = 0 + else: + if positive: + interval = _int_competing_interval( + candidate, + behavior_anchor_seconds=behavior, + uid=uid, + ) + if interval is None: + skipped["int_positive_without_crossing_other"] += 1 + continue + anchor, censor, competing_intervals = interval + resolution = behavior + basis = "latest_other_onset_crossing_wearer_offset" + elif subtype_value in INT_HARD_NEGATIVES: + if source in INT_MINED_SOURCES: + anchor, censor = _candidate_region(candidate, uid=uid) + resolution = censor + competing_intervals = 1 + basis = "mined_competing_other_onset" + else: + interval = _int_competing_interval( + candidate, + behavior_anchor_seconds=behavior, + uid=uid, + ) + if interval is None: + skipped["int_hn_without_competing_other"] += 1 + continue + anchor, censor, competing_intervals = interval + resolution = behavior + basis = "latest_other_onset_crossing_non_yield_offset" + else: + skipped["int_unsupported_hard_negative"] += 1 + continue + + if resolution <= anchor: + skipped[f"{family}_nonpositive_opportunity_duration"] += 1 + continue + if censor <= anchor: + skipped[f"{family}_nonpositive_censor_duration"] += 1 + continue + events.append( + OpportunityEvent( + uid=uid, + session=session, + family=family, + positive=positive, + subtype=subtype_value, + candidate_source=source, + opportunity_anchor_seconds=anchor, + resolution_seconds=resolution, + censor_seconds=censor, + behavior_anchor_seconds=behavior if positive else None, + anchor_basis=basis, + competing_intervals=competing_intervals, + ) + ) + + diagnostics: dict[str, Any] = { + "num_review_rows": len(seen), + "num_opportunity_events": len(events), + "num_provisional_llm": provisional, + "skipped": dict(sorted(skipped.items())), + } + return events, diagnostics + + +def _first_causal_source_row(anchor_seconds: float) -> tuple[int, float]: + """First 5-Hz state whose right-edge cutoff observes the opportunity.""" + + cutoff_step = int(math.ceil(anchor_seconds / STEP_SECONDS - 1e-7)) + source_row = cutoff_step - 1 + return source_row, cutoff_step * STEP_SECONDS + + +def _score_event( + event: OpportunityEvent, + arrays: dict[str, tuple[np.ndarray, np.ndarray]], + *, + entry_max_seconds: float, + yield_max_seconds: float, +) -> tuple[float, dict[str, Any]] | None: + prediction, valid = arrays[event.session] + source_row, source_cutoff = _first_causal_source_row( + event.opportunity_anchor_seconds + ) + if source_row < 0 or source_row >= len(prediction): + return None + # A post-opportunity state that already contains the positive behavior is + # outcome observation, not prediction. Drop these quantized fast events. + if event.positive and event.behavior_anchor_seconds is not None: + if source_cutoff >= event.behavior_anchor_seconds - 1e-7: + return None + + max_seconds = entry_max_seconds if event.family == "entry" else yield_max_seconds + # A positive ENTRY may use the 5-Hz target bin containing the wearer onset. + # Hard-negative ENTRY and both YIELD classes stop strictly at their + # competing non-wearer boundary, as required by competing-risk censoring. + censor_limit = event.censor_seconds + if event.family == "entry" and event.positive: + # Quantize only to the right edge of the 5-Hz bin containing the + # onset. An onset already on-grid must not admit one post-onset bin. + censor_steps = int(math.ceil(censor_limit / STEP_SECONDS - 1e-7)) + censor_limit = censor_steps * STEP_SECONDS + horizon_limit = min(source_cutoff + max_seconds, censor_limit) + probabilities: list[float] = [] + used_horizons: list[float] = [] + target_rows: list[int] = [] + for horizon_index in range(NUM_HORIZONS): + horizon_seconds = (horizon_index + 1) * STEP_SECONDS + target_seconds = source_cutoff + horizon_seconds + if target_seconds > horizon_limit + 1e-7: + continue + target_row = source_row + horizon_index + 1 + if target_row >= len(prediction) or not valid[target_row, horizon_index]: + return None + probabilities.append( + float(prediction[target_row, WEARER_INDEX, horizon_index]) + ) + used_horizons.append(horizon_seconds) + target_rows.append(target_row) + if not probabilities: + return None + wearer = np.asarray(probabilities, dtype=np.float64) + action_scores = wearer if event.family == "entry" else 1.0 - wearer + best_index = int(np.argmax(action_scores)) + return float(action_scores[best_index]), { + "source_row": source_row, + "source_cutoff_seconds": source_cutoff, + "used_horizons_seconds": used_horizons, + "target_rows": target_rows, + "wearer_probabilities": probabilities, + "best_horizon_seconds": used_horizons[best_index], + "best_wearer_probability": probabilities[best_index], + } + + +def evaluate_family( + events: list[OpportunityEvent], + arrays: dict[str, tuple[np.ndarray, np.ndarray]], + family: Family, + recalls: list[float], + *, + entry_max_seconds: float, + yield_max_seconds: float, + fixed_thresholds: dict[float, float | None] | None = None, + threshold_source: str = "not_requested", +) -> tuple[dict[str, Any], list[dict[str, Any]]]: + rows: list[dict[str, Any]] = [] + invalid = 0 + for event in events: + if event.family != family: + continue + scored = _score_event( + event, + arrays, + entry_max_seconds=entry_max_seconds, + yield_max_seconds=yield_max_seconds, + ) + if scored is None: + invalid += 1 + continue + score, details = scored + rows.append( + { + "uid": event.uid, + "session": event.session, + "family": event.family, + "positive": event.positive, + "subtype": event.subtype, + "candidate_source": event.candidate_source, + "score": score, + "opportunity_anchor_seconds": event.opportunity_anchor_seconds, + "resolution_seconds": event.resolution_seconds, + "censor_seconds": event.censor_seconds, + "behavior_anchor_seconds": event.behavior_anchor_seconds, + "anchor_basis": event.anchor_basis, + "competing_intervals": event.competing_intervals, + **details, + } + ) + + scores = np.asarray([row["score"] for row in rows], dtype=np.float64) + positive = np.asarray([row["positive"] for row in rows], dtype=bool) + negative = ~positive + labels = positive.astype(np.int8) + subtypes = sorted({str(row["subtype"]) for row in rows if not row["positive"]}) + extra_groups = { + f"fpr_{subtype.lower()}": np.asarray( + [not row["positive"] and row["subtype"] == subtype for row in rows], + dtype=bool, + ) + for subtype in subtypes + } + operating_points: list[dict[str, Any]] = [] + if recalls: + operating_points = _operating_points( + scores, + positive, + negative, + recalls, + extra_groups=extra_groups, + fixed_thresholds=fixed_thresholds, + threshold_source=threshold_source, + ) + metric = { + "family": family, + "num_positive": int(positive.sum()), + "num_hard_negative": int(negative.sum()), + "num_invalid_or_post_outcome": invalid, + "average_precision": _safe_ap(labels, scores), + "operating_points": operating_points, + } + return metric, rows + + +def _thresholds(metric: dict[str, Any]) -> dict[float, float | None]: + return { + float(item["requested_positive_recall"]): item["threshold"] + for item in metric["operating_points"] + } + + +def _positive_grid_seconds(value: float, *, field: str) -> float: + if not math.isfinite(value) or value <= 0: + raise ValueError(f"{field} must be finite and positive") + steps = round(value / STEP_SECONDS) + if not math.isclose(value, steps * STEP_SECONDS, abs_tol=1e-7): + raise ValueError(f"{field} must lie on the {STEP_SECONDS:g}-s grid") + if value > NUM_HORIZONS * STEP_SECONDS: + raise ValueError(f"{field} cannot exceed {NUM_HORIZONS * STEP_SECONDS:g} s") + return float(value) + + +def _validate_calibration( + args: argparse.Namespace, + runs: list[tuple[str, Path]], + recalls: list[float], +) -> tuple[list[tuple[str, Path]], bool]: + paired = bool(args.calibration_reviews) == bool(args.calibration_sessions) + if not paired: + raise ValueError( + "--calibration-reviews and --calibration-sessions must be supplied together" + ) + has_calibration = bool(args.calibration_reviews and args.calibration_sessions) + if args.calibration_run and not has_calibration: + raise ValueError("--calibration-run requires calibration reviews and sessions") + if has_calibration: + unknown = sorted(set(args.calibration_sessions) - set(OFFICIAL_TRAIN_SESSIONS)) + if unknown: + raise ValueError(f"Calibration must use Sessions 4--12, got {unknown}") + overlap = sorted(set(args.sessions) & set(args.calibration_sessions)) + if overlap: + raise ValueError(f"Evaluation and calibration sessions overlap: {overlap}") + official_test = bool(set(args.sessions) & set(OFFICIAL_TEST_SESSIONS)) + if recalls and official_test and not has_calibration and not args.allow_in_sample_thresholds: + raise ValueError( + "Official Sessions 1--3 require training-side calibration for " + "--matched-recalls; omit it for threshold-free AP" + ) + calibration_runs = parse_runs(args.calibration_run) if args.calibration_run else runs + if [name for name, _ in calibration_runs] != [name for name, _ in runs]: + raise ValueError("--calibration-run names/order must match --run") + return calibration_runs, has_calibration + + +def _write_metrics(path: Path, results: list[dict[str, Any]]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + fields = [ + "run", "family", "num_positive", "num_hard_negative", + "num_invalid_or_post_outcome", "average_precision", "requested_recall", + "threshold", "threshold_source", "achieved_recall", + "false_positive_rate", "false_positive_count", + ] + with path.open("w", encoding="utf-8", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=fields) + writer.writeheader() + for run in results: + for metric in run["metrics"]: + points = metric["operating_points"] or [None] + for point in points: + writer.writerow( + { + "run": run["name"], + "family": metric["family"], + "num_positive": metric["num_positive"], + "num_hard_negative": metric["num_hard_negative"], + "num_invalid_or_post_outcome": metric[ + "num_invalid_or_post_outcome" + ], + "average_precision": metric["average_precision"], + "requested_recall": "" if point is None else point[ + "requested_positive_recall" + ], + "threshold": "" if point is None else point["threshold"], + "threshold_source": "" if point is None else point[ + "threshold_source" + ], + "achieved_recall": "" if point is None else point[ + "achieved_positive_recall" + ], + "false_positive_rate": "" if point is None else point[ + "negative_false_positive_rate" + ], + "false_positive_count": "" if point is None else point[ + "negative_false_positive_count" + ], + } + ) + + +def _write_events(path: Path, rows: list[dict[str, Any]]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + fields = [ + "run", "uid", "session", "family", "positive", "subtype", + "candidate_source", "score", "opportunity_anchor_seconds", + "source_row", "source_cutoff_seconds", "resolution_seconds", + "censor_seconds", "behavior_anchor_seconds", "anchor_basis", + "competing_intervals", "best_horizon_seconds", + "best_wearer_probability", "used_horizons_seconds", "target_rows", + "wearer_probabilities", + ] + with path.open("w", encoding="utf-8", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=fields) + writer.writeheader() + for row in rows: + serialized = dict(row) + for field in ("used_horizons_seconds", "target_rows", "wearer_probabilities"): + serialized[field] = json.dumps(row[field]) + writer.writerow(serialized) + + +def _print_examples(rows: list[dict[str, Any]], limit: int) -> None: + if limit <= 0: + return + print("\nOpportunity-conditioned examples (first run)") + for family in FAMILIES: + print(f"{family}:") + shown = 0 + for row in rows: + if row["family"] != family: + continue + label = "POS" if row["positive"] else "HN" + print( + f" {label} {row['session']} {row['uid']} " + f"anchor={row['opportunity_anchor_seconds']:.2f}s " + f"source_cutoff={row['source_cutoff_seconds']:.2f}s " + f"resolution={row['resolution_seconds']:.2f}s " + f"censor={row['censor_seconds']:.2f}s " + f"horizons={row['used_horizons_seconds']} " + f"score={row['score']:.4f} subtype={row['subtype']}" + ) + shown += 1 + if shown >= limit: + break + + +def main() -> None: + args = parse_args() + entry_max = _positive_grid_seconds(args.entry_max_seconds, field="entry max") + yield_max = _positive_grid_seconds(args.yield_max_seconds, field="yield max") + recalls = validate_recalls(args.matched_recalls) if args.matched_recalls else [] + runs = parse_runs(args.run) + calibration_runs, has_calibration = _validate_calibration(args, runs, recalls) + allow_provisional = bool(args.allow_provisional_llm_reviews) + + events, review_diagnostics = load_opportunity_events( + args.reviews, + args.sessions, + allow_provisional_llm=allow_provisional, + ) + if not events: + raise ValueError("No opportunity-conditioned events for evaluated sessions") + calibration_events = None + calibration_diagnostics = None + if has_calibration: + calibration_events, calibration_diagnostics = load_opportunity_events( + args.calibration_reviews, + args.calibration_sessions, + allow_provisional_llm=allow_provisional, + ) + if not calibration_events: + raise ValueError("No opportunity-conditioned calibration events") + + results: list[dict[str, Any]] = [] + all_rows: list[dict[str, Any]] = [] + first_rows: list[dict[str, Any]] = [] + for run_index, ((name, run_dir), (_, calibration_dir)) in enumerate( + zip(runs, calibration_runs) + ): + arrays = load_session_predictions(run_dir, args.sessions) + calibration_arrays = ( + load_session_predictions(calibration_dir, args.calibration_sessions) + if calibration_events is not None + else None + ) + metrics: list[dict[str, Any]] = [] + run_rows: list[dict[str, Any]] = [] + for family in FAMILIES: + fixed_thresholds = None + threshold_source = "evaluated_events_exploratory" + if calibration_events is not None and calibration_arrays is not None: + calibration_metric, _ = evaluate_family( + calibration_events, + calibration_arrays, + family, + recalls, + entry_max_seconds=entry_max, + yield_max_seconds=yield_max, + threshold_source="training_side_calibration", + ) + fixed_thresholds = _thresholds(calibration_metric) + threshold_source = "training_side_calibration" + metric, rows = evaluate_family( + events, + arrays, + family, + recalls, + entry_max_seconds=entry_max, + yield_max_seconds=yield_max, + fixed_thresholds=fixed_thresholds, + threshold_source=threshold_source, + ) + metrics.append(metric) + run_rows.extend({"run": name, **row} for row in rows) + if run_index == 0: + first_rows.extend(rows) + results.append({"name": name, "result_dir": str(run_dir), "metrics": metrics}) + all_rows.extend(run_rows) + + provisional_count = review_diagnostics["num_provisional_llm"] + ( + 0 if calibration_diagnostics is None + else calibration_diagnostics["num_provisional_llm"] + ) + payload = { + "schema_version": 1, + "protocol": "egospeak-opportunity-conditioned-floor-actions-v1", + "semantic_scope": ( + "LLM-assisted opportunity labels with dense wearer-activity scores; " + "this is provisional and no human review is implied." + if provisional_count + else "Human-reviewed opportunity labels with dense wearer-activity scores; " + "the model output is not a semantic intent probability." + ), + "provisional_llm": bool(provisional_count), + "prediction_layout": "[target_time, class, future_horizon]", + "class_names": list(CLASS_NAMES), + "sessions": list(args.sessions), + "calibration_sessions": args.calibration_sessions, + "entry_score": ( + "max wearer probability until competing resolution, " + f"capped at {args.entry_max_seconds:g} s" + ), + "yield_score": ( + "max (1 - wearer probability) until other offset, " + f"capped at {args.yield_max_seconds:g} s" + ), + "entry_max_seconds": entry_max, + "yield_max_seconds": yield_max, + "entry_headline_hard_negatives": sorted(ENTRY_CORE_HARD_NEGATIVES), + "entry_backchannel_policy": "separate diagnostic; excluded from headline", + "matched_recalls": recalls, + "review_diagnostics": review_diagnostics, + "calibration_review_diagnostics": calibration_diagnostics, + "runs": results, + } + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text( + json.dumps(payload, indent=2, ensure_ascii=False) + "\n", + encoding="utf-8", + ) + metrics_path = args.csv_output or args.output.with_name( + args.output.stem + "_metrics.csv" + ) + events_path = args.event_csv_output or args.output.with_name( + args.output.stem + "_events.csv" + ) + _write_metrics(metrics_path, results) + _write_events(events_path, all_rows) + _print_examples(first_rows, args.print_event_examples) + print(f"Saved {args.output}") + print(f"Saved {metrics_path}") + print(f"Saved {events_path}") + + +if __name__ == "__main__": + main() diff --git a/manuscript/main/main.tex b/manuscript/main/main.tex index ce1926c..8fe6d8f 100644 --- a/manuscript/main/main.tex +++ b/manuscript/main/main.tex @@ -328,47 +328,55 @@ configuration, trainable parameter counts, and inference latency.} We first retain dense future-activity evaluation for direct comparison with EgoSpeak, reporting AP at the 0.2--2.0~s horizons and their mean mAP. -For dense-only models, behavioral evaluation begins with an explicitly -activity-derived event proxy rather than treating wearer voice activity as a -semantic event detector. Following the prediction-region principle used by -VAP~\cite{ekstedt2022vap}, every adjudicated item contributes one score from a -fixed 0.6-s (three-frame) causal decision region. A positive uses the three -source frames immediately preceding its corrected behavior anchor. An -independently mined hard negative uses the first three source frames after its -opportunity begins; a raw wearer boundary relabeled as a hard negative instead -uses the same pre-boundary region as a positive. Items without a complete -three-frame region are excluded and counted. The longer interval supplied to -the annotator is evidence context only and is never a model-scoring window. +For dense-only models, behavioral evaluation is opportunity-conditioned and +uses the model's original wearer-activity forecasts rather than treating them +as semantic intent probabilities. An entry opportunity begins at a reviewed +non-wearer release. It resolves positively when the wearer enters, or +negatively when the same non-wearer resumes or a different non-wearer takes the +next floor. From the first causal 5-Hz state that observes the release, the +entry score is the maximum projected wearer probability before that competing +resolution, capped at 2.0~s. Positive entries without a reviewer-selected +release are reported as unanchored and excluded from this matched-opportunity +analysis; they remain available to the separate raw-onset analysis. -At each decision frame, we follow the target-aligned forecast tensor -diagonally to recover the wearer probabilities emitted at that same causal -source time. We average the 0.6--2.0-s horizons to obtain a sustained wearer- -activity score. This mean is the entry proxy, and one minus the mean is the -INT-yield proxy. Max-pooling the three decision-frame scores gives exactly one -score per positive event or hard-negative opportunity. This construction can -capture sustained activity versus a brief vocalization, but it cannot by -itself establish semantic floor intent; in particular, wearer backchannels and -within-turn pauses remain deliberately difficult negatives. A later linear -event probe, if used, is fit only on training-side annotations and compared at -matched capacity across representations, following the separated downstream- -probe logic of MuVAP~\cite{qi2026muvap}. +An INT opportunity begins when a non-wearer starts while the wearer holds the +floor. It resolves positively when the reviewed behavior is an INT-yield, and +negatively when the non-wearer contribution is a backchannel or failed +takeover and the wearer retains the floor. The yield score is the maximum of +one minus the projected wearer probability over the next 0.2--0.6~s, censored +strictly at the competing non-wearer offset. All prediction values are read +diagonally from one causal source row of the target-aligned forecast tensor. +No input after the observed action outcome is used. For current INT-positive +annotations, which do not select a competing onset explicitly, the +implementation uses the latest supplied non-wearer interval crossing the +wearer offset and surfaces cases with multiple eligible intervals for audit. + +The headline entry hard negatives are same-speaker mid-turn pauses and other- +to-other transfers. Wearer backchannels are not pooled into headline false- +entry rate: their high wearer probability is correct under the dense training +target. We instead retain entry-versus-backchannel as a separate semantic- +selectivity diagnostic. Thus, the semantic annotations define which matched +opportunities are evaluated, while every model is scored solely through the +same EgoSpeak-compatible dense output. A capacity-matched downstream probe, +following MuVAP~\cite{qi2026muvap}, is secondary and is not needed for this +zero-shot activity-derived comparison. \textbf{False-positive scoring rule.} -A false positive is a hard-negative opportunity whose fixed-region proxy score +A false positive is a hard-negative opportunity whose opportunity score exceeds the frozen threshold, regardless of whether the wearer subsequently -exhibits an onset or offset. Max-pooling means that multiple high-scoring -decision frames still count as one false-positive opportunity. Thus, false-positive -evaluation includes conversational situations in which the correct wearer -behavior is to remain silent or retain the floor. +exhibits an unrelated later onset or offset. Each opportunity contributes one +score, and competing-risk censoring prevents later conversational outcomes +from changing its label. Thus, false-positive evaluation includes situations +in which the correct wearer behavior is to remain silent or retain the floor. We report threshold-free proxy AP for entry and INT. The headline behavioral operating point additionally reports false-positive rate at matched recall $R=0.8$; thresholds are selected independently for entry and INT from training- side out-of-fold predictions. The other-to-other hard-negative subset tests whether local visual evidence specifically reduces erroneous wearer action -scores during multiparty floor transfers. \todo{Freeze the 0.6-s decision -window and 0.6--2.0-s aggregation range using training-side data, run window -sensitivity analysis, and verify identical event denominators across methods.} +scores during multiparty floor transfers. \todo{Freeze the 2.0-s entry and +0.6-s yield caps using training-side data, audit selected-release and competing- +onset coverage, and verify identical censored/valid denominators across methods.} A deployment-style committed-event analysis with one-to-one matching and signed latency is secondary and is omitted unless a threshold, persistence, @@ -393,8 +401,8 @@ verify identical targets, causal validity, and evaluation.} \begin{table}[t] \caption{Confirmatory EasyCom results. Entry and INT false-positive rates use -the fixed-region activity-derived proxy at matched recall $R=0.8$; E/I AP is -the corresponding event-balanced proxy AP.} +the opportunity-conditioned activity score at matched recall $R=0.8$; E/I AP +is the corresponding event-balanced opportunity AP.} \label{tab:main} \centering \small @@ -426,7 +434,7 @@ evidence primarily improves \todo{the interpretation supported by the data}. \textbf{Annotated behavior.} At matched wearer-entry recall $R=0.8$, the proposed adapter reduces entry FPR from \todo{$X$} to \todo{$Y$}; at matched INT-yield recall, it changes INT FPR -from \todo{$A$} to \todo{$B$}. Its entry/INT activity-proxy APs are +from \todo{$A$} to \todo{$B$}. Its entry/INT opportunity APs are \todo{$E/I$}. These recall-matched opportunity error rates test whether lower scores on confusing floor situations are obtained without suppressing legitimate entry or yield events. diff --git a/test_evaluate_egospeak_opportunity_events.py b/test_evaluate_egospeak_opportunity_events.py new file mode 100644 index 0000000..05741a5 --- /dev/null +++ b/test_evaluate_egospeak_opportunity_events.py @@ -0,0 +1,375 @@ +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_egospeak_opportunity_events as evaluator +from evaluate_egospeak_opportunity_events import ( + OpportunityEvent, + _first_causal_source_row, + _score_event, + evaluate_family, + load_opportunity_events, +) +from evaluate_floor_state_changes import NUM_CLASSES, NUM_HORIZONS + + +def reviewed_row( + uid: str, + *, + family: str, + verdict: str, + subtype: str | None, + anchor: float, + region_start: float, + region_end: float, + selected_release: float | None = None, + nearby: list[dict[str, object]] | None = None, + candidate_source: str | None = None, +) -> dict[str, object]: + event_type = None + if verdict == "POSITIVE": + event_type = "WEARER_ENTRY" if family == "entry" else "INT_YIELD" + return { + "candidate_id": uid, + "session": "Session_4", + "candidate_family": family, + "review": { + "human_status": "REVIEWED", + "human_verdict": verdict, + "human_event_type": event_type, + "human_subtype": subtype, + "selected_opportunity_seconds": selected_release, + "boundary_correction_seconds": 0, + }, + "candidate": { + "candidate_source": candidate_source or ( + "WEARER_ONSET" if family == "entry" else "WEARER_OFFSET" + ), + "anchor_dense_seconds": anchor, + "region_start_seconds": region_start, + "region_end_seconds": region_end, + "temporal_evidence": { + "behavior_anchor_seconds": anchor, + "wearer_offset_seconds": anchor, + "nearby_other_intervals": nearby or [], + }, + }, + } + + +def prediction_arrays(length: int = 80) -> tuple[np.ndarray, np.ndarray]: + return ( + np.zeros((length, NUM_CLASSES, NUM_HORIZONS), dtype=np.float32), + np.ones((length, NUM_HORIZONS), dtype=bool), + ) + + +class OpportunityEventTests(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_first_source_row_observes_non_grid_opportunity(self) -> None: + source_row, cutoff = _first_causal_source_row(1.05) + self.assertEqual(source_row, 5) + self.assertAlmostEqual(cutoff, 1.2) + self.assertEqual(_first_causal_source_row(1.0), (4, 1.0)) + + def test_loader_uses_selected_release_and_excludes_entry_backchannel(self) -> None: + rows = [ + reviewed_row( + "entry-positive", + family="entry", + verdict="POSITIVE", + subtype=None, + anchor=1.6, + region_start=0.0, + region_end=3.0, + selected_release=1.0, + ), + reviewed_row( + "entry-backchannel", + family="entry", + verdict="HARD_NEGATIVE", + subtype="ENTRY_WEARER_BACKCHANNEL", + anchor=3.0, + region_start=2.0, + region_end=4.0, + ), + ] + temporary, path = self.write_reviews(rows) + self.addCleanup(temporary.cleanup) + events, diagnostics = load_opportunity_events([path], ["Session_4"]) + self.assertEqual(len(events), 1) + self.assertEqual(events[0].opportunity_anchor_seconds, 1.0) + self.assertEqual(events[0].resolution_seconds, 1.6) + self.assertEqual( + diagnostics["skipped"]["entry_backchannel_separate_diagnostic"], 1 + ) + + def test_loader_selects_latest_crossing_other_for_int(self) -> None: + nearby = [ + { + "onset_seconds": 1.0, + "offset_seconds": 3.5, + "crosses_offset_with_tolerance": True, + }, + { + "onset_seconds": 2.0, + "offset_seconds": 3.8, + "crosses_offset_with_tolerance": True, + }, + ] + temporary, path = self.write_reviews([ + reviewed_row( + "yield-positive", + family="int", + verdict="POSITIVE", + subtype=None, + anchor=3.0, + region_start=0.0, + region_end=6.0, + nearby=nearby, + ) + ]) + self.addCleanup(temporary.cleanup) + events, _ = load_opportunity_events([path], ["Session_4"]) + self.assertEqual(len(events), 1) + self.assertEqual(events[0].opportunity_anchor_seconds, 2.0) + self.assertEqual(events[0].censor_seconds, 3.8) + self.assertEqual(events[0].competing_intervals, 2) + + def test_entry_hn_requires_matching_independently_mined_source(self) -> None: + rows = [ + reviewed_row( + "valid", + family="entry", + verdict="HARD_NEGATIVE", + subtype="ENTRY_OTHER_TO_OTHER", + anchor=2.0, + region_start=2.0, + region_end=2.6, + candidate_source="OTHER_TO_OTHER", + ), + reviewed_row( + "invalid", + family="entry", + verdict="HARD_NEGATIVE", + subtype="ENTRY_OTHER_TO_OTHER", + anchor=4.0, + region_start=1.0, + region_end=7.0, + candidate_source="WEARER_ONSET", + ), + ] + temporary, path = self.write_reviews(rows) + self.addCleanup(temporary.cleanup) + events, diagnostics = load_opportunity_events([path], ["Session_4"]) + self.assertEqual([event.uid for event in events], ["valid"]) + self.assertEqual( + diagnostics["skipped"]["entry_hn_without_matching_mined_release"], + 1, + ) + + def test_entry_score_uses_max_wearer_probability_before_resolution(self) -> None: + event = OpportunityEvent( + "entry", "Session_4", "entry", True, "WEARER_ENTRY", "WEARER_ONSET", + 1.0, 1.6, 1.6, 1.6, "reviewed_release", + ) + prediction, valid = prediction_arrays() + # source row 4, cutoff 1.0; h=.2/.4/.6 are admitted. Since the onset + # is exactly on-grid, the h=.8 post-onset bin must be excluded. + prediction[5, 1, 0] = 0.1 + prediction[6, 1, 1] = 0.2 + prediction[7, 1, 2] = 0.8 + prediction[8, 1, 3] = 0.3 + scored = _score_event( + event, + {"Session_4": (prediction, valid)}, + entry_max_seconds=2.0, + yield_max_seconds=0.6, + ) + self.assertIsNotNone(scored) + self.assertAlmostEqual(scored[0], 0.8) # type: ignore[index] + self.assertAlmostEqual(scored[1]["best_horizon_seconds"], 0.6) # type: ignore[index] + self.assertEqual(len(scored[1]["used_horizons_seconds"]), 3) # type: ignore[index] + + def test_entry_positive_admits_bin_containing_off_grid_onset(self) -> None: + event = OpportunityEvent( + "entry", "Session_4", "entry", True, "WEARER_ENTRY", "WEARER_ONSET", + 1.0, 1.65, 1.65, 1.65, "reviewed_release", + ) + prediction, valid = prediction_arrays() + prediction[8, 1, 3] = 0.9 # target cutoff 1.8 contains onset 1.65. + scored = _score_event( + event, + {"Session_4": (prediction, valid)}, + entry_max_seconds=2.0, + yield_max_seconds=0.6, + ) + self.assertIsNotNone(scored) + self.assertAlmostEqual(scored[0], 0.9) # type: ignore[index] + self.assertEqual(len(scored[1]["used_horizons_seconds"]), 4) # type: ignore[index] + + def test_entry_hard_negative_stops_before_competing_onset(self) -> None: + event = OpportunityEvent( + "negative", "Session_4", "entry", False, "ENTRY_OTHER_TO_OTHER", + "OTHER_TO_OTHER", 1.0, 1.6, 1.6, None, "mined_release", + ) + prediction, valid = prediction_arrays() + prediction[5, 1, 0] = 0.1 + prediction[6, 1, 1] = 0.2 + prediction[7, 1, 2] = 0.3 + prediction[8, 1, 3] = 0.95 # 1.8 s: after competing onset, must be ignored. + scored = _score_event( + event, + {"Session_4": (prediction, valid)}, + entry_max_seconds=2.0, + yield_max_seconds=0.6, + ) + self.assertIsNotNone(scored) + self.assertAlmostEqual(scored[0], 0.3) # type: ignore[index] + self.assertEqual(scored[1]["used_horizons_seconds"], [0.2, 0.4, 0.6000000000000001]) # type: ignore[index] + + def test_yield_score_is_near_term_inactivity_and_censored(self) -> None: + event = OpportunityEvent( + "yield", "Session_4", "int", True, "INT_YIELD", "WEARER_OFFSET", + 2.0, 2.4, 2.5, 2.4, "crossing_other", + ) + prediction, valid = prediction_arrays() + prediction[10, 1, 0] = 0.9 + prediction[11, 1, 1] = 0.2 + prediction[12, 1, 2] = 0.0 # target 2.6 crosses other offset 2.5. + scored = _score_event( + event, + {"Session_4": (prediction, valid)}, + entry_max_seconds=2.0, + yield_max_seconds=0.6, + ) + self.assertIsNotNone(scored) + self.assertAlmostEqual(scored[0], 0.8) # type: ignore[index] + self.assertEqual(scored[1]["used_horizons_seconds"], [0.2, 0.4]) # type: ignore[index] + + def test_positive_is_dropped_if_first_causal_state_observes_outcome(self) -> None: + event = OpportunityEvent( + "fast", "Session_4", "entry", True, "WEARER_ENTRY", "WEARER_ONSET", + 1.05, 1.1, 1.1, 1.1, "reviewed_release", + ) + prediction, valid = prediction_arrays() + self.assertIsNone( + _score_event( + event, + {"Session_4": (prediction, valid)}, + entry_max_seconds=2.0, + yield_max_seconds=0.6, + ) + ) + + def test_one_score_per_opportunity_and_ap(self) -> None: + events = [ + OpportunityEvent( + "p", "Session_4", "entry", True, "WEARER_ENTRY", "WEARER_ONSET", + 1.0, 1.6, 1.6, 1.6, "reviewed_release", + ), + OpportunityEvent( + "n", "Session_4", "entry", False, "ENTRY_OTHER_TO_OTHER", + "OTHER_TO_OTHER", 2.0, 2.6, 2.6, None, "mined_release", + ), + ] + prediction, valid = prediction_arrays() + prediction[7, 1, 2] = 0.9 + metric, rows = evaluate_family( + events, + {"Session_4": (prediction, valid)}, + "entry", + [], + entry_max_seconds=2.0, + yield_max_seconds=0.6, + ) + 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) + + def test_main_writes_summary_metrics_and_event_audit(self) -> None: + nearby = [{ + "onset_seconds": 5.0, + "offset_seconds": 6.0, + "crosses_offset_with_tolerance": True, + }] + rows = [ + reviewed_row( + "ep", family="entry", verdict="POSITIVE", subtype=None, + anchor=1.6, region_start=0.0, region_end=3.0, + selected_release=1.0, + ), + reviewed_row( + "en", family="entry", verdict="HARD_NEGATIVE", + subtype="ENTRY_OTHER_TO_OTHER", anchor=3.0, + region_start=3.0, region_end=3.8, + candidate_source="OTHER_TO_OTHER", + ), + reviewed_row( + "ip", family="int", verdict="POSITIVE", subtype=None, + anchor=5.4, region_start=3.0, region_end=7.0, nearby=nearby, + ), + reviewed_row( + "in", family="int", verdict="HARD_NEGATIVE", + subtype="INT_OTHER_BACKCHANNEL", anchor=7.0, + region_start=7.0, region_end=7.8, + candidate_source="OTHER_BACKCHANNEL", + ), + ] + 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, valid = prediction_arrays() + np.save(run_dir / "pred" / "Session_4.npy", prediction) + np.save(run_dir / "valid" / "Session_4.npy", valid) + output = root / "summary.json" + args = argparse.Namespace( + reviews=[reviews], + run=[f"egospeak={run_dir}"], + sessions=["Session_4"], + entry_max_seconds=2.0, + yield_max_seconds=0.6, + calibration_reviews=None, + calibration_run=None, + calibration_sessions=None, + matched_recalls=[], + allow_in_sample_thresholds=False, + output=output, + csv_output=None, + event_csv_output=None, + print_event_examples=0, + allow_provisional_llm_reviews=False, + ) + with mock.patch.object(evaluator, "parse_args", return_value=args): + evaluator.main() + payload = json.loads(output.read_text(encoding="utf-8")) + self.assertEqual( + payload["protocol"], + "egospeak-opportunity-conditioned-floor-actions-v1", + ) + self.assertTrue((root / "summary_metrics.csv").is_file()) + self.assertTrue((root / "summary_events.csv").is_file()) + self.assertEqual(len(payload["runs"][0]["metrics"]), 2) + + +if __name__ == "__main__": + unittest.main()