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 Any, TYPE_CHECKING, Dict 

29 

30if TYPE_CHECKING: 

31 from ..types import TriggerArray 

32 

33from .._jit import lazy_njit 

34import numpy as np 

35 

36from ..types import EventArray 

37from .common import EventDecoder, EventEncoder 

38from ._native_core import ( 

39 EventSoABuffers, 

40 TriggerSoABuffers, 

41 decode_all_soa, 

42 events_view, 

43 triggers_view, 

44 parse_step, 

45) 

46from ._native_evt import ( 

47 Evt2Input, 

48 Evt2Parser, 

49 Evt3Input, 

50 Evt3Parser, 

51 Evt21Input, 

52 Evt21Parser, 

53 Evt4Input, 

54 Evt4Parser, 

55) 

56from ._source import ByteSource 

57 

58_EMPTY_EVENTS = EventArray.empty() 

59 

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

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

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

63_GEN_RESOLUTION = { 

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

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

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

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

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

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

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

71} 

72 

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

74_SENSOR_RESOLUTION = { 

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

76} 

77 

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

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

80_SYSTEM_ID_RESOLUTION = { 

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

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

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

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

85} 

86 

87 

88def _resolution_from_generation(gen: Any) -> tuple[int, int] | None: 

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

90 if gen is None: 

91 return None 

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

93 if g in _GEN_RESOLUTION: 

94 return _GEN_RESOLUTION[g] 

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

96 if m: 

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

98 return _GEN_RESOLUTION.get(key) 

99 return None 

100 

101 

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

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

104_BACKENDS = { 

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

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

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

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

109} 

110 

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

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

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

114# for the three Prophesee EVT variants. 

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

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

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

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

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

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

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

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

123} 

124 

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

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

127 

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 def __init__(self, source: ByteSource, chunk_size: int = 1_000_000, read_external_triggers: bool = False): 

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

152 

153 self._format: str | None = None 

154 self._header: Dict[str, Any] = { 

155 "date": datetime.now(), 

156 "evt": None, 

157 "format": None, 

158 "generation": None, 

159 "serial_number": "00000000", 

160 "system_id": 49, 

161 "camera_integrator_name": "Prophesee", 

162 "integrator_name": "Prophesee", 

163 "sensor_name": None, 

164 "sensor_generation": None, 

165 "geometry": None, 

166 "plugin_name": None, 

167 "plugin_integrator_name": None, 

168 } 

169 

170 # Filled in init() 

171 self._buf: Any = None # keeps the underlying storage alive 

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

173 self._words: Any = None # uint16 view of the whole payload 

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

175 self._parser: Any = None 

176 

177 

178 # ------------------------------------------------------------------ # 

179 # Header 

180 # ------------------------------------------------------------------ # 

181 def _parse_header(self, buf: Any) -> int: 

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

183 

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

185 """ 

186 mv = memoryview(buf) 

187 n = len(mv) 

188 off = 0 

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

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

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

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

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

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

195 if rel < 0: 

196 break 

197 line = window[:rel] 

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

199 # "% endianness ...". 

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

201 off += rel + 1 

202 break 

203 self._consume_header_line(line) 

204 off += rel + 1 

205 return off 

206 

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

208 try: 

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

210 except UnicodeDecodeError: 

211 return 

212 if len(split) < 2: 

213 return 

214 key = split[1].lower() 

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

216 

217 value: str | int | datetime = raw 

218 try: 

219 if key == "date": 

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

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

222 value = int(raw) 

223 except ValueError: 

224 return 

225 self._header[key] = value 

226 

227 def _finalize_header(self) -> None: 

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

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

230 if isinstance(fmt, str): 

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

232 if s.startswith("height"): 

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

234 elif s.startswith("width"): 

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

236 else: 

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

238 if s in _EVT_FORMATS: 

239 self._format = s 

240 

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

242 if isinstance(geom, str): 

243 parts = geom.split("x") 

244 if len(parts) == 2: 

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

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

247 

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

249 if evt in _EVT_TOKEN_TO_NAME: 

250 self._format = _EVT_TOKEN_TO_NAME[evt] 

251 

252 if self._format is None: 

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

254 

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

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

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

258 res = self._infer_resolution() 

259 if res is not None: 

260 self._width, self._height = res 

261 

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

263 self._width = 2048 

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

265 self._height = 2048 

266 

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

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

269 

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

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

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

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

274 """ 

275 h = self._header 

276 

277 name = h.get("sensor_name") 

278 if isinstance(name, str): 

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

280 if res: 

281 return res 

282 

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

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

285 if res: 

286 return res 

