Coverage for src/evutils/io/_evt.py: 90%

1"""EVT (Prophesee EVT2 / EVT2.1 / EVT3) decoder backed by the native C parser. 

2 

3Reads the Prophesee ASCII header from a :class:`ByteSource`, then decodes the 

4binary EVT payload into ``EventArray`` chunks using the compiled parser in 

5``csrc`` (via :mod:`evutils.io._native_evt`). 

6 

7Input strategy: 

8 

9* If the source is *mappable* (mmap / in-memory), the whole payload is exposed 

10 as a single zero-copy ``uint16`` view and the parser walks it in windows -- 

11 no per-chunk copy, no vector-group carry across chunk boundaries. 

12* Otherwise the remaining stream is slurped into memory once and treated the 

13 same way. (Truly incremental streaming for live devices is future work; the 

14 parser ABI already supports it via ``result.current``.) 

15 

16The Prophesee formats are wired to native parsers, dispatched by the header 

17``format`` field: EVT3 (16-bit words), EVT2 / EVT4 (32-bit words) and EVT2.1 

18(64-bit words). EVT4 is not a standard Prophesee RAW variant -- it reuses EVT2's 

19CD/TIME_HIGH layout with distinct type codes plus vectorised CD; evutils defines 

20its own ``% evt 4.0`` header token for a self-consistent round-trip. All formats 

21have encoders. 

22""" 

23from __future__ import annotations 

24 

25import io 

26import re 

27from datetime import datetime 

28from typing import TYPE_CHECKING, Dict 

29 

30from ..jit import lazy_njit 

31import numpy as np 

32 

33from ..types import EventArray, TriggerArray 

34from .common import EventDecoder, EventEncoder 

35from ._native_core import ( 

36 EventSoABuffers, 

37 TriggerSoABuffers, 

38 decode_all_soa, 

39 events_view, 

40 triggers_view, 

41 parse_step, 

42 EVUTILS_PARSE_ERROR, 

43 EVUTILS_PARSE_WINDOW_DONE, 

44 EVUTILS_PARSE_OUTPUT_FULL, 

45 EVUTILS_PARSE_WARNING, 

46 NativeError, 

47 _handle_parse_warning, 

48) 

49from ._native_evt import ( 

50 Evt2Input, 

51 Evt2Parser, 

52 Evt3Input, 

53 Evt3Parser, 

54 Evt21Input, 

55 Evt21Parser, 

56 Evt4Input, 

57 Evt4Parser, 

58) 

59from ._source import ByteSource 

60 

61_EMPTY_EVENTS = EventArray.empty() 

62 

63# Prophesee sensor generation -> (width, height). Older EVT2 RAWs omit an 

64# explicit `format`/`geometry` field, so the geometry has to be inferred from 

65# the sensor identity -- the same thing the Metavision SDK does. 

66_GEN_RESOLUTION = { 

67 "1": (304, 240), "1.0": (304, 240), # Gen1 ATIS 

68 "2": (640, 480), "2.0": (640, 480), # Gen2 VGA 

69 "3": (640, 480), "3.0": (640, 480), # Gen3 VGA 

70 "3.1": (640, 480), # Gen3.1 VGA 

71 "4": (1280, 720), "4.0": (1280, 720), # Gen4 HD 

72 "4.1": (1280, 720), # Gen4.1 HD 

73 "4.2": (1280, 720), # Gen4.2 (IMX636) HD 

74} 

75 

76# Sensor model name -> (width, height). 

77_SENSOR_RESOLUTION = { 

78 "imx636": (1280, 720), # Gen4.2 HD 

79} 

80 

81# system_ID -> (width, height), for headers that carry nothing else. Values 

82# observed on Prophesee EVKs: 21-29 are Gen3/Gen3.1 VGA, 40-49 are Gen4.x HD. 

83_SYSTEM_ID_RESOLUTION = { 

84 21: (640, 480), 22: (640, 480), 23: (640, 480), 

85 28: (640, 480), 29: (640, 480), 

86 40: (1280, 720), 41: (1280, 720), 42: (1280, 720), 

87 48: (1280, 720), 49: (1280, 720), 

88} 

89 

90def _resolution_from_generation(gen: "int | str") -> tuple[int, int] | None: 

91 """Map a generation string (``"4.2"``, ``"gen31"``, ...) to a resolution.""" 

92 if gen is None: 

93 return None 

94 g = str(gen).strip().lower() 

95 if g in _GEN_RESOLUTION: 

96 return _GEN_RESOLUTION[g] 

