Coverage for tests/io/test_review_regressions.py: 96%
308 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-18 05:24 +0000
1"""Regression tests for correctness bugs found in review (July 2026).
3Each test pins one confirmed bug so the fix stays in place:
51. Double-seek timestamp corruption -- a second ``seek()`` must not inherit the
6 first seek's TIME_HIGH wrap correction (``_evt.py`` ``seek``).
72. Dense kernels scrambling coordinates -- structured arrays whose fields are
8 stored in a non-canonical order must still decode ``(t, x, y, p)`` correctly
9 (``jit.py`` ``lazy_njit_unwrapped_events``).
103. EVT3 timestamps overflowing at ~71.6 min -- both the scalar and the
11 vectorised event paths must keep full 64-bit timestamps past 2**32 µs
12 (``csrc/evt3.c`` state pipeline + ``EMIT_SOA``).
13"""
14import os
15import tempfile
16from pathlib import Path
18import numpy as np
19import pytest
21from evutils.io import EventReader, EventWriter
22from evutils.types import Event_dtype
25# --------------------------------------------------------------------------- #
26# Bug 1: double seek must not inherit the previous seek's wrap correction
27# --------------------------------------------------------------------------- #
28def _ramp(n=20_000, dt=1_000):
29 ev = np.zeros(n, dtype=Event_dtype)
30 ev["t"] = np.arange(n, dtype=np.int64) * dt
31 ev["x"] = np.arange(n) % 1280
32 ev["y"] = np.arange(n) % 720
33 ev["p"] = np.arange(n) % 2
34 return ev
37@pytest.mark.parametrize("fmt", ["evt3", "evt2", "evt21"])
38def test_double_seek_across_wrap_then_drain(tmp_path, fmt):
39 """First seek lands past the EVT3 TIME_HIGH wrap (2**24 µs); the second seek
40 to a pre-wrap target must then read a contiguous, correctly-timestamped tail.
42 Before the fix the stale ``_seek_correction`` from the first seek turned the
43 second seek's correction into ``W2 - W1``; the staged boundary chunk came out
44 right but every chunk after it was offset by ``-W1``.
45 """
46 ev = _ramp()
47 p = tmp_path / f"ramp.{fmt}.raw"
48 with EventWriter(str(p), format=fmt) as w:
49 w.write(ev)
51 with EventReader(str(p), n_events=1000) as r:
52 r.seek(t=18_000_000) # past 2**24 = 16_777_216 (EVT3 wrap)
53 T = 2_000_000 # backward, pre-wrap
54 exp = int(np.searchsorted(ev["t"], T))
55 r.seek(t=T)
56 got = []
57 while True:
58 c = r.read()
59 if len(c) == 0:
60 break
61 got.append(np.asarray(c.t).copy())
63 tail = np.concatenate(got)
64 assert np.array_equal(tail, ev["t"][exp:])
67# --------------------------------------------------------------------------- #
68# Bug 2: dense kernels must not scramble coordinates on non-canonical field order
69# --------------------------------------------------------------------------- #
70def test_dense_kernel_independent_of_field_order():
71 """A structured array stored as (x, y, t, p) must produce the same frame as
72 the canonical (t, x, y, p) layout -- the kernel takes (t, x, y, p)
73 positionally, so fields must be selected by name, not by dtype order.
74 """
75 from evutils.dense import frame_gray
77 canon = np.array(
78 [(5, 10, 20, 1), (7, 30, 40, 0)],
79 dtype=[("t", "<i8"), ("x", "<u2"), ("y", "<u2"), ("p", "i1")],
80 )
81 scrambled = np.array(
82 [(10, 20, 5, 1), (30, 40, 7, 0)],
83 dtype=[("x", "<u2"), ("y", "<u2"), ("t", "<i8"), ("p", "i1")],
84 )
86 f_canon = frame_gray(canon, width=100, height=100)
87 f_scram = frame_gray(scrambled, width=100, height=100)
89 assert np.array_equal(f_canon, f_scram)
90 assert f_canon[20, 10] == 255 # (y=20, x=10, p=1)
91 assert f_canon[40, 30] == 0 # (y=40, x=30, p=0)
94# --------------------------------------------------------------------------- #
95# Bug 3: EVT3 timestamps must survive past the uint32 (~71.6 min) ceiling
96# --------------------------------------------------------------------------- #
97def _evt3_word(packet_type, data):
98 return np.uint16(((packet_type & 0xF) << 12) | (data & 0x0FFF))
101def _write_evt3_overflow_stream(path, wraps=257):
102 """Craft a minimal EVT3 payload whose time base wraps ``wraps`` times, so the
103 accumulator exceeds 2**32 µs, then emit one scalar and one vector event that
104 share that timestamp. ``wraps=257`` gives ts = 257 * 2**24 + 100 > 2**32.
105 """
106 TH, TL, ADDR_Y, ADDR_X, VECT_BASE_X, VECT_12, OTHERS = 0x8, 0x6, 0x0, 0x2, 0x3, 0x4, 0xE
107 words = []
108 # Each (0xFFF, 0x000) TIME_HIGH pair makes the 24-bit field wrap once.
109 for _ in range(wraps):
110 words.append(_evt3_word(TH, 0xFFF))
111 words.append(_evt3_word(TH, 0x000))
112 words.append(_evt3_word(TL, 0x064)) # ts = wraps*2**24 + 100
113 words.append(_evt3_word(ADDR_Y, 5))
114 words.append(_evt3_word(ADDR_X, 10 | (1 << 11))) # scalar: x=10, p=1
115 words.append(_evt3_word(VECT_BASE_X, 0x100 | (1 << 11)))
116 words.append(_evt3_word(VECT_12, 0x001)) # one vector event at base
117 words += [_evt3_word(OTHERS, 0)] * 8 # look-ahead / tail padding
118 payload = np.array(words, dtype=np.uint16).tobytes()
120 header = (b"% evt 3.0\n% format EVT3;height=720;width=1280\n"
121 b"% geometry 1280x720\n% end\n")
122 with open(path, "wb") as f:
123 f.write(header)
124 f.write(payload)
125 return wraps * (1 << 24) + 100
128def test_evt3_timestamp_past_uint32_ceiling(tmp_path):
129 """Both the scalar (ADDR_X) and the vectorised (EMIT_SOA) event paths must
130 keep the full 64-bit timestamp once the time base passes 2**32 µs.
131 """
132 p = tmp_path / "evt3_overflow.raw"
133 expected = _write_evt3_overflow_stream(str(p))
134 assert expected > 2**32
136 with EventReader(str(p)) as r:
137 ev = r.read_all()
139 assert len(ev) == 2
140 assert int(ev.t[0]) == expected # scalar
141 assert int(ev.t[1]) == expected # vector -- would truncate to low 32 bits before the fix
144# --------------------------------------------------------------------------- #
145# Bug 4: corrupt packets -> robust warn+skip by default, raise under strict
146# --------------------------------------------------------------------------- #
147def _write_evt3_corrupt_stream(path):
148 """EVT3 payload with a stray VECT_8 (a vector continuation with no preceding
149 base) between two otherwise-valid scalar events. The decoder must skip the
150 bad word, keeping both valid events.
151 """
152 TH, TL, ADDR_Y, ADDR_X, VECT_8, OTHERS = 0x8, 0x6, 0x0, 0x2, 0x5, 0xE
153 words = [
154 _evt3_word(TH, 0x000),
155 _evt3_word(TL, 0x064), # ts = 100
156 _evt3_word(ADDR_Y, 5),
157 _evt3_word(ADDR_X, 10 | (1 << 11)), # valid event #1
158 _evt3_word(VECT_8, 0x0AA), # stray VECT_8 -> corrupt
159 _evt3_word(ADDR_X, 20 | (1 << 11)), # valid event #2
160 ]
161 words += [_evt3_word(OTHERS, 0)] * 8
162 payload = np.array(words, dtype=np.uint16).tobytes()
163 header = (b"% evt 3.0\n% format EVT3;height=720;width=1280\n"
164 b"% geometry 1280x720\n% end\n")
165 with open(path, "wb") as f:
166 f.write(header)
167 f.write(payload)
170def test_corrupt_packet_robust_by_default(tmp_path):
171 """Default decoder warns and skips the bad word, still returning both valid
172 events (Metavision's UNRELIABLE behaviour) -- it must not raise."""
173 import warnings
175 p = tmp_path / "evt3_corrupt.raw"
176 _write_evt3_corrupt_stream(str(p))
178 with warnings.catch_warnings(record=True) as caught:
179 warnings.simplefilter("always")
180 with EventReader(str(p)) as r:
181 ev = r.read_all()
183 assert [int(v) for v in ev.x] == [10, 20]
184 assert any("malformed" in str(w.message).lower() for w in caught)
187def test_corrupt_packet_raises_in_strict_mode(tmp_path):
188 """strict=True re-raises the malformed packet instead of skipping it
189 (Metavision's SAFE behaviour)."""
190 from evutils.io._evt import EventDecoder_EVT
192 p = tmp_path / "evt3_corrupt.raw"
193 _write_evt3_corrupt_stream(str(p))
194 # Feed the bytes in-memory: a mid-decode raise on an mmap source pins the
195 # payload view via the traceback, so the mmap close would mask the real
196 # error with BufferError. An in-memory BufferSource has no such close hazard.
197 raw = p.read_bytes()
199 with pytest.raises(Exception, match="(?i)malformed|strict"):
200 with EventReader(raw, file_decoder=EventDecoder_EVT, strict=True) as r:
201 r.read_all()
204# --------------------------------------------------------------------------- #
205# Seek gaps: seek must not break trigger decoding; seek(n=) past EOF is empty
206# --------------------------------------------------------------------------- #
207def _evt3_trigger_words(t, tid, value):
208 return np.array([
209 0x8000 | ((t >> 12) & 0xFFF), # TIME_HIGH
210 0x6000 | (t & 0xFFF), # TIME_LOW
211 0xA000 | ((tid & 0xF) << 8) | (value & 1),
212 ], dtype=np.uint16)
215def test_seek_preserves_triggers_evt3(tmp_path):
216 """A time seek with ext_trigger=True must still decode the external triggers
217 that follow the landing point (seek x triggers path)."""
218 ev = np.zeros(100, dtype=Event_dtype)
219 ev["t"] = np.arange(100, dtype=np.int64) * 100 # 0 .. 9900 µs
220 ev["x"] = np.arange(100) % 1280
221 ev["y"] = np.arange(100) % 720
222 ev["p"] = np.arange(100) % 2
223 triggers = [(10_000, 3, 1), (10_064, 3, 0), (20_000, 7, 1)]
225 p = tmp_path / "seek_trig.raw"
226 with EventWriter(str(p), format="evt3") as w:
227 w.write(ev)
228 words = np.concatenate([_evt3_trigger_words(t, i, v) for t, i, v in triggers])
229 with open(p, "ab") as f: # append crafted trigger words
230 f.write(words.tobytes())
232 T = 5_000
233 exp = int(np.searchsorted(ev["t"], T))
234 ev_got, tr_got = [], []
235 with EventReader(str(p), n_events=40, ext_trigger=True) as r:
236 r.seek(t=T)
237 while True:
238 e, tr = r.read()
239 if len(e) == 0 and len(tr) == 0:
240 break
241 if len(e):
242 ev_got.append(np.asarray(e.t).copy())
243 if len(tr):
244 tr_got.append(np.asarray(tr.t).copy())
246 ev_all = np.concatenate(ev_got)
247 tr_all = np.concatenate(tr_got)
248 assert int(ev_all[0]) == int(ev["t"][exp]) # events start at the target
249 assert np.array_equal(np.sort(tr_all), [10_000, 10_064, 20_000])
252def test_seek_by_event_index_past_eof(tmp_path):
253 """seek(n=) beyond the last event lands at EOF and the next read is empty."""
254 ev = np.zeros(1000, dtype=Event_dtype)
255 ev["t"] = np.arange(1000, dtype=np.int64) * 1_000
256 p = tmp_path / "seek_n_eof.raw"
257 with EventWriter(str(p), format="evt3") as w:
258 w.write(ev)
260 with EventReader(str(p), n_events=100) as r:
261 r.seek(n=10_000_000)
262 assert len(r.read()) == 0
265# --------------------------------------------------------------------------- #
266# Bug 5 (round 3): the dedicated delta_t C parser must honour the post-seek
267# TIME_HIGH wrap correction -- corrected end_ts in, corrected timestamps out.
268# --------------------------------------------------------------------------- #
269def test_parse_step_delta_t_applies_seek_correction(tmp_path):
270 """After a seek whose bookmark lies past the EVT3 wrap (2**24 µs), the
271 parser is reset and decodes in the raw (low) timeline; parse_step applies
272 ``_seek_correction`` on the way out, and parse_step_delta_t must do the
273 same -- translating the caller's absolute ``end_ts`` into the raw timeline
274 for the C call and shifting the decoded slice back. Before the fix the
275 window boundary was never reached and the emitted timestamps were low by
276 one wrap period.
277 """
278 from evutils.io._evt import EventDecoder_EVT
279 from evutils.io._native_core import (
280 EVUTILS_PARSE_WINDOW_DONE, EventSoABuffers, TriggerSoABuffers,
281 events_view,
282 )
283 from evutils.io._source import make_source
285 n, dt = 200_000, 100 # ts 0 .. 20_000_000 > 2**24
286 ev = _ramp(n, dt)
287 p = tmp_path / "wrap_dt.raw"
288 with EventWriter(str(p), format="evt3") as w:
289 w.write(ev)
291 d = EventDecoder_EVT(make_source(str(p)), chunk_size=2048)
292 d.init()
293 T = 19_800_000 # bookmark for this target sits past the wrap
294 res, rem, _ = d.seek(t=T)
295 assert d._seek_correction == 1 << 24 # one lost wrap to restore
296 assert int(res.ts) == T
298 # First event still in the stream (right after the returned remainder).
299 next_ts = int(rem.t[-1]) + dt
300 end_ts = next_ts + 50_000
301 out = EventSoABuffers(65_536)
302 tr = TriggerSoABuffers(1024)
303 while True:
304 out.c.capacity = out.capacity
305 appended, status = d.parse_step_delta_t(out, tr, end_ts)
306 if status == EVUTILS_PARSE_WINDOW_DONE or d.is_eof():
307 break
308 d.close()
310 t = events_view(out).t
311 assert len(t) == 500 # exactly one 50 ms window of the ramp
312 assert int(t[0]) == next_ts # absolute (corrected) timeline
313 assert int(t[-1]) < end_ts # boundary honoured in that timeline
314 assert bool(np.all(np.diff(t) == dt))
317# --------------------------------------------------------------------------- #
318# Bug 6 (round 3): index="metavision" must actually use the sidecar -- the
319# normalize_ts pre-anchor used to force-build an in-memory index over it.
320# --------------------------------------------------------------------------- #
321def _write_mv_sidecar(raw_path):
322 """Synthesize a Metavision ``.tmp_index`` sidecar for an EVT3 file, with
323 bookmarks taken from a real decode pass (exact offsets/counts)."""
324 from pathlib import Path
326 from evutils.io._evt import EventDecoder_EVT
327 from evutils.io._index import IncrementalSeekIndex, metavision_index_path
328 from evutils.io._source import make_source
330 d = EventDecoder_EVT(make_source(str(raw_path)))
331 d.init()
332 idx = IncrementalSeekIndex(
333 words=d._words, start_offset=d._start_offset,
334 input_cls=d._input_cls, parser_cls=type(d._parser),
335 tail_pad=d._tail_pad, word_dtype=d._word_dtype, chunk_cap=2048,
336 time_high=d._TIME_HIGH_TYPE[d._format], stride_words=512,
337 )
338 idx._build_until(target_t=2**62) # build to EOF
339 ts = np.asarray(idx._ts_list, dtype=np.int64)
340 off = np.asarray(idx._off_list, dtype=np.int64)
341 cum = np.asarray(idx._cum_list, dtype=np.int64)
342 counts = np.diff(np.append(cum, idx._cum))
343 word_size = np.dtype(d._word_dtype).itemsize
344 byte_off = d._payload_off + off * word_size
345 d.close()
347 rec_dtype = np.dtype([("ts", "<i8"), ("byte_offset", "<u8"), ("count", "<u4")])
348 recs = np.zeros(len(ts) + 1, dtype=rec_dtype)
349 recs["ts"][:-1] = ts
350 recs["byte_offset"][:-1] = byte_off
351 recs["count"][:-1] = counts
352 recs[-1] = (0x4D414749, 0, 0) # trailing completeness marker (dropped)
354 sidecar = metavision_index_path(str(raw_path))
355 size = Path(raw_path).stat().st_size
356 with open(sidecar, "wb") as f:
357 f.write(f"% size {size}\n% ts_shift_us 0\n% end\n".encode())
358 f.write(recs.tobytes())
361def test_metavision_sidecar_index_is_used(tmp_path):
362 """With index="metavision" a fresh sidecar must be the index actually
363 consulted (StaticSeekIndex, not the in-memory build), and the seek must
364 land exactly."""
365 from evutils.io._index import StaticSeekIndex
367 ev = _ramp()
368 p = tmp_path / "sidecar.raw"
369 with EventWriter(str(p), format="evt3") as w:
370 w.write(ev)
371 _write_mv_sidecar(p)
373 T = 12_345_000
374 exp = int(np.searchsorted(ev["t"], T))
375 with EventReader(str(p), n_events=1000, index="metavision") as r:
376 landed = r.seek(t=T)
377 c = r.read()
378 used = r._file_decoder._index
379 assert isinstance(used, StaticSeekIndex)
380 assert r._file_decoder._index_is_ours is False
381 assert landed.ts == int(ev["t"][exp])
382 assert int(c.t[0]) == int(ev["t"][exp])
385@pytest.mark.parametrize("fmt", ["evt3", "evt2"])
386def test_seek_multi_bookmark_exact(tmp_path, fmt):
387 """Built-index seeks must land exactly on files large enough for MANY
388 bookmarks. Regression for the bookmark-alignment bug: bookmarks recorded at
389 arbitrary parse-step boundaries leave a reset parser without a time base
390 (raw ts = low bits only), so the wrap-correction snap rounded to a whole
391 spurious wrap period. Bookmarks are now TIME_HIGH-aligned.
392 """
393 n, dt = 200_000, 100 # ~10 bookmarks at the default stride
394 ev = _ramp(n, dt)
395 p = tmp_path / f"multi_bm.{fmt}.raw"
396 with EventWriter(str(p), format=fmt) as w:
397 w.write(ev)
399 with EventReader(str(p), n_events=2000) as r:
400 for T in (3_000_000, 12_345_000, 19_800_000):
401 exp = int(np.searchsorted(ev["t"], T))
402 landed = r.seek(t=T)
403 c = r.read()
404 assert landed.ts == int(ev["t"][exp]), f"t={T}"
405 assert int(c.t[0]) == int(ev["t"][exp])
406 # index seek across bookmarks too
407 r.seek(n=150_000)
408 c = r.read()
409 assert int(c.t[0]) == int(ev["t"][150_000])
412def test_seek_then_read_normalize_ts_matches_read_then_seek(tmp_path):
413 """normalize_ts must be call-order independent: seeking before the first
414 read must normalize against the stream's first event, not against the
415 landing point (or 0)."""
416 ev = _ramp()
417 ev["t"] += 100_000 # stream starts at 100 ms
418 p = tmp_path / "norm_order.raw"
419 with EventWriter(str(p), format="evt3") as w:
420 w.write(ev)
422 T = 5_000_000
423 with EventReader(str(p), n_events=3000, normalize_ts=True) as r:
424 r.seek(t=T) # seek FIRST, no prior read
425 a = r.read()
426 with EventReader(str(p), n_events=3000, normalize_ts=True) as r:
427 r.read() # anchor by reading first
428 r.seek(t=T)
429 b = r.read()
431 assert int(a.t[0]) == int(b.t[0]) == T - 100_000
434# --------------------------------------------------------------------------- #
435# delta_t window dedup (_finalize_delta_t_window): the C path (EVT3) and the
436# generic searchsorted path (DAT/EVT2) must still produce identical windows.
437# --------------------------------------------------------------------------- #
439def _delta_t_windows(path, dt, **kw):
440 """Read one recording in delta_t mode, returning per-window (count, first,
441 last) plus the concatenated timestamps."""
442 counts, bounds, parts = [], [], []
443 with EventReader(str(path), delta_t=dt, **kw) as r:
444 while True:
445 c = r.read()
446 if len(c) == 0:
447 break
448 t = np.asarray(c.t)
449 counts.append(len(c))
450 bounds.append((int(t[0]), int(t[-1])))
451 parts.append(t.copy())
452 ts = np.concatenate(parts) if parts else np.empty(0, dtype=np.int64)
453 return counts, bounds, ts
456@pytest.mark.parametrize("fmt", ["evt3", "evt2", "dat"])
457def test_delta_t_windows_lossless_and_bounded(tmp_path, fmt):
458 """Concatenated delta_t windows equal a full read, and every window (bar a
459 count-cut one) spans strictly less than delta_t. Exercises both the C
460 (evt3) and generic (evt2/dat) window producers behind the shared finalize."""
461 n, dt_step = 60_000, 137
462 ev = _ramp(n, dt_step)
463 if fmt == "dat":
464 p = tmp_path / "win.dat"
465 with EventWriter(str(p), width=1280, height=720) as w:
466 w.write(ev)
467 else:
468 p = tmp_path / f"win_{fmt}.raw"
469 with EventWriter(str(p), format=fmt) as w:
470 w.write(ev)
472 with EventReader(str(p), mode="all") as r:
473 ref = np.asarray(r.read_all().t)
475 dt = 200_000
476 counts, bounds, ts = _delta_t_windows(p, dt)
477 assert np.array_equal(ts, ref)
478 assert sum(counts) == len(ref)
479 assert all(hi < lo + dt for lo, hi in bounds)
482@pytest.mark.parametrize("fmt", ["evt3", "dat"])
483def test_delta_t_count_cut_lossless(tmp_path, fmt):
484 """A tight n_events cap forces count-cut resumes; no events are lost and the
485 output still equals a full read (count-cut resume path in the shared
486 finalize)."""
487 n, dt_step = 60_000, 137
488 ev = _ramp(n, dt_step)
489 if fmt == "dat":
490 p = tmp_path / "cut.dat"
491 with EventWriter(str(p), width=1280, height=720) as w:
492 w.write(ev)
493 else:
494 p = tmp_path / f"cut_{fmt}.raw"
495 with EventWriter(str(p), format=fmt) as w:
496 w.write(ev)
498 with EventReader(str(p), mode="all") as r:
499 ref = np.asarray(r.read_all().t)
501 _, _, ts = _delta_t_windows(p, 200_000, n_events=1500)
502 assert np.array_equal(ts, ref)
505_FAN = Path(__file__).resolve().parents[2] / "data" / "fan"
508@pytest.mark.skipif(not (_FAN / "evt3_fan.raw").is_file(),
509 reason="data/fan/evt3_fan.raw fixture not present")
510def test_delta_t_dedup_real_evt3():
511 p = _FAN / "evt3_fan.raw"
512 with EventReader(str(p)) as r:
513 ref = np.asarray(r.read_all().t)
514 counts, bounds, ts = _delta_t_windows(p, 20_000)
515 assert np.array_equal(ts, ref)
516 assert all(hi < lo + 20_000 for lo, hi in bounds)
519@pytest.mark.skipif(not (_FAN / "dat_fan.dat").is_file(),
520 reason="data/fan/dat_fan.dat fixture not present")
521def test_delta_t_dedup_real_dat():
522 p = _FAN / "dat_fan.dat"
523 with EventReader(str(p)) as r:
524 ref = np.asarray(r.read_all().t)
525 counts, bounds, ts = _delta_t_windows(p, 20_000)
526 assert np.array_equal(ts, ref)
527 assert all(hi < lo + 20_000 for lo, hi in bounds)