From 18e87736857db88b73efd2959fa6daa57eb8431d Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Fri, 28 Aug 2026 09:26:45 +0800 Subject: [PATCH] Handle two-frame Ego4D bbox tails and preserve worker errors --- docs/ego4d_local_baselines.md | 19 +++++-- extract_ego4d_local_multigpu.py | 16 ++++-- extract_turn_local_features.py | 27 ++++++---- test_ego4d_box_frames.py | 93 +++++++++++++++++++++++++-------- test_ego4d_local_multigpu.py | 41 +++++++++++++++ 5 files changed, 156 insertions(+), 40 deletions(-) diff --git a/docs/ego4d_local_baselines.md b/docs/ego4d_local_baselines.md index cd1141b..14855c4 100644 --- a/docs/ego4d_local_baselines.md +++ b/docs/ego4d_local_baselines.md @@ -167,6 +167,8 @@ tail -f logs/ego4d_local_8gpu/<本次时间_PID>/gpu0.log 不用 `nohup` 或 shell 的 `&`，主命令会一直等待；Ctrl+C 或一个 worker 失败时，会停止本次启动的其他 worker 及其进程组，不杀无关训练。已完成的 `.pt` 保留，重跑同一命令会校验后跳过； 失败时先读对应日志。不要同时另开同输出根的重复提取任务。 +正常进度仍使用简短摘要；worker 失败时改为输出日志最后 64 KiB 中的最后一段 traceback +（没有 traceback 则输出末尾 40 行），不再只留下最后 300 个字符。原始日志仍完整保留。 最终输出是实体文件 `local_qwen_1s/.pt`，每个文件同时含 A/V，不需要第二个音频任务。 无框帧仍保留在 5Hz 特征/标签网格中，不能按预览筛选删除。 @@ -180,14 +182,21 @@ worker 及其进程组，不杀无关训练。已完成的 `.pt` 保留，重跑 `clip_end_frame=8999`、`clip_end_sec=299.9666666666666`，唯一越界索引是 `8999`。 这是无对应解码帧的末尾记录，不应把该框移到 `8998`，也不应将所有框减 1。 -提取器只在以下条件同时满足时忽略末尾记录：唯一越界索引为 `N`， +随后在 `ac7d3d16-d74d-4fe3-a1e0-cab0b1adf492` 核对到： +原视频 8998 帧、30 FPS，有效索引为 `0..8997`；`clip_end_frame=8998`、 +`clip_end_sec=299.9333333333333`，但 bbox 仍到 `8999`，越界索引为 `[8998, 8999]`。 +因此只兼容一个末尾索引还不够；这第二例同样是标注结束位置与媒体长度一致、bbox 尾部多出记录。 + +提取器只在以下条件同时满足时忽略末尾记录：所有越界索引都属于 `N、N+1`， `N=视频帧数=clip_end_frame`，标注帧起点为 0、时间起点在 1 微秒内为 0， 且标注时长与 `N / source_fps` 相差不超过 0.1 毫秒。 -缺少确认边界、较大越界或时长不匹配仍会报错，并打印具体范围；不自动跳过 clip。 +容差限制的是**末尾索引位置**，不是框数：即使仅一条记录，出现在 `N+2` 或更远处也不接受。 +负索引、缺少确认边界、较大越界或时长不匹配仍会报错，并打印具体范围；不自动跳过 clip。 +这是基于已核实样例的受限兼容规则，不代表所有 Ego4D 越界记录都可忽略。 日志显示忽略的末尾索引和框数；新 `.pt` 的 `box_frame_audit` 与预览目录的 `box_frame_audit.json` 保存原始范围、视频帧数、标注边界和忽略记录。 -不修改原视频、JSON、target 或已完成的特征。本例仍输出 1499 行，对应截止时间 +不修改原视频、JSON、target 或已完成的特征。首例已核对的 1499 行 target 仍对应 1499 行特征，截止时间为 `0.2..299.8s`，2 FPS 的最后一个输入采样时刻仍为 `299.5s`。 该处理不改编码器/池化配置，之前完成且通过校验的特征可以继续复用；同步修复后重跑原 8 卡命令即可。 @@ -271,5 +280,7 @@ python -m unittest -q test_easycom_head_objectives test_local_turn_features test CPU 回归包括 manifest 顺序/防交叉重复、完整批量提取的模型复用、断点文件校验、 单层/多层全局特征、测试集输入预检，以及合成 Ego4D 上四组各 4 epochs 的训练/推理与增量汇总。 另测试原 train/test 目录、433 clips 的 8 卡完整互斥分片、GPU mask、独立日志、失败/中断时清理自有进程。 -末尾框回归覆盖严格边界验证、其它越界拒绝、1499 行完整提取、审计保存和旧特征续跑。 +末尾框回归覆盖一帧/两帧越界、严格边界验证、较远稀疏越界拒绝、1499 行完整提取、 +审计保存和旧特征续跑；合成测试中，处理后的特征/时间戳/掩码/人物记录与先去掉越界框的结果完全相同。 +另覆盖启动期和运行期的长异常信息保留，正常进度仍保持简短。 真实 Ego4D 像素、提供框和 3B/L20 提取仍需按上面的 preview/smoke 步骤验证。 diff --git a/extract_ego4d_local_multigpu.py b/extract_ego4d_local_multigpu.py index f82bfac..026b3cc 100644 --- a/extract_ego4d_local_multigpu.py +++ b/extract_ego4d_local_multigpu.py @@ -107,11 +107,17 @@ def worker_environment(gpu, cpu_threads): return env -def log_tail(path): +def log_tail(path, *, failure=False): + """Compact progress, but retain traceback/long diagnostic lines on failure.""" try: with path.open("rb") as handle: - handle.seek(max(0, path.stat().st_size - 4096)) + handle.seek(max(0, path.stat().st_size - (65536 if failure else 4096))) lines = handle.read().decode("utf-8", errors="replace").replace("\r", "\n").splitlines() + if failure: + for index in range(len(lines) - 1, -1, -1): + if lines[index].startswith("Traceback (most recent call last):"): + return "\n".join(lines[index:]) + return "\n".join(lines[-40:]) or "no worker output" return next((line.strip() for line in reversed(lines) if line.strip()), "starting")[-300:] except FileNotFoundError: return "starting" @@ -166,14 +172,16 @@ def run_workers(plans, args, log_dir): time.sleep(args.startup_delay) for started, child, child_log in running: if child.poll() not in (None, 0): - raise RuntimeError(f"GPU {started['gpu']} failed (exit={child.returncode}); see {child_log}") + raise RuntimeError(f"GPU {started['gpu']} failed (exit={child.returncode}); see {child_log}\n" + f"{log_tail(child_log, failure=True)}") next_status, last_status = 0., {} while True: completed = 0 for plan, process, log in running: code = process.poll() if code not in (None, 0): - raise RuntimeError(f"GPU {plan['gpu']} failed (exit={code}); see {log}\n{log_tail(log)}") + raise RuntimeError(f"GPU {plan['gpu']} failed (exit={code}); see {log}\n" + f"{log_tail(log, failure=True)}") completed += code == 0 if time.monotonic() >= next_status or completed == len(running): for plan, process, log in running: diff --git a/extract_turn_local_features.py b/extract_turn_local_features.py index f2d85f1..ffede95 100644 --- a/extract_turn_local_features.py +++ b/extract_turn_local_features.py @@ -27,6 +27,9 @@ from local_turn_features import ( TESTED_TRANSFORMERS_VERSIONS = ("4.52.3", "4.57.6") +# Verified Ego4D exports can retain one or two bbox frame indices after the +# media endpoint. This bounds frame positions, not the number of people/boxes. +EGO4D_MAX_TERMINAL_FRAME_OVERRUN = 2 def validate_local_transformers_version(version: str) -> str: @@ -191,12 +194,13 @@ def load_ego4d_boxes(paths: list[Path], names: list[str]) -> dict[str, Any]: def validate_ego4d_box_frames(name: str, annotation: dict[str, Any], source): - """Ignore only a confirmed exclusive-end record; never shift/clamp boxes.""" + """Ignore a bounded, boundary-confirmed tail; never shift/clamp boxes.""" frames, bounds = annotation["frames"], annotation["bounds"] count = source.num_frames outside = sorted(frame for frame in frames if not 0 <= frame < count) audit = { - "policy": "clip-local-zero-based; ignore-confirmed-exclusive-end-only", + "policy": "clip-local-zero-based; ignore-confirmed-tail-up-to-2-frames", + "max_terminal_frame_overrun": EGO4D_MAX_TERMINAL_FRAME_OVERRUN, "source_frames": count, "source_fps": source.source_fps, "declared_bounds": dict(bounds), "annotation_frame_range": [min(frames), max(frames)] if frames else None, @@ -211,22 +215,27 @@ def validate_ego4d_box_frames(name: str, annotation: dict[str, Any], source): and source.source_fps > 0 and abs(start_sec) <= 1e-6 and abs((end_sec - start_sec) - count / source.source_fps) <= 1e-4 ) - if (outside != [count] or bounds.get("clip_start_frame") != 0 + # Allow only N and N+1. Counting outliers alone would accidentally permit + # sparse but arbitrarily distant boxes; the maximum index must be bounded. + small_terminal_tail = count <= outside[0] <= outside[-1] < count + EGO4D_MAX_TERMINAL_FRAME_OVERRUN + if (not small_terminal_tail or bounds.get("clip_start_frame") != 0 or bounds.get("clip_end_frame") != count or not duration_matches): raise ValueError( f"Ego4D annotation frame exceeds source clip: {name}; " f"source_frames={count}, source_fps={source.source_fps}, " f"annotation_range={audit['annotation_frame_range']}, " f"out_of_range_count={len(outside)}, first_10={outside[:10]}, " - f"declared_bounds={bounds}. Only frame == source_frames == clip_end_frame " - "with zero-origin, matching duration is an allowed terminal record; " + f"declared_bounds={bounds}. Only terminal indices in " + f"[{count}, {count + EGO4D_MAX_TERMINAL_FRAME_OVERRUN - 1}] " + "with source_frames == clip_end_frame, zero origin and matching duration are allowed; " "check annotation/video alignment instead of clamping or skipping the clip." ) - audit.update(ignored_frame_indices=[count], ignored_box_count=len(frames[count])) - print(f"{name}: ignoring exclusive-end bbox record at frame {count} " - f"({len(frames[count])} boxes); decoded frames are 0..{count - 1}. " + ignored_boxes = sum(len(frames[frame]) for frame in outside) + audit.update(ignored_frame_indices=outside, ignored_box_count=ignored_boxes) + print(f"{name}: ignoring terminal bbox frame indices {outside} " + f"({ignored_boxes} boxes); decoded frames are 0..{count - 1}. " "No frame shift or feature/target row removal.", flush=True) - return {frame: boxes for frame, boxes in frames.items() if frame != count}, audit + return {frame: boxes for frame, boxes in frames.items() if 0 <= frame < count}, audit def build_media_and_boxes(args: argparse.Namespace, name: str, end: float, ego_boxes: dict[str, Any]): diff --git a/test_ego4d_box_frames.py b/test_ego4d_box_frames.py index 185ae5f..4a6b3e8 100644 --- a/test_ego4d_box_frames.py +++ b/test_ego4d_box_frames.py @@ -1,4 +1,4 @@ -"""CPU regressions for a verified Ego4D exclusive-end bbox record.""" +"""CPU regressions for verified one/two-frame Ego4D terminal bbox overruns.""" from __future__ import annotations @@ -23,21 +23,22 @@ from test_local_turn_features import fixture UID = "31488ff6-05ef-4362-9786-35981b06f307" +TWO_FRAME_UID = "ac7d3d16-d74d-4fe3-a1e0-cab0b1adf492" -def terminal_clip(): +def terminal_clip(*, uid=UID, count=8999, tail_frames=1): return { - "clip_uid": UID, "clip_start_frame": 0, "clip_end_frame": 8999, - "clip_start_sec": 0, "clip_end_sec": 299.9666666666666, + "clip_uid": uid, "clip_start_frame": 0, "clip_end_frame": count, + "clip_start_sec": 0, "clip_end_sec": count / 30., "persons": [{"person_id": "p1", "tracking_paths": [{"track": [ {"frame": frame, "clip_frame": None, "x": 0, "y": 0, "width": 2, "height": 2} - for frame in (0, 8985, 8998, 8999) + for frame in (0, 8985, count - 1, *range(count, count + tail_frames)) ]}]}], } -def terminal_annotation(): - clip = terminal_clip() +def terminal_annotation(**kwargs): + clip = terminal_clip(**kwargs) return {"frames": extract.ego4d_clip_boxes(clip), "bounds": {key: clip[key] for key in ( "clip_start_frame", "clip_end_frame", "clip_start_sec", "clip_end_sec", @@ -69,8 +70,33 @@ class Ego4DBoxFrameTests(unittest.TestCase): self.assertIs(result, frames) self.assertEqual(audit["ignored_frame_indices"], []) + def test_confirmed_two_frame_tail_preserves_last_valid_box_and_counts_people(self): + annotation = terminal_annotation(uid=TWO_FRAME_UID, count=8998, tail_frames=2) + annotation["frames"][8999].append({"id": "p2", "xyxy": [1, 1, 2, 2]}) + original = copy.deepcopy(annotation) + with contextlib.redirect_stdout(io.StringIO()): + frames, audit = extract.validate_ego4d_box_frames( + TWO_FRAME_UID, annotation, SimpleNamespace(num_frames=8998, source_fps=30.)) + self.assertEqual(frames, {frame: original["frames"][frame] for frame in (0, 8985, 8997)}) + self.assertEqual(annotation, original) + self.assertEqual(audit["ignored_frame_indices"], [8998, 8999]) + self.assertEqual(audit["ignored_box_count"], 3) + self.assertEqual(audit["max_terminal_frame_overrun"], 2) + + def test_sparse_tail_is_bounded_by_position_not_outlier_count(self): + annotation = terminal_annotation(count=8998, tail_frames=2) + del annotation["frames"][8998] + source = SimpleNamespace(num_frames=8998, source_fps=30.) + with contextlib.redirect_stdout(io.StringIO()): + frames, audit = extract.validate_ego4d_box_frames(TWO_FRAME_UID, annotation, source) + self.assertEqual(audit["ignored_frame_indices"], [8999]) + self.assertEqual(sorted(frames), [0, 8985, 8997]) + annotation["frames"][9999] = annotation["frames"].pop(8999) + with self.assertRaisesRegex(ValueError, "out_of_range_count=1"): + extract.validate_ego4d_box_frames(TWO_FRAME_UID, annotation, source) + def test_larger_overrun_negative_and_truncated_source_still_fail(self): - for extra in (-1, 9000, 12000): + for extra in (-1, 9001, 12000): annotation = terminal_annotation() annotation["frames"][extra] = [] with self.subTest(frame=extra), self.assertRaisesRegex(ValueError, "source_frames=8999"): @@ -98,9 +124,15 @@ class Ego4DBoxFrameTests(unittest.TestCase): UID, terminal_annotation(), SimpleNamespace(num_frames=8999, source_fps=fps)) def test_full_extraction_keeps_1499_rows_saves_audit_and_reuses(self): + self._assert_full_extraction_and_reuse(UID, 8999, 1) + + def test_two_frame_tail_extraction_matches_manual_removal_of_unsampled_boxes(self): + self._assert_full_extraction_and_reuse(TWO_FRAME_UID, 8998, 2) + + def _assert_full_extraction_and_reuse(self, uid, count, tail_frames): image = Image.new("RGB", (2, 2)) - source = SimpleNamespace(num_frames=8999, source_fps=30., crop=None, dense_start=0., - path=Path(f"{UID}.mp4"), width=2, height=2) + source = SimpleNamespace(num_frames=count, source_fps=30., crop=None, dense_start=0., + path=Path(f"{uid}.mp4"), width=2, height=2) media = SimpleNamespace( clips=[source], metadata=lambda: {}, _source_frame_for_time=lambda t: (0, int(t * 30 + 1e-8), int(t * 30 + 1e-8) / 30), @@ -118,24 +150,25 @@ class Ego4DBoxFrameTests(unittest.TestCase): root = Path(temporary) for folder in ("global", "targets", "v1/clips", "v1/annotations"): (root / folder).mkdir(parents=True) - reference, _, targets = fixture(n=1499, name=UID, dataset="ego4d") - torch.save(reference, root / "global" / f"{UID}.pt") - np.save(root / "targets" / f"{UID}.npy", targets) - (root / "v1/clips" / f"{UID}.mp4").write_bytes(b"fake-video") - for split, clips in (("train", [terminal_clip()]), ("val", [])): + reference, _, targets = fixture(n=1499, name=uid, dataset="ego4d") + torch.save(reference, root / "global" / f"{uid}.pt") + np.save(root / "targets" / f"{uid}.npy", targets) + (root / "v1/clips" / f"{uid}.mp4").write_bytes(b"fake-video") + clip = terminal_clip(uid=uid, count=count, tail_frames=tail_frames) + for split, clips in (("train", [clip]), ("val", [])): (root / f"v1/annotations/av_{split}.json").write_text(json.dumps({"videos": [{"clips": clips}]})) - loaded = extract.load_ego4d_boxes([root / "v1/annotations/av_train.json"], [UID]) - self.assertEqual(loaded[UID], terminal_annotation()) + loaded = extract.load_ego4d_boxes([root / "v1/annotations/av_train.json"], [uid]) + self.assertEqual(loaded[uid], terminal_annotation(uid=uid, count=count, tail_frames=tail_frames)) argv = ["--dataset", "ego4d", "--data-root", str(root), "--feature-root", str(root / "global"), "--target-root", str(root / "targets"), "--output-root", str(root / "local"), - "--samples", UID, "--no-progress"] + "--samples", uid, "--no-progress"] with patch.dict(sys.modules, {"transformers": SimpleNamespace(__version__="4.57.6"), "streaming_turn_model_directly": fake_module}), \ patch.object(extract, "FrozenQwenLocalEncoder", return_value=encoder) as model, \ contextlib.redirect_stdout(io.StringIO()): extract.main(argv) model.assert_called_once() - destination = root / "local" / f"{UID}.pt" + destination = root / "local" / f"{uid}.pt" result = torch.load(destination, weights_only=False) self.assertEqual(result["visual"].shape, (1499, 6)) self.assertEqual(result["audio"].shape, (1499, 5)) @@ -143,10 +176,11 @@ class Ego4DBoxFrameTests(unittest.TestCase): self.assertTrue(torch.equal(result["label_indices"], reference["label_indices"])) self.assertAlmostEqual(result["times"][-1].item(), 299.8) self.assertFalse(result["partial"]) - self.assertEqual(result["box_frame_audit"]["ignored_frame_indices"], [8999]) + ignored = list(range(count, count + tail_frames)) + self.assertEqual(result["box_frame_audit"]["ignored_frame_indices"], ignored) self.assertEqual(result["persons"]["frame_times"][-1].item(), 299.5) self.assertEqual(result["persons"]["records"][-1]["source_frame_index"], 8985) - load_local_features(destination, reference, sample_id=UID, dataset="ego4d", modalities="av") + load_local_features(destination, reference, sample_id=uid, dataset="ego4d", modalities="av") # Previously completed valid sidecars have no audit. Their # vectors/config remain reusable: only unsampled boxes changed. del result["box_frame_audit"] @@ -155,8 +189,21 @@ class Ego4DBoxFrameTests(unittest.TestCase): extract.main(argv + ["--reuse-existing"]) model.assert_not_called() extract.main(argv + ["--preview-only", "--preview-count", "1"]) - audit = json.loads((root / "local" / f"{UID}_preview/box_frame_audit.json").read_text()) - self.assertEqual(audit["ignored_frame_indices"], [8999]) + audit = json.loads((root / "local" / f"{uid}_preview/box_frame_audit.json").read_text()) + self.assertEqual(audit["ignored_frame_indices"], ignored) + # The automatic tail handling must produce the same vectors, + # masks, timestamps and source-frame records as a clean fixture. + track = clip["persons"][0]["tracking_paths"][0] + track["track"] = [box for box in track["track"] if box["frame"] < count] + (root / "v1/annotations/av_train.json").write_text(json.dumps({"videos": [{"clips": [clip]}]})) + extract.main(argv + ["--output-root", str(root / "clean")]) + clean = torch.load(root / "clean" / f"{uid}.pt", weights_only=False) + for key in ("visual", "audio", "visual_valid", "audio_valid", "times", "label_indices"): + self.assertTrue(torch.equal(result[key], clean[key]), key) + for key in ("features", "frame_times", "frame_offsets"): + self.assertTrue(torch.equal(result["persons"][key], clean["persons"][key]), key) + self.assertEqual(result["persons"]["records"], clean["persons"]["records"]) + self.assertEqual(clean["box_frame_audit"]["ignored_frame_indices"], []) if __name__ == "__main__": diff --git a/test_ego4d_local_multigpu.py b/test_ego4d_local_multigpu.py index 6c2daab..655ff93 100644 --- a/test_ego4d_local_multigpu.py +++ b/test_ego4d_local_multigpu.py @@ -98,6 +98,47 @@ class MultiGPUExtractionTests(unittest.TestCase): self.assertEqual((self.root / "gpu2.log").read_text().strip(), "2 2") self.assertEqual((self.root / "gpu5.log").read_text().strip(), "5 2") + def test_failure_tail_keeps_traceback_and_long_diagnostics_but_progress_is_short(self): + path = self.root / "worker.log" + message = ("ValueError: Ego4D annotation frame exceeds source clip: " + self.names[0] + + "; source_frames=8998; " + "diagnostic " * 100 + "first_10=[8998, 8999]") + traceback = "Traceback (most recent call last):\n File 'extract.py', line 200\n" + message + path.write_text("old progress\n" * 10000 + traceback + "\n") + self.assertEqual(launch.log_tail(path, failure=True), traceback) + self.assertLessEqual(len(launch.log_tail(path)), 300) + path.write_text(message + "\n") + self.assertEqual(launch.log_tail(path, failure=True), message) + self.assertEqual(launch.log_tail(self.root / "missing.log", failure=True), "starting") + + def test_failure_diagnostics_survive_both_startup_and_polling_errors(self): + message = f"clip={self.names[0]}; source_frames=8998; " + "detail " * 100 + "outliers=[8998, 8999]" + for startup in (True, False): + with self.subTest(startup=startup): + folder = self.root / str(startup) + folder.mkdir() + # Simulate a flushed failing worker at startup or in the poll + # loop without scheduler-dependent sleep/timing assertions. + process = SimpleNamespace(returncode=1) + process.poll = lambda: 1 + process.wait = lambda **kwargs: 1 + process.pid = None # Stub only; must never be passed to a real signal call. + + def start(command, **kwargs): + kwargs["stdout"].write(("Traceback (most recent call last):\nValueError: " + message + "\n").encode()) + kwargs["stdout"].flush() + if not startup: + polls = iter([None, 1, 1]) + process.poll = lambda: next(polls, 1) + return process + + args = SimpleNamespace(cpu_threads=1, startup_delay=0., status_interval=.01) + plans = [{"gpu": 7, "samples": [self.names[0]], "command": ["unused"]}] + with patch("extract_ego4d_local_multigpu.subprocess.Popen", side_effect=start), \ + contextlib.redirect_stdout(io.StringIO()), self.assertRaises(RuntimeError) as raised: + launch.run_workers(plans, args, folder) + self.assertIn(message, str(raised.exception)) + self.assertIn("Traceback (most recent call last):", str(raised.exception)) + def test_failing_worker_stops_only_its_launched_peers(self): args = SimpleNamespace(cpu_threads=1, startup_delay=0., status_interval=.01) plans = [{"gpu": 0, "samples": ["dummy"], "command": [sys.executable, "-c", "import time; time.sleep(60)"]},