97 m = re.fullmatch(r"gen(\d)(\d?)", g) # "gen31" -> "3.1", "gen4" -> "4" 

98 if m: 

99 key = m.group(1) + ("." + m.group(2) if m.group(2) else "") 

100 return _GEN_RESOLUTION.get(key) 

101 return None 

102 

103# Per-format native backend: parser class, zero-copy input wrapper, and the 

104# numpy word dtype the binary payload is viewed as. 

105_BACKENDS = { 

106 "evt3": (Evt3Parser, Evt3Input, np.uint16), 

107 "evt2": (Evt2Parser, Evt2Input, np.uint32), 

108 "evt21": (Evt21Parser, Evt21Input, np.uint64), 

109 "evt4": (Evt4Parser, Evt4Input, np.uint32), 

110} 

111 

112# Canonical format name -> (`% evt` token, `% format` token). Drives both header 

113# emission (encoder writes `% evt 3.0` + `% format EVT3`) and header parsing 

114# (decoder maps either token back to the canonical name). Single source of truth 

115# for the three Prophesee EVT variants. 

116_EVT_FORMATS: dict[str, tuple[str, str]] = { 

117 "evt3": ("3.0", "EVT3"), 

118 "evt21": ("2.1", "EVT21"), 

119 "evt2": ("2.0", "EVT2"), 

120 # EVT4 is not a standard Prophesee RAW header variant; there is no public 

121 # `% evt 4.0` recording. These tokens are evutils' own convention so the 

122 # encoder/decoder round-trip is self-consistent. 

123 "evt4": ("4.0", "EVT4"), 

124} 

125 

126# `% evt <token>` value -> canonical name, e.g. "3.0" -> "evt3". 

127_EVT_TOKEN_TO_NAME = {token: name for name, (token, _) in _EVT_FORMATS.items()} 

128 

129class EventDecoder_EVT(EventDecoder): 

130 """Decode Prophesee EVT2, EVT2.1, and EVT3 streams into ``EventArray`` chunks. 

131 

132 Parameters 

133 ---------- 

134 source 

135 Byte source to read from. 

136 chunk_size 

137 Maximum number of events produced per :meth:`read_chunk` call (the 

138 native output-buffer capacity). Does not bound the file size. 

139 

140 References 

141 ---------- 

142 [1] Prophesee RAW file format 

143 https://docs.prophesee.ai/stable/data/file_formats/raw.html 

144 

145 """ 

146 

147 _TAIL_PAD = 8 # >= parser look-ahead padding (EVT3_INPUT_PADDING) 

148 SUPPORTS_EXT_TRIGGERS = True 

149 

150 #: EVT is variable-length with an incremental time base, so seeking uses a 

151 #: SeekIndex: a coarse jump to the nearest bookmark, a TIME_HIGH wrap 

152 #: correction, then a short forward decode to the exact target. The default 

153 #: index is built in memory (exact, lazily on first seek); a Metavision 

154 #: `.tmp_index` sidecar can be opted into (fast, but approximate near large 

155 #: event gaps -- the built index is exact there). 

156 SUPPORTS_SEEK = True 

157 

158 #: init() slurps the whole payload into memory (or mmaps it), so seek works 

159 #: even over a non-seekable source (e.g. a compressed stream). 

160 _buffers_in_memory = True 

161 

162 # Per-format TIME_HIGH record descriptor: (type-field right-shift, type code). 

163 # Used to skip leading records until the first TIME_HIGH establishes a valid 

164 # time base -- events before it carry an undefined timestamp (a stream sliced 

165 # after capture start begins mid-group). Matches OpenEB / the reference 

166 # decoders, which drop those events. 

167 _TIME_HIGH_TYPE = { 

168 "evt3": (12, 0x8), # 16-bit words, type in bits 12..15 

169 "evt2": (28, 0x8), # 32-bit words, type in bits 28..31 

170 "evt21": (28, 0x8), 

171 "evt4": (28, 0xE), # EVT4 TIME_HIGH code differs (0xE) 

172 } 

173 

174 #: Timestamp wrap period per format: the amount the overflow accumulator 

175 #: adds on each TIME_HIGH wrap (EVT3: ts_high is bits 12..23 -> 2**24; 

176 #: EVT2/2.1/4: the 28-bit field is shifted <<6 -> 2**34). Used to snap the 

177 #: post-seek wrap correction to an exact multiple (see seek()). 

178 _WRAP_PERIOD = { 

179 "evt3": 1 << 24, 

180 "evt2": 1 << 34, 

181 "evt21": 1 << 34, 

182 "evt4": 1 << 34, 

183 } 