287 

288 plugin = h.get("plugin_name") 

289 if isinstance(plugin, str): 

290 low = plugin.lower() 

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

292 if model in low: 

293 return res 

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

295 if m: 

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

297 res = _GEN_RESOLUTION.get(key) 

298 if res: 

299 return res 

300 

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

302 

303 # ------------------------------------------------------------------ # 

304 # Lifecycle 

305 # ------------------------------------------------------------------ # 

306 def init(self) -> None: 

307 """Initialize the EVT reader. 

308 

309 Returns 

310 ------- 

311 None 

312 

313 """ 

314 if self._is_initialized: 

315 return 

316 

317 if self._source.mappable(): 

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

319 else: 

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

321 

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

323 self._finalize_header() 

324 

325 if self._format not in _BACKENDS: 

326 raise NotImplementedError( 

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

328 ) 

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

330 self._input_cls = input_cls 

331 self._word_dtype = word_dtype 

332 

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

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

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

336 # loads in the C parser. 

337 itemsize = np.dtype(word_dtype).itemsize 

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

339 if n_words > 0: 

340 self._words = np.frombuffer( 

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

342 ) 

343 else: 

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

345 

346 self._offset = 0 

347 self._parser = parser_cls() 

348 cap = int(self._chunk_size) 

349 self._events = EventSoABuffers(cap) 

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

351 self._is_initialized = True 

352 

353 @property 

354 def _tail_pad(self) -> int: 

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

356 

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

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

359 

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

361 

362 Parameters 

363 ---------- 

364 events : EventSoABuffers 

365 Buffer to append events to. 

366 triggers : TriggerSoABuffers 

367 Buffer to append triggers to. 

368 

369 Returns 

370 ------- 

371 int 

372 Number of events appended. 

373 

374 """ 

375 if not self._is_initialized: 

376 self.init() 

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

378 self._eof = True 

379 return 0 

380 appended, self._offset = parse_step( 

381 self._words, self._offset, self._input_cls, self._parser, 

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

383 ) 

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

385 self._eof = True 

386 return appended 

387 

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

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

390 if not self._is_initialized: 

391 self.init() 

392 

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

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

395 self._eof = True 

396 if self.read_external_triggers: 

397 from ..types import TriggerArray 

398 return _EMPTY_EVENTS, TriggerArray.empty() 

399 return _EMPTY_EVENTS 

400 

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

402 ev.reset() 

403 tr.reset() 

404 

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

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

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

408 appended = 0 

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

410 appended = self.parse_step(ev, tr) 

411 

412 n = ev.size 

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

414 if self.read_external_triggers: 

415 from ..types import TriggerArray 

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

417 return ev_view, tr_view 

418 return ev_view 

419 

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

421 # single-buffer read_all() path. EVT2 is an exact upper bound (<=1 event per 

422 # 32-bit word); EVT3/EVT2.1 vary with vector density, so decode_all_soa grows 

423 # the buffer if the estimate is too small. 

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

425 

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

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

428 

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

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

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

432 """ 

433 if not self._is_initialized: 

434 self.init() 

435 

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

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

438 # carries the trigger stream through). 

439 if self.read_external_triggers: 

440 return super().read_all() 

441 

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

443 self._eof = True 

444 return _EMPTY_EVENTS 

445 

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

447 out, self._offset = decode_all_soa( 

448 self._words, self._offset, self._input_cls, self._parser, 

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

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

451 ) 

452 self._eof = True 

453 return out 

454 

455 def reset(self) -> None: 

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

457 

458 Returns 

459 ------- 

460 None 

461 

462 """ 

463 self._offset = 0 

464 self._eof = False 

465 if self._parser is not None: 

466 self._parser.reset() 

467 

468 def tell(self) -> int: 

469 """Get the current byte offset. 

470 

471 Returns 

472 ------- 

473 int 

474 Current byte offset. 

475 

476 """ 

477 if not self._is_initialized: 

478 return 0 

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

480 return self._payload_off + self._offset * word_size 

481 

482 def close(self) -> None: 

483 """Close the EVT reader. 

484 

485 Returns 

486 ------- 

487 None 

488 

