From b5ce64702a871d666a065ddf40e87949e027245c Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Fri, 28 Aug 2026 00:25:54 +0800 Subject: [PATCH] Add Ego4D local AV ablation workflow --- README.md | 10 ++ docs/ego4d_local_baselines.md | 164 ++++++++++++++++++++++++++ docs/local_feature_baseline.md | 6 +- easycom_head.py | 8 +- extract_turn_local_features.py | 171 +++++++++++++++++++++------ run_ego4d_local_baselines.py | 175 ++++++++++++++++++++++++++++ summarize_local_baselines.py | 32 ++++- test_ego4d_local_workflow.py | 206 +++++++++++++++++++++++++++++++++ 8 files changed, 731 insertions(+), 41 deletions(-) create mode 100644 docs/ego4d_local_baselines.md create mode 100644 run_ego4d_local_baselines.py create mode 100644 test_ego4d_local_workflow.py diff --git a/README.md b/README.md index b0a0a13..d820fe3 100644 --- a/README.md +++ b/README.md @@ -284,6 +284,16 @@ transition prevalence is 11.21718%, while transition AP is approximately over R1 by only 0.01162 points and transition AP by 0.16660 points. Seed 2 is still required before freezing this as a complete null ablation. +## Ego4D local-feature ablation family + +To test whether the recent-audio / per-person-visual gains transfer from EasyCom, +see [the Ego4D local-feature experiment guide](docs/ego4d_local_baselines.md). +It reuses existing global Omni features and the EgoSpeak manifest, extracts local +A/V directly from raw clips, and runs **G / G+A / G+V / G+A+V** for four epochs, +without a dev set. `run_ego4d_local_baselines.py` preflights all train/test files +and encoder contracts before training; the summary reports matched-seed deltas, +including AV minus audio. No new accuracy claim is implied by adding the code. + ## Current Ego4D context-length results The following single-seed snapshot was recorded on 2026-08-22. Both variants diff --git a/docs/ego4d_local_baselines.md b/docs/ego4d_local_baselines.md new file mode 100644 index 0000000..39d8c25 --- /dev/null +++ b/docs/ego4d_local_baselines.md @@ -0,0 +1,164 @@ +# Ego4D：近期音频／逐人视觉增强的四组对照 + +目标是检验 EasyCom 上的局部特征收益能否迁移到 Ego4D，而不是预先承诺提分。 +保留现有 Ego4D 全局 Omni 特征 G、EgoSpeak manifest 和 target；不重提 Thinker， +不重新编码视频，不重划 train/test。新增的两路侧路特征一次提取、三组复用： + +| 实验名 | 输入 | 回答的问题 | +|---|---|---| +| baseline | G | 同训练配置下的基准 | +| audio | G + A | 最近 1s 音频是否补回全局 readout 丢失的信息？ | +| visual | G + V | 人物局部视觉是否有效？ | +| av | G + A + V | 在音频增强之外，视觉是否仍有增益？重点比较 AV − A | + +`audio` **不是纯音频模型**；四组都包含 G。A 是同一 clip 的视频音轨， +16kHz mono 后按 `[max(0,t-1),t)` 切窗，每 200ms 用冻结 Qwen audio encoder 重编码。 +V 在原视频的固定 2FPS 网格读取可见人物框，冻结 Qwen vision encoder 后依次做 +空间、逐人、最近 1s 有效帧的均值池化。Qwen 3B 的 A/V 均保存原生 2048D； +训练时各投影到 256D，融合后沿用现有预测 head。无有效框时 V=0 且 validity=False， +不能拿缺框当损坏样本直接删掉，否则四组测试样本不一致。 + +默认统一 **4 epochs、无 dev、LSTR-MCE、无 transition auxiliary**；batch=512、 +lr=7e-5、weight_decay=5e-5、warmup=1.6 epochs、warmup_factor=0.3、dropout=0.2。 +训练只取 manifest 的 train，评测只取 test。现有正式 manifest 应显示 346 train / 87 test； +程序读取该文件，不会自行随机拆分。预检 test 只检查输入完整性，不计算指标或选模型。 + +## 1. 设置远程路径 + +在 `/data/wangx/Code/icassp2027-experiment` 的 Bash 中执行。只需把 `FEATURE_ROOT` +改成**实际已提完的 Ego4D 全局特征目录**；不能误用 EasyCom 的 Session 文件。 +建议本轮固定已有 8s 全局特征，四组不要混用不同 context、层数或 checkpoint。 + +```bash +cd /data/wangx/Code/icassp2027-experiment +export CUDA_VISIBLE_DEVICES=0 +export OMP_NUM_THREADS=4 MKL_NUM_THREADS=4 OPENBLAS_NUM_THREADS=4 + +DATA_ROOT=/data/wangx/Datasets/ego4d +MODEL=/data/wangx/Models/Qwen2.5-Omni-3B +FEATURE_ROOT=/替换为实际的/Ego4D全局特征目录 +TARGET_ROOT="${DATA_ROOT}/ego4d_target_frame" +MANIFEST="${DATA_ROOT}/ego4d_egospeak.tsv" +LOCAL_ROOT="${DATA_ROOT}/omni_features/local_qwen_1s" + +LOCAL_ARGS=( + --dataset ego4d --data-root "$DATA_ROOT" --manifest "$MANIFEST" + --feature-root "$FEATURE_ROOT" --target-root "$TARGET_ROOT" + --model-id "$MODEL" --device cuda:0 --dtype bfloat16 --flash-attention-2 + --video-fps 2 --visual-window-seconds 1 --audio-window-seconds 1 +) +``` + +原始输入应存在：`v1/clips/.mp4`、`v1/annotations/av_train.json` 和 `av_val.json`。 +这里 `av_val.json` 是原 AVD 标注的文件名，不意味着把它直接当作本实验的 test 划分； +train/test 始终按 EgoSpeak manifest。非标准导出路径可加 `--ego4d-annotations `。 +局部编码器支持现有 Transformers 4.57.6，也支持 4.52.3；同一次实验所有 clips 应统一版本。 + +## 2. 先检查一条，再做 4s smoke + +```bash +# 检查文件、全局时间网格和 target 长度；不加载模型，不解码，不写输出。 +python extract_turn_local_features.py "${LOCAL_ARGS[@]}" \ + --limit 1 --dry-run --output-root "$LOCAL_ROOT" + +# 不加载模型，写带框预览；确认框落在人脸上、clip-local 帧号正确。 +python extract_turn_local_features.py "${LOCAL_ARGS[@]}" \ + --limit 1 --max-steps 20 --preview-only \ + --output-root features/ego4d_local_preview + +# 同时提取 V 和 A；无需另跑音频脚本。 +python extract_turn_local_features.py "${LOCAL_ARGS[@]}" \ + --limit 1 --max-steps 20 \ + --output-root features/ego4d_local_smoke +``` + +预览位于 `features/ego4d_local_preview/_preview/`，包含图片与 `boxes.json`。 +最好再用 `--split test --limit 1` 检查一条 test clip；`--dry-run` 不能替代像素与框检查。 +4s smoke 应得到 V/A 各 `[20,2048]`、`partial=True`；此目录不能用于正式训练。 +V-valid 不必是 100%，Ego4D 有无可见人物和缺框的时刻；若有明显人物却长期无有效框，先停下来检查标注。 + +## 3. 提取完整 manifest：前台、模型只加载一次 + +```bash +python extract_turn_local_features.py "${LOCAL_ARGS[@]}" \ + --reuse-existing --output-root "$LOCAL_ROOT" +``` + +前台显示 clip 编号、视觉 frame 进度和音频 state 进度。中断后可重跑同一命令： +只复用完整且样本、时间、特征维度、版本及提取配置一致的 `.pt`；部分或不兼容文件会报错。 +输出原子替换，失败的临时文件不会被当作成功样本。不要用 `--overwrite` 当作“继续”。 +checkpoint 的 ID 是用户声明，不是权重文件校验和：不要在同一个 ID/路径下替换权重再复用旧特征。 + +可用 `--split train|test`、`--start-index N --end-index M` 或 `--limit N` 选择子集。 +索引是**先按 split 过滤，再按 manifest 顺序的 1-based 闭区间**；不传 split 时保留整个文件的行顺序。 +如果自行在多卡不同终端运行，必须选择互不重叠的区间和同一套编码器配置；默认不启动后台任务。 + +## 4. 跑 seed0 四组 + +```bash +EXP_ARGS=( + --data-root "$DATA_ROOT" --manifest "$MANIFEST" + --feature-root "$FEATURE_ROOT" --target-root "$TARGET_ROOT" + --local-feature-root "$LOCAL_ROOT" + --checkpoint-root checkpoints/ego4d_local_b1 + --result-root results/ego4d_local_b1 +) + +# 逐 clip 校验全部 train/test 输入、局部配置，打印命令但不训练。 +python run_ego4d_local_baselines.py "${EXP_ARGS[@]}" --seeds 0 --dry-run + +# 前台顺序训练/评测 baseline、audio、visual、av，并自动汇总。 +python run_ego4d_local_baselines.py "${EXP_ARGS[@]}" --seeds 0 + +# 补同设置 seed1/2，不覆盖 seed0。 +python run_ego4d_local_baselines.py "${EXP_ARGS[@]}" --seeds 1 2 +``` + +自动检测 G：`[T,L,D]` 使用 concat-MLP，`[T,D]` 使用原单层 head；不要求为 last4 重提全局特征。 +如果原 stacked baseline 使用 fixed-mean，可在四组统一加 `--layer-fusion fixed-mean`。 +不要拿不同 epoch/layer-fusion 的旧 baseline 直接计算新增分支的增量。 + +预检不会一次把 433 条全局特征全部装进 RAM；实际训练仍沿用原训练器，会加载整个 train split。 +last4 特征可能需要数十 GB 主机内存。多卡并行 head 训练也会复制主机内存，不是只看 GPU 是否空闲。 + +若只跑一组，增加 `--variants audio`；若某一组训练中断，默认拒绝覆盖现有 checkpoint/result。 +可先运行剩余组合，例如 `--variants visual av --seeds 0`，或为重跑选择新的输出根。 +baseline 单独运行不要求局部特征；整个四组 family 会在任何训练前检查所有必要局部特征。 + +## 5. 打印结果 + +```bash +python summarize_local_baselines.py \ + --root results/ego4d_local_b1 --dataset ego4d \ + --protocol causal_valid_only --paired-deltas + +# 对齐 EgoSpeak 的 full-timeline 表格时，单独报告这个协议。 +python summarize_local_baselines.py \ + --root results/ego4d_local_b1 --dataset ego4d \ + --protocol egospeak_full_timeline --paired-deltas +``` + +表格给出每个 seed、均值与 sample std，以及匹配 seed 的 `A−G`、`V−G`、`AV−G`、`AV−A`。 +不要混用两个评测协议。优先看 10-horizon mean mAP、wearer AP 和 AV−A， +不要只凭单 seed 的 0.2s mAP 宣称跨数据集提升。 + +## 论文可解释的范围 + +这轮回答“在冻结的全局 Omni readout 上加入近期编码器特征是否有用”，不是 Qwen 对 wav2vec +的受控替换实验，也不是逐人音视频对齐模型。使用提供的人物 tracks，忽略 wearer 框、讲话区间和 +说话人标签；缺框不向未来插值。但提供框本身不是在线检测，不能据此声称端到端在线视觉系统。 +新增分支带来额外参数，若收益稳定，再补参数量匹配的全局 head 控制。 + +本轮不改变已有的音频重采样、标签索引和 horizon 定义。5Hz 是输出网格； +不等于视觉获得 5FPS 唇动信息，也不等于已完成原始波形层面的端到端因果审计。 +如后续修正物理时间协议，必须在所有对照组统一修正并重新跑，不能只改增强组。 + +## 回归测试 + +```bash +python -m unittest -q test_easycom_head_objectives test_local_turn_features test_ego4d_local_workflow +``` + +CPU 回归包括 manifest 顺序/防交叉重复、完整批量提取的模型复用、断点文件校验、 +单层/多层全局特征、测试集输入预检，以及合成 Ego4D 上四组各 4 epochs 的训练/推理与增量汇总。 +真实 Ego4D 像素、提供框和 3B/L20 提取仍需按上面的 preview/smoke 步骤验证。 diff --git a/docs/local_feature_baseline.md b/docs/local_feature_baseline.md index 19614d4..3893a8d 100644 --- a/docs/local_feature_baseline.md +++ b/docs/local_feature_baseline.md @@ -125,7 +125,8 @@ python extract_turn_local_features.py "${LOCAL_ARGS[@]}" \ ``` 默认前台顺序执行，显示视觉帧进度和音频 state 进度；模型只加载一次。 -已有文件会报错，恢复时在 `--samples` 中只列尚未完成的 sessions。 +已有文件默认会报错；恢复时可只列尚未完成的 sessions，或增加 `--reuse-existing` +逐文件验证完整性、时间对齐和提取配置后跳过已完成文件。 `--overwrite` 是明确重算，不会重用不兼容文件。不要覆盖原始 feature root。 `persons` 还保留每帧变长的逐人特征、frame_offsets、框、帧号和时间，供后续 B2 复用。 若在不同挂载路径提取同一 checkpoint，使用相同 `--encoder-id`；不同权重不可共用这个 ID。 @@ -171,6 +172,9 @@ Loader 会检查 sample、时间、label_indices、维度、valid mask 和 encod ## Ego4D 接口 +完整命令见 [Ego4D 四组局部增强实验](ego4d_local_baselines.md)：新增 manifest 批量提取、 +续跑校验、`run_ego4d_local_baselines.py` 四组运行入口和匹配 seed 的增量汇总。 + 提取脚本同样支持 `--dataset ego4d`：输入根为 `/data/wangx/Datasets/ego4d`， 读取 `v1/clips/.mp4`、`v1/annotations/av_train.json` 与 `av_val.json`， `--samples` 传已有 Omni `.pt` 的 clip UID。`--target-root` 指向 `ego4d_target_frame`。 diff --git a/easycom_head.py b/easycom_head.py index 75477aa..385fd65 100644 --- a/easycom_head.py +++ b/easycom_head.py @@ -640,10 +640,12 @@ def normalize_sample_name( def load_ego4d_manifest_split( manifest: str | Path, - split: Literal["train", "test"], + split: Literal["train", "test", "all"], ) -> list[str]: - """Load one ordered UID split from ``ego4d_egospeak.tsv``.""" + """Load ordered UIDs from ``ego4d_egospeak.tsv``; all preserves row order.""" + if split not in {"train", "test", "all"}: + raise ValueError(f"Invalid Ego4D split: {split}") path = Path(manifest) if not path.is_file(): raise FileNotFoundError(path) @@ -671,7 +673,7 @@ def load_ego4d_manifest_split( f"Duplicate Ego4D UID {uid} at {path}:{row_number}" ) seen.add(uid) - if row_split == split: + if split == "all" or row_split == split: selected.append(uid) if not selected: diff --git a/extract_turn_local_features.py b/extract_turn_local_features.py index 12ed97f..34cf926 100644 --- a/extract_turn_local_features.py +++ b/extract_turn_local_features.py @@ -19,10 +19,10 @@ import numpy as np import torch from tqdm.auto import tqdm -from easycom_head import normalize_sample_name +from easycom_head import load_ego4d_manifest_split, normalize_sample_name from local_turn_features import ( LOCAL_FORMAT, causal_frame_ticks, crop_box, easycom_boxes, - pooled_recent_visual, reference_alignment, + load_local_features, pooled_recent_visual, reference_alignment, ) @@ -164,6 +164,7 @@ def ego4d_clip_boxes(clip: dict[str, Any]) -> dict[int, list[dict[str, Any]]]: def load_ego4d_boxes(paths: list[Path], names: list[str]) -> dict[str, Any]: result = {} + selected = set(names) for path in paths: print(f"Reading provided tracks: {path}", flush=True) with path.open() as handle: @@ -171,7 +172,7 @@ def load_ego4d_boxes(paths: list[Path], names: list[str]) -> dict[str, Any]: for video in payload["videos"]: for clip in video["clips"]: uid = clip["clip_uid"] - if uid in names: + if uid in selected: if uid in result: raise ValueError(f"Duplicate Ego4D annotation UID: {uid}") result[uid] = ego4d_clip_boxes(clip) @@ -338,6 +339,54 @@ def extract_local(media, annotations, encoder, times: torch.Tensor, args: argpar } +def select_samples(args: argparse.Namespace) -> list[str]: + if args.manifest is not None: + names = load_ego4d_manifest_split(args.manifest, args.split) + # One-based inclusive range in the selected split, then an optional limit. + names = names[args.start_index - 1:args.end_index] + if args.limit is not None: + names = names[:args.limit] + else: + names = [normalize_sample_name(value, args.dataset) for value in args.samples] + if not names: + raise ValueError("No samples selected") + if len(set(names)) != len(names): + raise ValueError("Duplicate samples") + return names + + +def requested_local_config(args: argparse.Namespace) -> dict[str, Any]: + """Settings that must match before reusing an existing complete sidecar.""" + return { + "encoder_id": args.encoder_id or args.model_id, + "dtype": args.dtype, + "attention": "flash_attention_2" if args.flash_attention_2 else "default", + "visual_fps": args.video_fps, "visual_window_seconds": args.visual_window_seconds, + "audio_window_seconds": args.audio_window_seconds, + "visual_pooling": "spatial-mean_then_equal-person-mean_then_equal-valid-frame-mean", + "audio_pooling": "valid-output-token-mean", "audio_source": "video-mixed-mono-16khz", + "box_kind": args.box_kind if args.dataset == "easycom" else "ego4d-provided-tracks", + "box_coordinate_space": args.box_coordinate_space, "crop_expansion": args.crop_expansion, + "max_pixels": args.max_pixels, "uses_speech_labels": False, + "causal_policy": "half-open-trailing-window", "box_protocol": "provided-boxes-not-online-detector", + } + + +def validate_reusable_sidecar(args, name, reference, version, expected_contract=None): + path = args.output_root / f"{name}.pt" + _, contract = load_local_features(path, reference, sample_id=name, dataset=args.dataset, + modalities="av", expected_contract=expected_contract) + config = contract["config"] + expected = requested_local_config(args) + if version is not None: + expected["transformers"] = version + mismatches = [key for key, value in expected.items() if config.get(key) != value] + if mismatches or not config.get("model_config_sha256") or not config.get("transformers"): + raise ValueError(f"Cannot reuse {path}: incompatible/incomplete encoder settings {mismatches}; " + "use a new output root or --overwrite") + return contract + + def parse_args(argv=None): p = argparse.ArgumentParser(description=__doc__) p.add_argument("--dataset", choices=("easycom", "ego4d"), default="easycom") @@ -345,7 +394,13 @@ def parse_args(argv=None): p.add_argument("--feature-root", required=True, type=Path) p.add_argument("--target-root", required=True, type=Path) p.add_argument("--output-root", required=True, type=Path) - p.add_argument("--samples", required=True, nargs="+", help="EasyCom session numbers or Ego4D clip UIDs") + selection = p.add_mutually_exclusive_group(required=True) + selection.add_argument("--samples", nargs="+", help="EasyCom session numbers or Ego4D clip UIDs") + selection.add_argument("--manifest", type=Path, help="Ego4D EgoSpeak TSV; never creates a new split") + p.add_argument("--split", choices=("all", "train", "test"), default="all") + p.add_argument("--start-index", type=int, default=1, help="One-based index AFTER split filtering") + p.add_argument("--end-index", type=int, help="Inclusive index AFTER split filtering") + p.add_argument("--limit", type=int, help="First N clips in the selected manifest range") p.add_argument("--ego4d-annotations", type=Path, nargs="+") p.add_argument("--model-id", default="Qwen/Qwen2.5-Omni-3B") p.add_argument("--encoder-id", help="Stable checkpoint identity for relocated copies; default: model-id") @@ -370,9 +425,25 @@ def parse_args(argv=None): p.add_argument("--decoder-gpu-id", type=int, default=0, help="Visible-device index, not physical GPU number") p.add_argument("--max-steps", type=int, help="Smoke test only; partial sidecars are rejected by training") p.add_argument("--preview-only", action="store_true", help="Save box overlays without loading the model") - p.add_argument("--overwrite", action="store_true") + existing = p.add_mutually_exclusive_group() + existing.add_argument("--overwrite", action="store_true") + existing.add_argument("--reuse-existing", action="store_true", help="Validate and skip complete matching sidecars") + p.add_argument("--dry-run", action="store_true", help="Validate references/targets and list work; no model, media decode or writes") p.add_argument("--no-progress", action="store_true") args = p.parse_args(argv) + if args.manifest is not None and args.dataset != "ego4d": + p.error("--manifest is only supported with --dataset ego4d") + if args.manifest is None and (args.split != "all" or args.start_index != 1 + or args.end_index is not None or args.limit is not None): + p.error("--split/--start-index/--end-index/--limit require --manifest") + if args.start_index < 1 or (args.end_index is not None and args.end_index < args.start_index): + p.error("Invalid manifest index range") + if args.limit is not None and args.limit <= 0: + p.error("--limit must be positive") + if args.reuse_existing and (args.max_steps is not None or args.preview_only): + p.error("--reuse-existing is for full extraction, not smoke tests or previews") + if args.dry_run and args.preview_only: + p.error("--dry-run and --preview-only are separate checks") if args.box_coordinate_space is None: args.box_coordinate_space = "encoded" if args.dataset == "easycom" else "camera" for name in ("video_fps", "audio_window_seconds", "visual_window_seconds", "crop_expansion"): @@ -389,30 +460,60 @@ def parse_args(argv=None): return args -def main() -> None: - args = parse_args() - if not args.preview_only: +def main(argv=None) -> None: + args = parse_args(argv) + version = None + if not args.preview_only and not args.dry_run: import transformers # Fail before indexing/decoding source media, not after expensive I/O. - validate_local_transformers_version(transformers.__version__) - names = [normalize_sample_name(value, args.dataset) for value in args.samples] - if len(set(names)) != len(names): - raise ValueError("Duplicate samples") - for name in names: - for path in (args.feature_root / f"{name}.pt", args.target_root / f"{name}.npy"): - if not path.is_file(): - raise FileNotFoundError(path) - if (args.output_root / f"{name}.pt").exists() and not args.overwrite and not args.preview_only: - raise FileExistsError(f"{name}.pt exists; use a new output root or --overwrite") - ego_boxes = {} + version = validate_local_transformers_version(transformers.__version__) + names = select_samples(args) + paths = [] if args.dataset == "ego4d": paths = args.ego4d_annotations or [args.data_root / "v1" / "annotations" / f"av_{split}.json" for split in ("train", "val")] - ego_boxes = load_ego4d_boxes(paths, names) + for path in paths: + if not path.is_file(): + raise FileNotFoundError(path) + for name in names: + required = [args.feature_root / f"{name}.pt", args.target_root / f"{name}.npy"] + if args.dataset == "ego4d": + required.append(args.data_root / "v1" / "clips" / f"{name}.mp4") + for path in required: + if not path.is_file(): + raise FileNotFoundError(path) + if ((args.output_root / f"{name}.pt").exists() and not args.overwrite + and not args.preview_only and not args.reuse_existing): + raise FileExistsError(f"{name}.pt exists; use --reuse-existing, a new output root or --overwrite") + pending, contract = [], None + # Stream through references on CPU; never keep all global states in memory. + for name in tqdm(names, desc="local preflight", unit="clip", disable=args.no_progress or args.dry_run): + reference = torch.load(args.feature_root / f"{name}.pt", map_location="cpu", weights_only=False) + times, _ = reference_alignment(reference) + target = np.load(args.target_root / f"{name}.npy", allow_pickle=False, mmap_mode="r") + if target.shape != (len(times), 3): + raise ValueError(f"Reference/target mismatch: {name}, {target.shape} vs {len(times)} states") + if args.reuse_existing and (args.output_root / f"{name}.pt").is_file(): + contract = validate_reusable_sidecar(args, name, reference, version, contract) + print(f"{name}: reuse ({len(times)} states)", flush=True) + else: + pending.append(name) + if args.dry_run: + steps = min(len(times), args.max_steps or len(times)) + print(f"{name}: extract ({steps} states, {steps / 5:g}s)", flush=True) + del reference, target + print(f"Selected {len(names)} clips; reuse={len(names) - len(pending)}, pending={len(pending)}", flush=True) + if args.dry_run: + print("Dry run only: annotation contents, pixel alignment and model/runtime are not checked.") + return + if not pending: + return + ego_boxes = load_ego4d_boxes(paths, pending) if args.dataset == "ego4d" else {} args.output_root.mkdir(parents=True, exist_ok=True) encoder = None - for name in names: + for position, name in enumerate(pending, 1): + print(f"[{position}/{len(pending)}] {name}", flush=True) path = args.feature_root / f"{name}.pt" reference = torch.load(path, map_location="cpu", weights_only=False) times, indices = reference_alignment(reference) @@ -430,30 +531,26 @@ def main() -> None: "annotations use --box-coordinate-space encoded; otherwise boxes shift right.", flush=True) if args.preview_only: preview_boxes(media, annotations, args, name, len(times) / 5) + ego_boxes.pop(name, None) + del media, annotations, reference continue if encoder is None: encoder = FrozenQwenLocalEncoder(args) + config = {**requested_local_config(args), + "model_config_sha256": encoder.model_config_sha256, + "transformers": encoder.transformers_version, + "visual_dim": encoder.visual_dim, "audio_dim": encoder.audio_dim} + new_contract = {"format": LOCAL_FORMAT, "modalities": "av", "config": config} + if contract is not None and contract != new_contract: + raise ValueError("Current encoder differs from reused sidecars; use a new output root") + contract = new_contract print(f"{name}: {len(times)} states; native visual={encoder.visual_dim}D audio={encoder.audio_dim}D", flush=True) output = extract_local(media, annotations, encoder, times, args) output.update({ "format": LOCAL_FORMAT, "dataset": args.dataset, "sample_id": name, "times": times, "label_indices": indices, "partial": len(times) != full_steps, "reference_features": str(path.resolve()), "direct_media": media.metadata(), - "config": { - "encoder_id": args.encoder_id or args.model_id, - "model_config_sha256": encoder.model_config_sha256, - "transformers": encoder.transformers_version, "dtype": args.dtype, - "attention": "flash_attention_2" if args.flash_attention_2 else "default", - "visual_dim": encoder.visual_dim, "audio_dim": encoder.audio_dim, - "visual_fps": args.video_fps, "visual_window_seconds": args.visual_window_seconds, - "audio_window_seconds": args.audio_window_seconds, - "visual_pooling": "spatial-mean_then_equal-person-mean_then_equal-valid-frame-mean", - "audio_pooling": "valid-output-token-mean", "audio_source": "video-mixed-mono-16khz", - "box_kind": args.box_kind if args.dataset == "easycom" else "ego4d-provided-tracks", - "box_coordinate_space": args.box_coordinate_space, "crop_expansion": args.crop_expansion, - "max_pixels": args.max_pixels, "uses_speech_labels": False, - "causal_policy": "half-open-trailing-window", "box_protocol": "provided-boxes-not-online-detector", - }, + "config": config, }) destination = args.output_root / f"{name}.pt" # Atomic replacement: a failed extraction never looks like a ready file. @@ -466,6 +563,8 @@ def main() -> None: temporary.unlink(missing_ok=True) print(f"Saved {destination}; V-valid={output['visual_valid'].float().mean():.1%}, " f"A-valid={output['audio_valid'].float().mean():.1%}, partial={output['partial']}", flush=True) + ego_boxes.pop(name, None) + del media, annotations, reference, output if __name__ == "__main__": diff --git a/run_ego4d_local_baselines.py b/run_ego4d_local_baselines.py new file mode 100644 index 0000000..97b98cb --- /dev/null +++ b/run_ego4d_local_baselines.py @@ -0,0 +1,175 @@ +"""Run Ego4D G / G+A / G+V / G+A+V: four epochs, no dev, manifest splits. + +All jobs run in the foreground. The full selected family is preflighted before +any training, including test feature contracts. No test metrics select epochs. +""" + +from __future__ import annotations + +import argparse +import os +from pathlib import Path +import shlex +import subprocess +import sys + +from easycom_head import load_ego4d_manifest_split, load_session_data + + +VARIANTS = ("baseline", "audio", "visual", "av") +REPO = Path(__file__).resolve().parent + + +def parse_args(argv=None): + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--data-root", type=Path, default=Path("/data/wangx/Datasets/ego4d")) + p.add_argument("--manifest", type=Path) + p.add_argument("--feature-root", required=True, type=Path, help="Your EXISTING global Omni features") + p.add_argument("--target-root", type=Path) + p.add_argument("--local-feature-root", type=Path) + p.add_argument("--checkpoint-root", type=Path, default=Path("checkpoints/ego4d_local_b1")) + p.add_argument("--result-root", type=Path, default=Path("results/ego4d_local_b1")) + p.add_argument("--variants", choices=VARIANTS, nargs="+", default=list(VARIANTS)) + p.add_argument("--seeds", type=int, nargs="+", default=[0, 1, 2]) + p.add_argument("--layer-fusion", choices=("auto", "single", "concat-mlp", "fixed-mean", "scalar-mixture"), + default="auto", help="auto: concat-mlp for stacked states, single for [T,D]") + p.add_argument("--local-projection-dim", type=int, default=256) + p.add_argument("--expected-mode", choices=("open-stream-replay", "causal-prefix"), default="open-stream-replay") + p.add_argument("--device", default="cuda:0", help="Visible device; select the physical GPU with CUDA_VISIBLE_DEVICES") + p.add_argument("--no-progress", action="store_true") + p.add_argument("--dry-run", action="store_true", help="Read/validate all inputs and print commands; no training or writes") + args = p.parse_args(argv) + args.manifest = args.manifest or args.data_root / "ego4d_egospeak.tsv" + args.target_root = args.target_root or args.data_root / "ego4d_target_frame" + args.local_feature_root = args.local_feature_root or args.data_root / "omni_features/local_qwen_1s" + for key in ("data_root", "manifest", "feature_root", "target_root", "local_feature_root", + "checkpoint_root", "result_root"): + setattr(args, key, getattr(args, key).expanduser().resolve()) + if len(set(args.variants)) != len(args.variants) or len(set(args.seeds)) != len(args.seeds): + p.error("Duplicate variants or seeds") + if any(seed < 0 for seed in args.seeds) or args.local_projection_dim < 0: + p.error("Seeds and local-projection-dim must be nonnegative") + if args.feature_root == args.local_feature_root: + p.error("Global and local feature roots must be different") + return args + + +def run_paths(args, variant, seed): + name = f"{variant}_seed{seed}" + return args.checkpoint_root / f"{name}.pth", args.result_root / name + + +def preflight(args) -> str: + train = load_ego4d_manifest_split(args.manifest, "train") + test = load_ego4d_manifest_split(args.manifest, "test") + names = train + test + print(f"EgoSpeak manifest: {len(train)} train / {len(test)} test; dev=none; epochs=4", flush=True) + use_local = any(variant != "baseline" for variant in args.variants) + wanted = set(args.variants) - {"baseline"} + local_modalities = next(iter(wanted)) if len(wanted) == 1 else "av" + roots = [(args.feature_root, ".pt"), (args.target_root, ".npy")] + if use_local: + roots.append((args.local_feature_root, ".pt")) + missing = [str(root / f"{name}{suffix}") for root, suffix in roots for name in names + if not (root / f"{name}{suffix}").is_file()] + if missing: + raise FileNotFoundError(f"Missing {len(missing)} input files (first 20):\n" + "\n".join(missing[:20])) + for variant in args.variants: + for seed in args.seeds: + checkpoint, result = run_paths(args, variant, seed) + if checkpoint.exists() or checkpoint.with_suffix(".pth.json").exists() or result.exists(): + raise FileExistsError(f"Refusing to overwrite {checkpoint} or {result}; " + "select unfinished variants/seeds or new output roots") + + shape, contract = None, None + counts = {"visual": [0, 0], "audio": [0, 0]} + total = 0 + for index, name in enumerate(names, 1): + data = load_session_data( + args.feature_root, args.target_root, name, dataset="ego4d", + expected_mode=args.expected_mode, + local_feature_root=args.local_feature_root if use_local else None, + local_modalities=local_modalities, local_contract=contract, + ) + if len(data.states) <= 10: + raise ValueError(f"{name}: no training rows with all ten future horizons") + current_shape = tuple(data.states.shape[1:]) + if shape is not None and current_shape != shape: + raise ValueError(f"Global feature shape differs: {name}: {current_shape} vs {shape}") + shape = current_shape + contract = data.feature_metadata.get("local_contract") + if contract is not None: + offset = 0 + for modality in ("visual", "audio"): + if local_modalities not in (modality, "av"): + continue + width = contract["config"][f"{modality}_dim"] + counts[modality][0] += int(data.local_features[:, offset + width].sum()) + counts[modality][1] += len(data.states) + offset += width + 1 + total += len(data.states) + del data # Validate one clip at a time, including all test clips. + if not args.no_progress and (index == 1 or index % 25 == 0 or index == len(names)): + print(f"preflight: {index}/{len(names)} clips", flush=True) + fusion = args.layer_fusion + if fusion == "auto": + fusion = "concat-mlp" if len(shape) == 2 else "single" + if fusion == "single" and len(shape) != 1: + raise ValueError("--layer-fusion single requires [T,D], not stacked global states") + if fusion != "single" and len(shape) != 2: + raise ValueError(f"--layer-fusion {fusion} requires stacked [T,L,D] global states") + print(f"Validated {total} states; global shape per row={shape}; fusion={fusion}", flush=True) + for modality, (valid, count) in counts.items(): + if count: + note = "; provided boxes, not an online detector" if modality == "visual" else "" + print(f"Local {modality} coverage: {valid / count:.2%} ({valid}/{count}){note}", flush=True) + print("Training loads the full train split into host RAM; stacked global features may require tens of GB.", flush=True) + return fusion + + +def experiment_commands(args, fusion): + common = ["--dataset", "ego4d", "--manifest", str(args.manifest), + "--feature-root", str(args.feature_root), "--target-root", str(args.target_root), + "--expected-mode", args.expected_mode, "--device", args.device] + if args.no_progress: + common.append("--no-progress") + for variant in args.variants: + for seed in args.seeds: + checkpoint, result = run_paths(args, variant, seed) + train = [sys.executable, str(REPO / "train_turn_head.py"), *common, + "--output", str(checkpoint), "--epochs", "4", "--batch-size", "512", + "--learning-rate", "7e-5", "--weight-decay", "5e-5", + "--warmup-epochs", "1.6", "--warmup-factor", "0.3", "--dropout", "0.2", + "--loss", "lstr-mce", "--transition-objective", "none", "--seed", str(seed)] + if fusion != "single": # Unspecified means single in train_turn_head.py. + train += ["--layer-fusion", fusion] + infer = [sys.executable, str(REPO / "infer_turn_head.py"), *common, + "--checkpoint", str(checkpoint), "--output-dir", str(result)] + if variant != "baseline": + train += ["--local-feature-root", str(args.local_feature_root), "--local-modalities", variant, + "--local-projection-dim", str(args.local_projection_dim)] + infer += ["--local-feature-root", str(args.local_feature_root)] + yield f"{variant}_seed{seed}", train, infer + + +def main(argv=None): + args = parse_args(argv) + fusion = preflight(args) + env = os.environ.copy() + for name in ("OMP_NUM_THREADS", "MKL_NUM_THREADS", "OPENBLAS_NUM_THREADS"): + env.setdefault(name, "4") + for name, train, infer in experiment_commands(args, fusion): + print(f"\n{name}: {'dry run' if args.dry_run else 'train, then evaluate'}", flush=True) + for command in (train, infer): + print(shlex.join(command), flush=True) + if not args.dry_run: + subprocess.run(command, cwd=REPO, env=env, check=True) + summary = [sys.executable, str(REPO / "summarize_local_baselines.py"), + "--root", str(args.result_root), "--dataset", "ego4d", "--paired-deltas"] + print("\n" + shlex.join(summary), flush=True) + if not args.dry_run: + subprocess.run(summary, cwd=REPO, env=env, check=True) + + +if __name__ == "__main__": + main() diff --git a/summarize_local_baselines.py b/summarize_local_baselines.py index f2cc4fb..c2eecba 100644 --- a/summarize_local_baselines.py +++ b/summarize_local_baselines.py @@ -11,15 +11,27 @@ def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--root", type=Path, default=Path("results/easycom_local_b1")) parser.add_argument("--protocol", choices=("causal_valid_only", "egospeak_full_timeline"), default="causal_valid_only") + parser.add_argument("--dataset", choices=("easycom", "ego4d"), help="Reject results from a different dataset") + parser.add_argument("--paired-deltas", action="store_true", help="Compare matching seeds to baseline, plus AV minus audio") args = parser.parse_args() files = sorted(args.root.glob("*/metrics.json")) if not files: raise SystemExit(f"No */metrics.json in {args.root}") print(f"Protocol: {args.protocol}; all AP/mAP numbers are percentages") print(f"{'run':<23} {'0.2s mAP':>11} {'BG AP':>11} {'wearer AP':>11} {'others AP':>11} {'mean mAP':>11}") - groups = {} + groups, by_seed = {}, {} + comparison_signature = None for path in files: metrics = json.loads(path.read_text()) + if args.dataset is not None and metrics.get("dataset") != args.dataset: + raise ValueError(f"Dataset mismatch in {path}: {metrics.get('dataset')}") + if args.paired_deltas: + signature = (metrics.get("dataset"), metrics.get("sessions"), metrics.get("checkpoint_epoch"), + metrics.get("checkpoint_train_sessions"), metrics.get("checkpoint_dev_sessions"), + metrics.get("loss"), metrics.get("layer_fusion")) + if comparison_signature is not None and signature != comparison_signature: + raise ValueError(f"Paired runs have different splits/epochs/loss/fusion: {path}") + comparison_signature = signature score = metrics[args.protocol] first = score["horizons"][0] ap = first["class_ap"] @@ -29,6 +41,9 @@ def main(): print(f"{name:<23}" + "".join(f" {v:11.5f}" for v in values)) group = re.sub(r"_seed\d+$", "", name) groups.setdefault(group, []).append(values) + match = re.fullmatch(r"(.+)_seed(\d+)", name) + if match: + by_seed.setdefault(match[1], {})[int(match[2])] = values print("\nMean ± sample standard deviation (AP/mAP percentage points)") for group, rows in sorted(groups.items()): stats = [] @@ -36,6 +51,21 @@ def main(): spread = f"{statistics.stdev(column):.5f}" if len(column) > 1 else "n/a" stats.append(f"{statistics.mean(column):.5f} ± {spread}") print(f"{group:<23}" + " | ".join(stats)) + if args.paired_deltas: + print("\nPaired deltas (percentage points; same column order; matching seeds only)") + for treatment, control in (("audio", "baseline"), ("visual", "baseline"), + ("av", "baseline"), ("av", "audio")): + left, right = by_seed.get(treatment, {}), by_seed.get(control, {}) + seeds = sorted(left.keys() & right.keys()) + if not seeds: + print(f"{treatment} - {control}: no matching seeds") + continue + differences = [[a - b for a, b in zip(left[seed], right[seed])] for seed in seeds] + stats = [] + for column in zip(*differences): + spread = f"{statistics.stdev(column):.5f}" if len(column) > 1 else "n/a" + stats.append(f"{statistics.mean(column):+.5f} ± {spread}") + print(f"{treatment} - {control} (seeds={seeds}): " + " | ".join(stats)) if __name__ == "__main__": diff --git a/test_ego4d_local_workflow.py b/test_ego4d_local_workflow.py new file mode 100644 index 0000000..2957b67 --- /dev/null +++ b/test_ego4d_local_workflow.py @@ -0,0 +1,206 @@ +"""CPU checks for manifest extraction, safe reuse and the four-run Ego4D family.""" + +from __future__ import annotations + +import contextlib +import copy +import io +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +from types import SimpleNamespace +import unittest +from unittest.mock import patch + +import numpy as np +import torch + +from easycom_head import load_ego4d_manifest_split +import extract_turn_local_features as extract +import run_ego4d_local_baselines as runner +from test_local_turn_features import fixture + + +class Ego4DLocalWorkflowTests(unittest.TestCase): + def setUp(self): + torch.set_num_threads(1) + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + self.names = [f"00000000-0000-0000-0000-{i:012d}" for i in (4, 1, 5)] + self.splits = ["train", "test", "train"] + for folder in ("features", "targets", "local", "v1/clips", "v1/annotations"): + (self.root / folder).mkdir(parents=True) + self.manifest = self.root / "ego4d_egospeak.tsv" + self.manifest.write_text("split\tuid\n" + "".join( + f"{split}\t{name}\n" for split, name in zip(self.splits, self.names))) + for name in self.names: + reference, side, labels = fixture(name=name, dataset="ego4d") + torch.save(reference, self.root / "features" / f"{name}.pt") + torch.save(side, self.root / "local" / f"{name}.pt") + np.save(self.root / "targets" / f"{name}.npy", labels) + (self.root / "v1/clips" / f"{name}.mp4").write_bytes(b"fake-media-not-decoded") + for annotation_split, split in (("train", "train"), ("val", "test")): + payload = {"videos": [{"clips": [{"clip_uid": name, "persons": []} + for name, part in zip(self.names, self.splits) if part == split]}]} + (self.root / "v1/annotations" / f"av_{annotation_split}.json").write_text(json.dumps(payload)) + + def extraction_args(self, extra=()): + return ["--dataset", "ego4d", "--data-root", str(self.root), + "--feature-root", str(self.root / "features"), "--target-root", str(self.root / "targets"), + "--manifest", str(self.manifest), "--output-root", str(self.root / "extracted"), *extra] + + def runner_args(self, extra=()): + return ["--data-root", str(self.root), "--feature-root", str(self.root / "features"), + "--target-root", str(self.root / "targets"), "--local-feature-root", str(self.root / "local"), + "--checkpoint-root", str(self.root / "checkpoints"), "--result-root", str(self.root / "results"), + "--seeds", "0", "--device", "cpu", "--no-progress", *extra] + + def test_manifest_order_split_and_selection(self): + self.assertEqual(load_ego4d_manifest_split(self.manifest, "all"), self.names) + args = extract.parse_args(self.extraction_args(["--split", "train", "--start-index", "2", "--limit", "1"])) + self.assertEqual(extract.select_samples(args), [self.names[2]]) + args = extract.parse_args(self.extraction_args(["--start-index", "2", "--end-index", "2"])) + self.assertEqual(extract.select_samples(args), [self.names[1]]) + args = extract.parse_args(self.extraction_args(["--start-index", "100"])) + with self.assertRaisesRegex(ValueError, "No samples"): + extract.select_samples(args) + with self.manifest.open("a") as handle: + handle.write(f"test\t{self.names[0]}\n") + with self.assertRaisesRegex(ValueError, "Duplicate"): + load_ego4d_manifest_split(self.manifest, "train") + + def test_invalid_cli_combinations(self): + for extra in (["--samples", self.names[0]], ["--dataset", "easycom"], ["--start-index", "0"], + ["--limit", "0"], ["--max-steps", "20", "--reuse-existing"], + ["--overwrite", "--reuse-existing"]): + with self.subTest(extra=extra), contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit): + extract.parse_args(self.extraction_args(extra)) + + def test_dry_run_no_model_decode_tracks_or_writes(self): + with patch("extract_turn_local_features.FrozenQwenLocalEncoder") as encoder, \ + patch("extract_turn_local_features.build_media_and_boxes") as media, \ + patch("extract_turn_local_features.load_ego4d_boxes") as boxes: + extract.main(self.extraction_args(["--dry-run"])) + encoder.assert_not_called() + media.assert_not_called() + boxes.assert_not_called() + self.assertFalse((self.root / "extracted").exists()) + np.save(self.root / "targets" / f"{self.names[1]}.npy", np.zeros((19, 3))) + with self.assertRaisesRegex(ValueError, "Reference/target mismatch"): + extract.main(self.extraction_args(["--dry-run"])) + + def test_one_encoder_per_batch_and_validated_reuse(self): + fake_encoder = SimpleNamespace(transformers_version="4.57.6", visual_dim=6, audio_dim=5, + model_config_sha256="same-checkpoint") + _, side, _ = fixture(dataset="ego4d") + media = SimpleNamespace(clips=[], metadata=lambda: {}) + with patch.dict(sys.modules, {"transformers": SimpleNamespace(__version__="4.57.6")}), \ + patch("extract_turn_local_features.FrozenQwenLocalEncoder", return_value=fake_encoder) as encoder, \ + patch("extract_turn_local_features.build_media_and_boxes", return_value=(media, [])) as build, \ + patch("extract_turn_local_features.extract_local", side_effect=lambda *a: copy.deepcopy(side)): + extract.main(self.extraction_args()) + encoder.assert_called_once() + self.assertEqual(build.call_count, 3) + encoder.reset_mock() + build.reset_mock() + extract.main(self.extraction_args(["--reuse-existing"])) + encoder.assert_not_called() + build.assert_not_called() + with self.assertRaisesRegex(ValueError, "audio_window_seconds"): + extract.main(self.extraction_args(["--reuse-existing", "--audio-window-seconds", "2"])) + first = self.root / "extracted" / f"{self.names[0]}.pt" + payload = torch.load(first, weights_only=False) + payload["partial"] = True + torch.save(payload, first) + with self.assertRaisesRegex(ValueError, "Partial smoke-test"): + extract.main(self.extraction_args(["--reuse-existing"])) + + def test_four_commands_keep_splits_epochs_and_feature_ablation(self): + args = runner.parse_args(self.runner_args()) + fusion = runner.preflight(args) + self.assertEqual(fusion, "concat-mlp") + commands = list(runner.experiment_commands(args, fusion)) + self.assertEqual([name for name, _, _ in commands], [v + "_seed0" for v in runner.VARIANTS]) + for name, train, infer in commands: + self.assertEqual(train[train.index("--epochs") + 1], "4") + for command in (train, infer): + self.assertEqual(command[command.index("--dataset") + 1], "ego4d") + self.assertEqual(command[command.index("--manifest") + 1], str(self.manifest.resolve())) + self.assertNotIn("--sessions", command) + self.assertNotIn("--dev-sessions", command) + self.assertEqual("--local-feature-root" in command, name != "baseline_seed0") + if name != "baseline_seed0": + self.assertEqual(train[train.index("--local-modalities") + 1], name.split("_")[0]) + + def test_family_preflight_catches_missing_test_sidecar_before_training(self): + (self.root / "local" / f"{self.names[1]}.pt").unlink() + with patch("run_ego4d_local_baselines.subprocess.run") as run: + with self.assertRaisesRegex(FileNotFoundError, self.names[1]): + runner.main(self.runner_args()) + run.assert_not_called() + self.assertEqual(runner.preflight(runner.parse_args(self.runner_args(["--variants", "baseline"]))), + "concat-mlp") + + def test_preflight_rejects_mixed_contracts_and_existing_outputs(self): + path = self.root / "local" / f"{self.names[1]}.pt" + side = torch.load(path, weights_only=False) + side["config"]["encoder_id"] = "different-test-encoder" + torch.save(side, path) + with self.assertRaisesRegex(ValueError, "contract differs"): + runner.preflight(runner.parse_args(self.runner_args())) + (self.root / "results" / "baseline_seed0").mkdir(parents=True) + with self.assertRaisesRegex(FileExistsError, "Refusing"): + runner.preflight(runner.parse_args(self.runner_args(["--variants", "baseline"]))) + + def test_single_layer_auto_and_dry_run(self): + for name in self.names: + path = self.root / "features" / f"{name}.pt" + reference = torch.load(path, weights_only=False) + reference["states"] = reference["states"].mean(1) + torch.save(reference, path) + args = runner.parse_args(self.runner_args()) + self.assertEqual(runner.preflight(args), "single") + for _, train, _ in runner.experiment_commands(args, "single"): + self.assertNotIn("--layer-fusion", train) + with self.assertRaisesRegex(ValueError, "requires stacked"): + runner.preflight(runner.parse_args(self.runner_args(["--layer-fusion", "concat-mlp"]))) + with patch("run_ego4d_local_baselines.subprocess.run") as run: + runner.main(self.runner_args(["--dry-run"])) + run.assert_not_called() + self.assertFalse((self.root / "checkpoints").exists()) + + def test_full_cpu_family_round_trip_and_summary(self): + env = dict(os.environ, OMP_NUM_THREADS="1", MKL_NUM_THREADS="1", OPENBLAS_NUM_THREADS="1") + completed = subprocess.run([sys.executable, str(runner.REPO / "run_ego4d_local_baselines.py"), + *self.runner_args()], env=env, capture_output=True, text=True, timeout=90) + self.assertEqual(completed.returncode, 0, completed.stdout + completed.stderr) + self.assertIn("av - audio (seeds=[0])", completed.stdout) + scores = {} + for variant in runner.VARIANTS: + path = self.root / "results" / f"{variant}_seed0" / "metrics.json" + metrics = json.loads(path.read_text()) + self.assertEqual(metrics["checkpoint_epoch"], 4) + self.assertEqual(metrics["dataset"], "ego4d") + self.assertEqual(metrics["checkpoint_dev_sessions"], []) + self.assertEqual(metrics["sessions"], [self.names[1]]) + self.assertEqual(metrics["checkpoint_train_sessions"], [self.names[0], self.names[2]]) + self.assertEqual(bool(metrics["local_contract"]), variant != "baseline") + scores[variant] = metrics["causal_valid_only"]["horizons"][0]["map"] + expected_delta = 100 * (scores["av"] - scores["audio"]) + self.assertIn(f"av - audio (seeds=[0]): {expected_delta:+.5f} ± n/a", completed.stdout) + # Do not silently pool a mismatched epoch as a paired comparison. + metrics["checkpoint_epoch"] = 25 + path.write_text(json.dumps(metrics)) + completed = subprocess.run([sys.executable, str(runner.REPO / "summarize_local_baselines.py"), + "--root", str(self.root / "results"), "--dataset", "ego4d", "--paired-deltas"], + env=env, capture_output=True, text=True, timeout=20) + self.assertNotEqual(completed.returncode, 0) + self.assertIn("different splits/epochs/loss/fusion", completed.stderr) + + +if __name__ == "__main__": + unittest.main()