184 

185 def __init__(self, source: ByteSource, chunk_size: int = 1_000_000, read_external_triggers: bool = False): 

186 # The C parsers reserve headroom for one full vector expansion (up to 64 

187 # events) below the buffer capacity; a smaller chunk would make the 

188 # reserved offset 0 and the parser could never emit anything. 

189 if chunk_size < 128: 

190 raise ValueError(f"chunk_size must be >= 128 (got {chunk_size})") 

191 super().__init__(source, chunk_size, read_external_triggers=read_external_triggers) 

192 

193 self._format: str | None = None 

194 self._header: "dict[str, str | int | float]" = { 

195 "date": datetime.now(), 

196 "evt": None, 

197 "format": None, 

198 "generation": None, 

199 "serial_number": "00000000", 

200 "system_id": 49, 

201 "camera_integrator_name": "Prophesee", 

202 "integrator_name": "Prophesee", 

203 "sensor_name": None, 

204 "sensor_generation": None, 

205 "geometry": None, 

206 "plugin_name": None, 

207 "plugin_integrator_name": None, 

208 } 

209 

210 # Filled in init() 

211 self._buf: bytes | bytearray | None = None # keeps the underlying storage alive 

212 self._payload_off: int = 0 # byte offset where the binary payload starts 

213 self._words: "np.ndarray | None" = None # uint16 view of the whole payload 

214 self._offset: int = 0 # current word offset into _words 

215 self._start_offset: int = 0 # word offset of the first TIME_HIGH 

216 self._parser: "Callable | None" = None 

217 

218 # Seek support (see seek()). 

219 self._index = None # SeekIndex, lazily obtained 

220 self._index_is_ours: bool = False # built in-memory (exact counts) 

221 self._seek_correction: int = 0 # TIME_HIGH wrap correction, added to decoded ts 

222 #: Full path to the raw file and whether to try a Metavision sidecar 

223 #: index (set by EventReader from its ``index=`` option). Off by default: 

224 #: the exact in-memory index is built lazily on the first seek instead. 

225 self._raw_path: "str | None" = None 

226 self._use_sidecar: bool = False 

227 #: Persist evutils' own exact index to a ``<raw>.evidx`` sidecar (built 

228 #: on the first seek, loaded on open when fresh). Set by EventReader from 

229 #: ``index="persist"``. 

230 self._persist_index: bool = False 

231 

232 # ------------------------------------------------------------------ # 

233 # Header 

234 # ------------------------------------------------------------------ # 

235 def _parse_header(self, buf: bytes) -> int: 

236 """Scan the leading ``%``-prefixed ASCII header of ``buf`` (a bytes-like). 

237 

238 Returns the byte offset of the first non-header byte (start of payload). 

239 """ 

240 mv = memoryview(buf) 

241 n = len(mv) 

242 off = 0 

243 # A header line always starts with "% " (0x25 0x20). The `% end` marker 

244 # is optional -- the payload begins at the first line that does *not* 

245 # start with "% ", so that two-byte prefix is the real terminator. 

246 while off + 1 < n and mv[off] == 0x25 and mv[off + 1] == 0x20: 

247 window = bytes(mv[off:off + 8192]) 

248 rel = window.find(b"\n") 

249 if rel < 0: 

250 break 

251 line = window[:rel] 

252 # Exact end-of-header marker. Must not match other keys such as 

253 # "% endianness ...". 

254 if line.strip() == b"% end": 

255 off += rel + 1 

256 break 

257 self._consume_header_line(line) 

258 off += rel + 1 

259 return off 

260 

261 def _consume_header_line(self, line: bytes) -> None: 

262 try: 

263 split = line.decode("utf-8").strip().split(" ") 

264 except UnicodeDecodeError: 

265 return 

266 if len(split) < 2: 

267 return 

268 key = split[1].lower() 

269 raw = " ".join(split[2:]) 

270 

271 value: str | int | datetime = raw 

272 try: 

273 if key == "date": 

274 value = datetime.strptime(raw, "%Y-%m-%d %H:%M:%S") 

275 elif key in ("height", "width", "system_id"): 

276 value = int(raw) 

277 except ValueError: 

278 return 

279 self._header[key] = value 

280 

281 def _finalize_header(self) -> None: 

282 """Resolve format / width / height from the parsed header fields.""" 

283 fmt = self._header.get("format") 

284 if isinstance(fmt, str): 

285 for s in fmt.split(";"): 

