Coverage for tests/io/test_evt.py: 74%
243 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-15 09:33 +0000
1import pytest
2import numpy as np
7####################################
8####################################
9# Test RAW module
10####################################
11####################################
14from typing import Any
16def test_RAW_writer_import() -> None:
17 '''
18 Testing if the RAW writer can be imported
19 '''
20 from evutils.io import EventWriter
21 assert EventWriter is not None
22 test_writer = EventWriter("test.raw")
23 assert test_writer is not None
24 test_writer.close()
27def test_RAW_reader_import(dummy_file_factory: Any) -> None:
28 '''
29 Testing if the RAW reader can be imported
30 '''
31 test_file = dummy_file_factory("test.raw")
33 from evutils.io import EventReader
34 assert EventReader is not None
35 test_reader = EventReader(test_file)
36 assert test_reader is not None
37 test_reader.close()
39### Test RAW writer
41@pytest.mark.parametrize("fmt", [None, 'evt2', 'evt21', 'evt3', 'evt4'])
42def test_RAW_writer(tmp_path: Any, test_events: Any, fmt: Any) -> None:
43 """Test that writing and reading RAW events yields identical results for supported formats."""
44 from evutils.io import EventWriter, EventReader
46 # Create in a subdirectory to test parent directory creation
47 d = tmp_path / "sub"
48 d.mkdir(exist_ok=True)
49 p = d / f"test_{fmt or 'default'}.raw"
51 kwargs = {'format': fmt} if fmt is not None else {}
52 with EventWriter(p, **kwargs) as writer:
53 writer.write(test_events)
55 fmt_name = fmt or 'default'
56 assert p.is_file(), f"format {fmt_name}: file was not created"
57 assert p.stat().st_size > 0, f"format {fmt_name}: created file is empty"
59 with EventReader(p) as reader:
60 events = reader.read()
62 assert np.array_equal(events, test_events), (
63 f"format {fmt_name}: read events do not match written events"
64 )
68def test_EVT2_matches_EVT3(real_event_files: Any) -> None:
69 '''EVT2 and EVT3 are the same recording transcoded, so decoding them must
70 yield identical events (timestamps equal up to a constant offset).'''
71 from evutils.io import EventReader
73 def get_hand_path(fmt: str) -> Any:
74 for ef in real_event_files[fmt]:
75 if 'hand' in ef.path.name:
76 return ef.path
77 return None
79 path2 = get_hand_path('evt2')
80 path3 = get_hand_path('evt3')
82 if path2 is None or path3 is None:
83 pytest.skip("Transcoded hand files not found")
85 def head(path: Any, n: int=200_000) -> tuple[Any, Any, Any, Any]:
86 parts = []
87 with EventReader(path, n_events=100_000) as reader:
88 for e in reader:
89 parts.append(e)
90 if sum(len(p) for p in parts) >= n:
91 break
92 t = np.concatenate([p["t"] for p in parts])[:n]
93 x = np.concatenate([p["x"] for p in parts])[:n]
94 y = np.concatenate([p["y"] for p in parts])[:n]
95 p_ = np.concatenate([p["p"] for p in parts])[:n]
96 return t, x, y, p_
98 t2, x2, y2, p2 = head(path2)
99 t3, x3, y3, p3 = head(path3)
101 ctx = f"{path2.name} (evt2) vs {path3.name} (evt3)"
102 assert np.array_equal(x2, x3), f"{ctx}: decoded x differs"
103 assert np.array_equal(y2, y3), f"{ctx}: decoded y differs"
104 assert np.array_equal(p2, p3), f"{ctx}: decoded polarity differs"
105 # Timestamps differ only by a constant (per-file time origin).
106 assert np.ptp(t3.astype(np.int64) - t2.astype(np.int64)) == 0, (
107 f"{ctx}: timestamp offset is not constant across events"
108 )
111@pytest.mark.parametrize("fmt", ['evt2', 'evt3'])
112def test_EVT_matches_expelliarmus(real_event_files: Any, fmt: Any) -> None:
113 """Our EventReader output must match expelliarmus byte-for-byte."""
114 pytest.importorskip("expelliarmus")
115 from expelliarmus import Wizard # type: ignore
116 from evutils.io import EventReader
118 if fmt not in real_event_files or not real_event_files[fmt]:
119 pytest.skip(f"No files for format {fmt}")
121 ef = next((f for f in real_event_files[fmt] if 'hand' in f.path.name), real_event_files[fmt][0])
123 ctx = f"{ef.path.name} ({fmt})"
125 # Read first 100k events with evutils
126 with EventReader(ef.path, n_events=100_000) as reader:
127 evutils_events = reader.read()
128 assert not isinstance(evutils_events, tuple), f"{ctx}: reader returned a (events, triggers) tuple"
130 # Read the same chunk with expelliarmus
131 wiz = Wizard(encoding=fmt)
132 wiz.set_file(str(ef.path))
133 exp_events = []
134 total = 0
135 for chunk in wiz.read_chunk():
136 exp_events.append(chunk)
137 total += len(chunk)
138 if total >= 100_000:
139 break
141 if not exp_events:
142 pytest.skip("Expelliarmus returned no events")
144 exp_events_arr = np.concatenate(exp_events)[:100_000]
146 assert len(evutils_events) == len(exp_events_arr), (
147 f"{ctx}: event count {len(evutils_events)} != expelliarmus {len(exp_events_arr)}"
148 )
149 assert np.array_equal(evutils_events["x"], exp_events_arr["x"]), f"{ctx}: x differs from expelliarmus"
150 assert np.array_equal(evutils_events["y"], exp_events_arr["y"]), f"{ctx}: y differs from expelliarmus"
151 assert np.array_equal(evutils_events["p"], exp_events_arr["p"]), f"{ctx}: polarity differs from expelliarmus"
153 if fmt == 'evt3':
154 # Expelliarmus has a known bug in EVT3 timestamp decoding where it occasionally
155 # drifts by exactly 4096us (1 TIME_HIGH tick). Our EVT3 decoder perfectly matches
156 # our EVT2 decoder, so we skip the timestamp assertion for expelliarmus EVT3.
157 pass
158 else:
159 assert np.array_equal(evutils_events["t"], exp_events_arr["t"]), (
160 f"{ctx}: timestamps differ from expelliarmus"
161 )
164@pytest.mark.parametrize("fmt", ['evt2', 'evt3'])
165def test_EVT_matches_evlib(real_event_files: Any, fmt: Any) -> None:
166 """Our EventReader output must match evlib."""
167 pytest.importorskip("evlib")
168 import evlib # type: ignore
169 from evutils.io import EventReader
171 if fmt not in real_event_files or not real_event_files[fmt]:
172 pytest.skip(f"No files for format {fmt}")
174 ef = next((f for f in real_event_files[fmt] if 'hand' in f.path.name), real_event_files[fmt][0])
176 ctx = f"{ef.path.name} ({fmt})"
178 # Read first 100k events with evutils
179 with EventReader(ef.path, n_events=100_000) as reader:
180 evutils_events = reader.read()
181 assert not isinstance(evutils_events, tuple), f"{ctx}: reader returned a (events, triggers) tuple"
183 # Read the same file with evlib (using polars)
184 df = evlib.load_events(str(ef.path))
185 # We only take the first 100_000 to match evutils and avoid massive memory consumption
186 df = df.head(100_000)
188 if hasattr(df, "collect"):
189 df = df.collect()
191 if len(df) == 0:
192 pytest.skip("evlib returned no events")
194 # evlib column mappings: 'x', 'y', 't', 'polarity'
195 # Polarity in evlib is often -1/1 or 0/1 depending on the rust backend.
196 # We map -1 back to 0 if necessary.
197 evlib_x = df["x"].to_numpy()
198 evlib_y = df["y"].to_numpy()
199 evlib_p = df["polarity"].to_numpy()
200 evlib_p = np.where(evlib_p == -1, 0, evlib_p)
202 # Time is a Polars duration. Casting to Int64 converts it to microseconds internally
203 import polars as pl # type: ignore
204 evlib_t = df.select(pl.col("t").dt.total_microseconds()).to_numpy()[:, 0]
206 # evlib sometimes reorders events that share the exact same timestamp, so we
207 # sort both arrays lexicographically by (t, x, y, p) to compare the actual contents.
208 def sort_events(t: Any, x: Any, y: Any, p: Any) -> tuple[Any, Any, Any, Any]:
209 idx = np.lexsort((p, y, x, t))
210 return t[idx], x[idx], y[idx], p[idx]
212 evutils_t, evutils_x, evutils_y, evutils_p = sort_events(
213 evutils_events["t"], evutils_events["x"], evutils_events["y"], evutils_events["p"]
214 )
215 evlib_t, evlib_x, evlib_y, evlib_p = sort_events(evlib_t, evlib_x, evlib_y, evlib_p)
217 assert len(evutils_events) == len(df), (
218 f"{ctx}: event count {len(evutils_events)} != evlib {len(df)}"
219 )
220 assert np.array_equal(evutils_x, evlib_x), f"{ctx}: x differs from evlib"
221 assert np.array_equal(evutils_y, evlib_y), f"{ctx}: y differs from evlib"
222 assert np.array_equal(evutils_p, evlib_p), f"{ctx}: polarity differs from evlib"
223 assert np.array_equal(evutils_t, evlib_t), f"{ctx}: timestamps differ from evlib"
226@pytest.mark.parametrize("fmt", ['evt3', 'evt2', 'evt21'])
227@pytest.mark.parametrize("length", [10, 100, 1000])
228def test_RAW_nevents_read(real_event_files: Any, fmt: Any, length: int) -> None:
229 """Test reading a specific number of events."""
230 from evutils.io import EventReader
232 if fmt not in real_event_files or not real_event_files[fmt]:
233 pytest.skip(f"No files for format {fmt}")
235 for ef in real_event_files[fmt]:
236 with EventReader(ef.path, n_events=length) as reader:
237 events = reader.read()
238 assert len(events) == length, (
239 f"{ef.path.name} ({fmt}): requested {length} events, got {len(events)}"
240 )
243@pytest.mark.parametrize("fmt", ['evt3', 'evt2', 'evt21'])
244@pytest.mark.parametrize("delta_t", [100, 1000, 10000])
245def test_RAW_delta_t_read(real_event_files: Any, fmt: Any, delta_t: int) -> None:
246 """Test reading a specific time slice (delta_t) of events."""
247 from evutils.io import EventReader
249 if fmt not in real_event_files or not real_event_files[fmt]:
250 pytest.skip(f"No files for format {fmt}")
252 for ef in real_event_files[fmt]:
253 with EventReader(ef.path, delta_t=delta_t) as reader:
254 events = reader.read()
255 ctx = f"{ef.path.name} ({fmt}, delta_t={delta_t})"
256 assert not isinstance(events, tuple), f"{ctx}: reader returned a (events, triggers) tuple"
257 assert len(events) > 0, f"{ctx}: no events in the time window"
258 # All events fall within the requested time window.
259 span = int(events["t"].max()) - int(events["t"].min())
260 assert span <= delta_t, f"{ctx}: event time span {span} exceeds delta_t {delta_t}"
262####################################
263# Timestamp wrap / edge cases
264####################################
266def _roundtrip(tmp_path: Any, ev: Any, fmt: str) -> Any:
267 from evutils.io import EventReader, EventWriter
268 p = tmp_path / f"wrap_{fmt}.raw"
269 with EventWriter(p, format=fmt) as w:
270 w.write(ev)
271 with EventReader(p) as r:
272 return r.read_all()
275def _wrap_events(t_values: Any) -> Any:
276 ev = np.zeros(len(t_values), dtype=np.dtype([('t', np.int64), ('x', np.uint16), ('y', np.uint16), ('p', np.uint8)]))
277 ev['t'] = np.asarray(t_values, dtype=np.int64)
278 ev['x'] = np.arange(len(t_values)) % 1280
279 ev['y'] = np.arange(len(t_values)) % 720
280 ev['p'] = np.arange(len(t_values)) % 2
281 return ev
284def test_EVT3_time_high_wrap(tmp_path: Any) -> None:
285 """EVT3's rolling time base is 24-bit (~16.7 s). Recordings longer than
286 that must round-trip exactly: the decoder counts TIME_HIGH wraps."""
287 step = 1_000_000 # 1 s steps << 2^24 us, crossing several wraps
288 t = np.arange(0, 90_000_000, step) # 90 s: 5+ wraps of 2^24 us
289 ev = _wrap_events(t)
290 out = _roundtrip(tmp_path, ev, "evt3")
291 assert np.array_equal(out["t"], ev['t']), "evt3 TIME_HIGH wrap: timestamps not round-tripped"
292 assert np.array_equal(out["x"], ev['x']), "evt3 TIME_HIGH wrap: x not round-tripped"
295@pytest.mark.parametrize("fmt", ["evt2", "evt21", "evt4"])
296def test_EVT2_time_high_wrap(tmp_path: Any, fmt: Any) -> None:
297 """EVT2/EVT2.1/EVT4 time bases are 34-bit (~4.8 h); crossing that boundary
298 must be tracked by the wrap accumulator."""
299 base = (1 << 34) - 50
300 t = base + np.arange(0, 100, 10) # crosses 2^34
301 ev = _wrap_events(t)
302 out = _roundtrip(tmp_path, ev, fmt)
303 assert np.array_equal(out["t"], ev['t']), f"{fmt} 34-bit time wrap: timestamps not round-tripped"
306@pytest.mark.parametrize("fmt", ["evt3", "evt2", "evt21", "evt4"])
307def test_EVT_long_recording_hours(tmp_path: Any, fmt: Any) -> None:
308 """Sparse multi-hour timestamps survive as long as gaps stay below the
309 format's wrap period (2^24 us for EVT3, 2^34 us for EVT2/2.1)."""
310 if fmt == "evt3":
311 t = np.arange(0, 3600_000_000, 10_000_000) # 1 h in 10 s steps
312 else:
313 t = np.arange(0, 10 * 3600_000_000, 3600_000_000) # 10 h in 1 h steps
314 ev = _wrap_events(t)
315 out = _roundtrip(tmp_path, ev, fmt)
316 assert np.array_equal(out["t"], ev['t']), f"{fmt} multi-hour recording: timestamps not round-tripped"
319@pytest.mark.parametrize("fmt", ["evt3", "evt2", "evt21", "evt4"])
320def test_EVT_truncated_payload(tmp_path: Any, fmt: Any) -> None:
321 """A file cut mid-word must not crash: all complete events decode."""
322 from evutils.io import EventReader, EventWriter
323 ev = _wrap_events(np.arange(1000) * 17)
324 p = tmp_path / f"trunc_{fmt}.raw"
325 with EventWriter(p, format=fmt) as w:
326 w.write(ev)
327 data = p.read_bytes()
328 p.write_bytes(data[:-3]) # not a multiple of any word size
330 with EventReader(p) as r:
331 out = r.read_all()
332 assert not isinstance(out, tuple), f"{fmt} truncated payload: reader returned a tuple"
333 assert 0 < len(out) <= 1000, f"{fmt} truncated payload: decoded {len(out)} events, expected 1..1000"
334 assert np.array_equal(out["t"], ev['t'][:len(out)]), (
335 f"{fmt} truncated payload: decoded timestamps do not match the complete-event prefix"
336 )
339def test_EVT_header_only_file(tmp_path: Any) -> None:
340 """A header with no payload reads as zero events (not an error)."""
341 from evutils.io import EventReader, EventWriter
342 p = tmp_path / "header_only.raw"
343 w = EventWriter(p, format="evt3")
344 w.init()
345 w.close()
346 with EventReader(p) as r:
347 n = len(r.read_all())
348 assert n == 0, f"header-only evt3 file decoded {n} events, expected 0"
351def test_EVT_from_bytes_and_stream(tmp_path: Any, test_events: Any) -> None:
352 """The reader accepts raw bytes and binary streams, not just paths."""
353 import io as _io
354 from evutils.io import EventReader, EventWriter
355 p = tmp_path / "stream.raw"
356 with EventWriter(p) as w:
357 w.write(test_events)
358 data = p.read_bytes()
360 for source in (data, _io.BytesIO(data)):
361 kind = type(source).__name__
362 with EventReader(source) as r: # type: ignore
363 assert np.array_equal(np.asarray(r.read_all()), test_events), (
364 f"reading from {kind} did not reproduce the written events"
365 )
368def test_EVT3_coordinate_extremes(tmp_path: Any) -> None:
369 """Max 11-bit coordinates and both polarities survive the trip."""
370 ev = _wrap_events(np.arange(6))
371 ev['x'] = [0, 2047, 0, 2047, 1, 2046]
372 ev['y'] = [0, 2047, 2047, 0, 2, 2045]
373 ev['p'] = [0, 1, 0, 1, 1, 0]
374 out = _roundtrip(tmp_path, ev, "evt3")
375 assert np.array_equal(out["x"], ev['x']), "evt3 coordinate extremes: x not round-tripped"
376 assert np.array_equal(out["y"], ev['y']), "evt3 coordinate extremes: y not round-tripped"
377 assert np.array_equal(out["p"], ev['p']), "evt3 coordinate extremes: polarity not round-tripped"
379def test_EVT_empty_file(tmp_path: Any) -> None:
380 from evutils.io import EventReader
381 p = tmp_path / "empty.raw"
382 p.touch()
383 with EventReader(p) as r:
384 n = len(r.read())
385 assert n == 0, f"empty file decoded {n} events, expected 0"
387def test_EVT_corrupted_header_no_end(tmp_path: Any) -> None:
388 from evutils.io import EventReader
389 p = tmp_path / "no_end_header.raw"
390 p.write_bytes(b"% camera_integrator_name Prophesee\n% date 2026-07-08 12:00:00\n\x00\x00")
391 with EventReader(p) as r:
392 n = len(r.read())
393 assert n == 0, f"header with no '% end' and no payload decoded {n} events, expected 0"