diff --git a/PROJECT_CONTEXT.md b/PROJECT_CONTEXT.md index 8d8fb33..c44d9a0 100644 --- a/PROJECT_CONTEXT.md +++ b/PROJECT_CONTEXT.md @@ -138,6 +138,14 @@ scenario definitions. These values remain timing-derived behavior proxies, not semantic EOT/INT or addressee ground truth. See the [`runbook`](docs/easycom_competing_risk_floor_evaluation.md). +Three explanation-oriented diagnostics are implemented but unmeasured: a +separately identified post-boundary reactive evaluator, a direct +competing-risk response/yield probe over the same frozen Omni/audio/oracle +inputs, and a privileged direct-to-wearer `Target_of_Speech` oracle with an +equal-width zero-channel control. They diagnose timing, objective and +addressee-information limitations respectively; none changes or freezes P0. +See [`docs/easycom_floor_event_diagnostics.md`](docs/easycom_floor_event_diagnostics.md). + ### Next experiment gates The user has temporarily chosen to run the dense-AP model screen while P0 is @@ -296,7 +304,7 @@ Do not reorganize them physically without compatibility wrappers. | `derive_easycom_interaction_events.py` | Deterministic transcript/timing proxies | | `adjudicate_easycom_interaction_events.py` | Optional ambiguous-event adjudication | | `audit_easycom_floor_events.py` | Count EOT/takeover/backchannel candidates | -| `prepare_easycom_oracle_features.py` | Build explicit past-only oracle person-activity sidecars for information-value screening | +| `prepare_easycom_oracle_features.py` | Build explicit past-only oracle person-activity sidecars and analysis-only addressee controls | ### Local features @@ -312,9 +320,13 @@ Do not reorganize them physically without compatibility wrappers. | `train_turn_head.py` | Frozen-backbone head training, optional local fusion/auxiliary losses | | `infer_turn_head.py` | Save EgoSpeak-compatible predictions and AP/mAP | | `evaluate_floor_state_changes.py` | CPU floor-state/event proxy evaluation | +| `train_easycom_floor_event_probe.py` | Analysis-only direct response/yield probe with shared competing-risk labels | +| `summarize_easycom_floor_event_probes.py` | Compare direct event-probe conditions under one evaluation contract | | `summarize_local_baselines.py` | Aggregate local baseline `metrics.json` files | | `run_easycom_track_gru_residual.sh` | Matched pooled and anonymous track-GRU training/inference launcher | | `run_easycom_oracle_information.sh` | Prepare and screen aggregate/per-person/shuffled/delayed oracle conditions on Session 12 | +| `run_easycom_floor_event_probes.sh` | Prepare, train and compare aggregate/addressee direct event probes | +| `run_easycom_post_boundary_eval.sh` | Compare existing dense runs after the aggregate-other boundary is observed | | `train_omni_turn_lora.py` | Thinker LoRA training from cached AV | | `eval_omni_turn_lora.py` | Thinker LoRA evaluation | diff --git a/docs/README.md b/docs/README.md index 7d2cbcd..d4934e3 100644 --- a/docs/README.md +++ b/docs/README.md @@ -32,6 +32,7 @@ tests. | [`easycom_muvap_stage3_baseline.md`](easycom_muvap_stage3_baseline.md) | MuVAP-only frozen VAP+ASD Stage-3 paper reconstruction on EgoSpeak/EasyCom targets | | [`easycom_oracle_information.md`](easycom_oracle_information.md) | Causal oracle per-person speaking-history upper bound and Session-12 screening commands | | [`easycom_competing_risk_floor_evaluation.md`](easycom_competing_risk_floor_evaluation.md) | Running and auditing the competing-risk entry/yield evaluator with training-side threshold calibration | +| [`easycom_floor_event_diagnostics.md`](easycom_floor_event_diagnostics.md) | Separating boundary observability, event-objective mismatch and addressee information with remote commands | ## Design history and implementation options diff --git a/docs/easycom_competing_risk_floor_evaluation.md b/docs/easycom_competing_risk_floor_evaluation.md index 3a69eff..d4a170f 100644 --- a/docs/easycom_competing_risk_floor_evaluation.md +++ b/docs/easycom_competing_risk_floor_evaluation.md @@ -46,6 +46,13 @@ For each event, predictions made from the same source time are max-pooled only over the observable event window. This is a ranking score, not a calibrated event probability. +The default `--decision-timing pre-boundary` uses the last row before the +aggregate-other transition and remains the primary anticipation protocol. The +optional `--decision-timing post-boundary` moves the decision source to the +first row after the transition. It is a separately identified reactive diagnostic, +not a replacement metric; see +[`easycom_floor_event_diagnostics.md`](easycom_floor_event_diagnostics.md). + ## Confirmatory test command Select each model's operating threshold on training-side predictions, then diff --git a/docs/easycom_floor_event_diagnostics.md b/docs/easycom_floor_event_diagnostics.md new file mode 100644 index 0000000..d307d43 --- /dev/null +++ b/docs/easycom_floor_event_diagnostics.md @@ -0,0 +1,142 @@ +# EasyCom floor-event diagnosis: timing, objective, and addressee + +These three experiments explain why a dense future-activity oracle can retain +reasonable mAP while response/yield event AP remains low. They are diagnostics, +not three proposed paper contributions. Keep the primary paper protocol as the +pre-boundary competing-risk evaluation until the evidence says otherwise. + +## What each experiment isolates + +| Diagnostic | Only intended change | If it improves clearly | +|---|---|---| +| pre-boundary vs post-boundary | whether the aggregate-other boundary has already been observed at the decision source | much of the error is boundary anticipation, not weak AV encoding | +| dense activity head vs direct event head | future-state loss vs response/yield loss, using the same frozen inputs | the dense objective/ranking conversion is the main mismatch | +| addressee vs zero-channel control | explicit past `Target_of_Speech == wearer_id` history | aggregate activity misses conversational routing information | + +Do not compare raw AP across the pre- and post-boundary protocols as if they +were the same task. Compare methods within one timing protocol, and use the +pre/post difference only as a difficulty decomposition. + +## Shared remote environment + +```bash +cd /data1/yfzx/wangx/Code/icassp2027-experiment + +export DATASET_ROOT=/data1/yfzx/wangx/Datasets/easycom/extracted_easycom +export FEATURE_ROOT=/data1/yfzx/wangx/Datasets/easycom/omni_features/open_stream_2fps_8s_last4_stack +export TARGET_ROOT=/data1/yfzx/wangx/Datasets/easycom/easycom_target_frame +export LOCAL_ROOT=/data1/yfzx/wangx/Datasets/easycom/omni_features/local_qwen_1s_encoded_tf4576 +export ORACLE_ROOT=/data1/yfzx/wangx/Datasets/easycom/omni_features/oracle_person_history_2s +``` + +Change only paths that differ on the remote host. The default feature mode is +`open-stream-replay`. + +## 1. Post-boundary reactive diagnostic for the three existing runs + +Point the wrapper at already aligned inference directories. Each directory +must contain `pred/Session_*.npy` and `valid/Session_*.npy`. + +```bash +export POOLED_AV_RUN=/path/to/pooled_av_result +export ORACLE_RUN=/path/to/aggregate_visible_oracle_result +export MINIROAD_RUN=/path/to/converted_miniroada_result + +bash run_easycom_post_boundary_eval.sh +``` + +Without `CALIBRATION_SESSIONS`, the wrapper explicitly marks threshold results +as exploratory and selects them in-sample. AP is still useful for this first +diagnosis. For a confirmatory result, provide training-side predictions in the +same run directories and use, for example: + +```bash +export CALIBRATION_SESSIONS="Session_12" +bash run_easycom_post_boundary_eval.sh +``` + +The JSON protocol ID becomes +`wearer_floor_event_post_boundary_diagnostic_v1`. Simultaneous wearer changes +at the observed boundary and zero-length windows before an immediate competing +boundary are excluded and counted explicitly. + +## 2. Direct response/yield event head + +The direct probe uses frozen Omni states, the same mixed-audio vectors and the +same `aggregate-visible` past-activity oracle. Its layer fusion, audio +projection, oracle person attention and shared MLP trunk are the same +`TurnPredictionHead` components as the dense oracle; only the final two-event +readout and event loss differ. It trains on Sessions 4--11, selects epoch and +thresholds on Session 12, and initially evaluates only Session 12. First run a +one-epoch Session-11/12 shape, mask and checkpoint smoke test: + +```bash +CUDA_VISIBLE_DEVICES=0 bash run_easycom_floor_event_probes.sh smoke +``` + +Then run the single aggregate oracle diagnosis: + +```bash +CUDA_VISIBLE_DEVICES=0 \ +VARIANTS="aggregate-visible" \ +bash run_easycom_floor_event_probes.sh all 0 +``` + +This is the clean comparison against the existing dense aggregate-visible +oracle head. A large direct-head gain supports objective mismatch; a small gain +means the oracle information itself is probably weak for the event decision. + +## 3. Addressee oracle with a capacity-matched control + +Run all three direct-probe variants: + +```bash +CUDA_VISIBLE_DEVICES=0 bash run_easycom_floor_event_probes.sh all 0 +``` + +The variants are: + +- `aggregate-visible`: 2 s aggregate activity plus observation mask; +- `aggregate-visible-address-control`: the same input plus an all-zero channel; +- `aggregate-visible-addressee`: the same input plus past bins in which a + visible non-wearer was explicitly annotated as addressing wearer ID 2. + +`Group`, `None`, unknown and other-person targets are not coerced into reliable +negative labels. The addressee channel is privileged, annotation-derived and +analysis-only; it is not an online feature or a deployable causal input. + +The runner writes one `summary.json` and `metrics.csv` per variant and a compact +`comparison_dev.csv`. Inspect that comparison before touching Sessions 1--3. + +## Optional frozen test evaluation + +Only after the variants, epochs and interpretation rule are fixed from Session +12, apply the saved probes to official test Sessions 1--3: + +```bash +CUDA_VISIBLE_DEVICES=0 bash run_easycom_floor_event_probes.sh test 0 +``` + +This reuses the saved Session-12 thresholds and writes `comparison_test.csv`. +Do not repeatedly inspect test results and redesign the probe. + +## Minimum interpretation rule + +For response, jointly inspect mean AP, 2 s AP, achieved response recall, false +entry rate, and short-resume trigger rate. For yield, inspect mean AP, 2 s AP, +achieved yield recall, and false-stop rate. Denominators and censoring counts +must match before comparing variants. + +The three outcomes lead to different paper decisions: + +1. **Only post-boundary improves:** retain the main anticipation task and frame + the difficulty as predicting a boundary before it is observable. Do not + claim that a larger event head solves grounding. +2. **Direct event supervision improves:** the main lesson is target alignment; + build the method around event-aware adaptation and keep dense mAP secondary. +3. **Addressee improves over both aggregate and zero control:** conversational + routing is the missing information. This motivates a deployable addressee or + addressedness estimator, but the oracle itself is not the method. +4. **None improves:** the current timing-derived event definition or available + observations are insufficient. Revisit annotation/task definition before a + heavier fusion architecture. diff --git a/docs/easycom_oracle_information.md b/docs/easycom_oracle_information.md index cd41f6e..31e5342 100644 --- a/docs/easycom_oracle_information.md +++ b/docs/easycom_oracle_information.md @@ -15,10 +15,12 @@ This is an information upper-bound experiment, not a deployable input setting. `0.2, 0.4, 0.6, ...` second cutoffs, with `Start_Frame <= marker < End_Frame`. This is right-edge point sampling, not an any-overlap reduction over a 200 ms bin. -- Each person vector stores the last 2 seconds of binary activity followed by - a visibility/availability mask. Activity is exposed only in bins where that - participant was visible according to the provided boxes. No future bin, - future wearer state, transcript text or `Target_of_Speech` value is input. +- Each original person vector stores the last 2 seconds of binary activity + followed by a visibility/availability mask. Activity is exposed only in bins + where that participant was visible according to the provided boxes. No future + bin, future wearer state or transcript text is input. The separately named + addressee diagnostic below additionally uses privileged `Target_of_Speech`; + it must not be described as an online/deployable feature. - Only non-wearer participant IDs present in the latest provided-box frame strictly before the cutoff receive slots. IDs route annotations during preparation but are not exposed as embeddings to the head. @@ -35,6 +37,8 @@ This is an information upper-bound experiment, not a deployable input setting. | `audio` | Omni `G` plus the existing recent local audio `A` | deployable baseline | | `visibility-only` | current visible-person set, but all activity bits are zero | parameter/person-count control | | `aggregate-visible` | one OR-pooled history over currently visible people | aggregate floor-information upper bound | +| `aggregate-visible-address-control` | aggregate-visible plus an all-zero history channel | capacity control for addressee input | +| `aggregate-visible-addressee` | aggregate-visible plus past direct-to-wearer annotation bins | analysis-only addressee information upper bound | | `per-person` | one correct activity history per currently visible participant | main oracle information test | | `shuffled-id` | preserves the number active at every lag but independently reassigns activity between people | destroys identity continuity | | `delayed` | correct per-person histories delayed by 1 second | destroys current temporal alignment without future leakage | @@ -43,6 +47,12 @@ The decisive comparison is `per-person` against `aggregate-visible`, `shuffled-id` and `delayed`, not only against `audio`. A gain over `audio` alone could come from perfect aggregate VAD rather than person continuity. +For the separate addressee question, the decisive comparison is +`aggregate-visible-addressee` against both `aggregate-visible` and +`aggregate-visible-address-control`. Group/None/unknown targets are audited and +not treated as trustworthy semantic negatives. The full diagnostic commands +are in [`easycom_floor_event_diagnostics.md`](easycom_floor_event_diagnostics.md). + ## Remote preparation and one-seed Session-12 screen The source local sidecars provide only mixed audio and the visibility/box diff --git a/evaluate_floor_state_changes.py b/evaluate_floor_state_changes.py index 9987ec3..4d30079 100644 --- a/evaluate_floor_state_changes.py +++ b/evaluate_floor_state_changes.py @@ -16,10 +16,11 @@ from __future__ import annotations import argparse import csv +from dataclasses import dataclass import json import math from pathlib import Path -from typing import Any, Iterable +from typing import Any, Iterable, Literal import numpy as np @@ -39,6 +40,24 @@ NUM_CLASSES = 3 WEARER_INDEX = 1 OTHERS_INDEX = 2 +DECISION_TIMINGS = ("pre-boundary", "post-boundary") +EventType = Literal["response", "yield"] + + +@dataclass(frozen=True) +class FloorEventCandidate: + """One observable boundary-conditioned event example. + + ``source_row`` is the row whose model state is allowed to make the + decision. It is the last pre-transition row in the primary anticipation + protocol and the first post-transition row in the reactive diagnostic. + """ + + source_row: int + label: bool + effective_steps: int + competing_row: int | None + short_resume_hard_negative: bool = False def normalize_session_name(value: str | int) -> str: @@ -125,6 +144,16 @@ def parse_args() -> argparse.Namespace: "short-resume hard-negative subgroup (default: 2.0)." ), ) + parser.add_argument( + "--decision-timing", + choices=DECISION_TIMINGS, + default="pre-boundary", + help=( + "pre-boundary keeps the primary anticipation protocol; " + "post-boundary is a reactive diagnostic whose source has already " + "observed the aggregate-other boundary (default: pre-boundary)." + ), + ) parser.add_argument( "--print-framewise", action="store_true", @@ -523,6 +552,7 @@ def _competing_risk_window( horizon_steps: int, *, competing_rising: bool, + first_competing_step: int = 2, ) -> tuple[int, int | None, bool] | None: """Return the observable half-open event window after ``source_row``. @@ -550,10 +580,15 @@ def _competing_risk_window( ): available_last_row -= 1 - # Row source+1 contains the anchor's post-transition state. A competing - # transition can occur no earlier than source+2. + if first_competing_step not in {1, 2}: + raise ValueError("first_competing_step must be 1 or 2") + + # In the primary pre-boundary protocol, row source+1 contains the anchor's + # post-transition state and a competing transition can occur no earlier + # than source+2. In the post-boundary diagnostic, source is already the + # post-transition row, so a competitor may occur at source+1. competing_row: int | None = None - for row in range(source_row + 2, available_last_row + 1): + for row in range(source_row + first_competing_step, 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 ( @@ -565,14 +600,140 @@ def _competing_risk_window( 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 collect_floor_event_candidates( + target: np.ndarray, + session_ids: np.ndarray, + horizon_index: int, + *, + event_type: EventType, + decision_timing: str = "pre-boundary", + midturn_pause_max_steps: int = NUM_HORIZONS, +) -> tuple[list[FloorEventCandidate], dict[str, int]]: + """Build label-only event examples without reading model predictions. + + This is shared by the dense-activity evaluator and the direct event-head + diagnostic so their anchors, horizons, competing risks and censoring are + identical. The post-boundary mode excludes a wearer transition that is + simultaneous with the observed aggregate-other boundary: that outcome has + already happened before the reactive decision source. + """ + + if decision_timing not in DECISION_TIMINGS: + raise ValueError(f"Unknown decision timing: {decision_timing}") + if event_type not in {"response", "yield"}: + raise ValueError(f"Unknown event type: {event_type}") + if horizon_index < 0: + raise ValueError("horizon_index must be non-negative") + if midturn_pause_max_steps <= 0: + raise ValueError("midturn_pause_max_steps must be positive") + if target.ndim != 2 or target.shape[1] != NUM_CLASSES: + raise ValueError(f"Expected target [T,{NUM_CLASSES}], got {target.shape}") + if session_ids.shape != (len(target),): + raise ValueError("session_ids must be [T]") + + horizon_steps = horizon_index + 1 + wearer = target[:, WEARER_INDEX].astype(bool) + others = target[:, OTHERS_INDEX].astype(bool) + if event_type == "response": + pre_sources = np.flatnonzero(others[:-1] & ~others[1:]) + required_wearer_state = False + wearer_rising = True + competing_rising = True + else: + pre_sources = np.flatnonzero(~others[:-1] & others[1:]) + required_wearer_state = True + wearer_rising = False + competing_rising = False + + candidates: list[FloorEventCandidate] = [] + source_boundaries = 0 + right_censored = 0 + simultaneous_boundary_outcomes_excluded = 0 + zero_window_competing_risk = 0 + for raw_pre_source in pre_sources: + pre_source = int(raw_pre_source) + boundary_row = pre_source + 1 + if session_ids[pre_source] != session_ids[boundary_row]: + continue + if bool(wearer[pre_source]) != required_wearer_state: + continue + source_boundaries += 1 + + source_row = pre_source + first_competing_step = 2 + if decision_timing == "post-boundary": + source_row = boundary_row + first_competing_step = 1 + if bool(wearer[source_row]) != required_wearer_state: + simultaneous_boundary_outcomes_excluded += 1 + continue + + window = _competing_risk_window( + others, + session_ids, + source_row, + horizon_steps, + competing_rising=competing_rising, + first_competing_step=first_competing_step, + ) + if window is None: + right_censored += 1 + continue + effective_steps, competing_row, nominal_complete = window + if effective_steps == 0: + # The floor changed again in the first post-decision bin. There is + # no attributable target row before that competing boundary. + zero_window_competing_risk += 1 + continue + outcome_row = _first_transition_row( + wearer, + source_row, + effective_steps, + rising=wearer_rising, + ) + if outcome_row is None and competing_row is None and not nominal_complete: + right_censored += 1 + continue + + label = outcome_row is not None + simultaneous_competing_outcome = bool( + competing_row is not None + and bool(wearer[competing_row - 1]) == required_wearer_state + and bool(wearer[competing_row]) != required_wearer_state + ) + short_resume = bool( + event_type == "response" + and not label + and competing_row is not None + and competing_row <= boundary_row + midturn_pause_max_steps + and not simultaneous_competing_outcome + ) + candidates.append( + FloorEventCandidate( + source_row=source_row, + label=label, + effective_steps=effective_steps, + competing_row=competing_row, + short_resume_hard_negative=short_resume, + ) + ) + + return candidates, { + "num_source_boundaries": source_boundaries, + "num_right_censored": right_censored, + "num_simultaneous_boundary_outcomes_excluded": ( + simultaneous_boundary_outcomes_excluded + ), + "num_zero_window_competing_risk": zero_window_competing_risk, + } + + def _event_metric( scores: list[float], labels: list[bool], @@ -611,6 +772,8 @@ def _event_window_diagnostics( competing_censored: int, right_censored: int, invalid_score: int, + simultaneous_boundary_outcomes_excluded: int = 0, + zero_window_competing_risk: int = 0, ) -> dict[str, Any]: steps = np.asarray(effective_steps, dtype=np.float64) evaluated = int(len(steps)) @@ -623,6 +786,10 @@ def _event_window_diagnostics( ), "num_right_censored": int(right_censored), "num_invalid_score_window": int(invalid_score), + "num_simultaneous_boundary_outcomes_excluded": int( + simultaneous_boundary_outcomes_excluded + ), + "num_zero_window_competing_risk": int(zero_window_competing_risk), "mean_effective_window_seconds": ( float(steps.mean() * STEP_SECONDS) if evaluated else None ), @@ -644,6 +811,7 @@ def evaluate_event_horizon( recalls: list[float], *, midturn_pause_max_steps: int, + decision_timing: str = "pre-boundary", operating_thresholds: dict[str, dict[float, float | None]] | None = None, threshold_source: str = "evaluated_sessions", ) -> dict[str, Any]: @@ -659,126 +827,78 @@ def evaluate_event_horizon( """ horizon_steps = horizon_index + 1 - wearer = target[:, WEARER_INDEX].astype(bool) - others = target[:, OTHERS_INDEX].astype(bool) + response_candidates, response_diagnostics = collect_floor_event_candidates( + target, + session_ids, + horizon_index, + event_type="response", + decision_timing=decision_timing, + midturn_pause_max_steps=midturn_pause_max_steps, + ) + yield_candidates, yield_diagnostics = collect_floor_event_candidates( + target, + session_ids, + horizon_index, + event_type="yield", + decision_timing=decision_timing, + midturn_pause_max_steps=midturn_pause_max_steps, + ) - 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), + def score_candidates( + candidates: list[FloorEventCandidate], + *, + offset: bool, + ) -> tuple[list[float], list[bool], list[int], list[bool], int, int]: + scores: list[float] = [] + labels: list[bool] = [] + effective_steps: list[int] = [] + short_resume: list[bool] = [] + competing_censored = 0 + invalid_score = 0 + for candidate in candidates: + score = _same_source_path_score( + prediction, + valid, + session_ids, + candidate.source_row, + candidate.effective_steps, + offset=offset, + ) + if score is None: + invalid_score += 1 + continue + scores.append(score) + labels.append(candidate.label) + effective_steps.append(candidate.effective_steps) + short_resume.append(candidate.short_resume_hard_negative) + competing_censored += int( + candidate.competing_row is not None and not candidate.label + ) + return ( + scores, + labels, 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 + short_resume, + competing_censored, + invalid_score, ) - 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_scores, + response_labels, + response_effective_steps, + short_resume_groups, + response_competing_censored, + response_invalid_score, + ) = score_candidates(response_candidates, offset=False) + ( + yield_scores, + yield_labels, + yield_effective_steps, + _yield_short_resume, + yield_competing_censored, + yield_invalid_score, + ) = score_candidates(yield_candidates, offset=True) response_metric = _event_metric( response_scores, @@ -794,11 +914,17 @@ def evaluate_event_horizon( ) response_metric.update( _event_window_diagnostics( - source_boundaries=response_source_boundaries, + source_boundaries=response_diagnostics["num_source_boundaries"], effective_steps=response_effective_steps, competing_censored=response_competing_censored, - right_censored=response_right_censored, + right_censored=response_diagnostics["num_right_censored"], invalid_score=response_invalid_score, + simultaneous_boundary_outcomes_excluded=response_diagnostics[ + "num_simultaneous_boundary_outcomes_excluded" + ], + zero_window_competing_risk=response_diagnostics[ + "num_zero_window_competing_risk" + ], ) ) yield_metric = _event_metric( @@ -812,17 +938,24 @@ def evaluate_event_horizon( ) yield_metric.update( _event_window_diagnostics( - source_boundaries=yield_source_boundaries, + source_boundaries=yield_diagnostics["num_source_boundaries"], effective_steps=yield_effective_steps, competing_censored=yield_competing_censored, - right_censored=yield_right_censored, + right_censored=yield_diagnostics["num_right_censored"], invalid_score=yield_invalid_score, + simultaneous_boundary_outcomes_excluded=yield_diagnostics[ + "num_simultaneous_boundary_outcomes_excluded" + ], + zero_window_competing_risk=yield_diagnostics[ + "num_zero_window_competing_risk" + ], ) ) return { "horizon_index": horizon_steps, "nominal_seconds": horizon_steps * STEP_SECONDS, + "decision_timing": decision_timing, "other_offset_to_wearer_response_event": response_metric, "other_onset_to_wearer_yield_event": yield_metric, } @@ -931,6 +1064,7 @@ def print_event_timeline_examples( *, examples_per_type: int, horizon_steps: int = NUM_HORIZONS, + decision_timing: str = "pre-boundary", ) -> None: """Print causal event windows for manual attribution inspection.""" @@ -942,51 +1076,30 @@ def print_event_timeline_examples( 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, - ), + specifications: tuple[tuple[EventType, bool, bool], ...] = ( + ("response", True, False), + ("yield", False, True), + ) + print( + "\nCompeting-risk event timeline audit " + f"(first run, {horizon_steps * STEP_SECONDS:g}-s horizon, " + f"{decision_timing})" ) - 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: + for event_name, 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 + candidates, _ = collect_floor_event_candidates( + target, + session_ids, + horizon_steps - 1, + event_type=event_name, + decision_timing=decision_timing, + midturn_pause_max_steps=NUM_HORIZONS, + ) + for candidate in candidates: + source_row = candidate.source_row + effective_steps = candidate.effective_steps + competing_row = candidate.competing_row outcome_row = _first_transition_row( wearer, source_row, @@ -997,37 +1110,44 @@ def print_event_timeline_examples( 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, - ) + status = "negative" + event_score = _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_start = source_row + if ( + decision_timing == "post-boundary" + and source_row > 0 + and session_ids[source_row - 1] == session_ids[source_row] + ): + display_start = source_row - 1 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"source={local_source} " + f"({(local_source + 1) * STEP_SECONDS:.1f}s cutoff) | " 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): + print( + "local_row | cutoff_s | wearer | others | " + "fixed-source score | mark" + ) + for row in range(display_start, display_last + 1): local_row = row - session_starts[session_id] step = row - source_row - if step == 0: + if step < 0: + score_text = "pre-boundary" + elif step == 0: score_text = "source" elif row == competing_row: score_text = "excluded" @@ -1038,13 +1158,19 @@ def print_event_timeline_examples( score_text = "invalid" marks: list[str] = [] if row == source_row: - marks.append("anchor") + marks.append( + "decision/observed-boundary" + if decision_timing == "post-boundary" + else "anchor" + ) + elif row == source_row - 1 and decision_timing == "post-boundary": + marks.append("pre-boundary") 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"{local_row:9d} | {(local_row + 1) * STEP_SECONDS:6.1f} | " f"{int(wearer[row]):6d} | {int(others[row]):6d} | " f"{score_text:18s} | {','.join(marks)}" ) @@ -1059,12 +1185,19 @@ 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)") + timing = payload.get("decision_timing", "pre-boundary") + label = ( + "primary anticipation protocol" + if timing == "pre-boundary" + else "post-boundary reactive diagnostic" + ) + print(f"\nWearer floor-event evaluation ({label})") 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." + "percentages; events remain timing proxies, not semantic EOT/INT. " + f"Decision timing: {timing}." ) if payload["calibration_sessions"]: print( @@ -1231,6 +1364,7 @@ def write_event_csv(path: Path, payload: dict[str, Any]) -> None: with path.open("w", newline="") as file: fieldnames = [ "run", + "decision_timing", "horizon_index", "nominal_seconds", "response_event_ap", @@ -1241,6 +1375,8 @@ def write_event_csv(path: Path, payload: dict[str, Any]) -> None: "response_competing_censored_count", "response_right_censored_count", "response_invalid_score_window_count", + "response_simultaneous_boundary_outcome_excluded_count", + "response_zero_window_competing_risk_count", "response_mean_effective_window_seconds", "yield_event_ap", "yield_count", @@ -1249,6 +1385,8 @@ def write_event_csv(path: Path, payload: dict[str, Any]) -> None: "yield_competing_censored_count", "yield_right_censored_count", "yield_invalid_score_window_count", + "yield_simultaneous_boundary_outcome_excluded_count", + "yield_zero_window_competing_risk_count", "yield_mean_effective_window_seconds", ] for recall in payload["matched_recalls"]: @@ -1274,6 +1412,9 @@ def write_event_csv(path: Path, payload: dict[str, Any]) -> None: yield_metric = item["other_onset_to_wearer_yield_event"] row: dict[str, Any] = { "run": name, + "decision_timing": payload.get( + "decision_timing", "pre-boundary" + ), "horizon_index": item["horizon_index"], "nominal_seconds": item["nominal_seconds"], "response_event_ap": response["average_precision"], @@ -1294,6 +1435,12 @@ def write_event_csv(path: Path, payload: dict[str, Any]) -> None: "response_invalid_score_window_count": response[ "num_invalid_score_window" ], + "response_simultaneous_boundary_outcome_excluded_count": response[ + "num_simultaneous_boundary_outcomes_excluded" + ], + "response_zero_window_competing_risk_count": response[ + "num_zero_window_competing_risk" + ], "response_mean_effective_window_seconds": response[ "mean_effective_window_seconds" ], @@ -1312,6 +1459,12 @@ def write_event_csv(path: Path, payload: dict[str, Any]) -> None: "yield_invalid_score_window_count": yield_metric[ "num_invalid_score_window" ], + "yield_simultaneous_boundary_outcome_excluded_count": yield_metric[ + "num_simultaneous_boundary_outcomes_excluded" + ], + "yield_zero_window_competing_risk_count": yield_metric[ + "num_zero_window_competing_risk" + ], "yield_mean_effective_window_seconds": yield_metric[ "mean_effective_window_seconds" ], @@ -1423,7 +1576,12 @@ def main() -> None: ) payload: dict[str, Any] = { - "protocol": "wearer_floor_event_competing_risk_v2", + "protocol": ( + "wearer_floor_event_competing_risk_v2" + if args.decision_timing == "pre-boundary" + else "wearer_floor_event_post_boundary_diagnostic_v1" + ), + "decision_timing": args.decision_timing, "sessions": sessions, "calibration_sessions": calibration_sessions, "classes": list(CLASS_NAMES), @@ -1442,10 +1600,12 @@ def main() -> None: "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 " + "initially on is one yield candidate. The decision source is the last " + "pre-boundary row for pre-boundary anticipation and the first post-boundary " + "row for the reactive diagnostic. A wearer change simultaneous with the " + "anchor is excluded from post-boundary evaluation because it precedes that " + "decision. A response window ends before the next aggregate-others onset; " + "a yield window ends before the next aggregate-others offset. 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 " @@ -1467,6 +1627,7 @@ def main() -> None: session_ids, sessions, examples_per_type=args.print_event_timelines, + decision_timing=args.decision_timing, ) printed_timelines = True calibration_thresholds: list[dict[str, Any]] | None = None @@ -1517,6 +1678,7 @@ def main() -> None: horizon_index, recalls, midturn_pause_max_steps=midturn_pause_max_steps, + decision_timing=args.decision_timing, ) thresholds = _event_operating_thresholds( calibration_event, recalls @@ -1570,6 +1732,7 @@ def main() -> None: horizon_index, recalls, midturn_pause_max_steps=midturn_pause_max_steps, + decision_timing=args.decision_timing, operating_thresholds=fixed_event_thresholds, threshold_source=threshold_source, ) diff --git a/prepare_easycom_oracle_features.py b/prepare_easycom_oracle_features.py index 24d0573..99732ee 100644 --- a/prepare_easycom_oracle_features.py +++ b/prepare_easycom_oracle_features.py @@ -46,8 +46,14 @@ ORACLE_CONDITIONS = ( "per-person", "shuffled-id", "delayed", + # Keep new analysis conditions after the historical five so their addition + # cannot change the established shuffled-id generation seeds. + "aggregate-visible-address-control", + "aggregate-visible-addressee", ) UNKNOWN_PERSON_IDS = {"", "-1", "none", "null", "unknown"} +UNKNOWN_ADDRESSEE_VALUES = {"", "-1", "none", "null", "unknown"} +GROUP_ADDRESSEE_VALUES = {"all", "everyone", "group"} def exact_steps(seconds: float, name: str) -> int: @@ -59,6 +65,33 @@ def exact_steps(seconds: float, name: str) -> int: return int(steps) +def _sampled_row_interval(row: Any, num_steps: int) -> tuple[int, int, int]: + """Map one transcript interval to the released EgoSpeak 5 Hz grid.""" + + dense_start = int(row.dense_start_frame) + 1 + dense_end = int(row.dense_end_frame_exclusive) + grid_frames = num_steps * FRAMES_PER_TASK_STEP + overrun = max(0, dense_end - grid_frames) + if overrun > FRAMES_PER_TASK_STEP - 1: + raise ValueError( + f"Transcript exceeds the final 5 Hz sample by {overrun} frames: " + f"{row.uid}" + ) + if dense_start >= dense_end: + return 0, 0, overrun + + first = (dense_start + FRAMES_PER_TASK_STEP - 1) // FRAMES_PER_TASK_STEP - 1 + stop = (dense_end + FRAMES_PER_TASK_STEP - 1) // FRAMES_PER_TASK_STEP - 1 + first = max(0, first) + stop = min(num_steps, stop) + if first > num_steps or stop < 0: + raise ValueError( + f"Transcript interval leaves target grid: {row.uid}, " + f"rows=[{first},{stop}), target_steps={num_steps}" + ) + return first, stop, overrun + + def participant_activity( utterances: Sequence[Any], *, @@ -78,33 +111,9 @@ def participant_activity( activity: dict[int, torch.BoolTensor] = {} wearer = torch.zeros(num_steps, dtype=torch.bool) terminal_frames_after_last_sample = 0 - grid_frames = num_steps * FRAMES_PER_TASK_STEP for row in utterances: - # Recover the source annotation marker from the zero-origin interval. - dense_start = int(row.dense_start_frame) + 1 - dense_end = int(row.dense_end_frame_exclusive) - if dense_end > grid_frames: - overrun = dense_end - grid_frames - if overrun > FRAMES_PER_TASK_STEP - 1: - raise ValueError( - f"Transcript exceeds the final 5 Hz sample by " - f"{overrun} frames: {row.uid}" - ) - terminal_frames_after_last_sample += overrun - if dense_start >= dense_end: - continue - - # Sample number k (one-origin) observes marker 4*k and maps to row k-1. - # The exclusive stop follows directly from 4*k < dense_end. - first = (dense_start + FRAMES_PER_TASK_STEP - 1) // FRAMES_PER_TASK_STEP - 1 - stop = (dense_end + FRAMES_PER_TASK_STEP - 1) // FRAMES_PER_TASK_STEP - 1 - first = max(0, first) - stop = min(num_steps, stop) - if first > num_steps or stop < 0: - raise ValueError( - f"Transcript interval leaves target grid: {row.uid}, " - f"rows=[{first},{stop}), target_steps={num_steps}" - ) + first, stop, overrun = _sampled_row_interval(row, num_steps) + terminal_frames_after_last_sample += overrun if first >= stop: continue destination = ( @@ -124,6 +133,71 @@ def participant_activity( } +def target_matches_participant( + targets: Sequence[int | str], participant_id: int +) -> bool: + """Match only an explicit numeric participant target.""" + + for value in targets: + if isinstance(value, bool): + continue + if isinstance(value, int) and value == participant_id: + return True + text = str(value).strip() + if text.lstrip("+-").isdigit() and int(text) == participant_id: + return True + return False + + +def participant_direct_addressee_activity( + utterances: Sequence[Any], + *, + num_steps: int, + wearer_id: int, +) -> tuple[dict[int, torch.BoolTensor], dict[str, Any]]: + """Return non-wearer speech bins explicitly addressed to the wearer. + + Group and unknown targets are audited but are not coerced to either a + positive or a trustworthy negative semantic label. The output is an + analysis-only privileged input derived from annotation. + """ + + directed: dict[int, torch.BoolTensor] = {} + category_counts = {"direct_wearer": 0, "group": 0, "other": 0, "unknown": 0} + direct_active_bins = 0 + for row in utterances: + if int(row.participant_id) == wearer_id: + continue + targets = tuple(row.target_of_speech) + normalized = {str(value).strip().lower() for value in targets} + if target_matches_participant(targets, wearer_id): + category = "direct_wearer" + elif normalized & GROUP_ADDRESSEE_VALUES: + category = "group" + elif not normalized or normalized.issubset(UNKNOWN_ADDRESSEE_VALUES): + category = "unknown" + else: + category = "other" + category_counts[category] += 1 + if category != "direct_wearer": + continue + first, stop, _overrun = _sampled_row_interval(row, num_steps) + if first >= stop: + continue + sequence = directed.setdefault( + int(row.participant_id), + torch.zeros(num_steps, dtype=torch.bool), + ) + before = int(sequence[first:stop].sum()) + sequence[first:stop] = True + direct_active_bins += int(sequence[first:stop].sum()) - before + return directed, { + "target_of_speech_utterance_counts": category_counts, + "direct_wearer_addressed_person_bins": direct_active_bins, + "target_of_speech_group_unknown_not_coerced": True, + } + + def causal_history( sequence: torch.BoolTensor, *, @@ -215,6 +289,7 @@ def oracle_person_rows( history_steps: int, delay_steps: int, seed: int, + direct_addressee_activity: dict[int, torch.BoolTensor] | None = None, ) -> tuple[torch.Tensor, torch.LongTensor, list[dict[str, Any]], dict[str, int]]: """Create a 5 Hz ragged set of causal oracle histories.""" @@ -222,6 +297,12 @@ def oracle_person_rows( raise ValueError(f"Unknown oracle condition: {condition}") if len(visible_ids) != len(source_records): raise ValueError("Visible ID/record row counts differ") + address_condition = condition in { + "aggregate-visible-address-control", + "aggregate-visible-addressee", + } + if condition == "aggregate-visible-addressee" and direct_addressee_activity is None: + raise ValueError("Addressee condition requires direct addressee activity") num_steps = len(visible_ids) zero_sequence = torch.zeros(num_steps, dtype=torch.bool) rng = np.random.default_rng(seed) @@ -251,17 +332,49 @@ def oracle_person_rows( if condition == "visibility-only": output = base.clone() output[:, :history_steps] = 0 - elif condition == "aggregate-visible": + elif condition in { + "aggregate-visible", + "aggregate-visible-address-control", + "aggregate-visible-addressee", + }: if len(base): - output = torch.cat( + aggregate = torch.cat( ( base[:, :history_steps].amax(dim=0, keepdim=True), base[:, history_steps:].amax(dim=0, keepdim=True), ), dim=-1, ) + if address_condition: + if condition == "aggregate-visible-addressee": + assert direct_addressee_activity is not None + addressed_rows = torch.stack( + [ + causal_history( + direct_addressee_activity.get( + person_id, zero_sequence + ), + row=row, + history_steps=history_steps, + observed_sequence=visibility.get( + person_id, zero_sequence + ), + )[:history_steps] + for person_id in person_ids + ], + dim=0, + ) + addressed = addressed_rows.amax(dim=0, keepdim=True) + else: + addressed = torch.zeros( + (1, history_steps), dtype=torch.float32 + ) + output = torch.cat((aggregate, addressed), dim=-1) + else: + output = aggregate else: - output = base + feature_dim = (3 if address_condition else 2) * history_steps + output = torch.empty((0, feature_dim), dtype=torch.float32) elif condition == "shuffled-id": output = base.clone() # Independently reassign each historical time bin across the @@ -282,10 +395,10 @@ def oracle_person_rows( output = base feature_rows.extend(output.unbind(dim=0)) - if condition == "aggregate-visible" and person_ids: + if condition.startswith("aggregate-visible") and person_ids: records.append( { - "id": "aggregate-visible", + "id": condition, "source_person_ids": person_ids, "source_visual_times": [ float(item.get("source_time", float("nan"))) @@ -308,7 +421,7 @@ def oracle_person_rows( ) offsets.append(offsets[-1] + len(output)) - feature_dim = 2 * history_steps + feature_dim = (3 if address_condition else 2) * history_steps features = ( torch.stack(feature_rows) if feature_rows @@ -402,6 +515,13 @@ def main() -> None: num_steps=len(times), wearer_id=args.camera_wearer_id, ) + direct_addressee_activity, addressee_audit = ( + participant_direct_addressee_activity( + utterances, + num_steps=len(times), + wearer_id=args.camera_wearer_id, + ) + ) all_other_activity = torch.zeros(len(times), dtype=torch.bool) for sequence in activity.values(): all_other_activity |= sequence @@ -503,8 +623,10 @@ def main() -> None: history_steps=history_steps, delay_steps=delay_steps, seed=condition_seed, + direct_addressee_activity=direct_addressee_activity, ) - visual = torch.zeros((len(times), 2 * history_steps)) + visual_dim = int(features.shape[1]) + visual = torch.zeros((len(times), visual_dim)) visual_valid = torch.zeros(len(times), dtype=torch.bool) for row in range(len(times)): start = int(frame_offsets[row]) @@ -515,6 +637,8 @@ def main() -> None: config = { "oracle_analysis": True, "uses_speech_labels": True, + "uses_target_of_speech": condition == "aggregate-visible-addressee", + "online_observable": False, "causal_policy": "egospeak-right-edge-current-state", "oracle_condition": condition, "oracle_scope": "currently-visible-nonwearer-participants", @@ -526,9 +650,19 @@ def main() -> None: ), "shuffle_seed": args.shuffle_seed if condition == "shuffled-id" else None, "speaker_activity_source": "EasyCom Speech_Transcriptions inclusive 20Hz intervals", + "addressee_source": ( + "EasyCom Target_of_Speech explicit wearer ID" + if condition == "aggregate-visible-addressee" + else None + ), + "addressee_semantics": ( + "direct-wearer-only; Group/None/unknown are not coerced" + if condition == "aggregate-visible-addressee" + else None + ), "visibility_source": "provided face boxes in source local sidecar", "camera_wearer_id": args.camera_wearer_id, - "visual_dim": 2 * history_steps, + "visual_dim": visual_dim, "audio_dim": audio_dim, "source_local_contract": source_contract, } @@ -556,6 +690,9 @@ def main() -> None: "source_local_features": str(source_local_path.resolve()), "audit": { "strictly_past_or_current_bins_only": True, + "target_of_speech_is_privileged_utterance_annotation": ( + condition == "aggregate-visible-addressee" + ), "transcript_target_mismatch_rows": mismatch_union, "wearer_mismatch_rows": wearer_mismatch, "others_mismatch_rows": other_mismatch, @@ -566,6 +703,7 @@ def main() -> None: "person_observations": len(features), "generation_seed": condition_seed, **transcript_audit, + **addressee_audit, **corruption, }, } diff --git a/run_easycom_floor_event_probes.sh b/run_easycom_floor_event_probes.sh new file mode 100755 index 0000000..d341cb0 --- /dev/null +++ b/run_easycom_floor_event_probes.sh @@ -0,0 +1,163 @@ +#!/usr/bin/env bash +set -euo pipefail + +stage="${1:-screen}" +shift || true +if (( $# )); then + seeds=("$@") +else + seeds=(0) +fi + +: "${DATASET_ROOT:?Set DATASET_ROOT to the extracted EasyCom root}" +: "${FEATURE_ROOT:?Set FEATURE_ROOT to the verified global Omni features}" +: "${TARGET_ROOT:?Set TARGET_ROOT to EasyCom target_perframe files}" +: "${LOCAL_ROOT:?Set LOCAL_ROOT to the complete Qwen local AV sidecars}" + +ORACLE_ROOT="${ORACLE_ROOT:-features/easycom_oracle_information_2s}" +CHECKPOINT_ROOT="${EVENT_CHECKPOINT_ROOT:-checkpoints/easycom_floor_event_probe}" +RESULT_ROOT="${EVENT_RESULT_ROOT:-results/easycom_floor_event_probe}" +TRAIN_SESSIONS="${TRAIN_SESSIONS:-4 5 6 7 8 9 10 11}" +DEV_SESSIONS="${DEV_SESSIONS:-12}" +EVAL_SESSIONS="${EVAL_SESSIONS:-12}" +TEST_SESSIONS="${TEST_SESSIONS:-1 2 3}" +VARIANTS="${VARIANTS:-aggregate-visible aggregate-visible-address-control aggregate-visible-addressee}" +EPOCHS="${EPOCHS:-20}" +EXPECTED_MODE="${EXPECTED_MODE:-open-stream-replay}" +DECISION_TIMING="${DECISION_TIMING:-pre-boundary}" +PYTHON_BIN="${PYTHON_BIN:-python}" + +read -r -a train_sessions <<<"${TRAIN_SESSIONS}" +read -r -a dev_sessions <<<"${DEV_SESSIONS}" +read -r -a eval_sessions <<<"${EVAL_SESSIONS}" +read -r -a test_sessions <<<"${TEST_SESSIONS}" +read -r -a variants <<<"${VARIANTS}" + +prepare_oracles() { + "${PYTHON_BIN}" prepare_easycom_oracle_features.py \ + --dataset-root "${DATASET_ROOT}" \ + --feature-root "${FEATURE_ROOT}" \ + --target-root "${TARGET_ROOT}" \ + --local-feature-root "${LOCAL_ROOT}" \ + --output-root "${ORACLE_ROOT}" \ + --conditions \ + aggregate-visible \ + aggregate-visible-address-control \ + aggregate-visible-addressee \ + --history-seconds 2.0 \ + --expected-mode "${EXPECTED_MODE}" \ + --reuse-existing +} + +smoke_probes() { + "${PYTHON_BIN}" prepare_easycom_oracle_features.py \ + --dataset-root "${DATASET_ROOT}" \ + --feature-root "${FEATURE_ROOT}" \ + --target-root "${TARGET_ROOT}" \ + --local-feature-root "${LOCAL_ROOT}" \ + --output-root "${ORACLE_ROOT}" \ + --sessions Session_11 Session_12 \ + --conditions \ + aggregate-visible \ + aggregate-visible-address-control \ + aggregate-visible-addressee \ + --history-seconds 2.0 \ + --expected-mode "${EXPECTED_MODE}" \ + --reuse-existing + for variant in "${variants[@]}"; do + "${PYTHON_BIN}" train_easycom_floor_event_probe.py \ + --feature-root "${FEATURE_ROOT}" \ + --target-root "${TARGET_ROOT}" \ + --oracle-feature-root "${ORACLE_ROOT}/${variant}" \ + --checkpoint "${CHECKPOINT_ROOT}_smoke/${variant}_seed0.pth" \ + --result-dir "${RESULT_ROOT}_smoke/${variant}_seed0" \ + --train-sessions Session_11 \ + --dev-sessions Session_12 \ + --eval-sessions Session_12 \ + --epochs 1 \ + --batch-size 128 \ + --decision-timing "${DECISION_TIMING}" \ + --expected-mode "${EXPECTED_MODE}" \ + --seed 0 \ + --device cuda:0 + done +} + +train_and_screen() { + summary_args=() + for seed in "${seeds[@]}"; do + for variant in "${variants[@]}"; do + checkpoint="${CHECKPOINT_ROOT}/${variant}_seed${seed}.pth" + result="${RESULT_ROOT}/${variant}_seed${seed}_dev" + echo "$(date): direct event probe ${variant}, seed=${seed}" + "${PYTHON_BIN}" train_easycom_floor_event_probe.py \ + --feature-root "${FEATURE_ROOT}" \ + --target-root "${TARGET_ROOT}" \ + --oracle-feature-root "${ORACLE_ROOT}/${variant}" \ + --checkpoint "${checkpoint}" \ + --result-dir "${result}" \ + --train-sessions "${train_sessions[@]}" \ + --dev-sessions "${dev_sessions[@]}" \ + --eval-sessions "${eval_sessions[@]}" \ + --epochs "${EPOCHS}" \ + --decision-timing "${DECISION_TIMING}" \ + --expected-mode "${EXPECTED_MODE}" \ + --seed "${seed}" \ + --device cuda:0 + summary_args+=(--run "${variant}_seed${seed}=${result}") + done + done + "${PYTHON_BIN}" summarize_easycom_floor_event_probes.py \ + "${summary_args[@]}" \ + --output "${RESULT_ROOT}/comparison_dev.csv" +} + +test_frozen() { + summary_args=() + for seed in "${seeds[@]}"; do + for variant in "${variants[@]}"; do + checkpoint="${CHECKPOINT_ROOT}/${variant}_seed${seed}.pth" + result="${RESULT_ROOT}/${variant}_seed${seed}_test" + echo "$(date): testing frozen direct event probe ${variant}, seed=${seed}" + "${PYTHON_BIN}" train_easycom_floor_event_probe.py \ + --feature-root "${FEATURE_ROOT}" \ + --target-root "${TARGET_ROOT}" \ + --oracle-feature-root "${ORACLE_ROOT}/${variant}" \ + --checkpoint "${checkpoint}" \ + --result-dir "${result}" \ + --eval-sessions "${test_sessions[@]}" \ + --eval-only \ + --decision-timing "${DECISION_TIMING}" \ + --expected-mode "${EXPECTED_MODE}" \ + --seed "${seed}" \ + --device cuda:0 + summary_args+=(--run "${variant}_seed${seed}=${result}") + done + done + "${PYTHON_BIN}" summarize_easycom_floor_event_probes.py \ + "${summary_args[@]}" \ + --output "${RESULT_ROOT}/comparison_test.csv" +} + +case "${stage}" in + smoke) + smoke_probes + ;; + prepare) + prepare_oracles + ;; + screen) + train_and_screen + ;; + all) + prepare_oracles + train_and_screen + ;; + test) + test_frozen + ;; + *) + echo "Usage: $0 {smoke|prepare|screen|all|test} [seed ...]" >&2 + exit 2 + ;; +esac diff --git a/run_easycom_post_boundary_eval.sh b/run_easycom_post_boundary_eval.sh new file mode 100755 index 0000000..3f4dcad --- /dev/null +++ b/run_easycom_post_boundary_eval.sh @@ -0,0 +1,35 @@ +#!/usr/bin/env bash +set -euo pipefail + +: "${TARGET_ROOT:?Set TARGET_ROOT to EasyCom target_perframe files}" +: "${POOLED_AV_RUN:?Set POOLED_AV_RUN to the aligned pooled-AV result directory}" +: "${ORACLE_RUN:?Set ORACLE_RUN to the aligned aggregate-visible oracle result directory}" +: "${MINIROAD_RUN:?Set MINIROAD_RUN to the converted MiniROAD result directory}" + +OUTPUT="${OUTPUT:-reports/easycom_floor_post_boundary_diagnostic/summary.json}" +SESSIONS="${SESSIONS:-Session_3 Session_2 Session_1}" +CALIBRATION_SESSIONS="${CALIBRATION_SESSIONS:-}" +PYTHON_BIN="${PYTHON_BIN:-python}" + +read -r -a sessions <<<"${SESSIONS}" +threshold_args=() +if [[ -n "${CALIBRATION_SESSIONS}" ]]; then + read -r -a calibration_sessions <<<"${CALIBRATION_SESSIONS}" + threshold_args=(--calibration-sessions "${calibration_sessions[@]}") +else + echo "No calibration sessions supplied: marking this post-boundary run exploratory/in-sample." >&2 + threshold_args=(--allow-in-sample-test-thresholds) +fi + +"${PYTHON_BIN}" evaluate_floor_state_changes.py \ + --target-root "${TARGET_ROOT}" \ + --run "pooled_av=${POOLED_AV_RUN}" \ + --run "oracle_aggregate_visible=${ORACLE_RUN}" \ + --run "egospeak_miniroada=${MINIROAD_RUN}" \ + --sessions "${sessions[@]}" \ + "${threshold_args[@]}" \ + --matched-recalls 0.7 0.8 0.9 \ + --decision-timing post-boundary \ + --midturn-pause-max-seconds 2.0 \ + --print-event-timelines 3 \ + --output "${OUTPUT}" diff --git a/summarize_easycom_floor_event_probes.py b/summarize_easycom_floor_event_probes.py new file mode 100755 index 0000000..af1c7c0 --- /dev/null +++ b/summarize_easycom_floor_event_probes.py @@ -0,0 +1,133 @@ +#!/usr/bin/env python3 +"""Compare direct EasyCom floor-event probe result directories.""" + +from __future__ import annotations + +import argparse +import csv +import json +from pathlib import Path +from typing import Any + +import numpy as np + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--run", + action="append", + required=True, + metavar="NAME=RESULT_DIR", + ) + parser.add_argument("--output", type=Path, required=True) + return parser.parse_args() + + +def parse_run(value: str) -> tuple[str, Path]: + if "=" not in value: + raise ValueError(f"Expected NAME=RESULT_DIR, got {value!r}") + name, raw_path = value.split("=", 1) + if not name.strip() or not raw_path.strip(): + raise ValueError(f"Expected NAME=RESULT_DIR, got {value!r}") + return name.strip(), Path(raw_path.strip()) + + +def mean_ap(rows: list[dict[str, Any]], event_type: str) -> float | None: + values = [ + float(row["average_precision"]) + for row in rows + if row["event_type"] == event_type + and row.get("average_precision") is not None + ] + return float(np.mean(values)) if values else None + + +def horizon_row( + rows: list[dict[str, Any]], event_type: str, horizon_index: int +) -> dict[str, Any]: + matches = [ + row + for row in rows + if row["event_type"] == event_type + and int(row["horizon_index"]) == horizon_index + ] + if len(matches) != 1: + raise ValueError( + f"Expected one {event_type} horizon {horizon_index}, got {len(matches)}" + ) + return matches[0] + + +def main() -> None: + args = parse_args() + parsed = [parse_run(value) for value in args.run] + if len({name for name, _ in parsed}) != len(parsed): + raise ValueError("Duplicate run name") + output_rows: list[dict[str, Any]] = [] + common_contract: tuple[Any, ...] | None = None + for name, directory in parsed: + path = directory / "summary.json" + if not path.is_file(): + raise FileNotFoundError(path) + payload = json.loads(path.read_text()) + if payload.get("protocol") != "easycom_direct_floor_event_probe_v1": + raise ValueError(f"Unexpected protocol in {path}") + contract = ( + tuple(payload["eval_sessions"]), + payload["decision_timing"], + float(payload["matched_recall"]), + ) + if common_contract is None: + common_contract = contract + elif contract != common_contract: + raise ValueError( + f"Incomparable evaluation contract in {path}: {contract} vs " + f"{common_contract}" + ) + rows = payload["eval_metrics"] + response_2s = horizon_row(rows, "response", 10) + yield_2s = horizon_row(rows, "yield", 10) + result = { + "run": name, + "oracle_condition": payload["oracle_condition"], + "decision_timing": payload["decision_timing"], + "eval_sessions": ",".join(payload["eval_sessions"]), + "best_epoch": payload["best_epoch"], + "mean_response_ap": mean_ap(rows, "response"), + "mean_yield_ap": mean_ap(rows, "yield"), + "response_2s_ap": response_2s["average_precision"], + "response_2s_recall": response_2s["achieved_positive_recall"], + "response_2s_false_entry_rate": response_2s[ + "negative_trigger_rate" + ], + "response_2s_short_resume_trigger_rate": response_2s[ + "short_resume_trigger_rate" + ], + "yield_2s_ap": yield_2s["average_precision"], + "yield_2s_recall": yield_2s["achieved_positive_recall"], + "yield_2s_false_stop_rate": yield_2s["negative_trigger_rate"], + } + output_rows.append(result) + + args.output.parent.mkdir(parents=True, exist_ok=True) + with args.output.open("w", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=list(output_rows[0])) + writer.writeheader() + writer.writerows(output_rows) + print( + "run | mean response AP | response 2s AP/error | " + "mean yield AP | yield 2s AP/error" + ) + for row in output_rows: + print( + f"{row['run']} | {row['mean_response_ap']} | " + f"{row['response_2s_ap']}/{row['response_2s_false_entry_rate']} | " + f"{row['mean_yield_ap']} | " + f"{row['yield_2s_ap']}/{row['yield_2s_false_stop_rate']}" + ) + print(f"comparison: {args.output}") + + +if __name__ == "__main__": + main() diff --git a/test_easycom_floor_event_probe.py b/test_easycom_floor_event_probe.py new file mode 100644 index 0000000..ad53b39 --- /dev/null +++ b/test_easycom_floor_event_probe.py @@ -0,0 +1,120 @@ +"""Focused tests for the direct competing-risk floor-event probe.""" + +from __future__ import annotations + +import unittest + +import numpy as np +import torch + +from train_easycom_floor_event_probe import ( + DirectFloorEventHead, + FloorEventDataset, + ProbeSession, + metric_rows, +) + + +def activity(wearer: int, others: int) -> list[int]: + return [int(not wearer and not others), wearer, others] + + +def probe_session() -> ProbeSession: + target = np.asarray( + [ + activity(0, 1), + activity(0, 0), + activity(1, 0), + activity(1, 0), + activity(1, 1), + activity(0, 1), + activity(0, 1), + activity(0, 0), + activity(0, 0), + activity(0, 1), + ], + dtype=np.float32, + ) + steps = len(target) + return ProbeSession( + name="Session_12", + states=torch.randn(steps, 2, 4), + audio=torch.cat((torch.randn(steps, 3), torch.ones(steps, 1)), dim=1), + oracle=torch.randn(steps, 6), + oracle_valid=torch.ones(steps, dtype=torch.bool), + target=target, + oracle_condition="aggregate-visible", + oracle_contract={"config": {"oracle_condition": "aggregate-visible"}}, + ) + + +class FloorEventProbeTests(unittest.TestCase): + def setUp(self) -> None: + torch.set_num_threads(1) + + def test_dataset_uses_requested_decision_source(self) -> None: + session = probe_session() + pre = FloorEventDataset( + [session], decision_timing="pre-boundary", midturn_pause_max_steps=10 + ) + post = FloorEventDataset( + [session], decision_timing="post-boundary", midturn_pause_max_steps=10 + ) + pre_response = next( + item + for item in pre.examples + if item.event_index == 0 and item.horizon_index == 1 and item.label + ) + post_response = next( + item + for item in post.examples + if item.event_index == 0 and item.horizon_index == 1 and item.label + ) + self.assertEqual(pre_response.source_row, 0) + self.assertEqual(post_response.source_row, 1) + + def test_head_outputs_two_events_by_ten_horizons(self) -> None: + model = DirectFloorEventHead( + state_shape=(2, 4), + audio_dim=3, + oracle_dim=6, + layer_fusion="concat-mlp", + local_projection_dim=4, + person_attention_dim=4, + person_attention_heads=1, + dropout=0.0, + ) + output = model( + torch.randn(5, 2, 4), + torch.cat((torch.randn(5, 3), torch.ones(5, 1)), dim=1), + torch.randn(5, 6), + torch.ones(5), + ) + self.assertEqual(output.shape, (5, 2, 10)) + + def test_fixed_threshold_is_applied_without_reselection(self) -> None: + predictions = { + "score": np.tile(np.asarray([0.9, 0.4]), 20), + "event": np.repeat(np.arange(2), 20), + "horizon": np.tile(np.repeat(np.arange(10), 2), 2), + "label": np.tile(np.asarray([True, False]), 20), + "short_resume": np.zeros(40, dtype=bool), + } + thresholds = { + (event, horizon): 0.95 + for event in range(2) + for horizon in range(10) + } + rows, _ = metric_rows( + predictions, + matched_recall=0.8, + fixed_thresholds=thresholds, + threshold_source="Session_12", + ) + self.assertEqual(len(rows), 20) + self.assertEqual(rows[0]["threshold"], 0.95) + self.assertEqual(rows[0]["achieved_positive_recall"], 0.0) + + +if __name__ == "__main__": + unittest.main() diff --git a/test_easycom_oracle_features.py b/test_easycom_oracle_features.py index e5ffc93..feb4778 100644 --- a/test_easycom_oracle_features.py +++ b/test_easycom_oracle_features.py @@ -16,6 +16,7 @@ from prepare_easycom_oracle_features import ( causal_history, oracle_person_rows, participant_activity, + participant_direct_addressee_activity, ) @@ -79,6 +80,79 @@ class OracleFeatureTests(unittest.TestCase): [0, 0, 0, 0, 0, 1, 0, 1], ) + def test_direct_addressee_keeps_group_and_unknown_out_of_positive_channel(self) -> None: + rows = [ + SimpleNamespace( + uid="direct", + participant_id=3, + target_of_speech=(2,), + dense_start_frame=0, + dense_end_frame_exclusive=8, + ), + SimpleNamespace( + uid="group", + participant_id=4, + target_of_speech=("Group",), + dense_start_frame=4, + dense_end_frame_exclusive=13, + ), + SimpleNamespace( + uid="unknown", + participant_id=5, + target_of_speech=("None",), + dense_start_frame=8, + dense_end_frame_exclusive=17, + ), + ] + directed, audit = participant_direct_addressee_activity( + rows, num_steps=5, wearer_id=2 + ) + self.assertEqual(directed[3].tolist(), [True, False, False, False, False]) + self.assertNotIn(4, directed) + self.assertNotIn(5, directed) + self.assertEqual( + audit["target_of_speech_utterance_counts"], + {"direct_wearer": 1, "group": 1, "other": 0, "unknown": 1}, + ) + self.assertTrue(audit["target_of_speech_group_unknown_not_coerced"]) + + def test_addressee_oracle_has_capacity_matched_zero_control(self) -> None: + activity = {3: torch.tensor([1, 1, 0], dtype=torch.bool)} + addressed = {3: torch.tensor([1, 0, 0], dtype=torch.bool)} + visibility = {3: torch.ones(3, dtype=torch.bool)} + visible = [[3], [3], [3]] + records = [ + [{"id": "3", "source_clip_index": 0, "source_time": row / 5}] + for row in range(3) + ] + control, control_offsets, _, _ = oracle_person_rows( + activity, + visibility, + visible, + records, + condition="aggregate-visible-address-control", + history_steps=2, + delay_steps=0, + seed=0, + direct_addressee_activity=addressed, + ) + oracle, oracle_offsets, _, _ = oracle_person_rows( + activity, + visibility, + visible, + records, + condition="aggregate-visible-addressee", + history_steps=2, + delay_steps=0, + seed=0, + direct_addressee_activity=addressed, + ) + self.assertEqual(control.shape, (3, 6)) + self.assertTrue(torch.equal(control_offsets, oracle_offsets)) + self.assertTrue(torch.equal(control[:, :4], oracle[:, :4])) + self.assertEqual(control[:, 4:].tolist(), [[0, 0], [0, 0], [0, 0]]) + self.assertEqual(oracle[:, 4:].tolist(), [[0, 1], [1, 0], [0, 0]]) + def test_shuffled_id_preserves_every_lag_count_but_breaks_histories(self) -> None: activity = { 3: torch.tensor([1, 1, 0, 0, 1, 1], dtype=torch.bool), diff --git a/test_evaluate_floor_state_changes.py b/test_evaluate_floor_state_changes.py index eb2fa28..baac1a6 100644 --- a/test_evaluate_floor_state_changes.py +++ b/test_evaluate_floor_state_changes.py @@ -14,6 +14,7 @@ import numpy as np from evaluate_floor_state_changes import ( NUM_CLASSES, NUM_HORIZONS, + collect_floor_event_candidates, evaluate_event_horizon, evaluate_horizon, load_arrays, @@ -29,6 +30,85 @@ def activity(wearer: int, others: int) -> list[int]: class FloorStateMetricTests(unittest.TestCase): + def test_post_boundary_candidate_uses_first_post_transition_row(self) -> None: + target = np.asarray( + [ + activity(0, 1), + activity(0, 0), # observed other offset + activity(1, 0), # wearer response + activity(1, 0), + ], + dtype=np.float32, + ) + session_ids = np.zeros(len(target), dtype=np.int32) + pre, _ = collect_floor_event_candidates( + target, + session_ids, + 1, + event_type="response", + decision_timing="pre-boundary", + ) + post, diagnostics = collect_floor_event_candidates( + target, + session_ids, + 1, + event_type="response", + decision_timing="post-boundary", + ) + self.assertEqual([(item.source_row, item.label) for item in pre], [(0, True)]) + self.assertEqual([(item.source_row, item.label) for item in post], [(1, True)]) + self.assertEqual(diagnostics["num_simultaneous_boundary_outcomes_excluded"], 0) + + def test_post_boundary_excludes_outcome_that_already_happened_at_anchor(self) -> None: + target = np.asarray( + [ + activity(0, 1), + activity(1, 0), # simultaneous other offset and wearer onset + activity(1, 0), + ], + dtype=np.float32, + ) + session_ids = np.zeros(len(target), dtype=np.int32) + pre, _ = collect_floor_event_candidates( + target, + session_ids, + 0, + event_type="response", + decision_timing="pre-boundary", + ) + post, diagnostics = collect_floor_event_candidates( + target, + session_ids, + 0, + event_type="response", + decision_timing="post-boundary", + ) + self.assertEqual(len(pre), 1) + self.assertTrue(pre[0].label) + self.assertEqual(post, []) + self.assertEqual(diagnostics["num_source_boundaries"], 1) + self.assertEqual(diagnostics["num_simultaneous_boundary_outcomes_excluded"], 1) + + def test_post_boundary_reports_competitor_with_no_observable_bin(self) -> None: + target = np.asarray( + [ + activity(0, 1), + activity(0, 0), + activity(0, 1), # immediate restart after reactive source + activity(0, 1), + ], + dtype=np.float32, + ) + candidates, diagnostics = collect_floor_event_candidates( + target, + np.zeros(len(target), dtype=np.int32), + 1, + event_type="response", + decision_timing="post-boundary", + ) + self.assertEqual(candidates, []) + self.assertEqual(diagnostics["num_zero_window_competing_risk"], 1) + def test_response_window_stops_before_next_other_onset(self) -> None: target = np.asarray( [ diff --git a/train_easycom_floor_event_probe.py b/train_easycom_floor_event_probe.py new file mode 100644 index 0000000..7e2d8de --- /dev/null +++ b/train_easycom_floor_event_probe.py @@ -0,0 +1,832 @@ +#!/usr/bin/env python3 +"""Train an analysis-only direct EasyCom response/yield event probe. + +The probe consumes frozen Omni states, the same mixed-audio vector and one +aggregate oracle-history vector at the causal decision source. Unlike the +main activity head, it is supervised directly on competing-risk-censored +response/yield labels. It is a diagnostic for objective and information +alignment, not a deployable model or a replacement for dense activity mAP. +""" + +from __future__ import annotations + +import argparse +import csv +from dataclasses import dataclass +import json +import math +from pathlib import Path +import random +from typing import Any, Iterable + +import numpy as np +import torch +from torch import nn +from torch.nn import functional as F +from torch.utils.data import DataLoader, Dataset + +from easycom_head import ( + NUM_HORIZONS, + STEP_SECONDS, + TurnPredictionHead, + load_session_data, +) +from evaluate_floor_state_changes import ( + DECISION_TIMINGS, + OFFICIAL_TEST_SESSIONS, + _rate_at_threshold, + _safe_ap, + _threshold_for_recall, + collect_floor_event_candidates, + normalize_session_name, +) + + +EVENT_NAMES = ("response", "yield") +PROBE_ORACLE_CONDITIONS = ( + "aggregate-visible", + "aggregate-visible-address-control", + "aggregate-visible-addressee", +) +DEFAULT_TRAIN_SESSIONS = tuple(f"Session_{index}" for index in range(4, 12)) +DEFAULT_DEV_SESSIONS = ("Session_12",) + + +@dataclass(frozen=True) +class ProbeExample: + session_index: int + source_row: int + event_index: int + horizon_index: int + label: bool + short_resume_hard_negative: bool + + +@dataclass +class ProbeSession: + name: str + states: torch.Tensor + audio: torch.Tensor + oracle: torch.Tensor + oracle_valid: torch.Tensor + target: np.ndarray + oracle_condition: str + oracle_contract: dict[str, Any] + + +class FloorEventDataset(Dataset): + def __init__( + self, + sessions: list[ProbeSession], + *, + decision_timing: str, + midturn_pause_max_steps: int, + ) -> None: + self.sessions = sessions + self.examples: list[ProbeExample] = [] + self.censoring: dict[str, dict[str, dict[str, int]]] = {} + for session_index, session in enumerate(sessions): + self.censoring[session.name] = {} + session_ids = np.zeros(len(session.target), dtype=np.int32) + for event_index, event_name in enumerate(EVENT_NAMES): + aggregate = { + "num_source_boundaries": 0, + "num_right_censored": 0, + "num_simultaneous_boundary_outcomes_excluded": 0, + "num_zero_window_competing_risk": 0, + } + for horizon_index in range(NUM_HORIZONS): + candidates, diagnostics = collect_floor_event_candidates( + session.target, + session_ids, + horizon_index, + event_type=event_name, # type: ignore[arg-type] + decision_timing=decision_timing, + midturn_pause_max_steps=midturn_pause_max_steps, + ) + for key, value in diagnostics.items(): + aggregate[key] += int(value) + self.examples.extend( + ProbeExample( + session_index=session_index, + source_row=item.source_row, + event_index=event_index, + horizon_index=horizon_index, + label=item.label, + short_resume_hard_negative=( + item.short_resume_hard_negative + ), + ) + for item in candidates + ) + self.censoring[session.name][event_name] = aggregate + if not self.examples: + raise ValueError("No observable floor-event examples") + + def __len__(self) -> int: + return len(self.examples) + + def __getitem__(self, index: int) -> tuple[torch.Tensor, ...]: + example = self.examples[index] + session = self.sessions[example.session_index] + row = example.source_row + return ( + session.states[row].float(), + session.audio[row].float(), + session.oracle[row].float(), + session.oracle_valid[row].float(), + torch.tensor(example.event_index, dtype=torch.long), + torch.tensor(example.horizon_index, dtype=torch.long), + torch.tensor(float(example.label), dtype=torch.float32), + torch.tensor( + float(example.short_resume_hard_negative), dtype=torch.float32 + ), + ) + + +class DirectFloorEventHead(nn.Module): + """Event readout with the same fusion trunk as the dense oracle head.""" + + def __init__( + self, + *, + state_shape: tuple[int, ...], + audio_dim: int, + oracle_dim: int, + layer_fusion: str, + local_projection_dim: int, + person_attention_dim: int, + person_attention_heads: int, + dropout: float, + ) -> None: + super().__init__() + if len(state_shape) not in {1, 2}: + raise ValueError(f"Expected state shape [D] or [L,D], got {state_shape}") + if layer_fusion not in {"fixed-mean", "concat-mlp"}: + raise ValueError(f"Unknown layer fusion: {layer_fusion}") + if min( + local_projection_dim, + person_attention_dim, + person_attention_heads, + audio_dim, + oracle_dim, + ) <= 0: + raise ValueError("All probe dimensions must be positive") + self.state_shape = state_shape + num_input_layers = 1 if len(state_shape) == 1 else state_shape[0] + effective_layer_fusion = "single" if len(state_shape) == 1 else layer_fusion + self.head = TurnPredictionHead( + hidden_size=state_shape[-1], + num_horizons=NUM_HORIZONS, + num_classes=len(EVENT_NAMES), + dropout=dropout, + num_input_layers=num_input_layers, + layer_fusion=effective_layer_fusion, + local_visual_dim=oracle_dim, + local_audio_dim=audio_dim, + local_projection_dim=local_projection_dim, + local_fusion="oracle-frame-set-attention", + person_attention_dim=person_attention_dim, + person_attention_heads=person_attention_heads, + ) + + def forward( + self, + states: torch.Tensor, + audio: torch.Tensor, + oracle: torch.Tensor, + oracle_valid: torch.Tensor, + ) -> torch.Tensor: + if oracle_valid.ndim == 1: + oracle_valid = oracle_valid[:, None] + logits = self.head( + states, + audio, + oracle[:, None, :], + oracle_valid.bool(), + ) + return logits.transpose(1, 2) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--feature-root", type=Path, required=True) + parser.add_argument("--target-root", type=Path, required=True) + parser.add_argument("--oracle-feature-root", type=Path, required=True) + parser.add_argument("--checkpoint", type=Path, required=True) + parser.add_argument("--result-dir", type=Path, required=True) + parser.add_argument( + "--train-sessions", nargs="+", default=list(DEFAULT_TRAIN_SESSIONS) + ) + parser.add_argument( + "--dev-sessions", nargs="+", default=list(DEFAULT_DEV_SESSIONS) + ) + parser.add_argument("--eval-sessions", nargs="+", default=None) + parser.add_argument("--eval-only", action="store_true") + parser.add_argument("--epochs", type=int, default=20) + parser.add_argument("--batch-size", type=int, default=256) + parser.add_argument("--learning-rate", type=float, default=1e-4) + parser.add_argument("--weight-decay", type=float, default=5e-5) + parser.add_argument("--dropout", type=float, default=0.2) + parser.add_argument("--local-projection-dim", type=int, default=256) + parser.add_argument("--person-attention-dim", type=int, default=64) + parser.add_argument("--person-attention-heads", type=int, default=4) + parser.add_argument( + "--layer-fusion", choices=("fixed-mean", "concat-mlp"), default="concat-mlp" + ) + parser.add_argument("--max-positive-weight", type=float, default=20.0) + parser.add_argument("--matched-recall", type=float, default=0.8) + parser.add_argument("--midturn-pause-max-seconds", type=float, default=2.0) + parser.add_argument( + "--decision-timing", choices=DECISION_TIMINGS, default="pre-boundary" + ) + parser.add_argument("--expected-mode", default="open-stream-replay") + parser.add_argument("--device", default="cuda:0") + parser.add_argument("--num-workers", type=int, default=0) + parser.add_argument("--seed", type=int, default=0) + parser.add_argument("--overwrite", action="store_true") + return parser.parse_args() + + +def normalize_sessions(values: Iterable[str | int]) -> list[str]: + sessions = [normalize_session_name(value) for value in values] + if len(sessions) != len(set(sessions)): + raise ValueError(f"Duplicate sessions: {sessions}") + return sessions + + +def load_probe_session(args: argparse.Namespace, session: str) -> ProbeSession: + data = load_session_data( + args.feature_root, + args.target_root, + session, + expected_mode=None if args.expected_mode == "any" else args.expected_mode, + dataset="easycom", + local_feature_root=args.oracle_feature_root, + local_modalities="av", + local_fusion="oracle-frame-set-attention", + ) + if data.local_features is None or data.local_person_features is None: + raise AssertionError("Oracle loader did not return local features") + people, mask = data.local_person_features.padded() + if people.shape[1] > 1: + raise ValueError( + f"{session}: direct probe requires an aggregate oracle condition " + f"with at most one row-level vector, got max_people={people.shape[1]}" + ) + counts = mask.sum(dim=1, keepdim=True) + oracle = (people * mask[..., None]).sum(dim=1) + oracle = oracle / counts.clamp_min(1).to(oracle.dtype) + oracle_valid = mask.any(dim=1) + contract = data.feature_metadata.get("local_contract") + if not isinstance(contract, dict): + raise ValueError(f"{session}: missing validated oracle contract") + config = contract.get("config") + condition = config.get("oracle_condition") if isinstance(config, dict) else None + if condition not in PROBE_ORACLE_CONDITIONS: + raise ValueError( + f"{session}: expected an aggregate-visible oracle condition, got {condition!r}" + ) + history_steps = config.get("history_steps") + if not isinstance(history_steps, int) or history_steps <= 0: + raise ValueError(f"{session}: invalid oracle history_steps") + expected_oracle_dim = ( + 3 * history_steps + if condition != "aggregate-visible" + else 2 * history_steps + ) + if oracle.shape[1] != expected_oracle_dim: + raise ValueError( + f"{session}: condition {condition} expects oracle_dim=" + f"{expected_oracle_dim}, got {oracle.shape[1]}" + ) + if condition == "aggregate-visible-address-control" and bool( + torch.count_nonzero(oracle[:, -history_steps:]) + ): + raise ValueError(f"{session}: addressee capacity-control channel is not zero") + if condition == "aggregate-visible-addressee" and config.get( + "uses_target_of_speech" + ) is not True: + raise ValueError(f"{session}: addressee oracle lacks privileged-input flag") + return ProbeSession( + name=session, + states=data.states, + audio=data.local_features, + oracle=oracle, + oracle_valid=oracle_valid, + target=data.targets.numpy(), + oracle_condition=condition, + oracle_contract=contract, + ) + + +def validate_session_shapes(sessions: list[ProbeSession]) -> dict[str, Any]: + if not sessions: + raise ValueError("No sessions loaded") + first = sessions[0] + state_shape = tuple(int(value) for value in first.states.shape[1:]) + audio_width = int(first.audio.shape[1]) + oracle_dim = int(first.oracle.shape[1]) + condition = first.oracle_condition + for session in sessions[1:]: + observed = ( + tuple(int(value) for value in session.states.shape[1:]), + int(session.audio.shape[1]), + int(session.oracle.shape[1]), + session.oracle_condition, + ) + expected = (state_shape, audio_width, oracle_dim, condition) + if observed != expected: + raise ValueError( + f"Feature contract changes at {session.name}: {observed} vs {expected}" + ) + if session.oracle_contract != first.oracle_contract: + raise ValueError( + f"Oracle feature contract changes at {session.name}; " + "regenerate sidecars under one configuration" + ) + if audio_width < 2: + raise ValueError("Expected [audio vector, valid bit]") + return { + "state_shape": state_shape, + "audio_dim": audio_width - 1, + "oracle_dim": oracle_dim, + "oracle_condition": condition, + "oracle_contract": first.oracle_contract, + } + + +def positive_weights( + dataset: FloorEventDataset, max_weight: float +) -> torch.Tensor: + positive = torch.zeros((len(EVENT_NAMES), NUM_HORIZONS), dtype=torch.float64) + total = torch.zeros_like(positive) + for item in dataset.examples: + total[item.event_index, item.horizon_index] += 1 + positive[item.event_index, item.horizon_index] += int(item.label) + negative = total - positive + weights = negative / positive.clamp_min(1) + return weights.clamp(min=1.0, max=max_weight).float() + + +@torch.no_grad() +def predict_dataset( + model: DirectFloorEventHead, + dataset: FloorEventDataset, + *, + batch_size: int, + device: torch.device, + num_workers: int, +) -> dict[str, np.ndarray]: + loader = DataLoader( + dataset, + batch_size=batch_size, + shuffle=False, + num_workers=num_workers, + pin_memory=device.type == "cuda", + ) + model.eval() + scores: list[np.ndarray] = [] + events: list[np.ndarray] = [] + horizons: list[np.ndarray] = [] + labels: list[np.ndarray] = [] + short_resume: list[np.ndarray] = [] + for batch in loader: + states, audio, oracle, oracle_valid, event, horizon, label, short = batch + logits = model( + states.to(device, non_blocking=True), + audio.to(device, non_blocking=True), + oracle.to(device, non_blocking=True), + oracle_valid.to(device, non_blocking=True), + ) + row = torch.arange(len(event), device=device) + selected = logits[ + row, + event.to(device, non_blocking=True), + horizon.to(device, non_blocking=True), + ] + scores.append(selected.sigmoid().cpu().numpy()) + events.append(event.numpy()) + horizons.append(horizon.numpy()) + labels.append(label.numpy().astype(bool)) + short_resume.append(short.numpy().astype(bool)) + return { + "score": np.concatenate(scores), + "event": np.concatenate(events), + "horizon": np.concatenate(horizons), + "label": np.concatenate(labels), + "short_resume": np.concatenate(short_resume), + } + + +def metric_rows( + predictions: dict[str, np.ndarray], + *, + matched_recall: float, + fixed_thresholds: dict[tuple[int, int], float | None] | None = None, + threshold_source: str, +) -> tuple[list[dict[str, Any]], dict[tuple[int, int], float | None]]: + rows: list[dict[str, Any]] = [] + thresholds: dict[tuple[int, int], float | None] = {} + for event_index, event_name in enumerate(EVENT_NAMES): + for horizon_index in range(NUM_HORIZONS): + mask = (predictions["event"] == event_index) & ( + predictions["horizon"] == horizon_index + ) + score = predictions["score"][mask] + label = predictions["label"][mask] + positive_scores = score[label] + negative_scores = score[~label] + key = (event_index, horizon_index) + threshold = ( + _threshold_for_recall(positive_scores, matched_recall) + if fixed_thresholds is None + else fixed_thresholds[key] + ) + thresholds[key] = threshold + hard_negative = predictions["short_resume"][mask] & ~label + rows.append( + { + "event_type": event_name, + "horizon_index": horizon_index + 1, + "nominal_seconds": (horizon_index + 1) * STEP_SECONDS, + "num_positive": int(label.sum()), + "num_negative": int((~label).sum()), + "positive_rate": float(label.mean()) if len(label) else None, + "average_precision": _safe_ap(label, score), + "matched_recall": matched_recall, + "threshold": threshold, + "threshold_source": threshold_source, + "achieved_positive_recall": _rate_at_threshold( + positive_scores, threshold + ), + "negative_trigger_rate": _rate_at_threshold( + negative_scores, threshold + ), + "short_resume_hard_negative_count": int(hard_negative.sum()), + "short_resume_trigger_rate": _rate_at_threshold( + score[hard_negative], threshold + ), + } + ) + return rows, thresholds + + +def mean_event_ap(rows: list[dict[str, Any]]) -> float: + values = [ + float(row["average_precision"]) + for row in rows + if row["average_precision"] is not None + ] + return float(np.mean(values)) if values else float("-inf") + + +def train_model( + args: argparse.Namespace, + model: DirectFloorEventHead, + train_dataset: FloorEventDataset, + dev_dataset: FloorEventDataset, + device: torch.device, +) -> tuple[dict[str, torch.Tensor], float, int, list[dict[str, Any]]]: + generator = torch.Generator().manual_seed(args.seed) + loader = DataLoader( + train_dataset, + batch_size=args.batch_size, + shuffle=True, + generator=generator, + num_workers=args.num_workers, + pin_memory=device.type == "cuda", + ) + weights = positive_weights(train_dataset, args.max_positive_weight).to(device) + optimizer = torch.optim.AdamW( + model.parameters(), + lr=args.learning_rate, + weight_decay=args.weight_decay, + ) + best_score = float("-inf") + best_epoch = 0 + best_state: dict[str, torch.Tensor] | None = None + history: list[dict[str, Any]] = [] + model.to(device) + for epoch in range(1, args.epochs + 1): + model.train() + total_loss = 0.0 + total_examples = 0 + for batch in loader: + states, audio, oracle, oracle_valid, event, horizon, label, _short = batch + states = states.to(device, non_blocking=True) + audio = audio.to(device, non_blocking=True) + oracle = oracle.to(device, non_blocking=True) + oracle_valid = oracle_valid.to(device, non_blocking=True) + event = event.to(device, non_blocking=True) + horizon = horizon.to(device, non_blocking=True) + label = label.to(device, non_blocking=True) + logits = model(states, audio, oracle, oracle_valid) + batch_rows = torch.arange(len(event), device=device) + selected = logits[batch_rows, event, horizon] + loss = F.binary_cross_entropy_with_logits( + selected, + label, + weight=torch.where(label.bool(), weights[event, horizon], 1.0), + ) + optimizer.zero_grad(set_to_none=True) + loss.backward() + optimizer.step() + total_loss += float(loss.detach()) * len(event) + total_examples += len(event) + + dev_predictions = predict_dataset( + model, + dev_dataset, + batch_size=args.batch_size, + device=device, + num_workers=args.num_workers, + ) + dev_rows, _ = metric_rows( + dev_predictions, + matched_recall=args.matched_recall, + threshold_source="development_sessions", + ) + score = mean_event_ap(dev_rows) + record = { + "epoch": epoch, + "train_loss": total_loss / max(total_examples, 1), + "dev_mean_event_ap": score, + } + history.append(record) + print( + f"epoch={epoch:02d} train_loss={record['train_loss']:.6f} " + f"dev_mean_event_ap={score:.6f}" + ) + if score > best_score: + best_score = score + best_epoch = epoch + best_state = { + key: value.detach().cpu().clone() + for key, value in model.state_dict().items() + } + if best_state is None: + raise RuntimeError("Training produced no checkpoint") + return best_state, best_score, best_epoch, history + + +def write_metrics_csv(path: Path, rows: list[dict[str, Any]]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=list(rows[0])) + writer.writeheader() + writer.writerows(rows) + + +def main() -> None: + args = parse_args() + if args.epochs <= 0 or args.batch_size <= 0: + raise ValueError("--epochs and --batch-size must be positive") + if not 0 < args.matched_recall <= 1: + raise ValueError("--matched-recall must be in (0,1]") + if args.midturn_pause_max_seconds <= 0: + raise ValueError("--midturn-pause-max-seconds must be positive") + if args.max_positive_weight < 1: + raise ValueError("--max-positive-weight must be at least 1") + train_sessions = normalize_sessions(args.train_sessions) + dev_sessions = normalize_sessions(args.dev_sessions) + eval_sessions = normalize_sessions(args.eval_sessions or dev_sessions) + if not args.eval_only: + overlap = set(train_sessions) & set(dev_sessions) + if overlap: + raise ValueError(f"Training/development sessions overlap: {sorted(overlap)}") + if set(train_sessions + dev_sessions) & set(OFFICIAL_TEST_SESSIONS): + raise ValueError("Official test Sessions 1--3 cannot train or select the probe") + if args.checkpoint.exists() and not args.overwrite: + raise FileExistsError( + f"Refusing to overwrite checkpoint {args.checkpoint}; pass --overwrite" + ) + elif not args.checkpoint.is_file(): + raise FileNotFoundError(args.checkpoint) + summary_path = args.result_dir / "summary.json" + if summary_path.exists() and not args.overwrite: + raise FileExistsError( + f"Refusing to overwrite result {summary_path}; pass --overwrite" + ) + + random.seed(args.seed) + np.random.seed(args.seed) + torch.manual_seed(args.seed) + if torch.cuda.is_available(): + torch.cuda.manual_seed_all(args.seed) + device = torch.device(args.device) + if device.type == "cuda" and not torch.cuda.is_available(): + raise RuntimeError(f"CUDA requested but unavailable: {args.device}") + midturn_steps = int( + math.ceil(args.midturn_pause_max_seconds / STEP_SECONDS) + ) + + requested = list( + dict.fromkeys( + eval_sessions + if args.eval_only + else train_sessions + dev_sessions + eval_sessions + ) + ) + loaded = { + session: load_probe_session(args, session) + for session in requested + } + feature_config = validate_session_shapes(list(loaded.values())) + model_config = { + "state_shape": feature_config["state_shape"], + "audio_dim": feature_config["audio_dim"], + "oracle_dim": feature_config["oracle_dim"], + "layer_fusion": args.layer_fusion, + "local_projection_dim": args.local_projection_dim, + "person_attention_dim": args.person_attention_dim, + "person_attention_heads": args.person_attention_heads, + "dropout": args.dropout, + } + print( + f"condition={feature_config['oracle_condition']} " + f"state_shape={feature_config['state_shape']} " + f"audio_dim={feature_config['audio_dim']} " + f"oracle_dim={feature_config['oracle_dim']} " + f"decision_timing={args.decision_timing}" + ) + + if args.eval_only: + checkpoint = torch.load(args.checkpoint, map_location="cpu", weights_only=False) + if checkpoint.get("format") != "easycom-direct-floor-event-probe-v1": + raise ValueError(f"Invalid probe checkpoint format: {args.checkpoint}") + saved_recall = float(checkpoint.get("matched_recall", float("nan"))) + if not math.isclose(saved_recall, args.matched_recall): + raise ValueError( + f"Checkpoint matched recall is {saved_recall}, requested " + f"{args.matched_recall}" + ) + train_sessions = list(checkpoint.get("train_sessions", [])) + dev_sessions = list(checkpoint.get("dev_sessions", [])) + if not dev_sessions: + raise ValueError("Probe checkpoint has no development sessions") + saved_config = checkpoint.get("model_config") + if not isinstance(saved_config, dict): + raise ValueError(f"Invalid probe checkpoint: {args.checkpoint}") + comparable_saved = dict(saved_config) + comparable_saved["state_shape"] = tuple(comparable_saved["state_shape"]) + if comparable_saved != model_config: + # Evaluation must follow the checkpoint architecture; only the + # feature dimensions/condition must match current data. + expected_features = { + key: comparable_saved[key] + for key in ("state_shape", "audio_dim", "oracle_dim") + } + observed_features = { + key: model_config[key] + for key in ("state_shape", "audio_dim", "oracle_dim") + } + if expected_features != observed_features: + raise ValueError( + f"Checkpoint/data feature mismatch: {expected_features} vs " + f"{observed_features}" + ) + model_config = comparable_saved + if checkpoint.get("oracle_condition") != feature_config["oracle_condition"]: + raise ValueError( + "Checkpoint/data oracle condition mismatch: " + f"{checkpoint.get('oracle_condition')!r} vs " + f"{feature_config['oracle_condition']!r}" + ) + if checkpoint.get("decision_timing") != args.decision_timing: + raise ValueError("Checkpoint/data decision timing mismatch") + model = DirectFloorEventHead(**model_config) + model.load_state_dict(checkpoint["model_state"]) + best_epoch = int(checkpoint["best_epoch"]) + best_score = float(checkpoint["best_dev_mean_event_ap"]) + history = checkpoint.get("history", []) + dev_rows = checkpoint["dev_metrics"] + raw_thresholds = checkpoint["dev_thresholds"] + thresholds = { + tuple(int(part) for part in key.split(",")): value + for key, value in raw_thresholds.items() + } + else: + train_dataset = FloorEventDataset( + [loaded[name] for name in train_sessions], + decision_timing=args.decision_timing, + midturn_pause_max_steps=midturn_steps, + ) + dev_dataset = FloorEventDataset( + [loaded[name] for name in dev_sessions], + decision_timing=args.decision_timing, + midturn_pause_max_steps=midturn_steps, + ) + print( + f"train_examples={len(train_dataset)} " + f"dev_examples={len(dev_dataset)}" + ) + model = DirectFloorEventHead(**model_config) + best_state, best_score, best_epoch, history = train_model( + args, model, train_dataset, dev_dataset, device + ) + model.load_state_dict(best_state) + model.to(device) + dev_predictions = predict_dataset( + model, + dev_dataset, + batch_size=args.batch_size, + device=device, + num_workers=args.num_workers, + ) + dev_rows, thresholds = metric_rows( + dev_predictions, + matched_recall=args.matched_recall, + threshold_source="development_sessions:" + ",".join(dev_sessions), + ) + args.checkpoint.parent.mkdir(parents=True, exist_ok=True) + checkpoint_payload = { + "format": "easycom-direct-floor-event-probe-v1", + "model_config": model_config, + "model_state": best_state, + "oracle_condition": feature_config["oracle_condition"], + "oracle_contract": feature_config["oracle_contract"], + "decision_timing": args.decision_timing, + "train_sessions": train_sessions, + "dev_sessions": dev_sessions, + "matched_recall": args.matched_recall, + "best_epoch": best_epoch, + "best_dev_mean_event_ap": best_score, + "history": history, + "dev_metrics": dev_rows, + "dev_thresholds": { + f"{event},{horizon}": value + for (event, horizon), value in thresholds.items() + }, + "args": { + key: str(value) if isinstance(value, Path) else value + for key, value in vars(args).items() + }, + } + torch.save(checkpoint_payload, args.checkpoint) + + eval_dataset = FloorEventDataset( + [loaded[name] for name in eval_sessions], + decision_timing=args.decision_timing, + midturn_pause_max_steps=midturn_steps, + ) + print(f"eval_sessions={','.join(eval_sessions)} eval_examples={len(eval_dataset)}") + model.to(device) + eval_predictions = predict_dataset( + model, + eval_dataset, + batch_size=args.batch_size, + device=device, + num_workers=args.num_workers, + ) + eval_rows, _ = metric_rows( + eval_predictions, + matched_recall=args.matched_recall, + fixed_thresholds=thresholds, + threshold_source="development_sessions:" + ",".join(dev_sessions), + ) + payload = { + "protocol": "easycom_direct_floor_event_probe_v1", + "analysis_only": True, + "interpretation": ( + "Direct competing-risk response/yield supervision over frozen Omni, " + "mixed audio and privileged aggregate oracle history. This diagnoses " + "objective/information alignment and is not a deployable system." + ), + "decision_timing": args.decision_timing, + "oracle_condition": feature_config["oracle_condition"], + "train_sessions": train_sessions, + "dev_sessions": dev_sessions, + "eval_sessions": eval_sessions, + "matched_recall": args.matched_recall, + "checkpoint": str(args.checkpoint.resolve()), + "best_epoch": best_epoch, + "best_dev_mean_event_ap": best_score, + "model_config": model_config, + "history": history, + "dev_metrics": dev_rows, + "eval_metrics": eval_rows, + "eval_censoring_audit": eval_dataset.censoring, + "num_eval_examples": len(eval_dataset), + } + args.result_dir.mkdir(parents=True, exist_ok=True) + summary_path.write_text(json.dumps(payload, indent=2) + "\n") + write_metrics_csv(args.result_dir / "metrics.csv", eval_rows) + print( + f"saved condition={feature_config['oracle_condition']} " + f"best_epoch={best_epoch} dev_mean_event_ap={best_score:.6f}" + ) + for event_name in EVENT_NAMES: + event_rows = [row for row in eval_rows if row["event_type"] == event_name] + print( + f"{event_name}: mean_AP={mean_event_ap(event_rows):.6f}, " + f"2s_AP={event_rows[-1]['average_precision']}, " + f"2s_error_rate@R={args.matched_recall:g}=" + f"{event_rows[-1]['negative_trigger_rate']}" + ) + print(f"summary: {summary_path}") + print(f"metrics: {args.result_dir / 'metrics.csv'}") + + +if __name__ == "__main__": + main()