286 if s.startswith("height"): 

287 self._height = int(s.split("=")[1]) 

288 elif s.startswith("width"): 

289 self._width = int(s.split("=")[1]) 

290 else: 

291 s = s.lower().replace(".", "") 

292 if s in _EVT_FORMATS: 

293 self._format = s 

294 

295 geom = self._header.get("geometry") 

296 if isinstance(geom, str): 

297 parts = geom.split("x") 

298 if len(parts) == 2: 

299 self._width = int(parts[0]) 

300 self._height = int(parts[1]) 

301 

302 evt = self._header.get("evt") 

303 if evt in _EVT_TOKEN_TO_NAME: 

304 self._format = _EVT_TOKEN_TO_NAME[evt] 

305 

306 if self._format is None: 

307 self._format = "evt3" # sensible default for Prophesee RAW 

308 

309 # EVT2.1 exists in two word layouts: "legacy" (two swapped 32-bit 

310 # halves, what the parser implements) and native little-endian 64-bit. 

311 # A non-legacy file would decode silently into garbage -- refuse it. 

312 if self._format == "evt21": 

313 endianness = str(self._header.get("endianness", "legacy")).lower() 

314 if endianness != "legacy": 

315 raise NotImplementedError( 

316 f"EVT2.1 with '% endianness {endianness}' is not supported " 

317 f"(only the 'legacy' swapped-halves layout is implemented)" 

318 ) 

319 

320 # Geometry not stated explicitly (older EVT2 RAWs): infer it from the 

321 # sensor identity, the way the Metavision SDK does. 

322 if self._width is None or self._height is None: 

323 res = self._infer_resolution() 

324 if res is not None: 

325 self._width, self._height = res 

326 

327 if self._width is None or not (0 < self._width <= 2048): 

328 self._width = 2048 

329 if self._height is None or not (0 < self._height <= 2048): 

330 self._height = 2048 

331 

332 def _infer_resolution(self) -> tuple[int, int] | None: 

333 """Guess (width, height) from the sensor identity fields in the header. 

334 

335 Tries, in order of reliability: an explicit sensor model, the 

336 ``sensor_generation`` / ``generation`` fields, the generation token 

337 embedded in ``plugin_name`` (e.g. ``hal_plugin_gen31_fx3``), and 

338 finally the ``system_ID``. Returns ``None`` if nothing matches. 

339 """ 

340 h = self._header 

341 

342 name = h.get("sensor_name") 

343 if isinstance(name, str): 

344 res = _SENSOR_RESOLUTION.get(name.strip().lower()) 

345 if res: 

346 return res 

347 

348 for key in ("sensor_generation", "generation"): 

349 res = _resolution_from_generation(h.get(key)) 

350 if res: 

351 return res 

352 

353 plugin = h.get("plugin_name") 

354 if isinstance(plugin, str): 

355 low = plugin.lower() 

356 for model, res in _SENSOR_RESOLUTION.items(): 

357 if model in low: 

358 return res 

359 m = re.search(r"gen(\d)(\d?)", low) # hal_plugin_gen31_fx3 -> "3.1" 

360 if m: 

361 key = m.group(1) + ("." + m.group(2) if m.group(2) else "") 

362 res = _GEN_RESOLUTION.get(key) 

363 if res: 

364 return res 

365 

366 return _SYSTEM_ID_RESOLUTION.get(h.get("system_id")) 

367 

368 # ------------------------------------------------------------------ # 

369 # Lifecycle 

370 # ------------------------------------------------------------------ # 

371 def init(self) -> None: 

372 """Initialize the EVT reader. 

373 

374 Returns 

375 ------- 

376 None 

377 

378 """ 

379 if self._is_initialized: 

380 return 

381 

382 if self._source.mappable(): 

383 self._buf = self._source.buffer() # zero-copy, whole file 

384 else: 

385 self._buf = memoryview(self._source.read(-1)) # slurp the stream 

386 

387 self._payload_off = self._parse_header(self._buf) 

388 self._finalize_header() 

389 

390 if self._format not in _BACKENDS: 

391 raise NotImplementedError( 

392 f"native decoder does not support format {self._format!r}" 

393 ) 

394 parser_cls, input_cls, word_dtype = _BACKENDS[self._format] 

395 self._input_cls = input_cls 

396 self._word_dtype = word_dtype 

397 

398 # View the binary payload as words of the format's native width. The 

399 # payload can be unaligned relative to the word size (the header length 

400 # is arbitrary); numpy tolerates this and x86 handles the unaligned 

