#!/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()