"""CPU regressions for a verified Ego4D exclusive-end bbox record.""" from __future__ import annotations import contextlib import copy import io import json from pathlib import Path import sys import tempfile from types import SimpleNamespace import unittest from unittest.mock import patch import numpy as np from PIL import Image import torch import extract_turn_local_features as extract from local_turn_features import load_local_features from test_local_turn_features import fixture UID = "31488ff6-05ef-4362-9786-35981b06f307" def terminal_clip(): return { "clip_uid": UID, "clip_start_frame": 0, "clip_end_frame": 8999, "clip_start_sec": 0, "clip_end_sec": 299.9666666666666, "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) ]}]}], } def terminal_annotation(): clip = terminal_clip() 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", )}} class Ego4DBoxFrameTests(unittest.TestCase): def setUp(self): torch.set_num_threads(1) self.source = SimpleNamespace(num_frames=8999, source_fps=30.) def test_confirmed_terminal_only_is_removed_without_mutating_input(self): annotation = terminal_annotation() original = copy.deepcopy(annotation) with contextlib.redirect_stdout(io.StringIO()) as log: frames, audit = extract.validate_ego4d_box_frames(UID, annotation, self.source) self.assertEqual(frames, {frame: original["frames"][frame] for frame in (0, 8985, 8998)}) self.assertEqual(annotation, original) self.assertEqual(audit["ignored_frame_indices"], [8999]) self.assertEqual(audit["ignored_box_count"], 1) self.assertEqual(audit["annotation_frame_range"], [0, 8999]) self.assertIn("No frame shift or feature/target row removal", log.getvalue()) def test_valid_and_empty_mappings_are_unchanged_even_without_bounds(self): for frames in ({}, {0: [], 8998: [{"id": "valid"}]}): with self.subTest(frames=frames): result, audit = extract.validate_ego4d_box_frames( UID, {"frames": frames, "bounds": {}}, self.source) self.assertIs(result, frames) self.assertEqual(audit["ignored_frame_indices"], []) def test_larger_overrun_negative_and_truncated_source_still_fail(self): for extra in (-1, 9000, 12000): annotation = terminal_annotation() annotation["frames"][extra] = [] with self.subTest(frame=extra), self.assertRaisesRegex(ValueError, "source_frames=8999"): extract.validate_ego4d_box_frames(UID, annotation, self.source) with self.assertRaisesRegex(ValueError, "source_frames=8998"): extract.validate_ego4d_box_frames( UID, terminal_annotation(), SimpleNamespace(num_frames=8998, source_fps=30.)) def test_terminal_requires_zero_origin_matching_declared_end_and_duration(self): for key, value in (("clip_start_frame", 1), ("clip_end_frame", 9000), ("clip_start_sec", .1), ("clip_end_sec", 300.), ("clip_end_sec", float("nan"))): annotation = terminal_annotation() annotation["bounds"][key] = value with self.subTest(key=key, value=value), self.assertRaisesRegex(ValueError, "declared_bounds"): extract.validate_ego4d_box_frames(UID, annotation, self.source) for key in terminal_annotation()["bounds"]: annotation = terminal_annotation() del annotation["bounds"][key] with self.subTest(missing=key), self.assertRaises(ValueError): extract.validate_ego4d_box_frames(UID, annotation, self.source) for fps in (29.97, 15., 0., float("nan")): with self.subTest(fps=fps), self.assertRaises(ValueError): extract.validate_ego4d_box_frames( UID, terminal_annotation(), SimpleNamespace(num_frames=8999, source_fps=fps)) def test_full_extraction_keeps_1499_rows_saves_audit_and_reuses(self): 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) media = SimpleNamespace( clips=[source], metadata=lambda: {}, _source_frame_for_time=lambda t: (0, int(t * 30 + 1e-8), int(t * 30 + 1e-8) / 30), frame_at_or_before=lambda t: image, audio_slice=lambda start, end: np.array([start, end]), ) fake_module = SimpleNamespace(CausalSourceAV=lambda *a, **kw: media, SourceClipSpec=lambda path: path, _easycom_directory_specs=None) encoder = SimpleNamespace( visual_dim=6, audio_dim=5, transformers_version="4.57.6", model_config_sha256="test", images=lambda crops: torch.ones(len(crops), 6), audio=lambda waveform: torch.full((5,), float(waveform[-1])), ) with tempfile.TemporaryDirectory() as temporary: 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", [])): (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()) 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"] 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" result = torch.load(destination, weights_only=False) self.assertEqual(result["visual"].shape, (1499, 6)) self.assertEqual(result["audio"].shape, (1499, 5)) self.assertTrue(torch.equal(result["times"], reference["times"])) 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]) 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") # Previously completed valid sidecars have no audit. Their # vectors/config remain reusable: only unsampled boxes changed. del result["box_frame_audit"] torch.save(result, destination) model.reset_mock() 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]) if __name__ == "__main__": unittest.main()