#!/usr/bin/env python3 """Evaluate an activity-derived ENTRY/YIELD proxy from dense forecasts. This evaluator does not treat wearer voice-activity probability as a semantic event detector. Instead, it asks whether the multi-horizon wearer trajectory separates adjudicated ENTRY/YIELD events from matched hard-negative opportunities. Every opportunity contributes at most one score. Positive events use a fixed causal decision region immediately before the corrected behavior anchor. Independently mined hard negatives use an equal-width region beginning at the annotated opportunity onset. A raw wearer boundary relabeled as a hard negative uses the pre-boundary region, just like its positive counterpart. Long annotation/evidence regions are never used as scoring windows. At every 5-Hz source row, the evaluator follows the target-aligned prediction tensor diagonally to recover wearer probabilities from one causal prediction time across selected future horizons. Their mean is a sustained-activity proxy for ENTRY; one minus that mean is the corresponding YIELD proxy. The maximum over the fixed decision region produces one event-level score. """ from __future__ import annotations import argparse import csv import json import math from pathlib import Path from typing import Any, Iterable import numpy as np from evaluate_annotated_floor_events import ( AnnotatedOpportunity, FAMILIES, Family, load_annotated_opportunities, 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, ) RAW_WEARER_BOUNDARY_SOURCES = frozenset({"WEARER_ONSET", "WEARER_OFFSET"}) DEFAULT_PROXY_HORIZONS = tuple(round(value * STEP_SECONDS, 10) for value in range(3, 11)) 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("--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=(), help=( "Optional recall operating points. Official Sessions 1--3 require " "disjoint training-side calibration when this option is non-empty." ), ) parser.add_argument( "--allow-in-sample-thresholds", action="store_true", help="Exploratory only: select requested operating points on evaluated events.", ) parser.add_argument("--decision-window-seconds", type=float, default=0.6) parser.add_argument( "--proxy-horizons-seconds", nargs="+", type=float, default=DEFAULT_PROXY_HORIZONS, metavar="SECONDS", help="Future horizons averaged into the sustained-activity proxy.", ) parser.add_argument( "--output", type=Path, default=Path("results/easycom_egospeak_event_proxy/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 _grid_steps(seconds: float, *, field: str) -> int: if not math.isfinite(seconds) or seconds <= 0: raise ValueError(f"{field} must be finite and positive") steps = int(round(seconds / STEP_SECONDS)) if not math.isclose(seconds, steps * STEP_SECONDS, abs_tol=1e-7): raise ValueError(f"{field} must lie on the {STEP_SECONDS:g}-s grid") return steps def horizon_indices(values: Iterable[float]) -> list[int]: result: list[int] = [] for value in values: steps = _grid_steps(float(value), field="proxy horizon") if not 1 <= steps <= NUM_HORIZONS: raise ValueError( f"proxy horizon must be between {STEP_SECONDS:g} and " f"{NUM_HORIZONS * STEP_SECONDS:g} seconds" ) result.append(steps - 1) if len(set(result)) != len(result): raise ValueError("proxy horizons must be unique") return sorted(result) def _anchor_row(opportunity: AnnotatedOpportunity) -> int: return int(math.floor(opportunity.anchor_seconds / STEP_SECONDS + 1e-7)) def decision_source_rows( opportunity: AnnotatedOpportunity, *, window_rows: int, ) -> tuple[int, int, str] | None: """Return an equal-width half-open source-row region for one opportunity.""" raw_boundary = opportunity.candidate_source in RAW_WEARER_BOUNDARY_SOURCES if not opportunity.candidate_source: # Backward-compatible fallback for older reviewed JSONL without source. raw_boundary = ( opportunity.region_start_seconds + 1e-7 < opportunity.anchor_seconds < opportunity.region_end_seconds - 1e-7 ) if opportunity.positive or raw_boundary: end = _anchor_row(opportunity) start = end - window_rows basis = "pre_behavior_anchor" else: start = opportunity.start_row end = start + window_rows basis = "hard_negative_opportunity_onset" if end > opportunity.end_row_exclusive: return None if start < 0 or end <= start: return None return start, end, basis def _source_trajectory( prediction: np.ndarray, valid: np.ndarray, source_row: int, selected_horizons: list[int], ) -> np.ndarray | None: values: list[float] = [] for horizon_index in selected_horizons: target_row = source_row + horizon_index + 1 if target_row < 0 or target_row >= len(prediction): return None if not valid[target_row, horizon_index]: return None values.append(float(prediction[target_row, WEARER_INDEX, horizon_index])) return np.asarray(values, dtype=np.float64) def score_activity_proxy( opportunity: AnnotatedOpportunity, arrays: dict[str, tuple[np.ndarray, np.ndarray]], *, window_rows: int, selected_horizons: list[int], ) -> tuple[float, dict[str, Any]] | None: region = decision_source_rows(opportunity, window_rows=window_rows) if region is None: return None source_start, source_end, basis = region prediction, valid = arrays[opportunity.session] frame_scores: list[float] = [] trajectories: list[list[float]] = [] for source_row in range(source_start, source_end): trajectory = _source_trajectory( prediction, valid, source_row, selected_horizons ) if trajectory is None: return None sustained_activity = float(np.mean(trajectory)) score = sustained_activity if opportunity.family == "entry" else 1.0 - sustained_activity frame_scores.append(score) trajectories.append(trajectory.tolist()) best_index = int(np.argmax(frame_scores)) return float(frame_scores[best_index]), { "decision_basis": basis, "source_start_row": source_start, "source_end_row_exclusive": source_end, "best_source_row": source_start + best_index, "frame_scores": frame_scores, "best_wearer_trajectory": trajectories[best_index], } def evaluate_family( opportunities: list[AnnotatedOpportunity], arrays: dict[str, tuple[np.ndarray, np.ndarray]], family: Family, recalls: list[float], *, window_rows: int, selected_horizons: list[int], 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 opportunity in opportunities: if opportunity.family != family: continue scored = score_activity_proxy( opportunity, arrays, window_rows=window_rows, selected_horizons=selected_horizons, ) if scored is None: invalid += 1 continue score, details = scored rows.append( { "uid": opportunity.uid, "session": opportunity.session, "family": family, "positive": opportunity.positive, "subtype": opportunity.subtype, "candidate_source": opportunity.candidate_source, "score": score, "anchor_seconds": opportunity.anchor_seconds, "evidence_region_start_seconds": opportunity.region_start_seconds, "evidence_region_end_seconds": opportunity.region_end_seconds, **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) subtype_names = sorted({str(row["subtype"]) for row in rows if not row["positive"]}) extra_groups = { f"fpr_{name.lower()}": np.asarray( [not row["positive"] and row["subtype"] == name for row in rows], dtype=bool, ) for name in subtype_names } 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_short": 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 _validate_calibration( args: argparse.Namespace, runs: list[tuple[str, Path]], recalls: list[float], ) -> tuple[list[tuple[str, Path]], bool]: supplied = [ bool(args.calibration_reviews), bool(args.calibration_sessions), ] if any(supplied) and not all(supplied): raise ValueError( "--calibration-reviews and --calibration-sessions must be supplied together" ) has_calibration = all(supplied) if args.calibration_run and not has_calibration: raise ValueError( "--calibration-run requires --calibration-reviews and " "--calibration-sessions" ) if has_calibration: unknown = sorted(set(args.calibration_sessions) - set(OFFICIAL_TRAIN_SESSIONS)) if unknown: raise ValueError(f"Calibration must use EasyCom 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 --matched-recalls for threshold-free AP" ) if args.calibration_run: calibration_runs = parse_runs(args.calibration_run) if [name for name, _ in calibration_runs] != [name for name, _ in runs]: raise ValueError("--calibration-run names/order must match --run") else: calibration_runs = runs return calibration_runs, has_calibration def _write_event_csv(path: Path, rows: list[dict[str, Any]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) fieldnames = [ "run", "uid", "session", "family", "positive", "subtype", "candidate_source", "score", "anchor_seconds", "decision_basis", "source_start_row", "source_end_row_exclusive", "best_source_row", "evidence_region_start_seconds", "evidence_region_end_seconds", "frame_scores", "best_wearer_trajectory", ] with path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames) writer.writeheader() for row in rows: serialized = dict(row) serialized["frame_scores"] = json.dumps(row["frame_scores"]) serialized["best_wearer_trajectory"] = json.dumps( row["best_wearer_trajectory"] ) writer.writerow(serialized) def _write_metric_csv(path: Path, results: list[dict[str, Any]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) fieldnames = [ "run", "family", "num_positive", "num_hard_negative", "num_invalid_or_short", "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=fieldnames) 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_short": metric["num_invalid_or_short"], "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 _print_examples(rows: list[dict[str, Any]], limit: int) -> None: if limit <= 0: return print("\nActivity-derived proxy examples (one score per opportunity)") 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"source_rows=[{row['source_start_row']},{row['source_end_row_exclusive']}) " f"best_source={row['best_source_row']} " f"evidence={row['evidence_region_start_seconds']:.2f}--" f"{row['evidence_region_end_seconds']:.2f}s " f"score={row['score']:.4f} subtype={row['subtype']}" ) shown += 1 if shown >= limit: break def main() -> None: args = parse_args() window_rows = _grid_steps(args.decision_window_seconds, field="decision window") selected_horizons = horizon_indices(args.proxy_horizons_seconds) 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) opportunities, review_diagnostics = load_annotated_opportunities( args.reviews, args.sessions, allow_provisional_llm=args.allow_provisional_llm_reviews, ) if not opportunities: raise ValueError("No scoring-eligible reviewed opportunities") calibration_opportunities = None calibration_diagnostics = None if has_calibration: calibration_opportunities, calibration_diagnostics = load_annotated_opportunities( args.calibration_reviews, args.calibration_sessions, allow_provisional_llm=args.allow_provisional_llm_reviews, ) if not calibration_opportunities: raise ValueError("No scoring-eligible calibration opportunities") results: list[dict[str, Any]] = [] event_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_opportunities 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_opportunities is not None and calibration_arrays is not None: calibration_metric, _ = evaluate_family( calibration_opportunities, calibration_arrays, family, recalls, window_rows=window_rows, selected_horizons=selected_horizons, threshold_source="training_side_calibration", ) fixed_thresholds = _thresholds(calibration_metric) threshold_source = "training_side_calibration" metric, rows = evaluate_family( opportunities, arrays, family, recalls, window_rows=window_rows, selected_horizons=selected_horizons, 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) event_rows.extend(run_rows) results.append({"name": name, "result_dir": str(run_dir), "metrics": metrics}) provisional = bool( review_diagnostics["num_provisional_llm"] + (0 if calibration_diagnostics is None else calibration_diagnostics["num_provisional_llm"]) ) payload = { "schema_version": 1, "protocol": "egospeak-wearer-trajectory-event-proxy-v1", "semantic_scope": ( "Activity-derived diagnostic, not a semantic event detector. " "Provisional LLM auto-accepted labels are present." if provisional else "Activity-derived diagnostic on human-reviewed opportunities; " "the wearer trajectory is not a semantic event detector." ), "provisional_llm": provisional, "prediction_layout": "[target_time, class, future_horizon]", "class_names": list(CLASS_NAMES), "sessions": list(args.sessions), "calibration_sessions": args.calibration_sessions, "decision_window_seconds": args.decision_window_seconds, "decision_window_rows": window_rows, "proxy_horizons_seconds": [ (index + 1) * STEP_SECONDS for index in selected_horizons ], "source_frame_proxy": ( "entry=mean wearer probability; int=1-mean wearer probability; " "mean over selected future horizons" ), "opportunity_pooling": "maximum over fixed-width decision source rows", "review_diagnostics": review_diagnostics, "calibration_review_diagnostics": calibration_diagnostics, "matched_recalls": recalls, "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", ) event_path = args.event_csv_output or args.output.with_name( args.output.stem + "_events.csv" ) metric_path = args.csv_output or args.output.with_name( args.output.stem + "_metrics.csv" ) _write_metric_csv(metric_path, results) _write_event_csv(event_path, event_rows) _print_examples(first_rows, args.print_event_examples) print(f"Saved {args.output}") print(f"Saved {metric_path}") print(f"Saved {event_path}") if __name__ == "__main__": main()