feat(testing): add max_frames parameter to E2EHarness

Caps the iteration length (and the matching GT slice) when set, so CI
tiers can stay fast on multi-thousand-frame sequences like EuRoC MH_01
(3682 frames ≈ 3+ hours at 3-5s/frame). Also useful for eyeballing a
new adapter's first N frames before committing to a full run.

Three new harness tests cover truncation, explicit None, and over-large
limits. No change to existing adapters or downstream tests.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Yuzviak
2026-04-17 17:42:38 +03:00
committed by Maksym Yuzviak
parent 5128ac17ba
commit fd54af2d9f
2 changed files with 48 additions and 2 deletions
+19 -2
View File
@@ -45,11 +45,24 @@ class HarnessResult:
class E2EHarness:
"""Drives FlightProcessor from a DatasetAdapter; collects results."""
"""Drives FlightProcessor from a DatasetAdapter; collects results.
def __init__(self, adapter: DatasetAdapter, flight_id: str = "e2e-flight") -> None:
`max_frames` caps the iteration (and the matching GT slice). Useful for:
- CI tiers where a full multi-thousand-frame sequence is too slow.
- Debugging a new adapter — eyeball the first N frames before committing
to a multi-hour run.
`None` (default) means "consume the full dataset".
"""
def __init__(
self,
adapter: DatasetAdapter,
flight_id: str = "e2e-flight",
max_frames: Optional[int] = None,
) -> None:
self._adapter = adapter
self._flight_id = flight_id
self._max_frames = max_frames
self._estimates: list[tuple[int, Optional[tuple[float, float, float]]]] = []
async def run(self) -> HarnessResult:
@@ -57,6 +70,10 @@ class E2EHarness:
frames = list(self._adapter.iter_frames())
gt_poses = list(self._adapter.iter_ground_truth())
if self._max_frames is not None:
frames = frames[: self._max_frames]
gt_poses = gt_poses[: self._max_frames]
for frame in frames:
image = self._load_or_synth_image(frame.image_path)
result = await processor.process_frame(
+29
View File
@@ -35,3 +35,32 @@ async def test_harness_captures_ground_truth_as_enu():
east_disp = result.ground_truth[-1, 0] - result.ground_truth[0, 0]
# Allow 5% tolerance for the lat/lon → ENU conversion approximation
assert abs(east_disp - 4.0) < 0.5
@pytest.mark.asyncio
async def test_harness_max_frames_truncates_iteration():
# Adapter says 10 frames; harness with max_frames=3 should stop at 3.
adapter = SyntheticAdapter(num_frames=10, fps=5.0)
harness = E2EHarness(adapter, max_frames=3)
result = await harness.run()
assert result.num_frames_submitted == 3
# GT aligned to the same truncation so downstream metrics match lengths
assert result.ground_truth.shape[0] == 3
@pytest.mark.asyncio
async def test_harness_max_frames_none_runs_full():
# Explicit None = no limit (same as omitting the parameter).
adapter = SyntheticAdapter(num_frames=4, fps=5.0)
harness = E2EHarness(adapter, max_frames=None)
result = await harness.run()
assert result.num_frames_submitted == 4
@pytest.mark.asyncio
async def test_harness_max_frames_larger_than_dataset_is_harmless():
# Limit above dataset size should not over-extend.
adapter = SyntheticAdapter(num_frames=4, fps=5.0)
harness = E2EHarness(adapter, max_frames=100)
result = await harness.run()
assert result.num_frames_submitted == 4