489 """ 

490 # Drop numpy views into the (possibly mmap-backed) storage so the source 

491 # can be closed without BufferError. 

492 self._words = None 

493 self._buf = None 

494 

495 

496# --------------------------------------------------------------------------- # 

497# Encoders (EVT3 / EVT2 / EVT2.1 writers) 

498# 

499# The writers are numba (there is no native encoder yet). They live here with 

500# the format module now that _raw.py is gone. 

501# --------------------------------------------------------------------------- # 

502EVT3_EVT_ADDR_Y = 0x0000 

503EVT3_EVT_ADDR_X = 0x2000 

504EVT3_VECT_BASE_X = 0x3000 

505EVT3_VECT_12 = 0x4000 

506EVT3_VECT_8 = 0x5000 

507EVT3_EVT_TIME_LOW = 0x6000 

508EVT3_CONTINUED_4 = 0x7000 

509EVT3_EVT_TIME_HIGH = 0x8000 

510EVT3_EXT_TRIGGER = 0xA000 

511EVT3_OTHERS = 0xE000 

512EVT3_CONTINUED_12 = 0xF000 

513 

514 

515@lazy_njit 

516def get_raw_evt3_buffer(events: np.ndarray, last_lower12_ts: int, last_upper12_ts: int, last_y: int, master: bool = True) -> tuple[np.ndarray, int, int, int]: 

517 """Encode events as EVT3. 

518 

519 Parameters 

520 ---------- 

521 events : np.ndarray 

522 Array of events to encode. 

523 last_lower12_ts : int 

524 Last lower 12-bit timestamp. 

525 last_upper12_ts : int 

526 Last upper 12-bit timestamp. 

527 last_y : int 

528 Last y coordinate. 

529 master : bool, optional 

530 Whether this is the master camera, by default True. 

531 

532 Returns 

533 ------- 

534 tuple 

535 A tuple containing the raw buffer, last lower 12-bit timestamp, last upper 12-bit timestamp, and last y coordinate. 

536 

537 """ 

538 # Pre-allocate large buffer 

539 buffer = np.zeros(len(events) * 8, dtype=np.uint8) 

540 

541 # Prepare the master/slave bit 

542 if master: 

543 master_slave = 0x000 

544 else: 

545 master_slave = 0x800 

546 

547 # Current position of the buffer 

548 i = 0 

549 

550 for ev in events: 

551 upper12_ts = (int(ev['t']) & 0x0FFF000) >> 12 

552 lower12_ts = int(ev['t']) & 0x00000FFF 

553 

554 # EVT_TIME_HIGH - Updates the higher 12-bit portion of the 24-bit time base 

555 if upper12_ts != last_upper12_ts: 

556 last_upper12_ts = upper12_ts 

557 value = EVT3_EVT_TIME_HIGH | (upper12_ts & 0xFFF) 

558 

559 buffer[i] = value & 0xFF 

560 buffer[i + 1] = (value >> 8) & 0xFF 

561 i += 2 

562 

563 # EVT_TIME_LOW - Updates the lower 12-bit portion of the 24-bit time base 

564 if lower12_ts != last_lower12_ts: 

565 last_lower12_ts = lower12_ts 

566 value = EVT3_EVT_TIME_LOW | (lower12_ts & 0xFFF) 

567 

568 buffer[i] = value & 0xFF 

569 buffer[i + 1] = (value >> 8) & 0xFF 

570 i += 2 

571 

572 # EVT_ADDR_Y - Y coordinate, and system type (master/slave camera) 

573 if last_y != ev['y']: 

574 last_y = ev['y'] 

575 value = (EVT3_EVT_ADDR_Y | master_slave | (int(ev['y']) & 0x7FF)) 

576 

577 buffer[i] = value & 0xFF 

578 buffer[i + 1] = (value >> 8) & 0xFF 

579 i += 2 

580 

581 # EVT_ADDR_X - Single valid event, X coordinate and polarity 

582 value = EVT3_EVT_ADDR_X | (int(ev['x']) & 0x7FF) | ((int(ev['p']) & 0x01) << 11) 

583 

584 buffer[i] = value & 0xFF 

585 buffer[i + 1] = (value >> 8) & 0xFF 

586 i += 2 

587 

588 return buffer[:i], last_lower12_ts, last_upper12_ts, last_y 

589 

590 

591@lazy_njit 

592def get_raw_evt2_buffer(events: np.ndarray, last_ts_high: int) -> tuple[np.ndarray, int]: 

593 """Encode events as EVT2 (32-bit words). 

594 

595 Timestamp is split into a 28-bit high part (EVT_TIME_HIGH word) and a 6-bit 

596 low part carried in each CD word. A TIME_HIGH word is emitted only when the 

597 high part changes. Layout: type[28:31], ts_low[22:27], x[11:21], y[0:10]. 

598 

599 Parameters 

600 ---------- 

601 events : np.ndarray 

602 Array of events to encode. 

603 last_ts_high : int 

604 Last high timestamp part. 

605 

606 Returns 

607 ------- 

608 tuple 

