Coverage for tests/io/test_playback.py: 100%
147 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-15 09:33 +0000
1"""Real-time playback pacing tests (EventReader(real_time=True)).
3Timing assertions are deliberately loose (lower bounds tight, upper bounds
4generous) so scheduler jitter on CI machines does not flake them.
5"""
6import time
8import numpy as np
9import pytest
11from typing import Any
13SPAN_US = 200_000 # recording length: 0.2 s
14N = 200 # one event per millisecond
17@pytest.fixture
18def recording(tmp_path: Any) -> Any:
19 from evutils.io import EventWriter
20 ev = np.zeros(N, dtype=np.dtype([('t', np.int64), ('x', np.uint16), ('y', np.uint16), ('p', np.uint8)]))
21 ev['t'] = np.arange(N, dtype=np.int64) * (SPAN_US // N)
22 ev['x'] = np.arange(N) % 1280
23 ev['y'] = np.arange(N) % 720
24 ev['p'] = np.arange(N) % 2
25 p = tmp_path / "playback.raw"
26 with EventWriter(p, format="evt2") as w:
27 w.write(ev)
28 return p, ev
31def _drain(reader: Any) -> tuple[int, float]:
32 start = time.perf_counter()
33 total = 0
34 for chunk in reader:
35 total += len(chunk)
36 return total, time.perf_counter() - start
39def test_playback_paces_iteration_to_recording_speed(recording: Any) -> None:
40 from evutils.io import EventReader
41 p, ev = recording
42 with EventReader(p, real_time=True, mode="delta_t", delta_t=20_000) as r:
43 total, elapsed = _drain(r)
44 assert total == len(ev)
45 # 0.2 s recording: must take at least ~the recording duration, and not
46 # wildly more (each chunk waits only up to its own timestamp).
47 assert 0.15 <= elapsed < 1.0
50def test_playback_speed_multiplier(recording: Any) -> None:
51 from evutils.io import EventReader
52 p, ev = recording
53 with EventReader(p, real_time=True, playback_speed=4.0, mode="delta_t", delta_t=20_000) as r:
54 total, elapsed = _drain(r)
55 assert total == len(ev)
56 # 0.2 s recording at 4x -> ~0.05 s
57 assert 0.03 <= elapsed < 0.15
60def test_playback_off_is_fast(recording: Any) -> None:
61 from evutils.io import EventReader
62 p, _ = recording
63 with EventReader(p, mode="delta_t", delta_t=20_000) as r:
64 _, elapsed = _drain(r)
65 assert elapsed < 0.05
68def _run_slow_consumer(reader: Any, work: float) -> tuple[int, float]:
69 """Iterate a reader spending ``work`` seconds of 'work' per chunk."""
70 start = time.perf_counter()
71 total = 0
72 for chunk in reader:
73 total += len(chunk)
74 time.sleep(work)
75 return total, time.perf_counter() - start
78def test_playback_absorbs_slow_consumer(tmp_path: Any) -> None:
79 """Consumer work longer than the chunk period: pacing must add no extra
80 delay on top (absolute anchor, not per-chunk sleep).
82 Measured differentially against an unpaced run of the *same* slow consumer,
83 so the shared consumer-sleep cost cancels and only the paced wrapper's own
84 overhead remains in the difference.
86 Deliberately uses a longer recording read in FEW, LARGE chunks. The paced
87 wrapper's residual overhead is *per chunk* (tens of ms each on a loaded
88 macOS CI runner); the signal we're testing for -- broken per-chunk pacing
89 would add roughly one whole recording length -- is *per recording*. Few big
90 chunks + coarse sleeps (which overshoot proportionally less) therefore make
91 the signal dwarf the overhead, where the old 0.2 s / 20 ms-chunk version
92 had the overhead exceed the whole signal and flaked.
93 """
94 from evutils.io import EventReader, EventWriter
96 # 0.8 s recording, 1 event/ms.
97 span_us, n = 800_000, 800
98 ev = np.zeros(n, dtype=np.dtype([('t', np.int64), ('x', np.uint16), ('y', np.uint16), ('p', np.uint8)]))
99 ev['t'] = np.arange(n, dtype=np.int64) * (span_us // n)
100 ev['x'] = np.arange(n) % 1280
101 ev['y'] = np.arange(n) % 720
102 ev['p'] = np.arange(n) % 2
103 p = tmp_path / "playback_long.raw"
104 with EventWriter(p, format="evt2") as w:
105 w.write(ev)
107 # delta_t=200 ms -> 4 chunks; consumer sleeps 250 ms/chunk (> the 200 ms
108 # chunk period), so it is the bottleneck and correct pacing waits for nothing.
109 with EventReader(p, mode="delta_t", delta_t=200_000) as r:
110 base_total, base = _run_slow_consumer(r, work=0.25)
111 with EventReader(p, real_time=True, mode="delta_t", delta_t=200_000) as r:
112 paced_total, paced = _run_slow_consumer(r, work=0.25)
114 assert base_total == len(ev)
115 assert paced_total == len(ev)
116 # Separation is structural: correct absolute-anchor pacing adds exactly ONE
117 # delta_t (0.2 s -- the first chunk waits out its own window; the consumer is
118 # behind for the rest), while broken per-chunk pacing adds FOUR delta_t
119 # (0.8 s, one per chunk). Threshold at ~2.5 delta_t sits cleanly between,
120 # with headroom for the paced wrapper's per-chunk overhead on loaded macOS CI.
121 assert paced <= base + 0.5
124def test_playback_paces_read_calls(recording: Any) -> None:
125 from evutils.io import EventReader
126 p, _ = recording
127 with EventReader(p, real_time=True, mode="delta_t", delta_t=50_000) as r:
128 start = time.perf_counter()
129 r.read() # anchor chunk: waits out its own 50 ms span
130 r.read()
131 elapsed = time.perf_counter() - start
132 assert elapsed >= 0.075 # two 50 ms windows on the playback clock
135def test_playback_with_async_read(recording: Any) -> None:
136 from evutils.io import EventReader
137 p, ev = recording
138 with EventReader(p, real_time=True, async_read=True, mode="delta_t", delta_t=20_000) as r:
139 total, elapsed = _drain(r)
140 assert total == len(ev)
141 assert 0.15 <= elapsed < 1.0
144def test_playback_early_break_releases_prefetch(recording: Any) -> None:
145 from evutils.io import EventReader
146 p, _ = recording
147 with EventReader(p, real_time=True, async_read=True, mode="delta_t", delta_t=20_000) as r:
148 for chunk in r:
149 break
150 # the paced wrapper must close the prefetch iterator on break,
151 # releasing the decoder for direct reads again
152 r.reset()
153 out = r.read_all()
154 assert len(out) == N
157def test_playback_reset_rearms_anchor(recording: Any) -> None:
158 from evutils.io import EventReader
159 p, _ = recording
160 with EventReader(p, real_time=True, playback_speed=10.0, mode="delta_t", delta_t=100_000) as r:
161 _drain(r)
162 r.reset()
163 start = time.perf_counter()
164 total, _ = _drain(r)
165 elapsed = time.perf_counter() - start
166 assert total == N
167 # second pass paces again from a fresh anchor (0.2 s / 10x = 20 ms)
168 assert elapsed >= 0.015
171def test_playback_invalid_speed_rejected(recording: Any) -> None:
172 from evutils.io import EventReader
173 p, _ = recording
174 with pytest.raises(ValueError):
175 EventReader(p, real_time=True, playback_speed=0)
176 with pytest.raises(ValueError):
177 EventReader(p, real_time=True, playback_speed=-1.5)
180# --------------------------------------------------------------------------- #
181# Idle-gap handling (max_gap)
182# --------------------------------------------------------------------------- #
183GAP_US = 500_000 # 0.5 s of silence before the action starts
186@pytest.fixture
187def gapped_recording(tmp_path: Any) -> Any:
188 """A tiny blip at t=0, then 0.5 s of silence, then a short burst.
190 Reproduces recordings whose content starts well after t=0 -- without
191 gap-skipping, real-time pacing stalls on one long sleep to reach the burst.
192 """
193 from evutils.io import EventWriter
194 t = np.concatenate([
195 np.array([0], dtype=np.int64), # lead-in blip
196 GAP_US + np.arange(50, dtype=np.int64) * 1000, # burst: 50 ms
197 ])
198 ev = np.zeros(len(t), dtype=np.dtype([('t', np.int64), ('x', np.uint16), ('y', np.uint16), ('p', np.uint8)]))
199 ev['t'] = t
200 ev['x'] = np.arange(len(t)) % 1280
201 ev['y'] = np.arange(len(t)) % 720
202 ev['p'] = np.arange(len(t)) % 2
203 p = tmp_path / "gapped.raw"
204 with EventWriter(p, format="evt2") as w:
205 w.write(ev)
206 return p, ev
209def test_max_gap_skips_dead_air(gapped_recording: Any) -> None:
210 """With max_gap set, the 0.5 s silent lead-in is skipped, not slept through."""
211 from evutils.io import EventReader
212 p, ev = gapped_recording
213 with EventReader(p, real_time=True, mode="delta_t", delta_t=20_000, max_gap=0.1) as r:
214 total, elapsed = _drain(r)
215 assert total == len(ev)
216 # Only the ~50 ms burst is paced; the 0.5 s gap must not be waited out.
217 assert elapsed < 0.3
220def test_max_gap_none_is_strict_realtime(gapped_recording: Any) -> None:
221 """max_gap=None keeps strict pacing: the full 0.5 s gap is honoured."""
222 from evutils.io import EventReader
223 p, ev = gapped_recording
224 with EventReader(p, real_time=True, mode="delta_t", delta_t=20_000, max_gap=None) as r:
225 total, elapsed = _drain(r)
226 assert total == len(ev)
227 # Must sleep through the ~0.5 s gap before the burst plays.
228 assert elapsed >= 0.45
231def test_max_gap_invalid_rejected(recording: Any) -> None:
232 from evutils.io import EventReader
233 p, _ = recording
234 with pytest.raises(ValueError):
235 EventReader(p, real_time=True, max_gap=0)
236 with pytest.raises(ValueError):
237 EventReader(p, real_time=True, max_gap=-1.0)