Add EasyCom encoder-only comparison workflow

This commit is contained in:
Xin Wang
2026-08-28 10:44:13 +08:00
parent ee5e735fc2
commit 321238a92f
8 changed files with 488 additions and 49 deletions

View File

@@ -11,6 +11,11 @@ The pooled local-feature B1 experiment (frozen Qwen face crops + recent audio,
native 2048D sidecars, optional trainable projection, four-epoch ablations) is native 2048D sidecars, optional trainable projection, four-epoch ablations) is
documented in [docs/local_feature_baseline.md](docs/local_feature_baseline.md). documented in [docs/local_feature_baseline.md](docs/local_feature_baseline.md).
For **EasyCom encoder-only A / V / AV** with no Thinker states, start with
[the EasyCom comparison commands](docs/easycom_encoder_only.md).
They reuse the verified local sidecars, train on Session 4–12, test on 3/2/1,
and keep four epochs with no dev set.
The main experiment is deliberately simple: The main experiment is deliberately simple:
```text ```text
@@ -301,11 +306,13 @@ GPU workers, with per-GPU logs, validated resume, and foreground supervision.
See the guide's L20 commands for the `/data1/yfzx/wangx` paths. See the guide's L20 commands for the `/data1/yfzx/wangx` paths.
For **encoder-only A / V / AV**, without Thinker states, see For **encoder-only A / V / AV**, without Thinker states, see
[the encoder-only comparison guide](docs/ego4d_encoder_only.md). [the Ego4D guide](docs/ego4d_encoder_only.md) or
[the EasyCom-first guide](docs/easycom_encoder_only.md).
`prepare_encoder_only_features.py` packs complete local vectors and validity `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 bits on CPU (pass `--dataset easycom` for EasyCom; default remains Ego4D).
no dev set. This native-concatenation baseline is not capacity- or `run_ego4d_encoder_only.py` / `run_easycom_encoder_only.py` train/evaluate
context-matched to the Thinker baseline. 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 ## Current Ego4D context-length results

View File

@@ -0,0 +1,143 @@
# EasyCom:先跑 encoder-only A / V / AV 对照
复用已经提取的 Qwen 局部编码器特征,不输入 Thinker hidden states,不重提音频或视频,
不创建软链接。原 B1 的 `audio / visual / av` 分别是 **G+A / G+V / G+A+V**;
这里的新结果命名为 `encoder_audio / encoder_visual / encoder_av`,分别只有 **A / V / AV**。
固定训练 Session **4–12(升序,与原 B1 一致)**,测试 **3/2/1**;4 epochs、无 dev,
评测最后一轮。沿用原来的 5Hz 行索引、十个 horizon、LSTR-MCE 和 AP 协议,不改变标签。
## 1. 确定已经验证过的局部特征目录
在 L20 机器更新本次代码后执行。下面从你之前跑完的 `av_seed0.pth` 读取实际路径,
避免猜错目录或混入 bbox 修正前的旧视觉特征;checkpoint 仅用于找路径,不作为新模型输入。
```bash
cd /data1/yfzx/wangx/Code/icassp2027-experiment
DATA_ROOT=/data1/yfzx/wangx/Datasets/easycom
LOCAL_ROOT=$(python - <<'PY'
from pathlib import Path
import torch
checkpoint = Path("checkpoints/easycom_local_b1_encoded_tf4576/av_seed0.pth")
p = torch.load(checkpoint, map_location="cpu", weights_only=False)
value = p["training_config"]["local_feature_root"]
if not value:
raise ValueError("This checkpoint has no local-feature root")
root = Path(value).expanduser().resolve()
if not root.is_dir():
raise FileNotFoundError(root)
print(root)
PY
) && printf 'LOCAL_ROOT=%s\n' "$LOCAL_ROOT"
```
如果 checkpoint 已移动,修改上述路径;也可以直接将 `LOCAL_ROOT` 设成以前
`easycom_local_b1_encoded_tf4576` 实验实际使用的局部特征目录。
必须包含完整的 `Session_1.pt` 到 `Session_12.pt`,不能用 `local_smoke_encoded`。
下面命令对空变量和缺文件会报错,不会自动重提特征或换成 Thinker 输入。
## 2. 先做 seed 0,三组前台顺序运行
单张空闲 L20 足够先跑这组小 head,无需启动八卡提取或安装新的依赖。
下面先 A、再 AV、最后 V;使用同一批冻结特征,仅改变输入模态。
```bash
(
set -euo pipefail
: "${DATA_ROOT:?Set DATA_ROOT first}"
: "${LOCAL_ROOT:?Set the verified local-feature directory first}"
export CUDA_VISIBLE_DEVICES=0
export OMP_NUM_THREADS=4 MKL_NUM_THREADS=4 OPENBLAS_NUM_THREADS=4
for modality in audio av visual
do
ENCODER_ROOT="${DATA_ROOT}/omni_features/encoder_only_${modality}_1s"
python prepare_encoder_only_features.py \
--dataset easycom \
--data-root "$DATA_ROOT" \
--local-feature-root "$LOCAL_ROOT" \
--modalities "$modality" \
--output-root "$ENCODER_ROOT"
python run_easycom_encoder_only.py \
--data-root "$DATA_ROOT" \
--feature-root "$ENCODER_ROOT" \
--modalities "$modality" \
--seeds 0
done
)
```
两条 Python 命令都支持 `--dry-run`。整理的 dry-run 只验证源文件,不生成训练输入;
训练的 dry-run 则需要已完成的导出目录,只打印训练/评测命令,不启动训练。
整理在 CPU 上完成:读取原生向量+有效性标志,校验全部 12 sessions、时间/索引、
配置一致性和 target 长度,逐文件保存,最后写出 `complete.txt`。
默认 target 是 `$DATA_ROOT/easycom_target_frame`,可用 `--target-root` 覆盖。
不使用 Ego4D manifest。新入口和 `--dataset easycom` 不影响已有 Ego4D 命令。
输出:
- 特征:`omni_features/encoder_only_audio_1s`、`encoder_only_av_1s`、`encoder_only_visual_1s`。
- 模型:`checkpoints/easycom_encoder_only_1s/encoder_<modality>_seed0.pth`。
- 评测:`results/easycom_encoder_only_1s/encoder_<modality>_seed0/metrics.json` 和 `pred/`。
导出目录、checkpoint 或结果目录已存在时**拒绝覆盖**。不要重新跑整个循环来补 seeds;
保留已完成的导出,只运行下面的训练入口。
若某次导出中断、没有完成标记,请使用新的 `--output-root`,不要手动补标记。
训练中断后换新的 `--checkpoint-root` / `--result-root`;不会自动恢复半成品训练。
## 3. 打印结果,随后补 seed 1/2
```bash
python summarize_local_baselines.py \
--root results/easycom_encoder_only_1s \
--dataset easycom \
--protocol causal_valid_only
```
请贴出三行 `encoder_audio_seed0`、`encoder_av_seed0`、`encoder_visual_seed0`。
原来的 81.72257 / 82.02661 / 73.47412 是 **G+A / G+A+V / G+V** 的 0.2s mAP,
不能标为 encoder-only。比较时使用相同 seed 和同一 `causal_valid_only` 协议。
第一轮确认能跑通后,补齐三个种子(不重新导出):
```bash
(
set -euo pipefail
: "${DATA_ROOT:?Set DATA_ROOT first}"
export CUDA_VISIBLE_DEVICES=0
export OMP_NUM_THREADS=4 MKL_NUM_THREADS=4 OPENBLAS_NUM_THREADS=4
for modality in audio av visual
do
python run_easycom_encoder_only.py \
--data-root "$DATA_ROOT" \
--feature-root "${DATA_ROOT}/omni_features/encoder_only_${modality}_1s" \
--modalities "$modality" \
--seeds 1 2
done
)
```
## 实验能回答什么
它测的是“Qwen 编码器近期 readout 本身的预测能力”,不是严格相同容量的 Thinker 消融。
3B 的 A/V 原生各 2048D;带有效性位后单模态 2049D、AV 4098D,
顺序是 `[visual, visual_valid, audio, audio_valid]`。不用 B1 的额外 256D 分支投影;
沿用原 MLP 类,隐藏宽度为输入维度整除 2,因此单模态 1024、AV 2049。
窗口、视觉 crop/视野和 head 参数量与原 Thinker 基线不同,应一起报告。
batch=512、lr=7e-5、weight_decay=5e-5、warmup_epochs=1.6、warmup_factor=0.3、
dropout=0.2、transition_objective=none。目录名 `1s` 不覆盖源文件的窗口配置,
真实窗口保存在导出 `config.encoder_contract.config`;时间含义完全沿用源特征。
本地 CPU 回归包括全部 12 sessions 的三模态导出、精确数值/时间对齐、缺文件和 partial 拒绝、
固定 split 顺序、无 Thinker 依赖,以及三个模态各 4 epochs 的合成数据训练/推理。
真实 EasyCom 精度和 L20 耗时由上述实验验证,不由这些测试推断。
```bash
python -m unittest -q test_easycom_encoder_only test_encoder_only_workflow
```

View File

@@ -1,5 +1,8 @@
# Ego4D:不输入 Thinker 状态的 encoder-only 对照 # Ego4D:不输入 Thinker 状态的 encoder-only 对照
先做 EasyCom 时请用 [EasyCom encoder-only 命令](easycom_encoder_only.md)。
整理脚本现支持 `--dataset easycom`,默认仍为 Ego4D,本页已有命令不变。
本实验复用已经提取完的 `local_qwen_1s/<uid>.pt`,仅将选定的 Qwen 编码器向量和 本实验复用已经提取完的 `local_qwen_1s/<uid>.pt`,仅将选定的 Qwen 编码器向量和
有效性标志作为 anticipation MLP 的输入。**不读取全局 Thinker `.pt`,不运行 Qwen, 有效性标志作为 anticipation MLP 的输入。**不读取全局 Thinker `.pt`,不运行 Qwen,
不重新处理视频,不使用软链接。** 原来的 `run_ego4d_local_baselines.py` 跑的是 不重新处理视频,不使用软链接。** 原来的 `run_ego4d_local_baselines.py` 跑的是

View File

@@ -4,6 +4,10 @@
不是已证明的改进,也不是 multi-speaker attention 模型。 不是已证明的改进,也不是 multi-speaker attention 模型。
保留原 `[T,4,2048]` 特征;不重新提取全局 Thinker,不改标签和 AP 协议。 保留原 `[T,4,2048]` 特征;不重新提取全局 Thinker,不改标签和 AP 协议。
若要比较**不输入 Thinker 的纯 A / V / AV**,使用
[EasyCom encoder-only 实验](easycom_encoder_only.md)。本页的 `audio/visual/av`
仍包含全局 G,不能将两种设置混为一组。
## 固定实验定义 ## 固定实验定义
| 分支 | 冻结特征 | 更新/池化 | 训练时处理 | | 分支 | 冻结特征 | 更新/池化 | 训练时处理 |

View File

@@ -1,6 +1,6 @@
"""Pack Ego4D local Qwen encoder vectors as standalone head inputs on CPU. """Pack EasyCom/Ego4D local Qwen encoder vectors as standalone head inputs on CPU.
No model, raw media, Thinker states or symlink index is loaded. All manifest No model, raw media, Thinker states or symlink index is loaded. All fixed-split
train/test sidecars are validated before creating a fresh output directory. train/test sidecars are validated before creating a fresh output directory.
""" """
@@ -16,14 +16,38 @@ import numpy as np
import torch import torch
from tqdm.auto import tqdm from tqdm.auto import tqdm
from easycom_head import load_ego4d_manifest_split, resolve_feature_path from easycom_head import (
DATASET_NAMES, OFFICIAL_TEST_SESSIONS, OFFICIAL_TRAIN_SESSIONS,
load_ego4d_manifest_split, resolve_feature_path,
)
from local_turn_features import MODALITIES, load_local_features from local_turn_features import MODALITIES, load_local_features
DEFAULT_DATA_ROOT = Path("/data1/yfzx/wangx/Datasets/ego4d") DEFAULT_DATA_ROOTS = {dataset: Path("/data1/yfzx/wangx/Datasets") / dataset for dataset in DATASET_NAMES}
DEFAULT_DATA_ROOT = DEFAULT_DATA_ROOTS["ego4d"] # Preserve the original Ego4D default/API.
PACKING = "native-vectors-plus-validity-bits" PACKING = "native-vectors-plus-validity-bits"
def encoder_sample_split(dataset, manifest=None):
"""Fixed benchmark membership; EasyCom order matches the original B1 run."""
if dataset == "easycom":
if manifest is not None:
raise ValueError("EasyCom uses fixed sessions, not an Ego4D manifest")
number = lambda name: int(name.split("_")[-1])
# B1 explicitly passes sessions 4..12. Preserve that concatenation
# order too, so seeded shuffling sees the same sample ordering.
train = sorted(OFFICIAL_TRAIN_SESSIONS, key=number)
test = list(OFFICIAL_TEST_SESSIONS)
return sorted(train + test, key=number), train, test
if dataset != "ego4d" or manifest is None:
raise ValueError("Ego4D requires its EgoSpeak manifest")
return tuple(load_ego4d_manifest_split(manifest, part) for part in ("all", "train", "test"))
def sample_unit(dataset):
return "session" if dataset == "easycom" else "clip"
def encoder_only_mode(modalities): def encoder_only_mode(modalities):
if modalities not in MODALITIES: if modalities not in MODALITIES:
raise ValueError(f"Unknown encoder modalities: {modalities}") raise ValueError(f"Unknown encoder modalities: {modalities}")
@@ -32,7 +56,8 @@ def encoder_only_mode(modalities):
def parse_args(argv=None): def parse_args(argv=None):
parser = argparse.ArgumentParser(description=__doc__) parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--data-root", type=Path, default=DEFAULT_DATA_ROOT) parser.add_argument("--dataset", choices=DATASET_NAMES, default="ego4d")
parser.add_argument("--data-root", type=Path)
parser.add_argument("--manifest", type=Path) parser.add_argument("--manifest", type=Path)
parser.add_argument("--local-feature-root", type=Path) parser.add_argument("--local-feature-root", type=Path)
parser.add_argument("--target-root", type=Path) parser.add_argument("--target-root", type=Path)
@@ -41,18 +66,24 @@ def parse_args(argv=None):
parser.add_argument("--dry-run", action="store_true", help="Validate all inputs; no output files or directory") 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") parser.add_argument("--no-progress", action="store_true")
args = parser.parse_args(argv) args = parser.parse_args(argv)
args.manifest = args.manifest or args.data_root / "ego4d_egospeak.tsv" args.data_root = args.data_root or DEFAULT_DATA_ROOTS[args.dataset]
if args.dataset == "ego4d":
args.manifest = args.manifest or args.data_root / "ego4d_egospeak.tsv"
elif args.manifest is not None:
parser.error("--manifest is only supported for Ego4D; EasyCom uses all 12 fixed sessions")
args.local_feature_root = args.local_feature_root or args.data_root / "omni_features/local_qwen_1s" 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.target_root = args.target_root or args.data_root / f"{args.dataset}_target_frame"
args.output_root = args.output_root or args.data_root / f"omni_features/encoder_only_{args.modalities}_1s" 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"): for key in ("data_root", "manifest", "local_feature_root", "target_root", "output_root"):
setattr(args, key, getattr(args, key).expanduser().resolve()) value = getattr(args, key)
if value is not None:
setattr(args, key, value.expanduser().resolve())
if args.output_root == args.local_feature_root: if args.output_root == args.local_feature_root:
parser.error("Encoder-only output must be separate from local source features") parser.error("Encoder-only output must be separate from local source features")
return args return args
def pack_sample(path, target_path, sample_id, modalities, expected_contract=None): def pack_sample(path, target_path, sample_id, modalities, expected_contract=None, *, dataset="ego4d"):
"""Read only selected encoder vectors; even a sidecar's reference path is unused.""" """Read only selected encoder vectors; even a sidecar's reference path is unused."""
source = torch.load(path, map_location="cpu", weights_only=False) source = torch.load(path, map_location="cpu", weights_only=False)
if not isinstance(source, dict): if not isinstance(source, dict):
@@ -63,7 +94,7 @@ def pack_sample(path, target_path, sample_id, modalities, expected_contract=None
"label_indices": source.get("label_indices"), "start": 0., "label_indices": source.get("label_indices"), "start": 0.,
} }
states, contract = load_local_features( states, contract = load_local_features(
path, reference, sample_id=sample_id, dataset="ego4d", modalities=modalities, path, reference, sample_id=sample_id, dataset=dataset, modalities=modalities,
expected_contract=expected_contract, expected_contract=expected_contract,
) )
target = np.load(target_path, allow_pickle=False, mmap_mode="r") target = np.load(target_path, allow_pickle=False, mmap_mode="r")
@@ -75,7 +106,7 @@ def pack_sample(path, target_path, sample_id, modalities, expected_contract=None
if not torch.isfinite(states).all(): if not torch.isfinite(states).all():
raise ValueError(f"{sample_id}: encoder vectors overflow bfloat16 storage") raise ValueError(f"{sample_id}: encoder vectors overflow bfloat16 storage")
return { return {
"format": "qwen-encoder-only-v1", "dataset": "ego4d", "sample_id": sample_id, "format": "qwen-encoder-only-v1", "dataset": dataset, "sample_id": sample_id,
"states": states, "times": source["times"], "label_indices": source["label_indices"], "states": states, "times": source["times"], "label_indices": source["label_indices"],
"mode": encoder_only_mode(modalities), "start": 0., "end": len(states) / 5, "mode": encoder_only_mode(modalities), "start": 0., "end": len(states) / 5,
"num_steps": len(states), "source_local_features": str(path), "num_steps": len(states), "source_local_features": str(path),
@@ -88,10 +119,11 @@ def pack_sample(path, target_path, sample_id, modalities, expected_contract=None
def main(argv=None): def main(argv=None):
args = parse_args(argv) args = parse_args(argv)
names = load_ego4d_manifest_split(args.manifest, "all") names, train, test = encoder_sample_split(args.dataset, args.manifest)
unit = sample_unit(args.dataset)
if args.output_root.exists() and not args.dry_run: if args.output_root.exists() and not args.dry_run:
raise FileExistsError(f"Refusing to overwrite {args.output_root}; use a fresh output directory") 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} paths = {name: resolve_feature_path(args.local_feature_root, name, dataset=args.dataset) for name in names}
targets = {name: args.target_root / f"{name}.npy" for name in names} targets = {name: args.target_root / f"{name}.npy" for name in names}
for path in targets.values(): for path in targets.values():
if not path.is_file(): if not path.is_file():
@@ -99,19 +131,19 @@ def main(argv=None):
contract, total, width = None, 0, None contract, total, width = None, 0, None
# Two passes keep memory bounded and prevent bad/partial test inputs from # Two passes keep memory bounded and prevent bad/partial test inputs from
# leaving an apparently ready export. No full-dataset tensors are retained. # 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): for name in tqdm(names, desc="encoder-only preflight", unit=unit, disable=args.no_progress):
packed, contract = pack_sample(paths[name], targets[name], name, args.modalities, contract) packed, contract = pack_sample(paths[name], targets[name], name, args.modalities, contract, dataset=args.dataset)
total += len(packed["states"]) total += len(packed["states"])
width = packed["states"].shape[-1] width = packed["states"].shape[-1]
del packed del packed
print(f"Validated {len(names)} clips / {total} states; {encoder_only_mode(args.modalities)}; " print(f"Validated {len(names)} {unit}s / {total} states; {encoder_only_mode(args.modalities)}; "
f"input width={width}; no Thinker features", flush=True) f"input width={width}; no Thinker features", flush=True)
if args.dry_run: if args.dry_run:
return return
args.output_root.mkdir(parents=True, exist_ok=False) args.output_root.mkdir(parents=True, exist_ok=False)
rows = {} rows = {}
for name in tqdm(names, desc="pack encoder-only", unit="clip", disable=args.no_progress): for name in tqdm(names, desc="pack encoder-only", unit=unit, disable=args.no_progress):
packed, _ = pack_sample(paths[name], targets[name], name, args.modalities, contract) packed, _ = pack_sample(paths[name], targets[name], name, args.modalities, contract, dataset=args.dataset)
destination = args.output_root / f"{name}.pt" destination = args.output_root / f"{name}.pt"
with tempfile.NamedTemporaryFile(dir=args.output_root, prefix=f".{name}.", suffix=".tmp", delete=False) as handle: with tempfile.NamedTemporaryFile(dir=args.output_root, prefix=f".{name}.", suffix=".tmp", delete=False) as handle:
temporary = Path(handle.name) temporary = Path(handle.name)
@@ -123,13 +155,14 @@ def main(argv=None):
rows[name] = len(packed["states"]) rows[name] = len(packed["states"])
del packed del packed
(args.output_root / "export_manifest.json").write_text(json.dumps({ (args.output_root / "export_manifest.json").write_text(json.dumps({
"manifest": str(args.manifest), "mode": encoder_only_mode(args.modalities), "dataset": args.dataset, "manifest": str(args.manifest) if args.manifest else None,
"train_samples": train, "test_samples": test, "mode": encoder_only_mode(args.modalities),
"local_feature_root": str(args.local_feature_root), "target_root": str(args.target_root), "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, "encoder_contract": contract, "input_width": width, "num_states": total, f"{unit}s": rows,
}, indent=2) + "\n") }, indent=2) + "\n")
# Written last. The runner also validates every train/test input instead of # Written last. The runner also validates every train/test input instead of
# trusting this marker alone. Compatible with the original inline command. # trusting this marker alone. Compatible with the original inline command.
(args.output_root / "complete.txt").write_text(f"{len(names)} clips\n") (args.output_root / "complete.txt").write_text(f"{len(names)} {unit}s\n")
print(f"Complete: {args.output_root}", flush=True) print(f"Complete: {args.output_root}", flush=True)

View File

@@ -0,0 +1,16 @@
"""EasyCom encoder-only: train Session 4..12, test 3/2/1, four epochs, no dev."""
import run_ego4d_encoder_only as workflow
from run_ego4d_encoder_only import experiment_commands, preflight, run_paths
def parse_args(argv=None):
return workflow.parse_args(argv, dataset="easycom")
def main(argv=None):
return workflow.main(argv, dataset="easycom")
if __name__ == "__main__":
main()

View File

@@ -1,7 +1,8 @@
"""Run raw encoder-only Ego4D heads: four epochs, no dev, manifest splits. """Run raw encoder-only heads: four epochs, no dev, fixed benchmark splits.
This is not a capacity/context-matched Thinker ablation: native encoder vectors 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. and validity flags are concatenated, so the existing MLP adapts to input width.
This entry defaults to Ego4D; run_easycom_encoder_only.py shares the workflow.
""" """
from __future__ import annotations from __future__ import annotations
@@ -15,32 +16,44 @@ import sys
from tqdm.auto import tqdm from tqdm.auto import tqdm
from easycom_head import load_ego4d_manifest_split, load_session_data from easycom_head import DATASET_NAMES, load_session_data
from local_turn_features import LOCAL_FORMAT, MODALITIES from local_turn_features import LOCAL_FORMAT, MODALITIES
from prepare_encoder_only_features import DEFAULT_DATA_ROOT, PACKING, encoder_only_mode from prepare_encoder_only_features import (
DEFAULT_DATA_ROOTS, PACKING, encoder_only_mode, encoder_sample_split, sample_unit,
)
REPO = Path(__file__).resolve().parent REPO = Path(__file__).resolve().parent
def parse_args(argv=None): def parse_args(argv=None, *, dataset="ego4d"):
parser = argparse.ArgumentParser(description=__doc__) if dataset not in DATASET_NAMES:
parser.add_argument("--data-root", type=Path, default=DEFAULT_DATA_ROOT) raise ValueError(f"Unknown dataset: {dataset}")
parser = argparse.ArgumentParser(
description=f"{dataset}: raw encoder-only A/V/AV heads; four epochs, no dev, fixed train/test splits."
)
parser.add_argument("--data-root", type=Path, default=DEFAULT_DATA_ROOTS[dataset])
parser.add_argument("--manifest", type=Path) parser.add_argument("--manifest", type=Path)
parser.add_argument("--target-root", 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("--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("--modalities", choices=tuple(MODALITIES), default="av")
parser.add_argument("--checkpoint-root", type=Path, default=Path("checkpoints/ego4d_encoder_only_1s")) parser.add_argument("--checkpoint-root", type=Path, default=Path(f"checkpoints/{dataset}_encoder_only_1s"))
parser.add_argument("--result-root", type=Path, default=Path("results/ego4d_encoder_only_1s")) parser.add_argument("--result-root", type=Path, default=Path(f"results/{dataset}_encoder_only_1s"))
parser.add_argument("--seeds", type=int, nargs="+", default=[0, 1, 2]) parser.add_argument("--seeds", type=int, nargs="+", default=[0, 1, 2])
parser.add_argument("--device", default="cuda:0") 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("--dry-run", action="store_true", help="Validate all train/test inputs and print commands without training")
parser.add_argument("--no-progress", action="store_true") parser.add_argument("--no-progress", action="store_true")
args = parser.parse_args(argv) args = parser.parse_args(argv)
args.manifest = args.manifest or args.data_root / "ego4d_egospeak.tsv" args.dataset = dataset
args.target_root = args.target_root or args.data_root / "ego4d_target_frame" if dataset == "ego4d":
args.manifest = args.manifest or args.data_root / "ego4d_egospeak.tsv"
elif args.manifest is not None:
parser.error("EasyCom uses train sessions 4..12 and test sessions 3/2/1, not --manifest")
args.target_root = args.target_root or args.data_root / f"{dataset}_target_frame"
for key in ("data_root", "manifest", "target_root", "feature_root", "checkpoint_root", "result_root"): for key in ("data_root", "manifest", "target_root", "feature_root", "checkpoint_root", "result_root"):
setattr(args, key, getattr(args, key).expanduser().resolve()) value = getattr(args, key)
if value is not None:
setattr(args, key, value.expanduser().resolve())
if len(set(args.seeds)) != len(args.seeds) or any(seed < 0 for seed in args.seeds): 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") parser.error("Seeds must be unique and nonnegative")
return args return args
@@ -52,19 +65,18 @@ def run_paths(args, seed):
def preflight(args): def preflight(args):
names = load_ego4d_manifest_split(args.manifest, "all") names, train, test = encoder_sample_split(args.dataset, args.manifest)
train = load_ego4d_manifest_split(args.manifest, "train") unit = sample_unit(args.dataset)
test = load_ego4d_manifest_split(args.manifest, "test")
marker = args.feature_root / "complete.txt" marker = args.feature_root / "complete.txt"
if not marker.is_file() or marker.read_text().strip() != f"{len(names)} clips": if not marker.is_file() or marker.read_text().strip() != f"{len(names)} {unit}s":
raise ValueError(f"Missing/incomplete encoder-only export: {marker}") raise ValueError(f"Missing/incomplete encoder-only export: {marker}")
for seed in args.seeds: for seed in args.seeds:
checkpoint, result = run_paths(args, seed) checkpoint, result = run_paths(args, seed)
if checkpoint.exists() or checkpoint.with_suffix(".pth.json").exists() or result.exists(): 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") raise FileExistsError(f"Refusing to overwrite {checkpoint} or {result}; choose unfinished seeds/new output roots")
expected, total, width = None, 0, None expected, total, width = None, 0, None
for name in tqdm(names, desc="encoder-only inputs", unit="clip", disable=args.no_progress): for name in tqdm(names, desc="encoder-only inputs", unit=unit, disable=args.no_progress):
data = load_session_data(args.feature_root, args.target_root, name, dataset="ego4d", data = load_session_data(args.feature_root, args.target_root, name, dataset=args.dataset,
expected_mode=encoder_only_mode(args.modalities)) expected_mode=encoder_only_mode(args.modalities))
config = data.feature_metadata.get("config") or {} config = data.feature_metadata.get("config") or {}
contract = config.get("encoder_contract") or {} contract = config.get("encoder_contract") or {}
@@ -83,20 +95,21 @@ def preflight(args):
if data.states.ndim != 2 or data.hidden_size != width or len(data.states) <= 10: 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") raise ValueError(f"{name}: invalid native encoder-vector/validity-bit shape")
if expected is not None and config != expected: if expected is not None and config != expected:
raise ValueError(f"{name}: encoder export contract differs across clips") raise ValueError(f"{name}: encoder export contract differs across {unit}s")
expected = config expected = config
total += len(data.states) total += len(data.states)
del data del data
print(f"Encoder-only {args.modalities}: {len(train)} train / {len(test)} test; " print(f"{args.dataset} 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) 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; " print("No Thinker features or local-fusion branch. Native concatenation changes head capacity; "
"context remains the saved encoder windows.", flush=True) "context remains the saved encoder windows.", flush=True)
def experiment_commands(args): def experiment_commands(args):
common = ["--dataset", "ego4d", "--manifest", str(args.manifest), common = ["--dataset", args.dataset, "--feature-root", str(args.feature_root), "--target-root", str(args.target_root),
"--feature-root", str(args.feature_root), "--target-root", str(args.target_root),
"--expected-mode", encoder_only_mode(args.modalities), "--device", args.device] "--expected-mode", encoder_only_mode(args.modalities), "--device", args.device]
if args.manifest is not None:
common += ["--manifest", str(args.manifest)]
if args.no_progress: if args.no_progress:
common.append("--no-progress") common.append("--no-progress")
for seed in args.seeds: for seed in args.seeds:
@@ -108,11 +121,15 @@ def experiment_commands(args):
"--transition-objective", "none", "--seed", str(seed)] "--transition-objective", "none", "--seed", str(seed)]
infer = [sys.executable, str(REPO / "infer_turn_head.py"), *common, infer = [sys.executable, str(REPO / "infer_turn_head.py"), *common,
"--checkpoint", str(checkpoint), "--output-dir", str(result)] "--checkpoint", str(checkpoint), "--output-dir", str(result)]
if args.dataset == "easycom":
_, train_sessions, test_sessions = encoder_sample_split("easycom")
train += ["--sessions", *train_sessions]
infer += ["--sessions", *test_sessions]
yield train, infer yield train, infer
def main(argv=None): def main(argv=None, *, dataset="ego4d"):
args = parse_args(argv) args = parse_args(argv, dataset=dataset)
preflight(args) preflight(args)
env = os.environ.copy() env = os.environ.copy()
for name in ("OMP_NUM_THREADS", "MKL_NUM_THREADS", "OPENBLAS_NUM_THREADS"): for name in ("OMP_NUM_THREADS", "MKL_NUM_THREADS", "OPENBLAS_NUM_THREADS"):
@@ -123,7 +140,7 @@ def main(argv=None):
if not args.dry_run: if not args.dry_run:
subprocess.run(command, cwd=REPO, env=env, check=True) subprocess.run(command, cwd=REPO, env=env, check=True)
summary = [sys.executable, str(REPO / "summarize_local_baselines.py"), summary = [sys.executable, str(REPO / "summarize_local_baselines.py"),
"--root", str(args.result_root), "--dataset", "ego4d"] "--root", str(args.result_root), "--dataset", args.dataset]
print(shlex.join(summary), flush=True) print(shlex.join(summary), flush=True)
if not args.dry_run: if not args.dry_run:
subprocess.run(summary, cwd=REPO, env=env, check=True) subprocess.run(summary, cwd=REPO, env=env, check=True)

View File

@@ -0,0 +1,216 @@
"""EasyCom encoder-only split, packing and real CPU train/infer checks."""
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_easycom_encoder_only as runner
from test_local_turn_features import fixture
class EasyComEncoderOnlyTests(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"Session_{i}" for i in range(1, 13)]
self.train = self.names[3:]
self.test = ["Session_3", "Session_2", "Session_1"]
self.source = self.root / "local"
self.targets = self.root / "easycom_target_frame"
self.source.mkdir()
self.targets.mkdir()
self.payloads = {}
for name in self.names:
# No Thinker features, manifest, media, annotations or model exists.
_, side, target = fixture(name=name, dataset="easycom")
side["reference_features"] = "/must-not-be-read/thinker.pt"
side["states"] = "POISON: not an encoder input"
torch.save(side, self.source / f"{name}.pt")
np.save(self.targets / f"{name}.npy", target)
self.payloads[name] = side
def output_root(self, modality="av"):
return self.root / f"encoder_{modality}"
def prepare_args(self, modality="av", extra=()):
return ["--dataset", "easycom", "--data-root", str(self.root),
"--local-feature-root", str(self.source), "--modalities", modality,
"--output-root", str(self.output_root(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_dataset_defaults_and_ascending_b1_training_order(self):
args = prepare.parse_args(["--dataset", "easycom"])
root = Path("/data1/yfzx/wangx/Datasets/easycom")
self.assertEqual(args.data_root, root)
self.assertEqual(args.target_root, root / "easycom_target_frame")
self.assertIsNone(args.manifest)
run = runner.parse_args(["--feature-root", str(self.output_root())])
self.assertEqual(run.dataset, "easycom")
self.assertEqual(run.data_root, root)
self.assertEqual(run.target_root, root / "easycom_target_frame")
self.assertIsNone(run.manifest)
self.assertEqual(run.result_root.name, "easycom_encoder_only_1s")
self.assertEqual(run.checkpoint_root.name, "easycom_encoder_only_1s")
self.assertEqual(prepare.encoder_sample_split("easycom"), (self.names, self.train, self.test))
# Preserve the original export command's Ego4D default.
self.assertEqual(prepare.parse_args([]).dataset, "ego4d")
def test_all_modalities_export_twelve_real_files_and_exact_native_inputs(self):
for modality, parts in MODALITIES.items():
with self.subTest(modality=modality):
self.export(modality)
output = self.output_root(modality)
self.assertEqual((output / "complete.txt").read_text(), "12 sessions\n")
manifest = json.loads((output / "export_manifest.json").read_text())
self.assertEqual(manifest["dataset"], "easycom")
self.assertIsNone(manifest["manifest"])
self.assertEqual(manifest["train_samples"], self.train)
self.assertEqual(manifest["test_samples"], self.test)
self.assertEqual(manifest["sessions"], dict.fromkeys(self.names, 20))
self.assertEqual(manifest["num_states"], 240)
for name in self.names:
path = output / f"{name}.pt"
self.assertTrue(path.is_file())
self.assertFalse(path.is_symlink())
packed = torch.load(path, weights_only=False)
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(packed["states"], expected))
self.assertTrue(torch.equal(packed["times"], original["times"]))
self.assertTrue(torch.equal(packed["label_indices"], original["label_indices"]))
self.assertEqual(packed["dataset"], "easycom")
self.assertIs(packed["config"]["uses_thinker_states"], False)
data = load_session_data(output, self.targets, name,
expected_mode=f"encoder-only-{modality}")
self.assertIsNone(data.local_features)
self.assertEqual(tuple(data.states.shape), tuple(expected.shape))
with self.assertRaises(FileExistsError):
prepare.main(self.prepare_args(modality))
def test_partial_wrong_dataset_and_missing_heldout_session_fail_before_write(self):
name = "Session_3"
path = self.source / f"{name}.pt"
for key, value in (("partial", True), ("dataset", "ego4d")):
side = copy.deepcopy(self.payloads[name])
side[key] = value
torch.save(side, path)
with self.subTest(key=key), self.assertRaises(ValueError):
prepare.main(self.prepare_args())
self.assertFalse(self.output_root().exists())
path.unlink()
with self.assertRaises(FileNotFoundError):
prepare.main(self.prepare_args())
self.assertFalse(self.output_root().exists())
def test_easycom_rejects_ego4d_manifest(self):
with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
prepare.parse_args(self.prepare_args(extra=["--manifest", "/ego4d.tsv"]))
with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
runner.parse_args(self.runner_args(extra=["--manifest", "/ego4d.tsv"]))
def test_dry_runs_write_nothing_and_use_fixed_sessions(self):
with contextlib.redirect_stdout(io.StringIO()):
prepare.main(self.prepare_args(extra=["--dry-run"]))
self.assertFalse(self.output_root().exists())
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("--warmup-epochs") + 1], "1.6")
self.assertEqual(train[train.index("--seed") + 1], str(seed))
self.assertEqual(train[train.index("--sessions") + 1:], self.train)
self.assertEqual(infer[infer.index("--sessions") + 1:], self.test)
for command in (train, infer):
self.assertEqual(command[command.index("--dataset") + 1], "easycom")
self.assertEqual(command[command.index("--expected-mode") + 1], "encoder-only-av")
for forbidden in ("--manifest", "--local-feature-root", "--layer-fusion", "--dev-sessions"):
self.assertNotIn(forbidden, command)
with patch.object(runner.workflow.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_wrong_marker_missing_test_input_and_existing_results(self):
self.export()
args = runner.parse_args(self.runner_args())
marker = self.output_root() / "complete.txt"
marker.write_text("12 clips\n")
with self.assertRaisesRegex(ValueError, "incomplete"):
runner.preflight(args)
marker.write_text("12 sessions\n")
_, output = runner.run_paths(args, 0)
output.mkdir(parents=True)
with self.assertRaises(FileExistsError):
runner.preflight(args)
output.rmdir()
(self.output_root() / "Session_1.pt").unlink()
with self.assertRaises(FileNotFoundError):
runner.preflight(args)
def test_cpu_round_trip_three_modalities_four_epochs_no_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)
result = subprocess.run(
[sys.executable, str(Path(runner.__file__).resolve()), *self.runner_args(modality)],
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["dataset"], "easycom")
self.assertEqual(saved["epoch"], 4)
self.assertEqual(saved["train_sessions"], self.train)
self.assertEqual(saved["dev_sessions"], [])
self.assertIsNone(saved["local_contract"])
self.assertIsNone(saved["training_config"]["local_feature_root"])
width = sum(self.payloads["Session_1"]["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.test)
self.assertEqual(metrics["checkpoint_epoch"], 4)
for name in self.test:
self.assertEqual(np.load(output / "pred" / f"{name}.npy").shape, (20, 3, 10))
self.assertEqual({p.stem for p in (output / "pred").glob("*.npy")}, set(self.test))
if __name__ == "__main__":
unittest.main()