diff --git a/README.md b/README.md index 6cc105b..f394b66 100644 --- a/README.md +++ b/README.md @@ -300,6 +300,13 @@ extraction, training and inference; no symlink index is needed. GPU workers, with per-GPU logs, validated resume, and foreground supervision. See the guide's L20 commands for the `/data1/yfzx/wangx` paths. +For **encoder-only A / V / AV**, without Thinker states, see +[the encoder-only comparison guide](docs/ego4d_encoder_only.md). +`prepare_encoder_only_features.py` packs complete local vectors and validity +bits on CPU; `run_ego4d_encoder_only.py` trains/evaluates four-epoch heads with +no dev set. This native-concatenation baseline is not capacity- or +context-matched to the Thinker baseline. + ## Current Ego4D context-length results The following single-seed snapshot was recorded on 2026-08-22. Both variants diff --git a/docs/ego4d_encoder_only.md b/docs/ego4d_encoder_only.md new file mode 100644 index 0000000..441a222 --- /dev/null +++ b/docs/ego4d_encoder_only.md @@ -0,0 +1,95 @@ +# Ego4D：不输入 Thinker 状态的 encoder-only 对照 + +本实验复用已经提取完的 `local_qwen_1s/.pt`，仅将选定的 Qwen 编码器向量和 +有效性标志作为 anticipation MLP 的输入。**不读取全局 Thinker `.pt`，不运行 Qwen， +不重新处理视频，不使用软链接。** 原来的 `run_ego4d_local_baselines.py` 跑的是 +G / G+A / G+V / G+A+V，并不是本实验的 encoder-only。 + +沿用原 EgoSpeak manifest 的 train/test、5Hz 行索引、十个预测 horizon、LSTR-MCE， +固定 4 epochs、无 dev set，保存并评测最后一轮，不按 test 选择 epoch。 + +## 1. 一次性整理输入（CPU） + +必须先完成全部 manifest clips 的局部特征提取；不要用 preview 或 partial smoke 文件。 +下面的默认路径就是 L20 上的 `/data1/yfzx/wangx`： + +```bash +cd /data1/yfzx/wangx/Code/icassp2027-experiment + +python prepare_encoder_only_features.py \ + --data-root /data1/yfzx/wangx/Datasets/ego4d \ + --local-feature-root /data1/yfzx/wangx/Datasets/ego4d/omni_features/local_qwen_1s \ + --modalities av \ + --dry-run + +# 校验通过后，去掉 --dry-run 写出实体文件。 +python prepare_encoder_only_features.py \ + --data-root /data1/yfzx/wangx/Datasets/ego4d \ + --local-feature-root /data1/yfzx/wangx/Datasets/ego4d/omni_features/local_qwen_1s \ + --modalities av +``` + +默认输出为 `omni_features/encoder_only_av_1s`。换成 `--modalities audio` 或 `visual` +会分别写入 `encoder_only_audio_1s` / `encoder_only_visual_1s`。 +可显式指定 `--manifest`、`--target-root`、`--output-root`。 +局部输入支持平铺目录和 `train/test` 子目录，不改变 manifest 中的 split。 + +整理前逐条检查全部 train/test 的完整性、时间/索引、选定向量/有效性掩码、编码器配置一致性 +及 target 长度，再创建新目录。整理时逐条重新校验并原子保存；不保留全数据集张量。 +同名目录已存在时拒绝覆盖。中断或磁盘写入失败的目录不会有 `complete.txt`，不能拿来训练； +修复原因后用新的 `--output-root`，不要手工补写完成标记。 + +每个输出文件明确标记 `mode=encoder-only-av/audio/visual`、`uses_thinker_states=False`， +保存原始 times/label_indices 和编码器配置。`export_manifest.json` 记录 clip 行数、总量、 +路径和输入维度；`complete.txt` 最后写出。更换 seed 只重新训练，不重复整理。 +此前内联 Python 命令完整生成的文件和 `complete.txt` 也兼容下面的训练入口。 + +## 2. 训练和评测（单张空闲 GPU、前台运行） + +```bash +CUDA_VISIBLE_DEVICES=0 \ +OMP_NUM_THREADS=4 MKL_NUM_THREADS=4 OPENBLAS_NUM_THREADS=4 \ +python run_ego4d_encoder_only.py \ + --data-root /data1/yfzx/wangx/Datasets/ego4d \ + --feature-root /data1/yfzx/wangx/Datasets/ego4d/omni_features/encoder_only_av_1s \ + --modalities av \ + --seeds 0 1 2 +``` + +可先加 `--dry-run`：只检查全部 train/test 输入和已有输出，打印命令，不训练或写结果。 +默认结果为 `results/ego4d_encoder_only_1s/encoder_av_seed0` 等，checkpoint 位于 +`checkpoints/ego4d_encoder_only_1s`。已有 checkpoint、训练摘要或结果目录时拒绝覆盖； +补跑时选择未完成的 seeds，或指定新的 `--checkpoint-root` / `--result-root`。 + +换模态时同时修改 `--feature-root` 和 `--modalities`，不能只改其中一个。 +不传 `--local-feature-root`、`--layer-fusion` 或 `--dev-sessions`；原训练/推理文件无需修改。 +训练超参是 batch=512、lr=7e-5、weight_decay=5e-5、warmup_epochs=1.6、 +warmup_factor=0.3、dropout=0.2、transition_objective=none。 +训练前仍逐条检查所有导出输入，不能仅靠完成标记绕过缺文件或模式检查。 + +入口结束后自动汇总，也可单独打印： + +```bash +python summarize_local_baselines.py \ + --root results/ego4d_encoder_only_1s --dataset ego4d +``` + +## 如何解释比较 + +以两路原生向量均为 2048D 为例，audio/visual 单模态输入各 2049D（一个有效标志）， +AV 为 4098D，顺序是 `[visual, visual_valid, audio, audio_valid]`。 +不做原 B1 的 trainable 256D 分支投影；仍用原 MLP 类，但输入宽度决定其隐藏层宽度， +即单模态 1024、AV 2049。因此应报告参数量，**不是相同 head 容量的 Thinker 消融**。 + +窗口和视觉视野也不同：这些 sidecars 是已保存的近期音频／人物局部 crop pooling， +不是把 Thinker 的相同输入 token 绕过 decoder。目录名中的 `1s` 只是默认实验命名， +真实窗口以 `config.encoder_contract.config` 为准。它回答“这些编码器 readout 本身能预测多好”， +不能单独证明 Thinker 有用或无用。严格归因仍需统一窗口、视野、pooling 与 head 容量。 +5Hz 时间及 horizon 语义沿用原实验，不因重新打包而改变。 + +```bash +python -m unittest -q test_encoder_only_workflow +``` + +CPU 测试覆盖三种模态的精确输入构造、缺失/不完整/混合配置拒绝、无 Thinker 依赖、 +输出防覆盖与完成标记，以及三组各 4 epochs 的真实小规模训练/推理。L20 真实数据效果需另跑。 diff --git a/docs/ego4d_local_baselines.md b/docs/ego4d_local_baselines.md index 14855c4..9643df2 100644 --- a/docs/ego4d_local_baselines.md +++ b/docs/ego4d_local_baselines.md @@ -1,5 +1,8 @@ # Ego4D：近期音频／逐人视觉增强的四组对照 +本页四组均包含全局 Thinker 输入 G。若要完全不输入 Thinker 状态，使用 +[encoder-only A / V / AV 对照](ego4d_encoder_only.md)，不要把本页的 audio/visual/av 当作纯编码器组。 + 目标是检验 EasyCom 上的局部特征收益能否迁移到 Ego4D，而不是预先承诺提分。 保留现有 Ego4D 全局 Omni 特征 G、EgoSpeak manifest 和 target；不重提 Thinker， 不重新编码视频，不重划 train/test。新增的两路侧路特征一次提取、三组复用： diff --git a/prepare_encoder_only_features.py b/prepare_encoder_only_features.py new file mode 100644 index 0000000..aacf2d5 --- /dev/null +++ b/prepare_encoder_only_features.py @@ -0,0 +1,137 @@ +"""Pack Ego4D local Qwen encoder vectors as standalone head inputs on CPU. + +No model, raw media, Thinker states or symlink index is loaded. All manifest +train/test sidecars are validated before creating a fresh output directory. +""" + +from __future__ import annotations + +import argparse +import json +import os +from pathlib import Path +import tempfile + +import numpy as np +import torch +from tqdm.auto import tqdm + +from easycom_head import load_ego4d_manifest_split, resolve_feature_path +from local_turn_features import MODALITIES, load_local_features + + +DEFAULT_DATA_ROOT = Path("/data1/yfzx/wangx/Datasets/ego4d") +PACKING = "native-vectors-plus-validity-bits" + + +def encoder_only_mode(modalities): + if modalities not in MODALITIES: + raise ValueError(f"Unknown encoder modalities: {modalities}") + return f"encoder-only-{modalities}" + + +def parse_args(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--data-root", type=Path, default=DEFAULT_DATA_ROOT) + parser.add_argument("--manifest", type=Path) + parser.add_argument("--local-feature-root", type=Path) + parser.add_argument("--target-root", type=Path) + parser.add_argument("--output-root", type=Path) + parser.add_argument("--modalities", choices=tuple(MODALITIES), default="av") + parser.add_argument("--dry-run", action="store_true", help="Validate all inputs; no output files or directory") + parser.add_argument("--no-progress", action="store_true") + args = parser.parse_args(argv) + args.manifest = args.manifest or args.data_root / "ego4d_egospeak.tsv" + args.local_feature_root = args.local_feature_root or args.data_root / "omni_features/local_qwen_1s" + args.target_root = args.target_root or args.data_root / "ego4d_target_frame" + args.output_root = args.output_root or args.data_root / f"omni_features/encoder_only_{args.modalities}_1s" + for key in ("data_root", "manifest", "local_feature_root", "target_root", "output_root"): + setattr(args, key, getattr(args, key).expanduser().resolve()) + if args.output_root == args.local_feature_root: + parser.error("Encoder-only output must be separate from local source features") + return args + + +def pack_sample(path, target_path, sample_id, modalities, expected_contract=None): + """Read only selected encoder vectors; even a sidecar's reference path is unused.""" + source = torch.load(path, map_location="cpu", weights_only=False) + if not isinstance(source, dict): + raise ValueError(f"Invalid local sidecar: {path}") + anchor = MODALITIES[modalities][0] + reference = { + "states": source.get(anchor), "times": source.get("times"), + "label_indices": source.get("label_indices"), "start": 0., + } + states, contract = load_local_features( + path, reference, sample_id=sample_id, dataset="ego4d", modalities=modalities, + expected_contract=expected_contract, + ) + target = np.load(target_path, allow_pickle=False, mmap_mode="r") + if target.shape != (len(states), 3): + raise ValueError(f"{sample_id}: feature/target mismatch: {tuple(states.shape)} vs {target.shape}") + if len(states) <= 10: + raise ValueError(f"{sample_id}: no examples with all ten future horizons") + states = states.to(torch.bfloat16) + if not torch.isfinite(states).all(): + raise ValueError(f"{sample_id}: encoder vectors overflow bfloat16 storage") + return { + "format": "qwen-encoder-only-v1", "dataset": "ego4d", "sample_id": sample_id, + "states": states, "times": source["times"], "label_indices": source["label_indices"], + "mode": encoder_only_mode(modalities), "start": 0., "end": len(states) / 5, + "num_steps": len(states), "source_local_features": str(path), + "config": { + "input_source": "qwen-encoders-only", "uses_thinker_states": False, + "packing": PACKING, "encoder_contract": contract, + }, + }, contract + + +def main(argv=None): + args = parse_args(argv) + names = load_ego4d_manifest_split(args.manifest, "all") + if args.output_root.exists() and not args.dry_run: + raise FileExistsError(f"Refusing to overwrite {args.output_root}; use a fresh output directory") + paths = {name: resolve_feature_path(args.local_feature_root, name, dataset="ego4d") for name in names} + targets = {name: args.target_root / f"{name}.npy" for name in names} + for path in targets.values(): + if not path.is_file(): + raise FileNotFoundError(path) + contract, total, width = None, 0, None + # Two passes keep memory bounded and prevent bad/partial test inputs from + # leaving an apparently ready export. No full-dataset tensors are retained. + for name in tqdm(names, desc="encoder-only preflight", unit="clip", disable=args.no_progress): + packed, contract = pack_sample(paths[name], targets[name], name, args.modalities, contract) + total += len(packed["states"]) + width = packed["states"].shape[-1] + del packed + print(f"Validated {len(names)} clips / {total} states; {encoder_only_mode(args.modalities)}; " + f"input width={width}; no Thinker features", flush=True) + if args.dry_run: + return + args.output_root.mkdir(parents=True, exist_ok=False) + rows = {} + for name in tqdm(names, desc="pack encoder-only", unit="clip", disable=args.no_progress): + packed, _ = pack_sample(paths[name], targets[name], name, args.modalities, contract) + destination = args.output_root / f"{name}.pt" + with tempfile.NamedTemporaryFile(dir=args.output_root, prefix=f".{name}.", suffix=".tmp", delete=False) as handle: + temporary = Path(handle.name) + try: + torch.save(packed, temporary) + os.replace(temporary, destination) + finally: + temporary.unlink(missing_ok=True) + rows[name] = len(packed["states"]) + del packed + (args.output_root / "export_manifest.json").write_text(json.dumps({ + "manifest": str(args.manifest), "mode": encoder_only_mode(args.modalities), + "local_feature_root": str(args.local_feature_root), "target_root": str(args.target_root), + "encoder_contract": contract, "input_width": width, "num_states": total, "clips": rows, + }, indent=2) + "\n") + # Written last. The runner also validates every train/test input instead of + # trusting this marker alone. Compatible with the original inline command. + (args.output_root / "complete.txt").write_text(f"{len(names)} clips\n") + print(f"Complete: {args.output_root}", flush=True) + + +if __name__ == "__main__": + main() diff --git a/run_ego4d_encoder_only.py b/run_ego4d_encoder_only.py new file mode 100644 index 0000000..5d1e771 --- /dev/null +++ b/run_ego4d_encoder_only.py @@ -0,0 +1,133 @@ +"""Run raw encoder-only Ego4D heads: four epochs, no dev, manifest splits. + +This is not a capacity/context-matched Thinker ablation: native encoder vectors +and validity flags are concatenated, so the existing MLP adapts to input width. +""" + +from __future__ import annotations + +import argparse +import os +from pathlib import Path +import shlex +import subprocess +import sys + +from tqdm.auto import tqdm + +from easycom_head import load_ego4d_manifest_split, load_session_data +from local_turn_features import LOCAL_FORMAT, MODALITIES +from prepare_encoder_only_features import DEFAULT_DATA_ROOT, PACKING, encoder_only_mode + + +REPO = Path(__file__).resolve().parent + + +def parse_args(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--data-root", type=Path, default=DEFAULT_DATA_ROOT) + parser.add_argument("--manifest", type=Path) + parser.add_argument("--target-root", type=Path) + parser.add_argument("--feature-root", type=Path, required=True, help="Prepared encoder-only inputs, not Thinker/local sidecars") + parser.add_argument("--modalities", choices=tuple(MODALITIES), default="av") + parser.add_argument("--checkpoint-root", type=Path, default=Path("checkpoints/ego4d_encoder_only_1s")) + parser.add_argument("--result-root", type=Path, default=Path("results/ego4d_encoder_only_1s")) + parser.add_argument("--seeds", type=int, nargs="+", default=[0, 1, 2]) + parser.add_argument("--device", default="cuda:0") + parser.add_argument("--dry-run", action="store_true", help="Validate all train/test inputs and print commands without training") + parser.add_argument("--no-progress", action="store_true") + args = parser.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" + for key in ("data_root", "manifest", "target_root", "feature_root", "checkpoint_root", "result_root"): + setattr(args, key, getattr(args, key).expanduser().resolve()) + if len(set(args.seeds)) != len(args.seeds) or any(seed < 0 for seed in args.seeds): + parser.error("Seeds must be unique and nonnegative") + return args + + +def run_paths(args, seed): + name = f"encoder_{args.modalities}_seed{seed}" + return args.checkpoint_root / f"{name}.pth", args.result_root / name + + +def preflight(args): + names = load_ego4d_manifest_split(args.manifest, "all") + train = load_ego4d_manifest_split(args.manifest, "train") + test = load_ego4d_manifest_split(args.manifest, "test") + marker = args.feature_root / "complete.txt" + if not marker.is_file() or marker.read_text().strip() != f"{len(names)} clips": + raise ValueError(f"Missing/incomplete encoder-only export: {marker}") + for seed in args.seeds: + checkpoint, result = run_paths(args, seed) + if checkpoint.exists() or checkpoint.with_suffix(".pth.json").exists() or result.exists(): + raise FileExistsError(f"Refusing to overwrite {checkpoint} or {result}; choose unfinished seeds/new output roots") + expected, total, width = None, 0, None + for name in tqdm(names, desc="encoder-only inputs", unit="clip", disable=args.no_progress): + data = load_session_data(args.feature_root, args.target_root, name, dataset="ego4d", + expected_mode=encoder_only_mode(args.modalities)) + config = data.feature_metadata.get("config") or {} + contract = config.get("encoder_contract") or {} + if (config.get("input_source") != "qwen-encoders-only" or config.get("uses_thinker_states") is not False + or config.get("packing") != PACKING or contract.get("format") != LOCAL_FORMAT + or contract.get("modalities") != args.modalities): + raise ValueError(f"{name}: not a matching encoder-only export") + encoder_config = contract.get("config") or {} + if (encoder_config.get("causal_policy") != "half-open-trailing-window" + or encoder_config.get("uses_speech_labels") is not False): + raise ValueError(f"{name}: missing causal/no-speech-label encoder contract") + dimensions = [encoder_config.get(f"{part}_dim") for part in MODALITIES[args.modalities]] + if any(not isinstance(d, int) or d <= 0 for d in dimensions): + raise ValueError(f"{name}: invalid encoder dimensions") + width = sum(d + 1 for d in dimensions) + if data.states.ndim != 2 or data.hidden_size != width or len(data.states) <= 10: + raise ValueError(f"{name}: invalid native encoder-vector/validity-bit shape") + if expected is not None and config != expected: + raise ValueError(f"{name}: encoder export contract differs across clips") + expected = config + total += len(data.states) + del data + print(f"Encoder-only {args.modalities}: {len(train)} train / {len(test)} test; " + f"{total} states, input width={width}, MLP hidden width={width // 2}; epochs=4, dev=none", flush=True) + print("No Thinker features or local-fusion branch. Native concatenation changes head capacity; " + "context remains the saved encoder windows.", flush=True) + + +def experiment_commands(args): + common = ["--dataset", "ego4d", "--manifest", str(args.manifest), + "--feature-root", str(args.feature_root), "--target-root", str(args.target_root), + "--expected-mode", encoder_only_mode(args.modalities), "--device", args.device] + if args.no_progress: + common.append("--no-progress") + for seed in args.seeds: + checkpoint, result = run_paths(args, 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)] + infer = [sys.executable, str(REPO / "infer_turn_head.py"), *common, + "--checkpoint", str(checkpoint), "--output-dir", str(result)] + yield train, infer + + +def main(argv=None): + args = parse_args(argv) + preflight(args) + env = os.environ.copy() + for name in ("OMP_NUM_THREADS", "MKL_NUM_THREADS", "OPENBLAS_NUM_THREADS"): + env.setdefault(name, "4") + for commands in experiment_commands(args): + for command in commands: + 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"] + print(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/test_encoder_only_workflow.py b/test_encoder_only_workflow.py new file mode 100644 index 0000000..b13223b --- /dev/null +++ b/test_encoder_only_workflow.py @@ -0,0 +1,233 @@ +"""CPU checks for encoder-only packing and the unchanged four-epoch head.""" + +from __future__ import annotations + +import contextlib +import copy +import io +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest +from unittest.mock import patch + +import numpy as np +import torch + +from easycom_head import load_session_data +from local_turn_features import MODALITIES +import prepare_encoder_only_features as prepare +import run_ego4d_encoder_only as runner +from test_local_turn_features import fixture + + +class EncoderOnlyWorkflowTests(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") + self.source = self.root / "local" + self.targets = self.root / "ego4d_target_frame" + self.targets.mkdir() + 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))) + self.paths, self.payloads = {}, {} + for name, split in zip(self.names, self.splits): + # No global feature directory, raw video, annotation or model exists. + _, side, labels = fixture(name=name, dataset="ego4d") + side["reference_features"] = "/must-not-be-read/thinker.pt" + side["states"] = "POISON: never use this field as the encoder input" + path = self.source / split / f"{name}.pt" + path.parent.mkdir(parents=True, exist_ok=True) + torch.save(side, path) + np.save(self.targets / f"{name}.npy", labels) + self.paths[name], self.payloads[name] = path, side + + def output_root(self, modality="av"): + return self.root / f"encoder_{modality}" + + def prepare_args(self, modality="av", extra=()): + return ["--data-root", str(self.root), "--local-feature-root", str(self.source), + "--output-root", str(self.output_root(modality)), "--modalities", modality, + "--no-progress", *extra] + + def runner_args(self, modality="av", extra=()): + return ["--data-root", str(self.root), "--feature-root", str(self.output_root(modality)), + "--modalities", modality, "--checkpoint-root", str(self.root / "checkpoints"), + "--result-root", str(self.root / "results"), "--seeds", "0", "--device", "cpu", + "--no-progress", *extra] + + def export(self, modality="av"): + with contextlib.redirect_stdout(io.StringIO()): + prepare.main(self.prepare_args(modality)) + + def test_pack_all_modalities_uses_only_native_vectors_and_validity(self): + name = self.names[0] + for modality, parts in MODALITIES.items(): + with self.subTest(modality=modality): + result, contract = prepare.pack_sample( + self.paths[name], self.targets / f"{name}.npy", name, modality) + original = self.payloads[name] + expected = torch.cat([tensor for part in parts for tensor in ( + original[part].float(), original[f"{part}_valid"][:, None].float(), + )], dim=-1).to(torch.bfloat16) + self.assertTrue(torch.equal(result["states"], expected)) + self.assertTrue(torch.equal(result["times"], original["times"])) + self.assertTrue(torch.equal(result["label_indices"], original["label_indices"])) + self.assertEqual(result["mode"], f"encoder-only-{modality}") + self.assertIs(result["config"]["uses_thinker_states"], False) + self.assertEqual(contract["modalities"], modality) + + def test_export_real_files_full_timeline_and_refuse_overwrite(self): + self.export() + output = self.output_root() + self.assertEqual((output / "complete.txt").read_text(), "3 clips\n") + manifest = json.loads((output / "export_manifest.json").read_text()) + self.assertEqual(manifest["num_states"], 60) + self.assertEqual(manifest["input_width"], 13) + self.assertEqual(manifest["clips"], dict.fromkeys(self.names, 20)) + for name in self.names: + path = output / f"{name}.pt" + self.assertFalse(path.is_symlink()) + data = load_session_data(output, self.targets, name, dataset="ego4d", expected_mode="encoder-only-av") + self.assertEqual(data.states.shape, (20, 13)) + self.assertIsNone(data.local_features) + before = (output / f"{self.names[0]}.pt").read_bytes() + with self.assertRaises(FileExistsError): + prepare.main(self.prepare_args()) + self.assertEqual((output / f"{self.names[0]}.pt").read_bytes(), before) + + def test_dry_run_does_not_write(self): + with patch.object(prepare.torch, "save") as save, contextlib.redirect_stdout(io.StringIO()): + prepare.main(self.prepare_args(extra=["--dry-run"])) + save.assert_not_called() + self.assertFalse(self.output_root().exists()) + + def test_corrupt_test_sidecars_rejected_before_output_creation(self): + name = self.names[1] + for issue in ("partial", "identity", "times", "indices", "contract", "nan", "invalid_mask"): + side = copy.deepcopy(self.payloads[name]) + if issue == "partial": + side["partial"] = True + elif issue == "identity": + side["sample_id"] = self.names[0] + elif issue == "times": + side["times"][0] = .4 + elif issue == "indices": + side["label_indices"][0] = 1 + elif issue == "contract": + side["config"]["encoder_id"] = "different-encoder" + elif issue == "nan": + side["audio"][0, 0] = float("nan") + else: + side["visual"][0, 0] = 1 # Invalid visual rows must remain zero. + torch.save(side, self.paths[name]) + with self.subTest(issue=issue), contextlib.redirect_stdout(io.StringIO()), self.assertRaises(ValueError): + prepare.main(self.prepare_args()) + self.assertFalse(self.output_root().exists()) + + def test_target_mismatch_and_missing_sidecar_fail_before_writes(self): + name = self.names[1] + np.save(self.targets / f"{name}.npy", np.zeros((19, 3))) + with contextlib.redirect_stdout(io.StringIO()), self.assertRaisesRegex(ValueError, "feature/target mismatch"): + prepare.main(self.prepare_args()) + self.assertFalse(self.output_root().exists()) + self.paths[name].unlink() + with self.assertRaises(FileNotFoundError): + prepare.main(self.prepare_args()) + + def test_interrupted_export_has_no_complete_marker_or_temporary_file(self): + with patch.object(prepare.torch, "save", side_effect=RuntimeError("test write failure")), \ + contextlib.redirect_stdout(io.StringIO()), self.assertRaisesRegex(RuntimeError, "write failure"): + prepare.main(self.prepare_args()) + self.assertFalse((self.output_root() / "complete.txt").exists()) + self.assertEqual(list(self.output_root().glob("*.tmp")), []) + + def test_runner_commands_and_dry_run_never_select_thinker_or_dev(self): + self.export() + args = runner.parse_args(self.runner_args(extra=["--seeds", "0", "1", "2"])) + commands = list(runner.experiment_commands(args)) + self.assertEqual(len(commands), 3) + for seed, (train, infer) in enumerate(commands): + self.assertEqual(train[train.index("--epochs") + 1], "4") + self.assertEqual(train[train.index("--seed") + 1], str(seed)) + self.assertEqual(train[train.index("--warmup-epochs") + 1], "1.6") + for command in (train, infer): + for forbidden in ("--local-feature-root", "--layer-fusion", "--dev-sessions", "--sessions"): + self.assertNotIn(forbidden, command) + self.assertEqual(command[command.index("--expected-mode") + 1], "encoder-only-av") + with patch.object(runner.subprocess, "run") as run, contextlib.redirect_stdout(io.StringIO()): + runner.main(self.runner_args(extra=["--dry-run"])) + run.assert_not_called() + self.assertFalse((self.root / "checkpoints").exists()) + + def test_runner_rejects_missing_test_features_marker_and_wrong_contract(self): + self.export() + args = runner.parse_args(self.runner_args()) + marker = self.output_root() / "complete.txt" + marker.write_text("2 clips\n") + with self.assertRaisesRegex(ValueError, "incomplete"): + runner.preflight(args) + marker.write_text("3 clips\n") + path = self.output_root() / f"{self.names[1]}.pt" + payload = torch.load(path, weights_only=False) + payload["config"]["uses_thinker_states"] = True + torch.save(payload, path) + with self.assertRaisesRegex(ValueError, "encoder-only export"): + runner.preflight(args) + path.unlink() + with self.assertRaises(FileNotFoundError): + runner.preflight(args) + + def test_invalid_cli_and_existing_outputs_are_rejected(self): + for extra in (["--seeds", "0", "0"], ["--seeds", "-1"], ["--modalities", "thinker"]): + with self.subTest(extra=extra), contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit): + runner.parse_args(self.runner_args(extra=extra)) + with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit): + prepare.parse_args(self.prepare_args(extra=["--output-root", str(self.source)])) + self.export() + args = runner.parse_args(self.runner_args(extra=["--seeds", "0", "1"])) + checkpoint, _ = runner.run_paths(args, 1) + checkpoint.parent.mkdir() + checkpoint.write_bytes(b"preserve") + with self.assertRaises(FileExistsError): + runner.preflight(args) + self.assertEqual(checkpoint.read_bytes(), b"preserve") + + def test_cpu_round_trip_all_three_modalities_four_epochs_without_dev(self): + env = os.environ.copy() + env.update(OMP_NUM_THREADS="1", MKL_NUM_THREADS="1", OPENBLAS_NUM_THREADS="1") + for modality, parts in MODALITIES.items(): + with self.subTest(modality=modality): + self.export(modality) + command = [sys.executable, str(Path(runner.__file__).resolve()), *self.runner_args(modality)] + result = subprocess.run(command, env=env, capture_output=True, text=True, timeout=120) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + args = runner.parse_args(self.runner_args(modality)) + checkpoint, output = runner.run_paths(args, 0) + saved = torch.load(checkpoint, map_location="cpu", weights_only=False) + self.assertEqual(saved["epoch"], 4) + self.assertEqual(saved["dev_sessions"], []) + self.assertEqual(set(saved["train_sessions"]), {self.names[0], self.names[2]}) + self.assertIsNone(saved["local_contract"]) + width = sum(self.payloads[self.names[0]]["config"][f"{part}_dim"] + 1 for part in parts) + self.assertEqual(saved["head_config"]["hidden_size"], width) + self.assertEqual(saved["head_config"]["local_visual_dim"], 0) + self.assertEqual(saved["head_config"]["local_audio_dim"], 0) + self.assertEqual(saved["head_config"]["layer_fusion"], "single") + self.assertFalse(any("local_projections" in key for key in saved["head"])) + metrics = json.loads((output / "metrics.json").read_text()) + self.assertEqual(metrics["sessions"], [self.names[1]]) + self.assertEqual(metrics["checkpoint_epoch"], 4) + self.assertEqual(np.load(output / "pred" / f"{self.names[1]}.npy").shape, (20, 3, 10)) + + +if __name__ == "__main__": + unittest.main()