609 A tuple containing the raw buffer and last high timestamp part. 

610 

611 """ 

612 n = len(events) 

613 buffer = np.empty(2 * n, dtype=np.uint32) # <= 1 TIME_HIGH + 1 CD per event 

614 i = 0 

615 for k in range(n): 

616 ts = np.int64(events[k]['t']) 

617 x = np.int64(events[k]['x']) & 0x7FF 

618 y = np.int64(events[k]['y']) & 0x7FF 

619 p = np.int64(events[k]['p']) & 0x1 

620 

621 ts_high = (ts >> 6) & 0x0FFFFFFF 

622 ts_low = ts & 0x3F 

623 

624 if ts_high != last_ts_high: 

625 buffer[i] = np.uint32((8 << 28) | ts_high) # EVT2_EVT_TIME_HIGH 

626 i += 1 

627 last_ts_high = int(ts_high) 

628 

629 buffer[i] = np.uint32((p << 28) | (ts_low << 22) | (x << 11) | y) 

630 i += 1 

631 

632 return buffer[:i], last_ts_high 

633 

634 

635@lazy_njit 

636def get_raw_evt21_buffer(events: np.ndarray, last_ts_high: int) -> tuple[np.ndarray, int]: 

637 """Encode events as EVT2.1 (64-bit words, legacy endianness). 

638 

639 Same descriptor layout as EVT2 in the low 32 bits (type[28:31], 

640 ts_low[22:27], x_base[11:21], y[0:10]); the high 32 bits are a validity 

641 bitmask for x_base..x_base+31. This writer emits one event per word (mask 

642 with a single bit set at x_base = x) -- valid EVT2.1, not yet vectorised. 

643 

644 Parameters 

645 ---------- 

646 events : np.ndarray 

647 Array of events to encode. 

648 last_ts_high : int 

649 Last high timestamp part. 

650 

651 Returns 

652 ------- 

653 tuple 

654 A tuple containing the raw buffer and last high timestamp part. 

655 

656 """ 

657 n = len(events) 

658 buffer = np.empty(2 * n, dtype=np.uint64) # <= 1 TIME_HIGH + 1 CD per event 

659 i = 0 

660 for k in range(n): 

661 ts = np.int64(events[k]['t']) 

662 x = np.int64(events[k]['x']) & 0x7FF 

663 y = np.int64(events[k]['y']) & 0x7FF 

664 p = np.int64(events[k]['p']) & 0x1 

665 

666 ts_high = (ts >> 6) & 0x0FFFFFFF 

667 ts_low = ts & 0x3F 

668 

669 if ts_high != last_ts_high: 

670 buffer[i] = np.uint64((8 << 28) | ts_high) # EVT21_EVT_TIME_HIGH 

671 i += 1 

672 last_ts_high = int(ts_high) 

673 

674 desc = (p << 28) | (ts_low << 22) | (x << 11) | y 

675 # High 32 bits: validity mask with bit 0 set (single event at x_base=x). 

676 buffer[i] = (np.uint64(1) << np.uint64(32)) | np.uint64(desc) 

677 i += 1 

678 

679 return buffer[:i], last_ts_high 

680 

681 

682@lazy_njit 

683def get_raw_evt4_buffer(events: np.ndarray, last_ts_high: int) -> tuple[np.ndarray, int]: 

684 """Encode events as EVT4 (32-bit words). 

685 

686 Same CD / TIME_HIGH bit layout as EVT2 (type[28:31], ts_low[22:27], 

687 x[11:21], y[0:10]) but with EVT4's type codes: CD_OFF=0xA / CD_ON=0xB and 

688 TIME_HIGH=0xE. One CD word per event (EVT4's vectorised CD_VEC form is a 

689 decode-side optimisation and is not emitted here). 

690 

691 Parameters 

692 ---------- 

693 events : np.ndarray 

694 Array of events to encode. 

695 last_ts_high : int 

696 Last high timestamp part. 

697 

698 Returns 

699 ------- 

700 tuple 

701 A tuple containing the raw buffer and last high timestamp part. 

702 

