557 lines
21 KiB
Python
Executable File
557 lines
21 KiB
Python
Executable File
#!/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()
|