"""Evaluate wearer state changes and conservative floor-context events. This evaluator consumes the ``pred/`` and ``valid/`` arrays already written by ``infer_turn_head.py``. It does not run a model and does not require a GPU. The activity head predicts future wearer activity, not an explicit event hazard. We therefore evaluate endpoint state changes conditional on the observed wearer state at the prediction source. Event windows use competing-risk censoring: a response window ends before the next aggregate- others onset, and a yield window ends before the next aggregate-others offset. The resulting events remain timing-derived behavior proxies; aggregate activity cannot establish semantic EOT/INT, speaker identity, or addressee. """ 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 # Keep this read-only evaluator independent of torch/model imports. OFFICIAL_TEST_SESSIONS = ("Session_3", "Session_2", "Session_1") OFFICIAL_TRAIN_SESSIONS = tuple(f"Session_{index}" for index in range(4, 13)) CLASS_NAMES = ( "background", "camera_wearer_speaking", "other_people_speaking", ) STEP_SECONDS = 0.2 NUM_HORIZONS = 10 NUM_CLASSES = 3 WEARER_INDEX = 1 OTHERS_INDEX = 2 def normalize_session_name(value: str | int) -> str: text = str(value) suffix = text.removeprefix("Session_") if not suffix.isdigit(): raise ValueError(f"Invalid EasyCom session: {value}") return f"Session_{int(suffix)}" def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--target-root", type=Path, required=True) parser.add_argument( "--run", action="append", required=True, metavar="NAME=RESULT_DIR", help=( "Named inference result directory containing pred/ and valid/. " "Repeat once per method." ), ) parser.add_argument( "--sessions", nargs="+", default=list(OFFICIAL_TEST_SESSIONS), help="EasyCom sessions (default: Session_3 Session_2 Session_1).", ) parser.add_argument( "--matched-recalls", nargs="+", type=float, default=(0.7, 0.8, 0.9), help="EOT-response/yield proxy recall operating points.", ) parser.add_argument( "--calibration-sessions", nargs="+", default=None, help=( "Training-side sessions used to select event thresholds. Required " "when evaluating official Sessions 1--3 unless the exploratory " "in-sample override is set." ), ) parser.add_argument( "--calibration-run", action="append", default=None, metavar="NAME=RESULT_DIR", help=( "Optional calibration result directory for each --run name. If " "omitted, calibration predictions are read from the matching " "--run directory." ), ) parser.add_argument( "--allow-in-sample-test-thresholds", action="store_true", help=( "Exploratory override allowing thresholds to be selected on the " "same official test sessions being evaluated. Never use for a " "confirmatory paper result." ), ) parser.add_argument( "--output", type=Path, default=Path("reports/easycom_floor_state_changes.json"), ) parser.add_argument( "--csv-output", type=Path, default=None, help="Optional event-level CSV path (default: OUTPUT stem + _events.csv).", ) parser.add_argument( "--midturn-pause-max-seconds", type=float, default=2.0, help=( "Maximum aggregate-others silent gap used only to identify the " "short-resume hard-negative subgroup (default: 2.0)." ), ) parser.add_argument( "--print-framewise", action="store_true", help="Also print the older endpoint/framewise diagnostic table.", ) parser.add_argument( "--print-event-timelines", type=int, default=0, metavar="N", help=( "Print up to N auditable 2-s timelines for each response/yield " "event type from the first run (default: 0)." ), ) return parser.parse_args() def parse_runs(values: Iterable[str]) -> list[tuple[str, Path]]: runs: list[tuple[str, Path]] = [] names: set[str] = set() for value in values: if "=" not in value: raise ValueError(f"--run must be NAME=RESULT_DIR, got {value!r}") name, raw_path = value.split("=", 1) name = name.strip() raw_path = raw_path.strip() if not name or not raw_path: raise ValueError(f"--run must be NAME=RESULT_DIR, got {value!r}") if name in names: raise ValueError(f"Duplicate run name: {name}") names.add(name) runs.append((name, Path(raw_path))) return runs def validate_recalls(values: Iterable[float]) -> list[float]: recalls = [float(value) for value in values] if not recalls: raise ValueError("At least one matched recall is required") if any(not 0.0 < value <= 1.0 for value in recalls): raise ValueError("Matched recalls must be in (0, 1]") if len(recalls) != len(set(recalls)): raise ValueError("Matched recalls must be unique") return sorted(recalls) def _load_target(path: Path) -> np.ndarray: if not path.is_file(): raise FileNotFoundError(path) target = np.load(path, allow_pickle=False) if target.ndim != 2 or target.shape[1] != NUM_CLASSES: raise ValueError(f"Expected [T,{NUM_CLASSES}] target, got {target.shape}: {path}") target = np.asarray(target, dtype=np.float32) if not np.isfinite(target).all(): raise ValueError(f"Target contains NaN/Inf: {path}") if not np.logical_or(target == 0.0, target == 1.0).all(): raise ValueError(f"Target is not binary: {path}") return target def _load_prediction(path: Path, expected_steps: int) -> np.ndarray: if not path.is_file(): raise FileNotFoundError(path) prediction = np.load(path, allow_pickle=False) expected = (expected_steps, NUM_CLASSES, NUM_HORIZONS) if prediction.shape != expected: raise ValueError(f"Expected prediction {expected}, got {prediction.shape}: {path}") prediction = np.asarray(prediction, dtype=np.float32) if not np.isfinite(prediction).all(): raise ValueError(f"Prediction contains NaN/Inf: {path}") if (prediction < -1e-6).any() or (prediction > 1.0 + 1e-6).any(): raise ValueError(f"Prediction is outside [0,1]: {path}") return np.clip(prediction, 0.0, 1.0) def _load_valid(path: Path, expected_steps: int) -> np.ndarray: if not path.is_file(): raise FileNotFoundError(path) valid = np.load(path, allow_pickle=False) expected = (expected_steps, NUM_HORIZONS) if valid.shape != expected: raise ValueError(f"Expected valid mask {expected}, got {valid.shape}: {path}") return np.asarray(valid, dtype=bool) def load_arrays( target_root: Path, run_dir: Path, sessions: list[str], ) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]: targets: list[np.ndarray] = [] predictions: list[np.ndarray] = [] valid_masks: list[np.ndarray] = [] session_ids: list[np.ndarray] = [] for session_id, session in enumerate(sessions): target = _load_target(target_root / f"{session}.npy") prediction = _load_prediction(run_dir / "pred" / f"{session}.npy", len(target)) valid = _load_valid(run_dir / "valid" / f"{session}.npy", len(target)) targets.append(target) predictions.append(prediction) valid_masks.append(valid) session_ids.append(np.full(len(target), session_id, dtype=np.int32)) return ( np.concatenate(targets, axis=0), np.concatenate(predictions, axis=0), np.concatenate(valid_masks, axis=0), np.concatenate(session_ids, axis=0), ) def _safe_ap(target: np.ndarray, score: np.ndarray) -> float | None: target = np.asarray(target, dtype=bool) score = np.asarray(score, dtype=np.float64) if not len(target) or not target.any() or target.all(): return None # Non-interpolated AP with tied scores grouped together, matching sklearn's # binary average_precision_score definition without importing the training # stack for this lightweight post-processing script. order = np.argsort(-score, kind="mergesort") ordered_score = score[order] ordered_target = target[order] distinct_ends = np.flatnonzero( np.r_[ordered_score[1:] != ordered_score[:-1], True] ) true_positive = np.cumsum(ordered_target, dtype=np.int64)[distinct_ends] predicted_positive = distinct_ends + 1 precision = true_positive / predicted_positive recall = true_positive / int(target.sum()) recall_increment = np.diff(np.r_[0.0, recall]) return float(np.sum(recall_increment * precision)) def _ranking_from_masks( score: np.ndarray, positive: np.ndarray, negative: np.ndarray, ) -> dict[str, Any]: positive_scores = score[positive] negative_scores = score[negative] labels = np.concatenate( ( np.ones(len(positive_scores), dtype=bool), np.zeros(len(negative_scores), dtype=bool), ) ) scores = np.concatenate((positive_scores, negative_scores)) return { "num_positive": int(len(positive_scores)), "num_negative": int(len(negative_scores)), "positive_rate": ( float(len(positive_scores) / len(scores)) if len(scores) else None ), "average_precision": _safe_ap(labels, scores), } def _threshold_for_recall(positive_scores: np.ndarray, recall: float) -> float | None: if not len(positive_scores): return None ordered = np.sort(np.asarray(positive_scores, dtype=np.float64))[::-1] rank = max(1, int(math.ceil(recall * len(ordered)))) return float(ordered[rank - 1]) def _rate_at_threshold(scores: np.ndarray, threshold: float | None) -> float | None: if threshold is None or not len(scores): return None return float((scores >= threshold).mean()) def _count_at_threshold(scores: np.ndarray, threshold: float | None) -> int | None: if threshold is None: return None return int((scores >= threshold).sum()) def _operating_points( score: np.ndarray, positive: np.ndarray, negative: np.ndarray, recalls: list[float], *, extra_groups: dict[str, np.ndarray] | None = None, fixed_thresholds: dict[float, float | None] | None = None, threshold_source: str = "evaluated_sessions", ) -> list[dict[str, Any]]: positive_scores = score[positive] negative_scores = score[negative] result: list[dict[str, Any]] = [] for requested_recall in recalls: if fixed_thresholds is None: threshold = _threshold_for_recall(positive_scores, requested_recall) else: if requested_recall not in fixed_thresholds: raise ValueError( f"Missing fixed threshold for recall {requested_recall}" ) threshold = fixed_thresholds[requested_recall] item: dict[str, Any] = { "requested_positive_recall": requested_recall, "threshold": threshold, "threshold_source": threshold_source, "achieved_positive_recall": _rate_at_threshold(positive_scores, threshold), "achieved_positive_count": _count_at_threshold( positive_scores, threshold ), "negative_false_positive_rate": _rate_at_threshold( negative_scores, threshold ), "negative_false_positive_count": _count_at_threshold( negative_scores, threshold ), } for name, mask in (extra_groups or {}).items(): item[name] = _rate_at_threshold(score[mask], threshold) item[f"{name}_count"] = int(mask.sum()) item[f"{name}_trigger_count"] = _count_at_threshold( score[mask], threshold ) result.append(item) return result def evaluate_horizon( target: np.ndarray, prediction: np.ndarray, valid: np.ndarray, session_ids: np.ndarray, horizon_index: int, recalls: list[float], *, operating_thresholds: dict[str, dict[float, float | None]] | None = None, threshold_source: str = "evaluated_sessions", ) -> dict[str, Any]: shift = horizon_index + 1 rows = np.arange(len(target), dtype=np.int64) source_rows = rows - shift usable = valid[:, horizon_index] & (source_rows >= 0) # Never form an artificial transition across concatenated session borders. bounded_source = np.maximum(source_rows, 0) usable &= session_ids == session_ids[bounded_source] target_rows = rows[usable] source_rows = source_rows[usable] source = target[source_rows] future = target[target_rows] wearer_score = prediction[target_rows, WEARER_INDEX, horizon_index] offset_score = 1.0 - wearer_score wearer_now = source[:, WEARER_INDEX].astype(bool) wearer_future = future[:, WEARER_INDEX].astype(bool) others_now = source[:, OTHERS_INDEX].astype(bool) others_future = future[:, OTHERS_INDEX].astype(bool) off_opportunity = ~wearer_now on_opportunity = wearer_now endpoint_onset = off_opportunity & wearer_future endpoint_wait = off_opportunity & ~wearer_future endpoint_offset = on_opportunity & ~wearer_future endpoint_continue = on_opportunity & wearer_future eot_response = ( off_opportunity & others_now & wearer_future & ~others_future ) other_hold_wait = ( off_opportunity & others_now & ~wearer_future & others_future ) early_entry = ( off_opportunity & others_now & wearer_future & others_future ) other_entry_yield = ( on_opportunity & ~others_now & ~wearer_future & others_future ) other_entry_continue = ( on_opportunity & ~others_now & wearer_future & others_future ) onset_endpoint = _ranking_from_masks( wearer_score, endpoint_onset, endpoint_wait ) offset_endpoint = _ranking_from_masks( offset_score, endpoint_offset, endpoint_continue ) eot_hold = _ranking_from_masks( wearer_score, eot_response, other_hold_wait ) eot_hold["num_early_entry_proxy"] = int(early_entry.sum()) eot_hold["operating_points"] = _operating_points( wearer_score, eot_response, other_hold_wait, recalls, extra_groups={ "all_wait_false_entry_rate": endpoint_wait, "early_entry_behavior_recall": early_entry, }, fixed_thresholds=( None if operating_thresholds is None else operating_thresholds["response"] ), threshold_source=threshold_source, ) yield_proxy = _ranking_from_masks( offset_score, other_entry_yield, other_entry_continue ) yield_proxy["operating_points"] = _operating_points( offset_score, other_entry_yield, other_entry_continue, recalls, fixed_thresholds=( None if operating_thresholds is None else operating_thresholds["yield"] ), threshold_source=threshold_source, ) return { "horizon_index": horizon_index + 1, "nominal_seconds": (horizon_index + 1) * STEP_SECONDS, "source_to_target_steps": shift, "num_usable": int(len(target_rows)), "wearer_off_to_on_endpoint": onset_endpoint, "wearer_on_to_off_endpoint": offset_endpoint, "eot_response_vs_other_hold_proxy": eot_hold, "other_entry_yield_vs_continue_proxy": yield_proxy, } def _same_source_path_score( prediction: np.ndarray, valid: np.ndarray, session_ids: np.ndarray, source_row: int, horizon_steps: int, *, offset: bool, ) -> float | None: """Pool fixed-source future activity scores through one horizon. ``prediction[source+k, :, k-1]`` is the score produced from the same source row for the target ``k`` steps later. Max pooling asks whether the behavior occurs anywhere in the event response window; it is a ranking score, not a calibrated event probability. """ target_rows = source_row + np.arange(1, horizon_steps + 1, dtype=np.int64) if source_row < 0 or target_rows[-1] >= len(prediction): return None if np.any(session_ids[target_rows] != session_ids[source_row]): return None horizon_indices = np.arange(horizon_steps, dtype=np.int64) if not valid[target_rows, horizon_indices].all(): return None scores = prediction[target_rows, WEARER_INDEX, horizon_indices] if offset: scores = 1.0 - scores return float(np.max(scores)) def _has_transition_within( values: np.ndarray, source_row: int, horizon_steps: int, *, rising: bool, ) -> bool: window = values[source_row : source_row + horizon_steps + 1] before = window[:-1] after = window[1:] transitions = (~before & after) if rising else (before & ~after) return bool(transitions.any()) def _first_transition_row( values: np.ndarray, source_row: int, horizon_steps: int, *, rising: bool, ) -> int | None: window = values[source_row : source_row + horizon_steps + 1] before = window[:-1] after = window[1:] transitions = (~before & after) if rising else (before & ~after) indices = np.flatnonzero(transitions) return int(source_row + indices[0] + 1) if len(indices) else None def _competing_risk_window( competing_state: np.ndarray, session_ids: np.ndarray, source_row: int, horizon_steps: int, *, competing_rising: bool, ) -> tuple[int, int | None, bool] | None: """Return the observable half-open event window after ``source_row``. The returned step count covers target rows ``source+1`` through ``source+steps``. If an incompatible aggregate-others transition occurs at row ``c``, that row is excluded and returned as the competing-risk row. A wearer transition simultaneous with ``c`` is therefore not attributed to the earlier anchor. The third value records whether the full nominal horizon is observable. This lets the caller retain an observed wearer outcome before a recording boundary while dropping an unresolved right-censored candidate. """ if source_row < 0 or horizon_steps < 1 or source_row + 1 >= len(competing_state): return None session_id = session_ids[source_row] if session_ids[source_row + 1] != session_id: return None nominal_last_row = source_row + horizon_steps available_last_row = min(nominal_last_row, len(competing_state) - 1) while ( available_last_row > source_row and session_ids[available_last_row] != session_id ): available_last_row -= 1 # Row source+1 contains the anchor's post-transition state. A competing # transition can occur no earlier than source+2. competing_row: int | None = None for row in range(source_row + 2, available_last_row + 1): before = bool(competing_state[row - 1]) after = bool(competing_state[row]) is_competing = ((not before) and after) if competing_rising else ( before and (not after) ) if is_competing: competing_row = row break if competing_row is not None: effective_steps = competing_row - source_row - 1 if effective_steps < 1: raise AssertionError("Competing-risk window must contain an observable step") return effective_steps, competing_row, True nominal_complete = available_last_row >= nominal_last_row return available_last_row - source_row, None, nominal_complete def _event_metric( scores: list[float], labels: list[bool], recalls: list[float], *, extra_negative_groups: dict[str, list[bool]] | None = None, fixed_thresholds: dict[float, float | None] | None = None, threshold_source: str = "evaluated_sessions", ) -> dict[str, Any]: score = np.asarray(scores, dtype=np.float64) positive = np.asarray(labels, dtype=bool) negative = ~positive metric = _ranking_from_masks(score, positive, negative) masks = { name: np.asarray(values, dtype=bool) & negative for name, values in (extra_negative_groups or {}).items() } metric["operating_points"] = _operating_points( score, positive, negative, recalls, extra_groups=masks, fixed_thresholds=fixed_thresholds, threshold_source=threshold_source, ) for name, mask in masks.items(): metric[f"num_{name}"] = int(mask.sum()) return metric def _event_window_diagnostics( *, source_boundaries: int, effective_steps: list[int], competing_censored: int, right_censored: int, invalid_score: int, ) -> dict[str, Any]: steps = np.asarray(effective_steps, dtype=np.float64) evaluated = int(len(steps)) return { "num_source_boundaries": int(source_boundaries), "num_evaluated": evaluated, "num_competing_risk_censored": int(competing_censored), "competing_risk_censored_rate": ( float(competing_censored / evaluated) if evaluated else None ), "num_right_censored": int(right_censored), "num_invalid_score_window": int(invalid_score), "mean_effective_window_seconds": ( float(steps.mean() * STEP_SECONDS) if evaluated else None ), "min_effective_window_seconds": ( float(steps.min() * STEP_SECONDS) if evaluated else None ), "max_effective_window_seconds": ( float(steps.max() * STEP_SECONDS) if evaluated else None ), } def evaluate_event_horizon( target: np.ndarray, prediction: np.ndarray, valid: np.ndarray, session_ids: np.ndarray, horizon_index: int, recalls: list[float], *, midturn_pause_max_steps: int, operating_thresholds: dict[str, dict[float, float | None]] | None = None, threshold_source: str = "evaluated_sessions", ) -> dict[str, Any]: """Evaluate each aggregate-other boundary once with competing risks. Other offset events ask whether an inactive wearer starts within the horizon but before the next other onset. Other onset events ask whether an active wearer stops within the horizon but before the next other offset. Competing boundary rows are excluded, so simultaneous wearer changes are not attributed to the earlier anchor. The labels remain observed-behavior proxies: aggregate activity cannot establish addressee, speaker identity, or semantic EOT/INT. """ horizon_steps = horizon_index + 1 wearer = target[:, WEARER_INDEX].astype(bool) others = target[:, OTHERS_INDEX].astype(bool) response_scores: list[float] = [] response_labels: list[bool] = [] short_resume_groups: list[bool] = [] response_effective_steps: list[int] = [] response_competing_censored = 0 response_right_censored = 0 response_invalid_score = 0 response_source_boundaries = 0 offset_sources = np.flatnonzero(others[:-1] & ~others[1:]) for source_row in offset_sources: if session_ids[source_row] != session_ids[source_row + 1]: continue if wearer[source_row]: continue response_source_boundaries += 1 window = _competing_risk_window( others, session_ids, int(source_row), horizon_steps, competing_rising=True, ) if window is None: response_right_censored += 1 continue effective_steps, competing_row, nominal_complete = window response_row = _first_transition_row( wearer, int(source_row), effective_steps, rising=True, ) if response_row is None and competing_row is None and not nominal_complete: response_right_censored += 1 continue score = _same_source_path_score( prediction, valid, session_ids, int(source_row), effective_steps, offset=False, ) if score is None: response_invalid_score += 1 continue response = response_row is not None response_scores.append(score) response_labels.append(response) response_effective_steps.append(effective_steps) response_competing_censored += int( competing_row is not None and response_row is None ) simultaneous_wearer_entry = bool( competing_row is not None and not wearer[competing_row - 1] and wearer[competing_row] ) short_resume_groups.append( (not response) and competing_row is not None and competing_row <= source_row + 1 + midturn_pause_max_steps and not simultaneous_wearer_entry ) yield_scores: list[float] = [] yield_labels: list[bool] = [] yield_effective_steps: list[int] = [] yield_competing_censored = 0 yield_right_censored = 0 yield_invalid_score = 0 yield_source_boundaries = 0 onset_sources = np.flatnonzero(~others[:-1] & others[1:]) for source_row in onset_sources: if session_ids[source_row] != session_ids[source_row + 1]: continue if not wearer[source_row]: continue yield_source_boundaries += 1 window = _competing_risk_window( others, session_ids, int(source_row), horizon_steps, competing_rising=False, ) if window is None: yield_right_censored += 1 continue effective_steps, competing_row, nominal_complete = window yield_row = _first_transition_row( wearer, int(source_row), effective_steps, rising=False, ) if yield_row is None and competing_row is None and not nominal_complete: yield_right_censored += 1 continue score = _same_source_path_score( prediction, valid, session_ids, int(source_row), effective_steps, offset=True, ) if score is None: yield_invalid_score += 1 continue yielded = yield_row is not None yield_scores.append(score) yield_labels.append(yielded) yield_effective_steps.append(effective_steps) yield_competing_censored += int( competing_row is not None and yield_row is None ) response_metric = _event_metric( response_scores, response_labels, recalls, extra_negative_groups={ "short_resume_hard_negative": short_resume_groups, }, fixed_thresholds=( None if operating_thresholds is None else operating_thresholds["response"] ), threshold_source=threshold_source, ) response_metric.update( _event_window_diagnostics( source_boundaries=response_source_boundaries, effective_steps=response_effective_steps, competing_censored=response_competing_censored, right_censored=response_right_censored, invalid_score=response_invalid_score, ) ) yield_metric = _event_metric( yield_scores, yield_labels, recalls, fixed_thresholds=( None if operating_thresholds is None else operating_thresholds["yield"] ), threshold_source=threshold_source, ) yield_metric.update( _event_window_diagnostics( source_boundaries=yield_source_boundaries, effective_steps=yield_effective_steps, competing_censored=yield_competing_censored, right_censored=yield_right_censored, invalid_score=yield_invalid_score, ) ) return { "horizon_index": horizon_steps, "nominal_seconds": horizon_steps * STEP_SECONDS, "other_offset_to_wearer_response_event": response_metric, "other_onset_to_wearer_yield_event": yield_metric, } def _mean_available(items: Iterable[float | None]) -> float | None: values = [float(item) for item in items if item is not None] return float(np.mean(values)) if values else None def summarize_horizons(horizons: list[dict[str, Any]]) -> dict[str, float | None]: return { "mean_onset_endpoint_ap": _mean_available( item["wearer_off_to_on_endpoint"]["average_precision"] for item in horizons ), "mean_offset_endpoint_ap": _mean_available( item["wearer_on_to_off_endpoint"]["average_precision"] for item in horizons ), "mean_eot_hold_proxy_ap": _mean_available( item["eot_response_vs_other_hold_proxy"]["average_precision"] for item in horizons ), "mean_other_entry_yield_proxy_ap": _mean_available( item["other_entry_yield_vs_continue_proxy"]["average_precision"] for item in horizons ), } def summarize_event_horizons( horizons: list[dict[str, Any]], ) -> dict[str, float | None]: return { "mean_other_offset_response_event_ap": _mean_available( item["other_offset_to_wearer_response_event"]["average_precision"] for item in horizons ), "mean_other_onset_yield_event_ap": _mean_available( item["other_onset_to_wearer_yield_event"]["average_precision"] for item in horizons ), } def _percentage(value: float | None) -> str: return "n/a" if value is None else f"{value * 100:.3f}" def _operating_point( metric: dict[str, Any], requested: float ) -> dict[str, Any] | None: for point in metric["operating_points"]: if math.isclose(point["requested_positive_recall"], requested): return point return None def _event_operating_thresholds( event_horizon: dict[str, Any], recalls: list[float] ) -> dict[str, dict[float, float | None]]: result: dict[str, dict[float, float | None]] = { "response": {}, "yield": {}, } for recall in recalls: response_point = _operating_point( event_horizon["other_offset_to_wearer_response_event"], recall ) yield_point = _operating_point( event_horizon["other_onset_to_wearer_yield_event"], recall ) assert response_point is not None and yield_point is not None result["response"][recall] = response_point["threshold"] result["yield"][recall] = yield_point["threshold"] return result def _framewise_operating_thresholds( horizon: dict[str, Any], recalls: list[float] ) -> dict[str, dict[float, float | None]]: result: dict[str, dict[float, float | None]] = { "response": {}, "yield": {}, } for recall in recalls: response_point = _operating_point( horizon["eot_response_vs_other_hold_proxy"], recall ) yield_point = _operating_point( horizon["other_entry_yield_vs_continue_proxy"], recall ) assert response_point is not None and yield_point is not None result["response"][recall] = response_point["threshold"] result["yield"][recall] = yield_point["threshold"] return result def print_event_timeline_examples( target: np.ndarray, prediction: np.ndarray, valid: np.ndarray, session_ids: np.ndarray, session_names: list[str], *, examples_per_type: int, horizon_steps: int = NUM_HORIZONS, ) -> None: """Print causal event windows for manual attribution inspection.""" if examples_per_type <= 0: return wearer = target[:, WEARER_INDEX].astype(bool) others = target[:, OTHERS_INDEX].astype(bool) session_starts = { session_id: int(np.flatnonzero(session_ids == session_id)[0]) for session_id in np.unique(session_ids) } specifications = ( ( "response", np.flatnonzero(others[:-1] & ~others[1:]), False, True, True, False, ), ( "yield", np.flatnonzero(~others[:-1] & others[1:]), True, False, False, True, ), ) print("\nCompeting-risk event timeline audit (first run, 2-s horizon)") print("C=competing boundary excluded from the earlier event; E=wearer outcome") for ( event_name, sources, required_wearer_state, competing_rising, wearer_rising, offset_score, ) in specifications: printed = 0 for source_value in sources: source_row = int(source_value) if session_ids[source_row] != session_ids[source_row + 1]: continue if bool(wearer[source_row]) != required_wearer_state: continue window = _competing_risk_window( others, session_ids, source_row, horizon_steps, competing_rising=competing_rising, ) if window is None: continue effective_steps, competing_row, nominal_complete = window outcome_row = _first_transition_row( wearer, source_row, effective_steps, rising=wearer_rising, ) if outcome_row is not None: status = "positive" elif competing_row is not None: status = "competing-censored negative" elif nominal_complete: status = "negative" else: status = "right-censored (excluded)" event_score = ( None if status.startswith("right-censored") else _same_source_path_score( prediction, valid, session_ids, source_row, effective_steps, offset=offset_score, ) ) session_id = int(session_ids[source_row]) local_source = source_row - session_starts[session_id] display_last = source_row + effective_steps if competing_row is not None: display_last = max(display_last, competing_row) print( f"\n{event_name} | {session_names[session_id]} | " f"source={local_source} ({local_source * STEP_SECONDS:.1f}s) | " f"status={status} | score={event_score}" ) print("local_row | time_s | wearer | others | fixed-source score | mark") for row in range(source_row, display_last + 1): local_row = row - session_starts[session_id] step = row - source_row if step == 0: score_text = "source" elif row == competing_row: score_text = "excluded" elif step <= NUM_HORIZONS and valid[row, step - 1]: raw_score = float(prediction[row, WEARER_INDEX, step - 1]) score_text = f"{(1.0 - raw_score) if offset_score else raw_score:.3f}" else: score_text = "invalid" marks: list[str] = [] if row == source_row: marks.append("anchor") if row == outcome_row: marks.append("E") if row == competing_row: marks.append("C/excluded") print( f"{local_row:9d} | {local_row * STEP_SECONDS:6.1f} | " f"{int(wearer[row]):6d} | {int(others[row]):6d} | " f"{score_text:18s} | {','.join(marks)}" ) printed += 1 if printed >= examples_per_type: break if printed == 0: print(f"\n{event_name} | no eligible example") def print_results(payload: dict[str, Any], *, print_framewise: bool) -> None: report_recall = min( payload["matched_recalls"], key=lambda value: abs(value - 0.8) ) print("\nWearer floor-event evaluation (primary event-level protocol)") print( "Each aggregate-other boundary is counted once. Response windows stop " "before the next other onset; yield windows stop before the next other " "offset. The competing boundary itself is excluded. AP/rates are " "percentages; events remain timing proxies, not semantic EOT/INT." ) if payload["calibration_sessions"]: print( "Operating thresholds were selected on: " + ", ".join(payload["calibration_sessions"]) ) else: print( "Operating thresholds were selected on the evaluated sessions " "(exploratory/in-sample)." ) print( "run | h | other-offset response AP | response/no-response/short-resume | " f"achieved response recall @ requested {report_recall:.0%} | " "no-response trigger rate(count) | short-resume trigger rate(count) | " "other-onset yield AP | yield/continue | achieved yield recall | " "continue false-stop rate(count)" ) for name, run in payload["runs"].items(): for item in run["event_horizons"]: response = item["other_offset_to_wearer_response_event"] response_point = _operating_point(response, report_recall) yield_metric = item["other_onset_to_wearer_yield_event"] yield_point = _operating_point(yield_metric, report_recall) print( f"{name} | {item['nominal_seconds']:.1f}s | " f"{_percentage(response['average_precision'])} | " f"{response['num_positive']}/{response['num_negative']}/" f"{response['num_short_resume_hard_negative']} | " f"{_percentage(None if response_point is None else response_point['achieved_positive_recall'])} | " f"{_percentage(None if response_point is None else response_point['negative_false_positive_rate'])} " f"({None if response_point is None else response_point['negative_false_positive_count']}) | " f"{_percentage(None if response_point is None else response_point.get('short_resume_hard_negative'))} " f"({None if response_point is None else response_point.get('short_resume_hard_negative_trigger_count')}) | " f"{_percentage(yield_metric['average_precision'])} | " f"{yield_metric['num_positive']}/{yield_metric['num_negative']} | " f"{_percentage(None if yield_point is None else yield_point['achieved_positive_recall'])} | " f"{_percentage(None if yield_point is None else yield_point['negative_false_positive_rate'])} " f"({None if yield_point is None else yield_point['negative_false_positive_count']})" ) summary = run["event_summary"] print( f"{name} event mean | response=" f"{_percentage(summary['mean_other_offset_response_event_ap'])}, " f"yield={_percentage(summary['mean_other_onset_yield_event_ap'])}" ) if not print_framewise: return print("\nWearer endpoint/framewise evaluation (secondary diagnostic)") print("AP/rates are percentages; EOT/HOLD and yield are temporal proxies.") print( "run | h | OFF->ON AP | ON->OFF AP | EOT/HOLD AP | " f"HOLD false-entry @ EOT recall {report_recall:.0%} | " "other-entry yield AP | proxy counts EOT/HOLD/yield/continue" ) for name, run in payload["runs"].items(): for item in run["horizons"]: floor_metric = item["eot_response_vs_other_hold_proxy"] yield_metric = item["other_entry_yield_vs_continue_proxy"] point = _operating_point(floor_metric, report_recall) hold_rate = ( point["negative_false_positive_rate"] if point is not None else None ) print( f"{name} | {item['nominal_seconds']:.1f}s | " f"{_percentage(item['wearer_off_to_on_endpoint']['average_precision'])} | " f"{_percentage(item['wearer_on_to_off_endpoint']['average_precision'])} | " f"{_percentage(floor_metric['average_precision'])} | " f"{_percentage(hold_rate)} | " f"{_percentage(yield_metric['average_precision'])} | " f"{floor_metric['num_positive']}/{floor_metric['num_negative']}/" f"{yield_metric['num_positive']}/{yield_metric['num_negative']}" ) summary = run["summary"] print( f"{name} mean | onset={_percentage(summary['mean_onset_endpoint_ap'])}, " f"offset={_percentage(summary['mean_offset_endpoint_ap'])}, " f"EOT/HOLD={_percentage(summary['mean_eot_hold_proxy_ap'])}, " f"yield={_percentage(summary['mean_other_entry_yield_proxy_ap'])}" ) def write_framewise_csv(path: Path, payload: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", newline="") as file: fieldnames = [ "run", "horizon_index", "nominal_seconds", "onset_endpoint_ap", "offset_endpoint_ap", "eot_hold_proxy_ap", "eot_count", "hold_count", "early_entry_count", "other_entry_yield_proxy_ap", "yield_count", "continue_count", ] for recall in payload["matched_recalls"]: fieldnames += [ f"eot_threshold_{recall:g}", f"eot_threshold_source_{recall:g}", f"eot_recall_{recall:g}", f"hold_false_entry_rate_at_eot_recall_{recall:g}", f"other_entry_yield_threshold_{recall:g}", f"other_entry_yield_threshold_source_{recall:g}", f"other_entry_yield_recall_{recall:g}", f"continue_false_stop_rate_at_yield_recall_{recall:g}", ] writer = csv.DictWriter(file, fieldnames=fieldnames) writer.writeheader() for name, run in payload["runs"].items(): for item in run["horizons"]: floor_metric = item["eot_response_vs_other_hold_proxy"] yield_metric = item["other_entry_yield_vs_continue_proxy"] row: dict[str, Any] = { "run": name, "horizon_index": item["horizon_index"], "nominal_seconds": item["nominal_seconds"], "onset_endpoint_ap": item["wearer_off_to_on_endpoint"]["average_precision"], "offset_endpoint_ap": item["wearer_on_to_off_endpoint"]["average_precision"], "eot_hold_proxy_ap": floor_metric["average_precision"], "eot_count": floor_metric["num_positive"], "hold_count": floor_metric["num_negative"], "early_entry_count": floor_metric["num_early_entry_proxy"], "other_entry_yield_proxy_ap": yield_metric["average_precision"], "yield_count": yield_metric["num_positive"], "continue_count": yield_metric["num_negative"], } for recall in payload["matched_recalls"]: response_point = _operating_point(floor_metric, recall) yield_point = _operating_point(yield_metric, recall) assert response_point is not None and yield_point is not None row[f"eot_threshold_{recall:g}"] = response_point["threshold"] row[f"eot_threshold_source_{recall:g}"] = response_point[ "threshold_source" ] row[f"eot_recall_{recall:g}"] = response_point[ "achieved_positive_recall" ] row[f"hold_false_entry_rate_at_eot_recall_{recall:g}"] = ( response_point["negative_false_positive_rate"] ) row[f"other_entry_yield_threshold_{recall:g}"] = yield_point[ "threshold" ] row[ f"other_entry_yield_threshold_source_{recall:g}" ] = yield_point["threshold_source"] row[f"other_entry_yield_recall_{recall:g}"] = yield_point[ "achieved_positive_recall" ] row[ f"continue_false_stop_rate_at_yield_recall_{recall:g}" ] = yield_point["negative_false_positive_rate"] writer.writerow(row) def write_event_csv(path: Path, payload: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", newline="") as file: fieldnames = [ "run", "horizon_index", "nominal_seconds", "response_event_ap", "response_count", "no_response_count", "short_resume_hard_negative_count", "response_source_boundary_count", "response_competing_censored_count", "response_right_censored_count", "response_invalid_score_window_count", "response_mean_effective_window_seconds", "yield_event_ap", "yield_count", "continue_count", "yield_source_boundary_count", "yield_competing_censored_count", "yield_right_censored_count", "yield_invalid_score_window_count", "yield_mean_effective_window_seconds", ] for recall in payload["matched_recalls"]: fieldnames += [ f"response_threshold_{recall:g}", f"response_threshold_source_{recall:g}", f"response_recall_{recall:g}", f"no_response_trigger_rate_at_recall_{recall:g}", f"no_response_trigger_count_at_recall_{recall:g}", f"short_resume_trigger_rate_at_recall_{recall:g}", f"short_resume_trigger_count_at_recall_{recall:g}", f"yield_threshold_{recall:g}", f"yield_threshold_source_{recall:g}", f"yield_recall_{recall:g}", f"continue_false_stop_rate_at_recall_{recall:g}", f"continue_false_stop_count_at_recall_{recall:g}", ] writer = csv.DictWriter(file, fieldnames=fieldnames) writer.writeheader() for name, run in payload["runs"].items(): for item in run["event_horizons"]: response = item["other_offset_to_wearer_response_event"] yield_metric = item["other_onset_to_wearer_yield_event"] row: dict[str, Any] = { "run": name, "horizon_index": item["horizon_index"], "nominal_seconds": item["nominal_seconds"], "response_event_ap": response["average_precision"], "response_count": response["num_positive"], "no_response_count": response["num_negative"], "short_resume_hard_negative_count": response[ "num_short_resume_hard_negative" ], "response_source_boundary_count": response[ "num_source_boundaries" ], "response_competing_censored_count": response[ "num_competing_risk_censored" ], "response_right_censored_count": response[ "num_right_censored" ], "response_invalid_score_window_count": response[ "num_invalid_score_window" ], "response_mean_effective_window_seconds": response[ "mean_effective_window_seconds" ], "yield_event_ap": yield_metric["average_precision"], "yield_count": yield_metric["num_positive"], "continue_count": yield_metric["num_negative"], "yield_source_boundary_count": yield_metric[ "num_source_boundaries" ], "yield_competing_censored_count": yield_metric[ "num_competing_risk_censored" ], "yield_right_censored_count": yield_metric[ "num_right_censored" ], "yield_invalid_score_window_count": yield_metric[ "num_invalid_score_window" ], "yield_mean_effective_window_seconds": yield_metric[ "mean_effective_window_seconds" ], } for recall in payload["matched_recalls"]: response_point = _operating_point(response, recall) yield_point = _operating_point(yield_metric, recall) assert response_point is not None and yield_point is not None row[f"response_threshold_{recall:g}"] = response_point[ "threshold" ] row[f"response_threshold_source_{recall:g}"] = response_point[ "threshold_source" ] row[f"response_recall_{recall:g}"] = response_point[ "achieved_positive_recall" ] row[f"no_response_trigger_rate_at_recall_{recall:g}"] = ( response_point["negative_false_positive_rate"] ) row[f"no_response_trigger_count_at_recall_{recall:g}"] = ( response_point["negative_false_positive_count"] ) row[f"short_resume_trigger_rate_at_recall_{recall:g}"] = ( response_point["short_resume_hard_negative"] ) row[f"short_resume_trigger_count_at_recall_{recall:g}"] = ( response_point["short_resume_hard_negative_trigger_count"] ) row[f"yield_threshold_{recall:g}"] = yield_point["threshold"] row[f"yield_threshold_source_{recall:g}"] = yield_point[ "threshold_source" ] row[f"yield_recall_{recall:g}"] = yield_point[ "achieved_positive_recall" ] row[f"continue_false_stop_rate_at_recall_{recall:g}"] = ( yield_point["negative_false_positive_rate"] ) row[f"continue_false_stop_count_at_recall_{recall:g}"] = ( yield_point["negative_false_positive_count"] ) writer.writerow(row) def main() -> None: args = parse_args() runs = parse_runs(args.run) calibration_run_values = parse_runs(args.calibration_run or []) recalls = validate_recalls(args.matched_recalls) if args.midturn_pause_max_seconds <= 0: raise ValueError("--midturn-pause-max-seconds must be positive") if args.print_event_timelines < 0: raise ValueError("--print-event-timelines must be non-negative") midturn_pause_max_steps = max( 1, int(math.ceil(args.midturn_pause_max_seconds / STEP_SECONDS)) ) sessions = [normalize_session_name(value) for value in args.sessions] if len(sessions) != len(set(sessions)): raise ValueError("Duplicate sessions") calibration_sessions = ( [normalize_session_name(value) for value in args.calibration_sessions] if args.calibration_sessions else None ) if calibration_sessions and len(calibration_sessions) != len( set(calibration_sessions) ): raise ValueError("Duplicate calibration sessions") if calibration_run_values and calibration_sessions is None: raise ValueError("--calibration-run requires --calibration-sessions") run_names = {name for name, _ in runs} if calibration_run_values: calibration_names = {name for name, _ in calibration_run_values} if calibration_names != run_names: raise ValueError( "--calibration-run names must exactly match --run names; " f"runs={sorted(run_names)}, calibration={sorted(calibration_names)}" ) if calibration_sessions and set(calibration_sessions) & set(sessions): if not args.allow_in_sample_test_thresholds: raise ValueError( "Calibration and evaluation sessions overlap. Use disjoint " "training-side calibration sessions, or set the exploratory " "--allow-in-sample-test-thresholds override." ) if ( set(sessions) & set(OFFICIAL_TEST_SESSIONS) and calibration_sessions is None and not args.allow_in_sample_test_thresholds ): raise ValueError( "Official test evaluation requires --calibration-sessions so event " "thresholds are selected on training-side data. Use " "--allow-in-sample-test-thresholds only for an explicitly " "exploratory audit." ) if ( set(sessions) & set(OFFICIAL_TEST_SESSIONS) and calibration_sessions is not None and not set(calibration_sessions).issubset(OFFICIAL_TRAIN_SESSIONS) ): raise ValueError( "Official test thresholds may only be calibrated on training-side " "Sessions 4--12; got " + ", ".join(calibration_sessions) ) calibration_run_dirs = ( dict(calibration_run_values) if calibration_run_values else dict(runs) ) payload: dict[str, Any] = { "protocol": "wearer_floor_event_competing_risk_v2", "sessions": sessions, "calibration_sessions": calibration_sessions, "classes": list(CLASS_NAMES), "matched_recalls": recalls, "prediction_layout": "[target_time, class, future_horizon]", "current_state_source": ( "Ground-truth wearer/others activity at the source row. This is an " "oracle-current-state diagnostic, not a standalone event-hazard head." ), "interpretation": ( "OFF->ON and ON->OFF are endpoint state changes. EOT/HOLD, early-entry, " "and other-entry yield are timing-derived behavior proxies, not semantic " "judgments. Event operating points report false-entry/false-stop counts " "and rates at matched observed-behavior recall." ), "event_protocol": ( "Each aggregate-others 1->0 boundary with wearer initially off is one " "response candidate; each aggregate-others 0->1 boundary with wearer " "initially on is one yield candidate. A response window is half-open and " "ends before the next aggregate-others onset; a yield window ends before " "the next aggregate-others offset. A simultaneous wearer change at that " "competing boundary is excluded from the earlier event. Labels are " "observed wearer transitions within the remaining window. Fixed-source " "future activity scores are max-pooled over the same remaining window " "only as ranking scores. Candidates without enough follow-up and without " "an observed competing boundary are right-censored." ), "midturn_pause_max_seconds": args.midturn_pause_max_seconds, "runs": {}, } printed_timelines = False for name, run_dir in runs: target, prediction, valid, session_ids = load_arrays( args.target_root, run_dir, sessions ) if args.print_event_timelines and not printed_timelines: print_event_timeline_examples( target, prediction, valid, session_ids, sessions, examples_per_type=args.print_event_timelines, ) printed_timelines = True calibration_thresholds: list[dict[str, Any]] | None = None calibration_framewise_thresholds: list[dict[str, Any]] | None = None threshold_source = "evaluated_sessions" if calibration_sessions is not None: calibration_dir = calibration_run_dirs[name] ( calibration_target, calibration_prediction, calibration_valid, calibration_session_ids, ) = load_arrays( args.target_root, calibration_dir, calibration_sessions, ) threshold_source = "calibration_sessions:" + ",".join( calibration_sessions ) calibration_thresholds = [] calibration_framewise_thresholds = [] for horizon_index in range(NUM_HORIZONS): calibration_horizon = evaluate_horizon( calibration_target, calibration_prediction, calibration_valid, calibration_session_ids, horizon_index, recalls, ) framewise_thresholds = _framewise_operating_thresholds( calibration_horizon, recalls ) calibration_framewise_thresholds.append( { "horizon_index": horizon_index + 1, "nominal_seconds": (horizon_index + 1) * STEP_SECONDS, "response": framewise_thresholds["response"], "yield": framewise_thresholds["yield"], } ) calibration_event = evaluate_event_horizon( calibration_target, calibration_prediction, calibration_valid, calibration_session_ids, horizon_index, recalls, midturn_pause_max_steps=midturn_pause_max_steps, ) thresholds = _event_operating_thresholds( calibration_event, recalls ) calibration_thresholds.append( { "horizon_index": horizon_index + 1, "nominal_seconds": (horizon_index + 1) * STEP_SECONDS, "response": thresholds["response"], "yield": thresholds["yield"], } ) horizons: list[dict[str, Any]] = [] event_horizons: list[dict[str, Any]] = [] for horizon_index in range(NUM_HORIZONS): fixed_framewise_thresholds = None fixed_event_thresholds = None if calibration_thresholds is not None: assert calibration_framewise_thresholds is not None event_threshold_row = calibration_thresholds[horizon_index] framewise_threshold_row = calibration_framewise_thresholds[ horizon_index ] fixed_event_thresholds = { "response": event_threshold_row["response"], "yield": event_threshold_row["yield"], } fixed_framewise_thresholds = { "response": framewise_threshold_row["response"], "yield": framewise_threshold_row["yield"], } horizons.append( evaluate_horizon( target, prediction, valid, session_ids, horizon_index, recalls, operating_thresholds=fixed_framewise_thresholds, threshold_source=threshold_source, ) ) event_horizons.append( evaluate_event_horizon( target, prediction, valid, session_ids, horizon_index, recalls, midturn_pause_max_steps=midturn_pause_max_steps, operating_thresholds=fixed_event_thresholds, threshold_source=threshold_source, ) ) payload["runs"][name] = { "result_dir": str(run_dir.expanduser().resolve()), "calibration_result_dir": ( str(calibration_run_dirs[name].expanduser().resolve()) if calibration_sessions is not None else None ), "event_thresholds": calibration_thresholds, "framewise_thresholds": calibration_framewise_thresholds, "horizons": horizons, "summary": summarize_horizons(horizons), "event_horizons": event_horizons, "event_summary": summarize_event_horizons(event_horizons), } args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(payload, indent=2) + "\n") event_csv_path = args.csv_output or args.output.with_name( args.output.stem + "_events.csv" ) framewise_csv_path = args.output.with_name( args.output.stem + "_framewise.csv" ) write_event_csv(event_csv_path, payload) write_framewise_csv(framewise_csv_path, payload) print_results(payload, print_framewise=args.print_framewise) print(f"\nJSON: {args.output}") print(f"Event CSV: {event_csv_path}") print(f"Framewise CSV: {framewise_csv_path}") if __name__ == "__main__": main()