703 """ 

704 n = len(events) 

705 buffer = np.empty(2 * n, dtype=np.uint32) # <= 1 TIME_HIGH + 1 CD per event 

706 i = 0 

707 for k in range(n): 

708 ts = np.int64(events[k]['t']) 

709 x = np.int64(events[k]['x']) & 0x7FF 

710 y = np.int64(events[k]['y']) & 0x7FF 

711 p = np.int64(events[k]['p']) & 0x1 

712 

713 ts_high = (ts >> 6) & 0x0FFFFFFF 

714 ts_low = ts & 0x3F 

715 

716 if ts_high != last_ts_high: 

717 buffer[i] = np.uint32((0xE << 28) | ts_high) # EVT4_EVT_TIME_HIGH 

718 i += 1 

719 last_ts_high = int(ts_high) 

720 

721 # type = CD_OFF (0xA) for p=0, CD_ON (0xB) for p=1. 

722 buffer[i] = np.uint32(((0xA | p) << 28) | (ts_low << 22) | (x << 11) | y) 

723 i += 1 

724 

725 return buffer[:i], last_ts_high 

726 

727 

728class EventEncoder_EVT(EventEncoder): 

729 """Encoder for Prophesee RAW/EVT files. 

730 

731 Parameters 

732 ---------- 

733 writable 

734 Destination stream to write to. 

735 width, height : int 

736 Frame geometry written into the header. 

737 dt : datetime, optional 

738 Recording timestamp (defaults to now). 

739 serial : str 

740 Camera serial number written into the header. 

741 format : {"evt3", "evt21", "evt2", "evt4"} 

742 Output format. All are supported; EVT2.1 and EVT4 are written one event 

743 per word (valid but not vectorised). 

744 

745 References 

746 ---------- 

747 [1] Prophesee RAW file format 

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

749 

750 """ 

751 

752 def __init__(self, writable: io.BufferedWriter, width: int = 1280, height: int = 720, 

753 dt: datetime | None = None, serial: str = "00000000", format: str = "evt3"): 

754 super().__init__(writable, width, height, dt) 

755 

756 format = format.lower().replace(".", "") 

757 if format not in _EVT_FORMATS: 

758 raise ValueError(f"Unsupported format {format}. Supported formats are {list(_EVT_FORMATS)}") 

759 self._format = format 

760 

761 self._system_id = 49 

762 

763 self._last_upper12_ts = -1 

764 self._last_lower12_ts = -1 

765 self._last_y = -1 

766 self._last_ts_high = -1 # EVT2 / EVT2.1 time-high state 

767 

768 self._serial_number = serial 

769 

770 self._formatted_datetime = self._dt.strftime("%Y-%m-%d %H:%M:%S") 

771 

772 def init(self) -> None: 

773 """Initialize the EVT writer. 

774 

775 Returns 

776 ------- 

777 None 

778 

779 """ 

780 if self._is_initialized: 

781 return 

782 

783 # EVT2.1 packs its 64-bit words as two swapped 32-bit halves ("legacy"). 

784 endianness = "% endianness legacy\n" if self._format == "evt21" else "" 

785 self._fd.write( 

786f"""% camera_integrator_name Prophesee 

787% date {self._formatted_datetime} 

788{endianness}% evt {_EVT_FORMATS[self._format][0]} 

789% format {_EVT_FORMATS[self._format][1]};height={self._height};width={self._width} 

790% generation 4.2 

791% geometry {self._width}x{self._height} 

792% integrator_name Prophesee 

793% plugin_integrator_name Prophesee 

794% plugin_name hal_plugin_prophesee 

795% sensor_generation 4.2 

796% serial_number {self._serial_number} 

797% system_ID {self._system_id} 

798% end 

799""".encode('utf-8')) 

800 self._is_initialized = True 

801 

802 def write(self, events: 'np.ndarray | EventArray', triggers: 'np.ndarray | TriggerArray | None' = None) -> int: 

803 """Write events to the EVT file. 

804 

805 Parameters 

806 ---------- 

807 events : np.ndarray or EventArray 

808 Array of events to write. 

809 

810 Returns 

811 ------- 

812 int 

813 Number of written events. 

814 

815 """ 

816 assert self._fd is not None 

817 

818 if not self._is_initialized: 

819 self.init() 

820 

821 # Accept EventArray transparently (SoA -> AoS for the numba encoder). 

822 if isinstance(events, EventArray): 

823 events = events.to_aos() 

824 

825 if self._format == "evt3": 

826 buffer, self._last_lower12_ts, self._last_upper12_ts, self._last_y = get_raw_evt3_buffer( 

827 events, 

828 self._last_lower12_ts, 

829 self._last_upper12_ts, 

830 self._last_y) 

831 elif self._format == "evt2": 

832 buffer, self._last_ts_high = get_raw_evt2_buffer(events, self._last_ts_high) 

833 elif self._format == "evt21": 

834 buffer, self._last_ts_high = get_raw_evt21_buffer(events, self._last_ts_high) 

835 elif self._format == "evt4": 

836 buffer, self._last_ts_high = get_raw_evt4_buffer(events, self._last_ts_high)