401 # loads in the C parser. 

402 itemsize = np.dtype(word_dtype).itemsize 

403 n_words = (len(self._buf) - self._payload_off) // itemsize 

404 if n_words > 0: 

405 self._words = np.frombuffer( 

406 self._buf, dtype=word_dtype, count=n_words, offset=self._payload_off 

407 ) 

408 else: 

409 self._words = np.empty(0, dtype=word_dtype) 

410 

411 self._start_offset = self._find_first_time_high() 

412 self._offset = self._start_offset 

413 self._parser = parser_cls() 

414 cap = int(self._chunk_size) 

415 self._events = EventSoABuffers(cap) 

416 self._triggers = TriggerSoABuffers(max(cap // 16, 1)) 

417 self._is_initialized = True 

418 

419 def _find_first_time_high(self) -> int: 

420 """Word offset of the first TIME_HIGH record in the payload. 

421 

422 Records before the first TIME_HIGH have no established time base (their 

423 timestamp would decode to 0), which happens when a stream is sliced 

424 after capture start and begins mid-group. The reference decoders drop 

425 those records; starting decode at the first TIME_HIGH does the same 

426 while keeping the raw, absolute timestamps that follow. 

427 

428 Scans in exponentially growing blocks -- the first TIME_HIGH is almost 

429 always within the first few thousand words, so the whole payload is 

430 rarely touched. Returns 0 if the format has no TIME_HIGH descriptor or 

431 none is found (leave the stream untouched). 

432 """ 

433 desc = self._TIME_HIGH_TYPE.get(self._format or "") 

434 if desc is None or self._words is None: 

435 return 0 

436 shift, code = desc 

437 words = self._words 

438 n = len(words) 

439 start = 0 

440 block = 1 << 16 

441 while start < n: 

442 stop = min(start + block, n) 

443 seg = words[start:stop] 

444 hits = ((seg >> shift) & 0xF) == code 

445 idx = int(np.argmax(hits)) 

446 if hits[idx]: 

447 return start + idx 

448 start = stop 

449 block = min(block * 4, 1 << 24) 

450 return 0 

451 

452 @property 

453 def _tail_pad(self) -> int: 

454 return self._TAIL_PAD if self._format == "evt3" else 0 

455 

456 @property 

457 def _exact_window(self) -> bool: 

458 """True when the parser emits exactly one event per record, so it fills 

459 an output buffer to precisely its capacity. Only EVT2 qualifies: EVT3, 

460 EVT2.1 *and EVT4* all expand vector groups (a single word can emit up to 

461 32 events and the parser stops a few short of capacity), so they cannot 

462 use EventReader's zero-copy n_events fast path. Note EVT4's encoder only 

463 writes scalar CD, but the decoder must stay correct for vectorised EVT4 

464 input too.""" 

465 return self._format == "evt2" 

466 

467 def parse_step(self, events: EventSoABuffers, triggers: TriggerSoABuffers) -> int: 

468 """Run the parser once, appending decoded events into ``events``. 

469 

470 Advances the internal word offset and sets EOF when the input is drained. 

471 

472 Parameters 

473 ---------- 

474 events : EventSoABuffers 

475 Buffer to append events to. 

476 triggers : TriggerSoABuffers 

477 Buffer to append triggers to. 

478 

479 Returns 

480 ------- 

481 int 

482 Number of events appended. 

483 

484 """ 

485 if not self._is_initialized: 

486 self.init() 

487 if self._words is None or self._offset >= len(self._words): 

488 self._eof = True 

489 return 0 

490 before = events.size 

491 appended, self._offset = parse_step( 

492 self._words, self._offset, self._input_cls, self._parser, 

493 events, triggers, tail_pad=self._tail_pad, word_dtype=self._word_dtype, 

494 strict=self._strict, 

495 ) 

496 if appended and self._seek_correction: 

497 # Restore the TIME_HIGH wrap accumulation lost by the post-seek 

498 # parser.reset() (see seek()); applied here so every read path 

499 # (read_chunk and the reader's accumulator _pull) sees absolute ts. 

500 events_view(events).t[before:before + appended] += self._seek_correction 

501 if self._offset >= len(self._words): 

502 self._eof = True 

503 return appended 

504 

505 @property 

506 def _has_delta_t_parser(self) -> bool: 

507 """True when a dedicated C delta_t parser exists for this format, so the 

508 reader can decode a whole time window in one GIL-free C call (no 

509 Python-side searchsorted / overshoot carry). EVT3 only, for now.""" 

510 return self._format == "evt3" and hasattr(self._parser, "parse_delta_t_soa") 

511 

512 def parse_step_delta_t(self, events: EventSoABuffers, triggers: TriggerSoABuffers, 

513 end_ts: int) -> tuple[int, int]: 

514 """Run the dedicated C delta_t parser once, appending events with 

515 timestamp ``< end_ts`` into ``events``. 

516 

517 Returns ``(appended, status)`` where status is one of 

518 ``EVUTILS_PARSE_WINDOW_DONE`` (the time boundary was reached -- the 

519 window is complete), ``EVUTILS_PARSE_OUTPUT_FULL`` (grow and call again), 

520 or ``EVUTILS_PARSE_OK`` (input drained for this call). At true EOF the 

521 final incomplete vector group is flushed via the tail pad, mirroring 

522 :meth:`parse_step`. 

523 """ 

524 if not self._is_initialized: 

525 self.init() 

526 words = self._words 

527 if words is None or self._offset >= len(words): 

528 self._eof = True 

529 return 0, EVUTILS_PARSE_WINDOW_DONE 

530 before = events.size 

531 # A post-seek parser.reset() loses the TIME_HIGH wrap accumulation: the 

532 # C parser decodes in the raw (low) timeline while the caller's end_ts 

533 # lives in the corrected absolute one. Translate the boundary into the 

534 # raw timeline for the C call, then shift the decoded slice back 

535 # (mirrors parse_step's correction). 

536 corr = self._seek_correction 

537 raw_end_ts = end_ts - corr 

538 inp = self._input_cls(words[self._offset:]) 

539 while True: 

540 res = self._parser.parse_delta_t_soa(inp, events, triggers, raw_end_ts) 

541 consumed = inp.consumed(res) 

542 if res.status == EVUTILS_PARSE_WARNING: 

543 _handle_parse_warning(self._offset + consumed, self._strict, fmt=self._format) 

544 if consumed > 0: 

545 self._offset += consumed 

546 inp = self._input_cls(words[self._offset:]) 

547 continue 

548 elif res.status == EVUTILS_PARSE_ERROR: 

549 raise NativeError(f"{self._format} delta_t parse error at word {self._offset + consumed}") 

550 break 

551 status = int(res.status) 

552 # Input drained with no progress => only the sub-PADDING tail remains; 

553 # flush it (incomplete final group) and mark EOF, as parse_step does. 

554 if consumed == 0 and status not in (EVUTILS_PARSE_WINDOW_DONE, EVUTILS_PARSE_OUTPUT_FULL): 

555 if self._tail_pad and self._word_dtype is not None: 

556 tail = words[self._offset:] 

557 if len(tail): 

558 scratch = np.zeros(len(tail) + self._tail_pad, dtype=self._word_dtype) 

559 scratch[: len(tail)] = tail 

560 self._parser.parse_delta_t_soa( 

561 self._input_cls(scratch), events, triggers, raw_end_ts) 

562 self._offset = len(words) 

563 self._eof = True 

564 return self._apply_dt_correction(events, before, corr), status 

565 self._offset += consumed 

566 if self._offset >= len(words): 

567 self._eof = True 

568 return self._apply_dt_correction(events, before, corr), status 

569 

570 @staticmethod 

571 def _apply_dt_correction(events: EventSoABuffers, before: int, corr: int) -> int: 

572 """Shift the slice appended since ``before`` by the seek wrap correction 

573 and return the appended count.""" 

574 appended = events.size - before 

575 if appended and corr: 

576 events_view(events).t[before:before + appended] += corr 

577 return appended 

578 

579 def read_chunk(self, delta_t_hint: int | None = None, 

580 n_events_hint: int | None = None) -> 'EventArray | tuple[EventArray, TriggerArray]': 

581 if not self._is_initialized: 

582 self.init() 

583 

584 # Nothing left: signal EOF with an empty array (never a stale buffer). 

585 if self._words is None or self._offset >= len(self._words): 

586 self._eof = True 

587 if self.read_external_triggers: 

588 return _EMPTY_EVENTS, TriggerArray.empty() 

589 return _EMPTY_EVENTS 

590 

591 ev, tr = self._events, self._triggers 

592 ev.reset() 

593 tr.reset() 

594 

595 # Parse until we produce something or genuinely exhaust the input. A 

596 # window can consume words yet emit no events (pure timing packets), so 

597 # we must not treat an empty result as EOF unless the input is drained. 

598 appended = 0 

599 while appended == 0 and self._offset < len(self._words): 

600 if not self.read_external_triggers: 

601 # Unrequested triggers are discarded anyway; reset the sink so a 

602 # trigger-dense region can never fill it and stall the parser. 

603 tr.reset() 

604 elif tr.size > 0: 

605 break # trigger-only progress: hand the triggers out 

606 before_off = self._offset 

607 appended = self.parse_step(ev, tr) 

608 if appended == 0 and self._offset == before_off: 

609 break # zero progress (full buffers); never spin 

610 

611 n = ev.size 

612 ev_view = events_view(ev) if n > 0 else _EMPTY_EVENTS 

613 # The seek wrap-correction is applied inside parse_step (above), so it is 

614 # already reflected in ev_view -- do not add it again here. 

615 if self.read_external_triggers: 

616 tr_view = triggers_view(tr) if tr.size > 0 else TriggerArray.empty() 

617 return ev_view, tr_view 

618 return ev_view 

619 

620 # Initial output-capacity estimate (events per input word) for the 

621 # single-buffer read_all() path. EVT2/EVT4 are exact upper bounds (<=1 event 

622 # per 32-bit word); EVT3/EVT2.1 vary with vector density. The estimate only 

623 # needs to be a rough ballpark: if it is too small decode_all_soa grows the 

624 # buffer once, extrapolating the true count from the fraction of input 

625 # consumed so far (no repeated reallocation even for dense EVT2.1 vector 

626 # streams, which reach ~14+ events per word). 

627 _READ_ALL_EST = {"evt3": 1.0, "evt2": 1.0, "evt21": 1.5, "evt4": 1.0} 

628 

629 def read_all(self) -> 'EventArray | tuple[EventArray, TriggerArray]': 

630 """Decode the whole remaining payload into one buffer (no per-chunk copy). 

631 

632 See :func:`evutils.io._native_core.decode_all_soa`. Note this materialises 

633 every event at once; for very large recordings that do not fit in memory, 

634 iterate with :meth:`read_chunk` (via ``EventReader``) instead. 

635 """ 

636 if not self._is_initialized: 

637 self.init() 

638 

639 # decode_all_soa is an events-only fast path; with external triggers 

640 # requested, fall back to the chunked base implementation (which 

641 # carries triggers and applies the seek correction via read_chunk). 

642 if self.read_external_triggers: 

643 return super().read_all() 

644 

645 if self._words is None or self._offset >= len(self._words): 

646 self._eof = True 

647 return _EMPTY_EVENTS 

648 

649 assert self._format is not None # set by init() 

650 out, self._offset = decode_all_soa( 

651 self._words, self._offset, self._input_cls, self._parser, 

652 est_events_per_word=self._READ_ALL_EST.get(self._format, 1.0), 

653 tail_pad=self._tail_pad, word_dtype=self._word_dtype, 

654 strict=self._strict, 

655 ) 

656 if len(out) and self._seek_correction: 

657 out.t += self._seek_correction 

658 self._eof = True 

659 return out 

660 

661 def reset(self) -> None: 

662 """Reset the EVT reader to the beginning. 

663 

664 Returns 

665 ------- 

666 None 

667 

668 """ 

669 self._offset = self._start_offset 

670 self._eof = False 

671 self._seek_correction = 0 

672 if self._parser is not None: 

673 self._parser.reset() 

674 

675 # ------------------------------------------------------------------ # 

676 # Seeking 

677 # ------------------------------------------------------------------ # 

678 def _ensure_index(self, need_counts: bool = False): 

679 """Obtain a :class:`SeekIndex` (Metavision sidecar or built in memory). 

680 

681 ``need_counts`` forces a built (evutils-counted) index, whose cumulative 

682 counts match this decoder exactly -- required for event-index seeks. 

683 """ 

684 from ._index import ( 

685 IncrementalSeekIndex, 

686 evutils_index_path, 

687 load_seek_index, 

688 metavision_index_path, 

689 read_metavision_index, 

690 save_seek_index, 

691 ) 

692 

693 if self._index is not None and not (need_counts and not self._index_is_ours): 

694 return self._index 

695 

696 word_size = np.dtype(self._word_dtype).itemsize 

697 idx = None 

698 # Persist mode: load our own exact `.evidx` sidecar if one is present and 

699 # fresh (it carries evutils cumulative counts, so it satisfies both time 

700 # and event-index seeks). 

701 if self._persist_index and self._raw_path is not None: 

702 idx = load_seek_index( 

703 evutils_index_path(self._raw_path), self._raw_path) 

704 if idx is not None: 

705 self._index_is_ours = True 

706 if (idx is None and self._use_sidecar and self._raw_path is not None 

707 and not need_counts): 

708 idx = read_metavision_index( 

709 metavision_index_path(self._raw_path), self._raw_path, 

710 self._payload_off, word_size, 

711 ) 

712 self._index_is_ours = False 

713 if idx is None: 

714 built = IncrementalSeekIndex( 

715 words=self._words, start_offset=self._start_offset, 

716 input_cls=self._input_cls, parser_cls=type(self._parser), 

717 tail_pad=self._tail_pad, word_dtype=self._word_dtype, 

718 # TIME_HIGH-aligned bookmarks: a reset parser at a bookmark must 

719 # re-establish its time base before its first event, else the 

720 # seek-time wrap-correction snap is invalid (see _index.py). 

721 time_high=self._TIME_HIGH_TYPE.get(self._format or ""), 

722 ) 

723 self._index_is_ours = True 

724 if self._persist_index and self._raw_path is not None: 

725 # Build the whole exact index now and write the sidecar for next 

726 # time. A failed write (read-only dir, ...) is non-fatal: fall 

727 # back to the freshly built index. 

728 idx = built.to_static() 

729 try: 

730 save_seek_index( 

731 idx, evutils_index_path(self._raw_path), self._raw_path) 

732 except OSError: 

733 pass 

734 else: 

735 idx = built 

736 self._index = idx 

737 return idx 

738 

739 def _decode_step(self) -> tuple[EventArray, 'TriggerArray']: 

740 """One decode step from the current offset (no correction). 

741 Returns a tuple of EventArray and TriggerArray views. 

742 """ 

743 if self._words is None or self._offset >= len(self._words): 

744 self._eof = True 

745 return _EMPTY_EVENTS, TriggerArray.empty() 

746 

747 ev, tr = self._events, self._triggers 

748 ev.reset() 

749 tr.reset() 

750 appended = 0 

751 while appended == 0 and self._offset < len(self._words): 

752 tr.reset() 

753 before_off = self._offset 

754 appended = self.parse_step(ev, tr) 

755 if appended == 0 and self._offset == before_off: 

756 break 

757 

758 return (events_view(ev) if ev.size > 0 else _EMPTY_EVENTS, 

759 triggers_view(tr) if tr.size > 0 else TriggerArray.empty()) 

760 

761 def seek(self, t: int | None = None, n: int | None = None) -> tuple["SeekResult", "EventArray", "TriggerArray | None"]: 

762 """Seek to an absolute timestamp (µs) or event index. See base class. 

763 

764 Jumps to the nearest index bookmark at/before the target, restores the 

765 TIME_HIGH wrap accumulation, then decodes forward to land exactly on the 

766 target (returning the boundary chunk's remainder to the caller). 

767 """ 

768 from .common import SeekResult 

769 if not self._is_initialized: 

770 self.init() 

771 axis, val = self._seek_axis(t, n) 

772 

773 index = self._ensure_index(need_counts=(axis == "n")) 

774 

775 if index.n_events == 0 or self._words is None: 

776 self._offset = len(self._words) if self._words is not None else 0 

777 self._eof = True 

778 self._seek_correction = 0 

779 idx = val if axis == "n" else 0 

780 return SeekResult(ts=val, index=idx, eof=True), _EMPTY_EVENTS, None 

781 

782 word_off, cum, base_ts = (index.bookmark_for_time(val) if axis == "t" 

783 else index.bookmark_for_event(val)) 

784 

785 self._offset = word_off 

786 self._parser.reset() 

787 self._seek_correction = 0 

788 self._eof = False 

789 seen = cum 

790 

791 correction: int | None = None 

792 while True: 

793 raw_ev, raw_tr = self._decode_step() 

794 if len(raw_ev) == 0 and len(raw_tr) == 0: 

795 # Target is at/after the end of the stream. 

796 self._seek_correction = correction or 0 

797 self._eof = True 

798 idx = val if axis == "n" else seen 

799 return SeekResult(ts=val, index=idx, eof=True), _EMPTY_EVENTS, None 

800 if correction is None: 

801 if len(raw_ev) > 0: 

802 period = self._WRAP_PERIOD.get(self._format or "", 1 << 24) 

803 correction = int(round((base_ts - int(raw_ev.t[0])) / period)) * period 

804 else: 

805 correction = 0 # Cannot compute correction without events 

806 

807 # Apply correction 

808 t_corr = raw_ev.t + correction