Coverage for tests/io/test_seek.py: 91%
297 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-18 05:24 +0000
1"""Tests for timestamp / event-index random access (``EventReader.seek``).
3Synthetic, CI-safe: a known ramp (``t = i*dt``) is written in each seekable
4format and seeked by time, event index, relatively and backward. The ramp
5timestamps exceed the EVT3 TIME_HIGH wrap period (2**24 µs), so the post-seek
6wrap correction is exercised. A separate test validates the Metavision
7``.tmp_index`` reader against the real recordings in ``data/`` when present.
8"""
9from pathlib import Path
11import numpy as np
12import pytest
14from evutils.io import EventReader, EventWriter
15from evutils.types import Event_dtype
17# (format, extension). HDF5 needs the optional h5py dependency.
18FORMATS = [
19 ("evt3", "raw"), ("evt4", "raw"), ("evt2", "raw"), ("evt21", "raw"),
20 ("dat", "dat"), ("csv", "csv"), ("npz", "npz"), ("hdf5", "hdf5"),
21]
23N = 20_000
24DT = 1_000 # max t = 20_000_000 > 2**24 (16_777_216): EVT wrap is exercised.
27def _ramp(n: int = N, dt: int = DT):
28 ev = np.zeros(n, dtype=Event_dtype)
29 ev["t"] = np.arange(n, dtype=np.int64) * dt
30 ev["x"] = np.arange(n) % 1280
31 ev["y"] = np.arange(n) % 720
32 ev["p"] = np.arange(n) % 2
33 return ev
36def _write(path, fmt, ev, tr=None):
37 if fmt in ("evt3", "evt4", "evt2", "evt21"):
38 with EventWriter(path, format=fmt) as w:
39 w.write(ev)
40 if tr is not None:
41 w.write_triggers(tr)
42 else: # dispatched by extension; needs geometry for evt/hdf5-style writers
43 with EventWriter(path, width=1280, height=720) as w:
44 w.write(ev)
45 if tr is not None:
46 w.write_triggers(tr)
49@pytest.fixture(params=FORMATS, ids=[f[0] for f in FORMATS])
50def recording(request, tmp_path):
51 fmt, ext = request.param
52 if fmt == "hdf5":
53 pytest.importorskip("h5py")
54 ev = _ramp()
55 p = tmp_path / f"ramp_{fmt}.{ext}"
56 _write(p, fmt, ev)
57 return p, ev
60def test_seek_by_time(recording):
61 p, ev = recording
62 T = 10_000_000
63 exp = int(np.searchsorted(ev["t"], T))
64 with EventReader(p, n_events=3000) as r:
65 landed = r.seek(t=T)
66 c = r.read()
67 assert landed.ts == int(ev["t"][exp])
68 assert int(c.t[0]) == int(ev["t"][exp])
69 assert int(c.t[0]) >= T
72def test_seek_by_event_index(recording):
73 p, ev = recording
74 n = 7000
75 with EventReader(p, n_events=3000) as r:
76 landed = r.seek(n=n)
77 c = r.read()
78 assert landed.ts == int(ev["t"][n])
79 assert int(c.t[0]) == int(ev["t"][n])
80 assert int(c.x[0]) == int(ev["x"][n])
81 assert int(c.y[0]) == int(ev["y"][n])
84def test_seek_backward(recording):
85 p, ev = recording
86 with EventReader(p, n_events=3000) as r:
87 r.seek(n=7000)
88 r.seek(n=200) # backward
89 c = r.read()
90 assert int(c.t[0]) == int(ev["t"][200])
93def test_seek_relative(recording):
94 p, ev = recording
95 with EventReader(p, n_events=3000) as r:
96 r.seek(t=5_000_000)
97 landed = r.seek(t=5_000_000, relative=True) # -> 10_000_000
98 c = r.read()
99 exp = int(np.searchsorted(ev["t"], 10_000_000))
100 assert landed.ts == int(ev["t"][exp])
101 assert int(c.t[0]) == int(ev["t"][exp])
104def test_seek_then_drain_is_contiguous_tail(recording):
105 p, ev = recording
106 T = 8_000_000
107 exp = int(np.searchsorted(ev["t"], T))
108 with EventReader(p, n_events=1500) as r:
109 r.seek(t=T)
110 got = []
111 while True:
112 c = r.read()
113 if len(c) == 0:
114 break
115 got.append(np.asarray(c.t).copy())
116 tail = np.concatenate(got)
117 assert np.array_equal(tail, ev["t"][exp:])
120def test_seek_in_delta_t_mode(recording):
121 p, ev = recording
122 T = 10_000_000
123 exp = int(np.searchsorted(ev["t"], T))
124 with EventReader(p, delta_t=1_000_000) as r:
125 r.seek(t=T)
126 c = r.read()
127 assert len(c) > 0
128 assert int(c.t[0]) == int(ev["t"][exp])
129 assert int(c.t[-1]) < T + 1_000_000
132def test_seek_index_disabled_matches(recording):
133 """index=False (no sidecar) yields the same result as the default."""
134 p, ev = recording
135 T = 12_345_000
136 with EventReader(p, n_events=3000, index=False) as r:
137 a = r.seek(t=T)
138 with EventReader(p, n_events=3000) as r:
139 b = r.seek(t=T)
140 assert a.ts == b.ts == int(ev["t"][int(np.searchsorted(ev["t"], T))])
143def test_seek_out_of_range(recording):
144 p, ev = recording
145 with EventReader(p, n_events=3000) as r:
146 r.seek(t=10**15) # far past the end
147 assert len(r.read()) == 0
150def test_seek_requires_exactly_one_axis(recording):
151 p, _ = recording
152 with EventReader(p, n_events=3000) as r:
153 with pytest.raises(ValueError, match="exactly one"):
154 r.seek()
155 with pytest.raises(ValueError, match="exactly one"):
156 r.seek(t=1, n=1)
159# --------------------------------------------------------------------------- #
160# Non-seekable source: linear iterate-and-skip fallback
161# --------------------------------------------------------------------------- #
163class _NonSeekable:
164 """A read-only, non-seekable byte stream (like a pipe) over a bytes payload."""
166 def __init__(self, data: bytes):
167 self._data = data
168 self._pos = 0
170 def read(self, size: int = -1) -> bytes:
171 if size < 0:
172 size = len(self._data) - self._pos
173 chunk = self._data[self._pos:self._pos + size]
174 self._pos += len(chunk)
175 return chunk
177 def readable(self) -> bool:
178 return True
180 def seekable(self) -> bool:
181 return False
183 def seek(self, *args):
184 import io as _io
185 raise _io.UnsupportedOperation("underlying stream is not seekable")
187 def tell(self) -> int:
188 return self._pos
191def test_seek_linear_fallback_non_seekable(tmp_path):
192 from evutils.io._csv import EventDecoder_Csv
194 ev = _ramp()
195 p = tmp_path / "ramp.csv"
196 _write(p, "csv", ev)
197 data = Path(p).read_bytes()
199 T = 6_000_000 # forward-only (backward on a non-seekable stream can't rewind)
200 exp = int(np.searchsorted(ev["t"], T))
201 with EventReader(_NonSeekable(data), n_events=3000,
202 file_decoder=EventDecoder_Csv) as r:
203 landed = r.seek(t=T)
204 c = r.read()
205 assert landed.ts == int(ev["t"][exp])
206 assert int(c.t[0]) == int(ev["t"][exp])
209# --------------------------------------------------------------------------- #
210# Metavision .tmp_index interop (gated on the local data/ fixtures)
211# --------------------------------------------------------------------------- #
213_DATA = Path(__file__).resolve().parents[2] / "data"
216@pytest.mark.skipif(not (_DATA / "short.raw.tmp_index").is_file(),
217 reason="data/short.raw(.tmp_index) fixtures not present")
218def test_metavision_index_matches_built_index():
219 """A sidecar-driven seek must land identically to a built-index seek."""
220 from evutils.io._evt import EventDecoder_EVT
221 from evutils.io._index import metavision_index_path, read_metavision_index
222 from evutils.io._source import make_source
224 raw = _DATA / "short.raw"
225 d = EventDecoder_EVT(make_source(str(raw)))
226 d.init()
227 ws = np.dtype(d._word_dtype).itemsize
228 idx = read_metavision_index(metavision_index_path(str(raw)), str(raw),
229 d._payload_off, ws)
230 d.close()
231 assert idx is not None and len(idx) > 0
232 assert bool(np.all(np.diff(idx.ts) >= 0)) # monotonic timeline
234 T = int(idx.ts[len(idx) // 2]) + 12_345
236 with EventReader(str(raw), n_events=1000, index="metavision") as r: # sidecar
237 la = r.seek(t=T)
238 ca = r.read()
239 with EventReader(str(raw), n_events=1000, index=False) as r: # exact built
240 lb = r.seek(t=T)
241 cb = r.read()
243 assert la.ts == lb.ts
244 assert int(ca.t[0]) == int(cb.t[0]) >= T
245 assert int(ca.x[0]) == int(cb.x[0])
246 assert int(ca.y[0]) == int(cb.y[0])
250def test_seek_with_normalize_ts(tmp_path):
251 fmt, ext = "evt3", "raw"
252 ev = _ramp()
253 ev["t"] += 100_000 # start at 100k
254 p = tmp_path / f"ramp_norm_{fmt}.{ext}"
255 _write(p, fmt, ev)
257 T = 5_000_000
258 with EventReader(p, n_events=3000, normalize_ts=True) as r:
259 r.read() # Anchor first_ts so normalize_ts knows the start of the file
260 landed = r.seek(t=T)
261 c = r.read()
263 # EventReader.seek() expects/returns absolute timestamps
264 assert landed.ts == T
265 # EventReader.read() yields normalized timestamps (T - 100_000)
266 assert int(c.t[0]) >= T - 100_000
268def test_repeated_seek_past_wrap(tmp_path):
269 fmt, ext = "evt3", "raw"
270 ev = _ramp()
271 p = tmp_path / f"ramp_wrap_{fmt}.{ext}"
272 _write(p, fmt, ev)
274 T1 = 17_000_000 # Past EVT3 wrap (16.7M)
275 T2 = 18_000_000
276 with EventReader(p, n_events=3000) as r:
277 r.seek(t=T1)
278 r.seek(t=T2)
279 r.seek(t=T1)
280 c = r.read()
281 assert int(c.t[0]) >= T1
283def test_seek_past_eof(tmp_path):
284 fmt, ext = "evt3", "raw"
285 ev = _ramp()
286 p = tmp_path / f"ramp_eof_{fmt}.{ext}"
287 _write(p, fmt, ev)
289 with EventReader(p, n_events=3000) as r:
290 r.seek(t=100_000_000) # Far past end
291 c = r.read()
292 assert len(c) == 0
295# --------------------------------------------------------------------------- #
296# SeekResult return unification + event_index / last_seek
297# --------------------------------------------------------------------------- #
299def test_seek_returns_seekresult(recording):
300 """seek() returns a SeekResult exposing .ts / .index / .eof (and stores it
301 on .last_seek); .event_index tracks the landing index."""
302 from evutils.io.common import SeekResult
304 p, ev = recording
305 n = 7000
306 with EventReader(p, n_events=3000) as r:
307 res = r.seek(n=n)
308 assert isinstance(res, SeekResult)
309 assert res.ts == int(ev["t"][n])
310 assert res.index == n
311 assert res.eof is False
312 assert res.ts == res[0] # ts is the first positional field
313 assert r.last_seek == res # stored on the reader
314 assert r.event_index == n # event-coordinate cursor
316 far = r.seek(t=10**15) # past the end
317 assert far.eof is True
318 assert r.last_seek == far
321def test_seek_result_index_matches_read(recording):
322 """The reported .index is the true 0-based index of the first event read.
324 CSV time-axis seeks use a byte-offset binary search that does not know the
325 event index, so they report ``index == 0`` (only the landing is asserted).
326 """
327 p, ev = recording
328 T = 9_000_000
329 exp = int(np.searchsorted(ev["t"], T))
330 with EventReader(p, n_events=1000) as r:
331 res = r.seek(t=T)
332 c = r.read()
333 if p.suffix != ".csv":
334 assert res.index == exp
335 assert int(c.t[0]) == int(ev["t"][exp])
338# --------------------------------------------------------------------------- #
339# evutils own `.evidx` persistable sidecar (index="persist")
340# --------------------------------------------------------------------------- #
342@pytest.mark.parametrize("fmt", ["evt3", "evt2"])
343def test_evidx_persist_roundtrip(tmp_path, fmt):
344 """index='persist' builds the exact index on the first seek and writes a
345 `.evidx` sidecar; a re-open loads it (StaticSeekIndex) and lands identically."""
346 from evutils.io._index import evutils_index_path, StaticSeekIndex
348 n, dt = 200_000, 100
349 ev = _ramp(n, dt)
350 p = tmp_path / f"persist_{fmt}.raw"
351 with EventWriter(str(p), format=fmt) as w:
352 w.write(ev)
353 side = evutils_index_path(str(p))
354 assert not side.exists()
356 T = 12_345_000
357 exp = int(np.searchsorted(ev["t"], T))
359 # First open: sidecar is created on the first seek, built exactly.
360 with EventReader(str(p), n_events=3000, index="persist") as r:
361 a = r.seek(t=T)
362 ca = r.read()
363 assert r._file_decoder._index_is_ours is True
364 assert side.exists()
365 assert a.ts == int(ev["t"][exp])
366 assert int(ca.t[0]) == int(ev["t"][exp])
368 # Second open: the sidecar is loaded instead of rebuilt.
369 with EventReader(str(p), n_events=3000, index="persist") as r:
370 b = r.seek(t=T)
371 cb = r.read()
372 assert isinstance(r._file_decoder._index, StaticSeekIndex)
373 assert b.ts == a.ts
374 assert int(cb.t[0]) == int(ca.t[0])
376 # Event-index seek works from the loaded (count-carrying) sidecar too.
377 with EventReader(str(p), n_events=2000, index="persist") as r:
378 r.seek(n=150_000)
379 cc = r.read()
380 assert int(cc.t[0]) == int(ev["t"][150_000])
383def test_evidx_persist_matches_built_index(tmp_path):
384 """A persisted-index seek lands identically to the default built index."""
385 ev = _ramp(200_000, 100)
386 p = tmp_path / "persist_match.raw"
387 with EventWriter(str(p), format="evt3") as w:
388 w.write(ev)
390 T = 7_654_000
391 with EventReader(str(p), n_events=1000, index="persist") as r:
392 a = r.seek(t=T) # builds + saves
393 with EventReader(str(p), n_events=1000, index="persist") as r:
394 b = r.seek(t=T) # loads sidecar
395 with EventReader(str(p), n_events=1000, index=False) as r:
396 c = r.seek(t=T) # exact built, no sidecar
397 assert a.ts == b.ts == c.ts
400def test_evidx_stale_sidecar_ignored(tmp_path):
401 """A `.evidx` whose stored raw size no longer matches is ignored (rebuilt)."""
402 import struct
403 from evutils.io._index import evutils_index_path, load_seek_index
405 ev = _ramp(50_000, 200)
406 p = tmp_path / "stale.raw"
407 with EventWriter(str(p), format="evt3") as w:
408 w.write(ev)
409 side = evutils_index_path(str(p))
411 with EventReader(str(p), n_events=1000, index="persist") as r:
412 r.seek(t=3_000_000)
413 assert side.exists()
414 assert load_seek_index(side, str(p)) is not None # fresh -> loads
416 # Corrupt the stored raw size (header offset 12) -> stale.
417 b = bytearray(side.read_bytes())
418 struct.pack_into("<q", b, 12, 424242)
419 side.write_bytes(bytes(b))
420 assert load_seek_index(side, str(p)) is None # stale -> ignored
422 # A persist re-open silently rebuilds and still lands correctly.
423 T = 3_000_000
424 exp = int(np.searchsorted(ev["t"], T))
425 with EventReader(str(p), n_events=1000, index="persist") as r:
426 res = r.seek(t=T)
427 c = r.read()
428 assert res.ts == int(ev["t"][exp])
429 assert int(c.t[0]) == int(ev["t"][exp])