Coverage for tests/io/test_triggers.py: 99%
151 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-15 09:33 +0000
1"""External-trigger decoding tests for the EVT2 / EVT2.1 / EVT3 / EVT4 parsers.
3The encoders do not write trigger packets, so these tests hand-craft
4EXT_TRIGGER words from the Prophesee format specs and splice them into
5encoder-generated files:
7* EVT2 (uint32): type=0xA in bits 31..28, ts low 6 bits in 27..22,
8 id in bits 11..8, value in bit 0.
9* EVT2.1 (uint64): type=0xA in bits 31..28, ts low 6 bits in 27..22,
10 id in bits 44..40, value in bit 32.
11* EVT3 (uint16): type=0xA in bits 15..12, id in bits 11..8, value in bit 0;
12 timestamp comes from the running TIME_HIGH/TIME_LOW state.
13* EVT4 (uint32): type=0x9 in bits 31..28, ts low 6 bits in 27..22,
14 id in bits 12..8 (5-bit), value in bit 0; TIME_HIGH is type=0xE.
15"""
16import warnings
18import numpy as np
20from typing import Any
23####################################
24# word crafting helpers
25####################################
27def _evt2_time_high(t: int) -> int:
28 return (0x8 << 28) | ((t >> 6) & 0x0FFFFFFF)
31def _evt2_trigger(t: int, tid: int, value: int) -> int:
32 return (0xA << 28) | ((t & 0x3F) << 22) | ((tid & 0xF) << 8) | (value & 1)
35def _evt2_cd(t: int, x: int, y: int, p: int) -> int:
36 return ((p & 1) << 28) | ((t & 0x3F) << 22) | ((x & 0x7FF) << 11) | (y & 0x7FF)
39def _evt21_trigger_words(t: int, tid: int, value: int) -> Any:
40 th = (0x8 << 28) | ((t >> 6) & 0x0FFFFFFF)
41 tr = (0xA << 28) | ((t & 0x3F) << 22) | ((tid & 0x1F) << 40) | ((value & 1) << 32)
42 return np.array([th, tr], dtype=np.uint64)
45def _evt4_time_high(t: int) -> int:
46 return (0xE << 28) | ((t >> 6) & 0x0FFFFFFF)
49def _evt4_trigger(t: int, tid: int, value: int) -> int:
50 # EVT4 EXT_TRIGGER=0x9: value bit 0, id bits 12..8 (5-bit), ts low bits 27..22.
51 return (0x9 << 28) | ((t & 0x3F) << 22) | ((tid & 0x1F) << 8) | (value & 1)
54def _evt3_trigger_words(t: int, tid: int, value: int) -> Any:
55 return np.array([
56 0x8000 | ((t >> 12) & 0xFFF), # TIME_HIGH
57 0x6000 | (t & 0xFFF), # TIME_LOW
58 0xA000 | ((tid & 0xF) << 8) | (value & 1),
59 ], dtype=np.uint16)
62def _events(n: int = 100) -> Any:
63 ev = np.zeros(n, dtype=np.dtype([('t', np.int64), ('x', np.uint16), ('y', np.uint16), ('p', np.uint8)]))
64 ev['t'] = np.arange(n, dtype=np.int64) * 100 # 0 .. 9900 us
65 ev['x'] = np.arange(n) % 1280
66 ev['y'] = np.arange(n) % 720
67 ev['p'] = np.arange(n) % 2
68 return ev
71# Trigger fixtures appended after the events, so timestamps must not run
72# backwards: all >= the last event timestamp.
73TRIGGERS = [ # (t, id, value)
74 (10_000, 3, 1),
75 (10_064, 3, 0),
76 (20_000, 7, 1),
77]
80def _write_events_with_triggers(tmp_path: Any, fmt: str, trigger_words: Any) -> Any:
81 """Encoder-generated file (header + events) with raw words appended."""
82 from evutils.io import EventWriter
83 p = tmp_path / f"trig_{fmt}.raw"
84 with EventWriter(p, format=fmt) as w:
85 w.write(_events())
86 with open(p, "ab") as f:
87 f.write(trigger_words.tobytes())
88 return p
91####################################
92# read paths
93####################################
95def test_evt3_trigger_decode(tmp_path: Any) -> None:
96 from evutils.io import EventReader
97 words = np.concatenate([_evt3_trigger_words(t, i, v) for t, i, v in TRIGGERS])
98 p = _write_events_with_triggers(tmp_path, "evt3", words)
100 with EventReader(p, ext_trigger=True) as r:
101 out = r.read_all()
102 assert isinstance(out, tuple)
103 ev, tr = out
104 assert np.array_equal(ev.t, _events()['t'])
105 assert np.array_equal(tr.t, [t for t, _, _ in TRIGGERS])
106 assert np.array_equal(tr.id, [i for _, i, _ in TRIGGERS])
107 assert np.array_equal(tr.p, [v for _, _, v in TRIGGERS])
110def test_evt21_trigger_decode(tmp_path: Any) -> None:
111 from evutils.io import EventReader
112 words = np.concatenate([_evt21_trigger_words(t, i, v) for t, i, v in TRIGGERS])
113 p = _write_events_with_triggers(tmp_path, "evt21", words)
115 with EventReader(p, ext_trigger=True) as r:
116 ev, tr = r.read_all()
117 assert np.array_equal(ev.t, _events()['t'])
118 assert np.array_equal(tr.t, [t for t, _, _ in TRIGGERS])
119 assert np.array_equal(tr.id, [i for _, i, _ in TRIGGERS])
120 assert np.array_equal(tr.p, [v for _, _, v in TRIGGERS])
123def test_evt2_trigger_timestamps(tmp_path: Any) -> None:
124 from evutils.io import EventReader
125 words = np.array(
126 [w for t, i, v in TRIGGERS for w in (_evt2_time_high(t), _evt2_trigger(t, i, v))],
127 dtype=np.uint32,
128 )
129 p = _write_events_with_triggers(tmp_path, "evt2", words)
131 with EventReader(p, ext_trigger=True) as r:
132 ev, tr = r.read_all()
133 assert np.array_equal(ev.t, _events()['t'])
134 assert np.array_equal(tr.t, [t for t, _, _ in TRIGGERS])
137def test_evt2_trigger_id_and_value(tmp_path: Any) -> None:
138 from evutils.io import EventReader
139 words = np.array(
140 [w for t, i, v in TRIGGERS for w in (_evt2_time_high(t), _evt2_trigger(t, i, v))],
141 dtype=np.uint32,
142 )
143 p = _write_events_with_triggers(tmp_path, "evt2", words)
145 with EventReader(p, ext_trigger=True) as r:
146 _, tr = r.read_all()
147 assert np.array_equal(tr.id, [i for _, i, _ in TRIGGERS])
148 assert np.array_equal(tr.p, [v for _, _, v in TRIGGERS])
151def test_evt4_trigger_decode(tmp_path: Any) -> None:
152 from evutils.io import EventReader
153 words = np.array(
154 [w for t, i, v in TRIGGERS for w in (_evt4_time_high(t), _evt4_trigger(t, i, v))],
155 dtype=np.uint32,
156 )
157 p = _write_events_with_triggers(tmp_path, "evt4", words)
159 with EventReader(p, ext_trigger=True) as r:
160 ev, tr = r.read_all()
161 assert np.array_equal(ev.t, _events()['t'])
162 assert np.array_equal(tr.t, [t for t, _, _ in TRIGGERS])
163 assert np.array_equal(tr.id, [i for _, i, _ in TRIGGERS])
164 assert np.array_equal(tr.p, [v for _, _, v in TRIGGERS])
167def test_evt4_trigger_5bit_id(tmp_path: Any) -> None:
168 """EVT4 trigger id is 5 bits (0..31), wider than EVT2's 4-bit field."""
169 from evutils.io import EventReader
170 trigs = [(10_000, 17, 1), (10_064, 31, 0)] # ids > 15 need the 5th bit
171 words = np.array(
172 [w for t, i, v in trigs for w in (_evt4_time_high(t), _evt4_trigger(t, i, v))],
173 dtype=np.uint32,
174 )
175 p = _write_events_with_triggers(tmp_path, "evt4", words)
176 with EventReader(p, ext_trigger=True) as r:
177 _, tr = r.read_all()
178 assert np.array_equal(tr.id, [i for _, i, _ in trigs])
179 assert np.array_equal(tr.p, [v for _, _, v in trigs])
182####################################
183# windowed reads: triggers land in the window covering their timestamp
184####################################
186def test_evt2_trigger_windowing(tmp_path: Any) -> None:
187 """Hand-craft a full interleaved stream (events + triggers) and read it in
188 delta_t windows: each window's triggers must fall inside its time span."""
189 from evutils.io import EventReader, EventWriter
191 ev = _events(100) # t = 0 .. 9900
192 trig = [(1_500, 1, 1), (4_500, 2, 0), (8_500, 3, 1)]
194 # header only
195 p = tmp_path / "windowed_evt2.raw"
196 w = EventWriter(p, format="evt2")
197 w.init()
198 w.close()
200 # interleave CD and trigger words in timestamp order
201 items = sorted(
202 [("cd", int(e['t']), e) for e in ev] + [("tr", t, (t, i, v)) for t, i, v in trig],
203 key=lambda kv: kv[1],
204 )
205 words, last_high = [], -1
206 for kind, t, payload in items:
207 if (t >> 6) != last_high:
208 last_high = t >> 6
209 words.append(_evt2_time_high(t))
210 if kind == "cd":
211 e = payload
212 words.append(_evt2_cd(t, int(e['x']), int(e['y']), int(e['p'])))
213 else:
214 words.append(_evt2_trigger(*payload))
215 with open(p, "ab") as f:
216 f.write(np.array(words, dtype=np.uint32).tobytes())
218 seen_ev, seen_tr = 0, []
219 with EventReader(p, ext_trigger=True, mode="delta_t", delta_t=2_000) as r:
220 for chunk, triggers in r:
221 if len(chunk) == 0 and len(triggers) == 0:
222 continue
223 if len(chunk) > 0 and len(triggers) > 0:
224 lo, hi = int(chunk.t[0]), int(chunk.t[-1])
225 # window is 2 ms wide; triggers must lie within its span
226 assert all(lo - 2_000 < t <= hi + 2_000 for t in triggers.t)
227 seen_ev += len(chunk)
228 seen_tr.extend(zip(triggers.t.tolist(), triggers.id.tolist(), triggers.p.tolist()))
230 assert seen_ev == len(ev)
231 assert seen_tr == trig
234####################################
235# plumbing
236####################################
238def test_triggers_dropped_when_disabled(tmp_path: Any) -> None:
239 """ext_trigger=False (default): plain EventArray, trigger words skipped."""
240 from evutils.io import EventReader
241 words = np.concatenate([_evt3_trigger_words(t, i, v) for t, i, v in TRIGGERS])
242 p = _write_events_with_triggers(tmp_path, "evt3", words)
244 with EventReader(p) as r:
245 out = r.read_all()
246 assert not isinstance(out, tuple)
247 assert len(out) == len(_events())
250def test_no_triggers_yields_empty_array(tmp_path: Any) -> None:
251 from evutils.io import EventReader, EventWriter
252 p = tmp_path / "no_trig.raw"
253 with EventWriter(p, format="evt3") as w:
254 w.write(_events())
255 with EventReader(p, ext_trigger=True) as r:
256 ev, tr = r.read_all()
257 assert len(ev) == len(_events())
258 assert len(tr) == 0
261def test_ext_trigger_unsupported_decoder_warns(tmp_path: Any) -> None:
262 from evutils.io import EventReader, EventWriter
263 p = tmp_path / "events.csv"
264 with EventWriter(p) as w:
265 w.write(_events())
266 with warnings.catch_warnings(record=True) as caught:
267 warnings.simplefilter("always")
268 r = EventReader(p, ext_trigger=True)
269 r.close()
270 assert any("does not support" in str(w.message) for w in caught)