Coverage for src/evutils/io/_aedat.py: 87%
297 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-15 09:33 +0000
1"""AEDAT (jAER / cAER / DV) file decoder, versions 1.0 -- 4.0.
3All four on-disk layouts used by iniVation event cameras are supported for
4reading (writing is not implemented yet):
6* **AEDAT 1.0** (jAER, 2008): optional ``#``-prefixed ASCII header, then
7 6-byte big-endian records: ``uint16`` address + ``uint32`` timestamp (µs).
8 DVS128 address layout: ``p = bit 0``, ``x = bits 1-7``, ``y = bits 8-14``.
9* **AEDAT 2.0** (jAER, 2010): ``#!AER-DAT2.0`` header, then 8-byte big-endian
10 records: ``uint32`` address + ``uint32`` timestamp (µs). The address layout
11 depends on the camera -- see the ``layout`` parameter (default ``"davis"``:
12 ``p = bit 11``, ``x = bits 12-21``, ``y = bits 22-30``; APS/IMU words with
13 bit 31 set are skipped).
14* **AEDAT 3.1** (cAER): header terminated by ``#!END-HEADER``, then
15 little-endian packets with a 28-byte header; polarity-event packets carry
16 8-byte events (``uint32`` data + ``uint32`` timestamp): validity ``bit 0``,
17 ``p = bit 1``, ``y = bits 2-16``, ``x = bits 17-31``. The 31-bit packet
18 timestamp is extended by the header's TS-overflow counter to 64 bits.
19* **AEDAT 4.0** (DV framework): ``#!AER-DAT4.0`` version line, a FlatBuffer
20 ``IOHeader`` (compression type, stream table), then packets of size-prefixed
21 ``EventPacket`` FlatBuffers (identifier ``EVTS``), optionally LZ4- or
22 Zstd-compressed, holding 16-byte event structs (``int64`` t, ``int16`` x,
23 ``int16`` y, ``uint8`` p). Compressed files need the optional ``lz4`` /
24 ``zstandard`` package (``pip install evutils[aedat]``).
26The byte order and record layouts for 1.0/2.0/3.1 follow the official
27iniVation file-format documentation (jAER writes big-endian); the 4.0 layout
28follows dv-processing (cross-checked against the evlib reference reader).
30Decoding streams packet-by-packet / chunk-by-chunk -- the whole recording is
31never materialised.
33References
34----------
35[1] https://docs.inivation.com/software/software-advanced-usage/file-formats/
36[2] https://github.com/tallamjr/evlib (aedat_reader.rs, aedat4_reader.rs)
37"""
38from __future__ import annotations
40import re
41import struct
42from typing import Any, Callable
44import numpy as np
46from ..types import EventArray, TriggerArray
47from .common import EventDecoder, EventEncoder
48from ._source import ByteSource
50_EMPTY_EVENTS = EventArray.empty()
52# ---------------------------------------------------------------------------#
53# Record dtypes
54# ---------------------------------------------------------------------------#
55#: AEDAT 1.0: big-endian uint16 address + uint32 timestamp (6 bytes).
56_V1_DTYPE = np.dtype([("a", ">u2"), ("t", ">u4")])
57#: AEDAT 2.0: big-endian uint32 address + uint32 timestamp (8 bytes).
58_V2_DTYPE = np.dtype([("a", ">u4"), ("t", ">u4")])
59#: AEDAT 3.1 polarity event: little-endian uint32 data + uint32 timestamp.
60_V3_EVENT_DTYPE = np.dtype([("d", "<u4"), ("t", "<u4")])
61#: AEDAT 3.1 packet header (28 bytes, little-endian).
62_V3_HEADER = struct.Struct("<hhiiiiii")
63#: AEDAT 4.0 event struct: int64 t, int16 x, int16 y, uint8 p, 3 pad bytes.
64_V4_EVENT_DTYPE = np.dtype({
65 "names": ["t", "x", "y", "p"],
66 "formats": ["<i8", "<i2", "<i2", "u1"],
67 "offsets": [0, 8, 10, 12],
68 "itemsize": 16,
69})
71_V3_POLARITY_EVENT = 1 # cAER event type id for polarity events
73# AEDAT 2.0 address layouts: name -> (extractor, aps_mask_bit31).
74_V2_LAYOUTS = {
75 # DAVIS (jAER): bit 31 = readout type (1 = APS/IMU, skip), p = bit 11,
76 # x = bits 12-21 (10 bits), y = bits 22-30 (9 bits).
77 "davis": lambda a: (((a >> 12) & 0x3FF), ((a >> 22) & 0x1FF), ((a >> 11) & 0x1)),
78 # DVS128: same address layout as AEDAT 1.0, in the low 16 bits.
79 "dvs128": lambda a: (((a >> 1) & 0x7F), ((a >> 8) & 0x7F), (a & 0x1)),
80}
83# ---------------------------------------------------------------------------#
84# Minimal FlatBuffer field access (AEDAT 4.0). The two schemas involved are
85# tiny and fixed, so the offsets are walked by hand instead of depending on a
86# flatbuffers runtime (cf. the evlib reference reader).
87# ---------------------------------------------------------------------------#
88def _u16(b: bytes, o: int) -> int:
89 return int(struct.unpack_from("<H", b, o)[0])
92def _i32(b: bytes, o: int) -> int:
93 return int(struct.unpack_from("<i", b, o)[0])
96def _u32(b: bytes, o: int) -> int:
97 return int(struct.unpack_from("<I", b, o)[0])
100def _i64(b: bytes, o: int) -> int:
101 return int(struct.unpack_from("<q", b, o)[0])
104def _fb_field_pos(buf: bytes, table: int, index: int) -> int | None:
105 """Absolute position of field ``index`` of the table at ``table``, or None
106 if the field is absent from the vtable."""
107 vtable = table - _i32(buf, table)
108 vtable_size = _u16(buf, vtable)
109 slot = vtable + 4 + 2 * index
110 if slot + 2 > vtable + vtable_size:
111 return None
112 voffset = _u16(buf, slot)
113 return table + voffset if voffset else None
116def _parse_io_header(buf: bytes) -> tuple[int, int, str]:
117 """Parse the AEDAT4 ``IOHeader`` FlatBuffer.
119 Returns ``(compression, data_table_position, info_node_xml)``.
120 Fields: 0 = compression (i32), 1 = dataTablePosition (i64), 2 = infoNode.
121 """
122 table = _u32(buf, 0)
123 compression = 0
124 data_table_position = -1
125 info_node = ""
127 pos = _fb_field_pos(buf, table, 0)
128 if pos is not None:
129 compression = _i32(buf, pos)
130 pos = _fb_field_pos(buf, table, 1)
131 if pos is not None:
132 data_table_position = _i64(buf, pos)
133 pos = _fb_field_pos(buf, table, 2)
134 if pos is not None:
135 str_pos = pos + _u32(buf, pos)
136 str_len = _u32(buf, str_pos)
137 info_node = bytes(buf[str_pos + 4:str_pos + 4 + str_len]).decode("utf-8", "replace")
138 return compression, data_table_position, info_node
141def _parse_event_packet(body: bytes) -> np.ndarray | None:
142 """Extract the event structs from a size-prefixed ``EventPacket`` FlatBuffer.
144 Returns a ``_V4_EVENT_DTYPE`` view, or ``None`` when the body is not an
145 ``EVTS`` packet (frame/IMU/trigger streams).
146 """
147 if len(body) < 12 or body[8:12] != b"EVTS":
148 return None
149 root = 4 + _u32(body, 4)
150 pos = _fb_field_pos(body, root, 0) # single field: `elements` vector
151 if pos is None:
152 return _np_empty_v4()
153 vector = pos + _u32(body, pos)
154 count = _u32(body, vector)
155 start = vector + 4
156 if start + count * _V4_EVENT_DTYPE.itemsize > len(body):
157 raise ValueError("truncated AEDAT4 EventPacket")
158 return np.frombuffer(body, dtype=_V4_EVENT_DTYPE, count=count, offset=start)
161def _np_empty_v4() -> np.ndarray:
162 return np.empty(0, dtype=_V4_EVENT_DTYPE)
165def _parse_streams_xml(xml: str) -> tuple[set[int], int | None, int | None]:
166 """Extract event-stream ids and the sensor geometry from the ``IOHeader``
167 infoNode XML (a DV config tree).
169 Returns ``(event_stream_ids, width, height)``. Parsing is best-effort:
170 on any failure the id set is empty and the caller falls back to
171 identifying event packets by their FlatBuffer identifier.
172 """
173 ids: set[int] = set()
174 width: int | None = None
175 height: int | None = None
176 try:
177 import xml.etree.ElementTree as ET
178 root = ET.fromstring(xml)
179 for node in root.iter("node"):
180 name = node.get("name", "")
181 if not name.lstrip("-").isdigit():
182 continue
183 type_id = None
184 size_x = size_y = None
185 for attr in node.iter("attr"):
186 key = attr.get("key")
187 if key == "typeIdentifier":
188 type_id = (attr.text or "").strip()
189 elif key == "sizeX":
190 size_x = int((attr.text or "0").strip())
191 elif key == "sizeY":
192 size_y = int((attr.text or "0").strip())
193 if type_id == "EVTS":
194 ids.add(int(name))
195 if width is None and size_x:
196 width, height = size_x, size_y
197 except Exception:
198 return set(), None, None
199 return ids, width, height
202def _decompress_lz4(body: bytes) -> bytes:
203 try:
204 import lz4.frame
205 except ImportError as exc:
206 raise ImportError(
207 "This AEDAT4 file uses LZ4-compressed packets: install "
208 "`evutils[aedat]` (or `pip install lz4`)."
209 ) from exc
210 return bytes(lz4.frame.decompress(body))
213def _decompress_zstd(body: bytes) -> bytes:
214 try:
215 from compression import zstd # Python >= 3.14
216 return bytes(zstd.decompress(body))
217 except ImportError:
218 pass
219 try:
220 import zstandard
221 except ImportError as exc:
222 raise ImportError(
223 "This AEDAT4 file uses Zstd-compressed packets: install "
224 "`evutils[aedat]` (or `pip install zstandard`)."
225 ) from exc
226 return bytes(zstandard.ZstdDecompressor().decompress(body))
229#: DV CompressionType enum -> decompressor. LZ4_HIGH/ZSTD_HIGH share decoders.
230_DECOMPRESSORS: dict[int, Callable[[bytes], bytes]] = {
231 0: lambda b: b, # NONE
232 1: _decompress_lz4, # LZ4
233 2: _decompress_lz4, # LZ4_HIGH
234 3: _decompress_zstd, # ZSTD
235 4: _decompress_zstd, # ZSTD_HIGH
236}
239class EventDecoder_Aedat(EventDecoder):
240 """Decode AEDAT 1.0 / 2.0 / 3.1 / 4.0 files into ``EventArray`` chunks.
242 The version is detected from the ``#!AER-DATx.y`` header line (a file
243 with a bare ``#`` header, or none at all, is treated as AEDAT 1.0, per
244 the jAER convention).
246 Parameters
247 ----------
248 source
249 Byte source to read from.
250 chunk_size
251 Maximum number of events produced per :meth:`read_chunk` call
252 (AEDAT 3.1/4.0 packets are never split, so a chunk can exceed this
253 by at most one packet's worth).
254 layout : {"davis", "dvs128"}, default "davis"
255 AEDAT 2.0 address layout (the 2.0 container does not name the
256 camera). Ignored for the other versions.
258 """
260 def __init__(self, source: ByteSource, chunk_size: int = 1_000_000,
261 layout: str = "davis"):
262 super().__init__(source, chunk_size)
263 if layout not in _V2_LAYOUTS:
264 raise ValueError(f"layout must be one of {sorted(_V2_LAYOUTS)}, got {layout!r}")
265 self._layout = layout
267 self._buf: Any = None
268 self._version: int = 0 # 1, 2, 3 or 4 (x10: 31 -> 3, 40 -> 4)
269 self._payload_off: int = 0 # first byte after the ASCII header
270 self._cursor: int = 0 # current byte offset into _buf
272 # v1/v2 32-bit timestamp unwrap state.
273 self._ts_wraps: int = 0
274 self._last_raw_ts: int = -1
276 # v4 packet-region metadata.
277 self._v4_compression: int = 0
278 self._v4_region_end: int = 0
279 self._v4_stream_ids: set[int] = set()
281 # ------------------------------------------------------------------ #
282 # Header
283 # ------------------------------------------------------------------ #
284 def _parse_header(self) -> None:
285 """Detect the version and consume the ASCII header."""
286 buf = self._buf
287 head = bytes(buf[:16])
289 if head.startswith(b"#!AER-DAT4.0"):
290 self._version = 4
291 self._parse_v4_header()
292 return
293 if head.startswith(b"#!AER-DAT3"):
294 self._version = 3
295 elif head.startswith(b"#!AER-DAT2.0"):
296 self._version = 2
297 else:
298 # "#!AER-DAT1.0", a bare "#" header, or no header at all (raw
299 # DVS128 dumps): all AEDAT 1.0 per the jAER convention.
300 self._version = 1
302 # Consume '#'-prefixed header lines; harvest sizeX/sizeY when present.
303 n = len(buf)
304 off = 0
305 header_lines = []
306 while off < n and buf[off] == 0x23: # '#'
307 window = bytes(buf[off:off + 8192])
308 rel = window.find(b"\n")
309 if rel < 0:
310 off = n
311 break
312 line = window[:rel]
313 header_lines.append(line)
314 off += rel + 1
315 if self._version == 3 and line.strip() == b"#!END-HEADER":
316 break
317 self._payload_off = off
319 header_text = b"\n".join(header_lines).decode("ascii", "replace")
320 for key, attr in (("sizeX", "_width"), ("sizeY", "_height")):
321 m = re.search(rf"{key}\s*[=:]?\s*(\d+)", header_text)
322 if m:
323 setattr(self, attr, int(m.group(1)))
325 def _parse_v4_header(self) -> None:
326 """Parse the AEDAT4 version line + IOHeader FlatBuffer."""
327 buf = self._buf
328 # 14-byte version line, then a u32-size-prefixed IOHeader FlatBuffer.
329 header_size = _u32(bytes(buf[14:18]), 0)
330 io_start, io_end = 18, 18 + header_size
331 if io_end > len(buf):
332 raise ValueError("truncated AEDAT4 IOHeader")
333 compression, data_table_pos, info_node = _parse_io_header(bytes(buf[io_start:io_end]))
335 if compression not in _DECOMPRESSORS:
336 raise ValueError(f"unknown AEDAT4 compression type {compression}")
337 self._v4_compression = compression
338 self._payload_off = io_end
339 # Packets occupy the bytes between the IOHeader and the trailing
340 # FileDataTable (when its position is known).
341 if 0 <= data_table_pos <= len(buf):
342 self._v4_region_end = int(data_table_pos)
343 else:
344 self._v4_region_end = len(buf)
346 ids, width, height = _parse_streams_xml(info_node)
347 self._v4_stream_ids = ids
348 if width:
349 self._width = width
350 if height:
351 self._height = height
353 # ------------------------------------------------------------------ #
354 # Lifecycle
355 # ------------------------------------------------------------------ #
356 def init(self) -> None:
357 """Read the header and position the cursor at the first record/packet."""
358 if self._is_initialized:
359 return
361 if self._source.mappable():
362 self._buf = self._source.buffer()
363 else:
364 self._buf = memoryview(self._source.read(-1))
366 self._parse_header()
367 self._cursor = self._payload_off
368 self._is_initialized = True
370 # ------------------------------------------------------------------ #
371 # Per-version batch decoding
372 # ------------------------------------------------------------------ #
373 def _unwrap_ts(self, ts_raw: np.ndarray) -> np.ndarray:
374 """Extend raw 32-bit µs timestamps to int64, accumulating wraps
375 (a drop of more than 2^31 between consecutive timestamps counts as a
376 wrap; smaller decreases are genuine jitter and pass through).
377 """
378 t = ts_raw.astype(np.int64)
379 if len(t) == 0:
380 return t
381 prev = np.empty_like(t)
382 prev[0] = self._last_raw_ts if self._last_raw_ts >= 0 else int(t[0])
383 prev[1:] = t[:-1]
384 wraps = self._ts_wraps + np.cumsum((prev - t) > (1 << 31))
385 self._ts_wraps = int(wraps[-1])
386 self._last_raw_ts = int(t[-1])
387 return t + (wraps << np.int64(32))
389 def _batch_v1_v2(self) -> EventArray | None:
390 """Decode the next ``chunk_size`` fixed-size records (AEDAT 1.0/2.0)."""
391 dtype = _V1_DTYPE if self._version == 1 else _V2_DTYPE
392 remaining = (len(self._buf) - self._cursor) // dtype.itemsize
393 if remaining <= 0:
394 return None
395 count = min(self._chunk_size, remaining)
396 rec = np.frombuffer(self._buf, dtype=dtype, count=count, offset=self._cursor)
397 self._cursor += count * dtype.itemsize
399 a = rec["a"]
400 t = self._unwrap_ts(rec["t"])
401 if self._version == 1:
402 x = (a >> 1) & np.uint16(0x7F)
403 y = (a >> 8) & np.uint16(0x7F)
404 p = (a & np.uint16(0x1)).astype(np.uint8)
405 else:
406 # Skip non-DVS words (APS samples / IMU, flagged by bit 31).
407 dvs = (a >> np.uint32(31)) == 0
408 if not dvs.all():
409 a, t = a[dvs], t[dvs]
410 x, y, p = _V2_LAYOUTS[self._layout](a)
411 p = p.astype(np.uint8)
412 return EventArray(t, x.astype(np.uint16), y.astype(np.uint16), p)
414 def _batch_v3(self) -> EventArray | None:
415 """Decode the next AEDAT 3.1 packet holding polarity events."""
416 buf = self._buf
417 n = len(buf)
418 while self._cursor + _V3_HEADER.size <= n:
419 (ev_type, _source, ev_size, ts_offset, ts_overflow,
420 capacity, number, _valid) = _V3_HEADER.unpack_from(buf, self._cursor)
421 body_start = self._cursor + _V3_HEADER.size
422 body_end = body_start + capacity * ev_size
423 if ev_size <= 0 or body_end > n:
424 # Corrupt/truncated trailing packet: stop cleanly.
425 self._cursor = n
426 return None
427 self._cursor = body_end
429 if ev_type != _V3_POLARITY_EVENT or number == 0:
430 continue # frame / IMU / trigger packet: skip
432 if ev_size != _V3_EVENT_DTYPE.itemsize or ts_offset != 4:
433 raise ValueError(
434 f"unsupported AEDAT3 polarity event layout "
435 f"(eventSize={ev_size}, tsOffset={ts_offset})"
436 )
437 rec = np.frombuffer(buf, dtype=_V3_EVENT_DTYPE, count=number, offset=body_start)
438 d = rec["d"]
439 valid = (d & np.uint32(0x1)) != 0
440 if not valid.all():
441 rec, d = rec[valid], d[valid]
442 if len(rec) == 0:
443 continue
444 # 31-bit in-packet timestamp + 64-bit overflow counter.
445 t = rec["t"].astype(np.int64) + (np.int64(ts_overflow) << np.int64(31))
446 x = ((d >> np.uint32(17)) & np.uint32(0x7FFF)).astype(np.uint16)
447 y = ((d >> np.uint32(2)) & np.uint32(0x7FFF)).astype(np.uint16)
448 p = ((d >> np.uint32(1)) & np.uint32(0x1)).astype(np.uint8)
449 return EventArray(t, x, y, p)
450 self._cursor = n
451 return None
453 def _batch_v4(self) -> EventArray | None:
454 """Decode the next AEDAT4 packet carrying events."""
455 buf = self._buf
456 end = self._v4_region_end
457 decompress = _DECOMPRESSORS[self._v4_compression]
458 while self._cursor + 8 <= end:
459 stream_id = _i32(buf, self._cursor)
460 size = _i32(buf, self._cursor + 4)
461 body_start = self._cursor + 8
462 body_end = body_start + size
463 if size < 0 or body_end > end:
464 # Truncated trailing packet (or we ran into the data table).
465 self._cursor = end
466 return None
467 self._cursor = body_end
469 # When the header named the event streams, filter cheaply by id;
470 # otherwise decompress and check the FlatBuffer identifier.
471 if self._v4_stream_ids and stream_id not in self._v4_stream_ids:
472 continue
473 body = decompress(bytes(buf[body_start:body_end]))
474 rec = _parse_event_packet(body)
475 if rec is None or len(rec) == 0:
476 continue
477 return EventArray(
478 rec["t"],
479 rec["x"].astype(np.uint16),
480 rec["y"].astype(np.uint16),
481 rec["p"].astype(np.uint8),
482 )
483 self._cursor = end
484 return None
486 # ------------------------------------------------------------------ #
487 # EventDecoder interface
488 # ------------------------------------------------------------------ #
489 def read_chunk(self, delta_t_hint: int | None = None,
490 n_events_hint: int | None = None) -> EventArray:
491 if not self._is_initialized:
492 self.init()
494 if self._version in (1, 2):
495 batch = self._batch_v1_v2()
496 elif self._version == 3:
497 batch = self._batch_v3()
498 else:
499 batch = self._batch_v4()
501 if batch is None:
502 self._eof = True
503 return _EMPTY_EVENTS
504 return batch
506 def reset(self) -> None:
507 """Reset the reader to the first event."""
508 self._cursor = self._payload_off
509 self._ts_wraps = 0
510 self._last_raw_ts = -1
511 self._eof = False
513 def tell(self) -> int:
514 """Get the current byte offset.
516 Returns
517 -------
518 int
519 Current byte offset.
521 """
522 return self._cursor
524 def close(self) -> None:
525 """Release the buffer view so the source can be closed."""
526 self._buf = None
529class EventEncoder_Aedat(EventEncoder):
530 """An encoder for writing events to AEDAT files.
532 .. note::
533 Not implemented yet -- constructing this class raises
534 :class:`NotImplementedError`. (Reading AEDAT 1.0--4.0 is supported
535 via :class:`EventDecoder_Aedat`.)
537 Parameters
538 ----------
539 writable : io.BufferedIOBase
540 Destination for writing events.
541 **kwargs
542 Additional encoder arguments (``width``, ``height``, ``dt``, ...).
544 Raises
545 ------
546 NotImplementedError
547 Always, until the format is implemented.
549 """
551 _NOT_IMPLEMENTED = (
552 "Writing AEDAT files is not implemented yet. "
553 "Write a supported format instead (RAW/EVT, DAT, AER, HDF5, NPZ, CSV)."
554 )
556 def __init__(self, writable: Any, **kwargs: Any):
557 raise NotImplementedError(self._NOT_IMPLEMENTED)
559 def init(self) -> None:
560 """Initialize the file for writing."""
561 raise NotImplementedError(self._NOT_IMPLEMENTED)
563 def write(self, events: 'np.ndarray | EventArray', triggers: 'np.ndarray | TriggerArray | None' = None) -> int:
564 """Write a chunk of events."""
565 raise NotImplementedError(self._NOT_IMPLEMENTED)