Coverage for src/evutils/io/_aedat.py: 77%
430 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-18 05:24 +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 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}
82# ---------------------------------------------------------------------------#
83# Minimal FlatBuffer field access (AEDAT 4.0). The two schemas involved are
84# tiny and fixed, so the offsets are walked by hand instead of depending on a
85# flatbuffers runtime (cf. the evlib reference reader).
86# ---------------------------------------------------------------------------#
87def _u16(b: bytes, o: int) -> int:
88 return int(struct.unpack_from("<H", b, o)[0])
90def _i32(b: bytes, o: int) -> int:
91 return int(struct.unpack_from("<i", b, o)[0])
93def _u32(b: bytes, o: int) -> int:
94 return int(struct.unpack_from("<I", b, o)[0])
96def _i64(b: bytes, o: int) -> int:
97 return int(struct.unpack_from("<q", b, o)[0])
99def _fb_field_pos(buf: bytes, table: int, index: int) -> int | None:
100 """Absolute position of field ``index`` of the table at ``table``, or None
101 if the field is absent from the vtable."""
102 vtable = table - _i32(buf, table)
103 vtable_size = _u16(buf, vtable)
104 slot = vtable + 4 + 2 * index
105 if slot + 2 > vtable + vtable_size:
106 return None
107 voffset = _u16(buf, slot)
108 return table + voffset if voffset else None
110def _parse_io_header(buf: bytes) -> tuple[int, int, str]:
111 """Parse the AEDAT4 ``IOHeader`` FlatBuffer.
113 Returns ``(compression, data_table_position, info_node_xml)``.
114 Fields: 0 = compression (i32), 1 = dataTablePosition (i64), 2 = infoNode.
115 """
116 table = _u32(buf, 0)
117 compression = 0
118 data_table_position = -1
119 info_node = ""
121 pos = _fb_field_pos(buf, table, 0)
122 if pos is not None:
123 compression = _i32(buf, pos)
124 pos = _fb_field_pos(buf, table, 1)
125 if pos is not None:
126 data_table_position = _i64(buf, pos)
127 pos = _fb_field_pos(buf, table, 2)
128 if pos is not None:
129 str_pos = pos + _u32(buf, pos)
130 str_len = _u32(buf, str_pos)
131 info_node = bytes(buf[str_pos + 4:str_pos + 4 + str_len]).decode("utf-8", "replace")
132 return compression, data_table_position, info_node
134def _parse_event_packet(body: bytes) -> np.ndarray | None:
135 """Extract the event structs from a size-prefixed ``EventPacket`` FlatBuffer.
137 Returns a ``_V4_EVENT_DTYPE`` view, or ``None`` when the body is not an
138 ``EVTS`` packet (frame/IMU/trigger streams).
139 """
140 if len(body) < 12 or body[8:12] != b"EVTS":
141 return None
142 root = 4 + _u32(body, 4)
143 pos = _fb_field_pos(body, root, 0) # single field: `elements` vector
144 if pos is None:
145 return _np_empty_v4()
146 vector = pos + _u32(body, pos)
147 count = _u32(body, vector)
148 start = vector + 4
149 if start + count * _V4_EVENT_DTYPE.itemsize > len(body):
150 raise ValueError("truncated AEDAT4 EventPacket")
151 return np.frombuffer(body, dtype=_V4_EVENT_DTYPE, count=count, offset=start)
153def _np_empty_v4() -> np.ndarray:
154 return np.empty(0, dtype=_V4_EVENT_DTYPE)
156def _parse_streams_xml(xml: str) -> tuple[set[int], int | None, int | None]:
157 """Extract event-stream ids and the sensor geometry from the ``IOHeader``
158 infoNode XML (a DV config tree).
160 Returns ``(event_stream_ids, width, height)``. Parsing is best-effort:
161 on any failure the id set is empty and the caller falls back to
162 identifying event packets by their FlatBuffer identifier.
163 """
164 ids: set[int] = set()
165 width: int | None = None
166 height: int | None = None
167 try:
168 import xml.etree.ElementTree as ET
169 root = ET.fromstring(xml)
170 for node in root.iter("node"):
171 name = node.get("name", "")
172 if not name.lstrip("-").isdigit():
173 continue
174 type_id = None
175 size_x = size_y = None
176 for attr in node.iter("attr"):
177 key = attr.get("key")
178 if key == "typeIdentifier":
179 type_id = (attr.text or "").strip()
180 elif key == "sizeX":
181 size_x = int((attr.text or "0").strip())
182 elif key == "sizeY":
183 size_y = int((attr.text or "0").strip())
184 if type_id == "EVTS":
185 ids.add(int(name))
186 if width is None and size_x:
187 width, height = size_x, size_y
188 except Exception:
189 return set(), None, None
190 return ids, width, height
192def _decompress_lz4(body: bytes) -> bytes:
193 try:
194 import lz4.frame
195 except ImportError as exc:
196 raise ImportError(
197 "This AEDAT4 file uses LZ4-compressed packets: install "
198 "`evutils[aedat]` (or `pip install lz4`)."
199 ) from exc
200 return bytes(lz4.frame.decompress(body))
202def _decompress_zstd(body: bytes) -> bytes:
203 try:
204 from compression import zstd # Python >= 3.14
205 return bytes(zstd.decompress(body))
206 except ImportError:
207 pass
208 try:
209 import zstandard
210 except ImportError as exc:
211 raise ImportError(
212 "This AEDAT4 file uses Zstd-compressed packets: install "
213 "`evutils[aedat]` (or `pip install zstandard`)."
214 ) from exc
215 return bytes(zstandard.ZstdDecompressor().decompress(body))
217#: DV CompressionType enum -> decompressor. LZ4_HIGH/ZSTD_HIGH share decoders.
218_DECOMPRESSORS: dict[int, Callable[[bytes], bytes]] = {
219 0: lambda b: b, # NONE
220 1: _decompress_lz4, # LZ4
221 2: _decompress_lz4, # LZ4_HIGH
222 3: _decompress_zstd, # ZSTD
223 4: _decompress_zstd, # ZSTD_HIGH
224}
226class EventDecoder_Aedat(EventDecoder):
227 SUPPORTS_EXT_TRIGGERS = True
229 #: init() slurps the whole payload into memory (or mmaps it).
230 _buffers_in_memory = True
231 """Decode AEDAT 1.0 / 2.0 / 3.1 / 4.0 files into ``EventArray`` chunks.
233 The version is detected from the ``#!AER-DATx.y`` header line (a file
234 with a bare ``#`` header, or none at all, is treated as AEDAT 1.0, per
235 the jAER convention).
237 Parameters
238 ----------
239 source
240 Byte source to read from.
241 chunk_size
242 Maximum number of events produced per :meth:`read_chunk` call
243 (AEDAT 3.1/4.0 packets are never split, so a chunk can exceed this
244 by at most one packet's worth).
245 layout : {"davis", "dvs128"}, default "davis"
246 AEDAT 2.0 address layout (the 2.0 container does not name the
247 camera). Ignored for the other versions.
249 """
251 def __init__(self, source: ByteSource, chunk_size: int = 1_000_000,
252 layout: str = "davis"):
253 super().__init__(source, chunk_size)
254 if layout not in _V2_LAYOUTS:
255 raise ValueError(f"layout must be one of {sorted(_V2_LAYOUTS)}, got {layout!r}")
256 self._layout = layout
258 self._buf: bytes | bytearray | None = None
259 self._version: int = 0 # 1, 2, 3 or 4 (x10: 31 -> 3, 40 -> 4)
260 self._payload_off: int = 0 # first byte after the ASCII header
261 self._cursor: int = 0 # current byte offset into _buf
263 # v1/v2 32-bit timestamp unwrap state.
264 self._ts_wraps: int = 0
265 self._last_raw_ts: int = -1
267 # v4 packet-region metadata.
268 self._v4_compression: int = 0
269 self._v4_region_end: int = 0
270 self._v4_stream_ids: set[int] = set()
272 # ------------------------------------------------------------------ #
273 # Header
274 # ------------------------------------------------------------------ #
275 def _parse_header(self) -> None:
276 """Detect the version and consume the ASCII header."""
277 buf = self._buf
278 head = bytes(buf[:16])
280 if head.startswith(b"#!AER-DAT4.0"):
281 self._version = 4
282 self._parse_v4_header()
283 return
284 if head.startswith(b"#!AER-DAT3"):
285 self._version = 3
286 elif head.startswith(b"#!AER-DAT2.0"):
287 self._version = 2
288 else:
289 # "#!AER-DAT1.0", a bare "#" header, or no header at all (raw
290 # DVS128 dumps): all AEDAT 1.0 per the jAER convention.
291 self._version = 1
293 # Consume '#'-prefixed header lines; harvest sizeX/sizeY when present.
294 n = len(buf)
295 off = 0
296 header_lines = []
297 while off < n and buf[off] == 0x23: # '#'
298 window = bytes(buf[off:off + 8192])
299 rel = window.find(b"\n")
300 if rel < 0:
301 off = n
302 break
303 line = window[:rel]
304 header_lines.append(line)
305 off += rel + 1
306 if self._version == 3 and line.strip() == b"#!END-HEADER":
307 break
308 self._payload_off = off
310 header_text = b"\n".join(header_lines).decode("ascii", "replace")
311 for key, attr in (("sizeX", "_width"), ("sizeY", "_height")):
312 m = re.search(rf"{key}\s*[=:]?\s*(\d+)", header_text)
313 if m:
314 setattr(self, attr, int(m.group(1)))
316 def _parse_v4_header(self) -> None:
317 """Parse the AEDAT4 version line + IOHeader FlatBuffer."""
318 buf = self._buf
319 # 14-byte version line, then a u32-size-prefixed IOHeader FlatBuffer.
320 header_size = _u32(bytes(buf[14:18]), 0)
321 io_start, io_end = 18, 18 + header_size
322 if io_end > len(buf):
323 raise ValueError("truncated AEDAT4 IOHeader")
324 compression, data_table_pos, info_node = _parse_io_header(bytes(buf[io_start:io_end]))
326 if compression not in _DECOMPRESSORS:
327 raise ValueError(f"unknown AEDAT4 compression type {compression}")
328 self._v4_compression = compression
329 self._payload_off = io_end
330 # Packets occupy the bytes between the IOHeader and the trailing
331 # FileDataTable (when its position is known).
332 if 0 <= data_table_pos <= len(buf):
333 self._v4_region_end = int(data_table_pos)
334 else:
335 self._v4_region_end = len(buf)
337 ids, width, height = _parse_streams_xml(info_node)
338 self._v4_stream_ids = ids
339 if width:
340 self._width = width
341 if height:
342 self._height = height
344 # ------------------------------------------------------------------ #
345 # Lifecycle
346 # ------------------------------------------------------------------ #
347 def init(self) -> None:
348 """Read the header and position the cursor at the first record/packet."""
349 if self._is_initialized:
350 return
352 if self._source.mappable():
353 self._buf = self._source.buffer()
354 else:
355 self._buf = memoryview(self._source.read(-1))
357 self._parse_header()
358 self._cursor = self._payload_off
359 self._is_initialized = True
361 # ------------------------------------------------------------------ #
362 # Per-version batch decoding
363 # ------------------------------------------------------------------ #
364 def _unwrap_ts(self, ts_raw: np.ndarray) -> np.ndarray:
365 """Extend raw 32-bit µs timestamps to int64, accumulating wraps
366 (a drop of more than 2^31 between consecutive timestamps counts as a
367 wrap; smaller decreases are genuine jitter and pass through).
368 """
369 t = ts_raw.astype(np.int64)
370 if len(t) == 0:
371 return t
372 prev = np.empty_like(t)
373 prev[0] = self._last_raw_ts if self._last_raw_ts >= 0 else int(t[0])
374 prev[1:] = t[:-1]
375 wraps = self._ts_wraps + np.cumsum((prev - t) > (1 << 31))
376 self._ts_wraps = int(wraps[-1])
377 self._last_raw_ts = int(t[-1])
378 return t + (wraps << np.int64(32))
380 def _batch_v1_v2(self) -> tuple[EventArray, TriggerArray] | None:
381 """Decode the next ``chunk_size`` fixed-size records (AEDAT 1.0/2.0)."""
382 from ..types import TriggerArray
383 dtype = _V1_DTYPE if self._version == 1 else _V2_DTYPE
384 remaining = (len(self._buf) - self._cursor) // dtype.itemsize
385 if remaining <= 0:
386 return None
387 count = min(self._chunk_size, remaining)
388 rec = np.frombuffer(self._buf, dtype=dtype, count=count, offset=self._cursor)
389 self._cursor += count * dtype.itemsize
391 a = rec["a"]
392 t = self._unwrap_ts(rec["t"])
393 if self._version == 1:
394 x = (a >> 1) & np.uint16(0x7F)
395 y = (a >> 8) & np.uint16(0x7F)
396 p = (a & np.uint16(0x1)).astype(np.uint8)
397 events = EventArray(t, x.astype(np.uint16), y.astype(np.uint16), p)
398 triggers = TriggerArray.empty()
399 else:
400 # Skip non-DVS words (APS samples / IMU, flagged by bit 31).
401 dvs = (a >> np.uint32(31)) == 0
402 if not dvs.all():
403 a, t = a[dvs], t[dvs]
405 # Bits 11-10: 00=OFF, 10=ON, 01/11=External Event
406 subtype = (a >> np.uint32(10)) & np.uint32(0x3)
407 is_trigger = (subtype == 1) | (subtype == 3)
408 is_event = ~is_trigger
410 a_ev, t_ev = a[is_event], t[is_event]
411 x, y, p = _V2_LAYOUTS[self._layout](a_ev)
412 events = EventArray(t_ev, x.astype(np.uint16), y.astype(np.uint16), p.astype(np.uint8))
414 a_tr, t_tr = a[is_trigger], t[is_trigger]
415 tr_p = ((a_tr >> np.uint32(11)) & np.uint32(0x1)).astype(np.uint8)
416 triggers = TriggerArray(t_tr, tr_p, np.zeros_like(tr_p))
418 return events, triggers
420 def _batch_v3(self) -> tuple[EventArray, TriggerArray] | None:
421 """Decode the next AEDAT 3.1 packet holding polarity or trigger events."""
422 from ..types import TriggerArray
423 buf = self._buf
424 n = len(buf)
425 while self._cursor + _V3_HEADER.size <= n:
426 (ev_type, _source, ev_size, ts_offset, ts_overflow,
427 capacity, number, _valid) = _V3_HEADER.unpack_from(buf, self._cursor)
428 body_start = self._cursor + _V3_HEADER.size
429 body_end = body_start + capacity * ev_size
430 if ev_size <= 0 or body_end > n:
431 self._cursor = n
432 return None
433 self._cursor = body_end
435 if number == 0:
436 continue
438 if ev_type == _V3_POLARITY_EVENT:
439 if ev_size != _V3_EVENT_DTYPE.itemsize or ts_offset != 4:
440 raise ValueError(f"unsupported AEDAT3 polarity event layout (eventSize={ev_size})")
441 rec = np.frombuffer(buf, dtype=_V3_EVENT_DTYPE, count=number, offset=body_start)
442 d = rec["d"]
443 valid = (d & np.uint32(0x1)) != 0
444 if not valid.all():
445 rec, d = rec[valid], d[valid]
446 if len(rec) == 0:
447 continue
448 t = rec["t"].astype(np.int64) + (np.int64(ts_overflow) << np.int64(31))
449 x = ((d >> np.uint32(17)) & np.uint32(0x7FFF)).astype(np.uint16)
450 y = ((d >> np.uint32(2)) & np.uint32(0x7FFF)).astype(np.uint16)
451 p = ((d >> np.uint32(1)) & np.uint32(0x1)).astype(np.uint8)
452 return EventArray(t, x, y, p), TriggerArray.empty()
454 elif ev_type == 0: # SPECIAL_EVENT
455 rec = np.frombuffer(buf, dtype=_V3_EVENT_DTYPE, count=number, offset=body_start)
456 d = rec["d"]
457 valid = (d & np.uint32(0x1)) != 0
458 if not valid.all():
459 rec, d = rec[valid], d[valid]
460 type_id = (d >> np.uint32(1)) & np.uint32(0x7F)
461 is_ext = (type_id >= 2) & (type_id <= 13)
462 if is_ext.any():
463 tr_rec, tr_d, tr_type = rec[is_ext], d[is_ext], type_id[is_ext]
464 tr_t = tr_rec["t"].astype(np.int64) + (np.int64(ts_overflow) << np.int64(31))
465 tr_p = (tr_type % 2 == 0).astype(np.uint8) # Even types are rising (1), odd are falling (0)
466 return _EMPTY_EVENTS, TriggerArray(tr_t, tr_p, np.zeros_like(tr_p))
467 self._cursor = n
468 return None
470 def _batch_v4(self) -> tuple[EventArray, TriggerArray] | None:
471 """Decode the next AEDAT4 packet carrying events or triggers."""
472 from ..types import TriggerArray
473 buf = self._buf
474 end = self._v4_region_end
475 decompress = _DECOMPRESSORS[self._v4_compression]
476 while self._cursor + 8 <= end:
477 stream_id = _i32(buf, self._cursor)
478 size = _i32(buf, self._cursor + 4)
479 body_start = self._cursor + 8
480 body_end = body_start + size
481 if size < 0 or body_end > end:
482 self._cursor = end
483 return None
484 self._cursor = body_end
486 # Unconditionally decompress if we're reading triggers (don't skip non-EVTS streams just yet)
487 body = decompress(bytes(buf[body_start:body_end]))
489 if len(body) >= 12 and body[8:12] == b"EVTS":
490 rec = _parse_event_packet(body)
491 if rec is not None and len(rec) > 0:
492 return EventArray(
493 rec["t"],
494 rec["x"].astype(np.uint16),
495 rec["y"].astype(np.uint16),
496 rec["p"].astype(np.uint8),
497 ), TriggerArray.empty()
498 elif len(body) >= 12 and body[8:12] == b"TRIG":
499 # Trigger packet parsing:
500 # Trigger struct: int64 t, int8 type, 7 pad bytes = 16 bytes.
501 # Types: 0=TIMESTAMP_RESET, 1=EXTERNAL_INPUT_RISING_EDGE, 2=EXTERNAL_INPUT_FALLING_EDGE, 3=EXTERNAL_INPUT_PULSE, etc.
502 root = 4 + _u32(body, 4)
503 pos = _fb_field_pos(body, root, 0)
504 if pos is not None:
505 vector = pos + _u32(body, pos)
506 count = _u32(body, vector)
507 start = vector + 4
508 if start + count * 16 <= len(body):
509 tr_rec = np.frombuffer(body, dtype=np.dtype([("t", "<i8"), ("type", "i1"), ("pad", "V7")]), count=count, offset=start)
510 t = tr_rec["t"]
511 type_id = tr_rec["type"]
512 # Rising edge is typically 1 (odd), falling edge is 2 (even).
513 # Let's map it simply:
514 tr_p = (type_id % 2 != 0).astype(np.uint8)
515 return _EMPTY_EVENTS, TriggerArray(t, tr_p, np.zeros_like(tr_p))
516 self._cursor = end
517 return None
519 # ------------------------------------------------------------------ #
520 # EventDecoder interface
521 # ------------------------------------------------------------------ #
522 def read_chunk(self, delta_t_hint: int | None = None,
523 n_events_hint: int | None = None) -> 'EventArray | tuple[EventArray, TriggerArray]':
524 from ..types import TriggerArray
525 if not self._is_initialized:
526 self.init()
528 if self._version in (1, 2):
529 batch = self._batch_v1_v2()
530 elif self._version == 3:
531 batch = self._batch_v3()
532 else:
533 batch = self._batch_v4()
535 if batch is None:
536 self._eof = True
537 if self.read_external_triggers:
538 return _EMPTY_EVENTS, TriggerArray.empty()
539 return _EMPTY_EVENTS
541 events, triggers = batch
542 if self.read_external_triggers:
543 return events, triggers
544 return events
546 def reset(self) -> None:
547 """Reset the reader to the first event."""
548 self._cursor = self._payload_off
549 self._ts_wraps = 0
550 self._last_raw_ts = -1
551 self._eof = False
553 def tell(self) -> int:
554 """Get the current byte offset.
556 Returns
557 -------
558 int
559 Current byte offset.
561 """
562 return self._cursor
564 def close(self) -> None:
565 """Release the buffer view so the source can be closed."""
566 self._buf = None
568def _fb_io_header(compression: int, data_table_pos: int, info_node: bytes) -> bytes:
569 """Hand-built IOHeader FlatBuffer (not size-prefixed)."""
570 import struct
571 buf = bytearray()
572 buf += struct.pack("<I", 16) # root table offset
573 buf += struct.pack("<HHHHH", 10, 20, 4, 8, 16) # vtable @4: 3 fields
574 buf += b"\x00\x00" # pad, table at 16
575 buf += struct.pack("<i", 12) # soffset: table(16) - vtable(4)
576 buf += struct.pack("<i", compression) # field 0 @20 (voffset 4)
577 buf += struct.pack("<q", data_table_pos) # field 1 @24 (voffset 8)
578 buf += struct.pack("<I", 36 - 32) # field 2 @32: string @36
579 buf += struct.pack("<I", len(info_node)) + info_node + b"\x00"
580 return bytes(buf)
582def _fb_event_packet(events: 'EventArray') -> bytes:
583 """Hand-built size-prefixed EventPacket FlatBuffer (identifier EVTS)."""
584 import struct
585 n = len(events)
586 # AEDAT 4.0 event struct layout: int64 t, int16 x, int16 y, uint8 p, 3 pad bytes
587 rec = np.zeros(n, dtype=_V4_EVENT_DTYPE)
588 rec["t"] = events["t"]
589 rec["x"] = events["x"]
590 rec["y"] = events["y"]
591 rec["p"] = events["p"]
592 events_bytes = rec.tobytes()
594 buf = bytearray()
595 buf += struct.pack("<I", 0) # size prefix (patched below)
596 buf += struct.pack("<I", 16) # root table offset, relative to pos 4
597 buf += b"EVTS"
598 buf += struct.pack("<HHH", 6, 8, 4) # vtable: size 6, table size 8, field0 @4
599 buf += b"\x00\x00" # pad to 4-aligned table at 20
600 buf += struct.pack("<i", 8) # soffset: table(20) - vtable(12)
601 buf += struct.pack("<I", 4) # field0: vector offset rel to pos 24
602 buf += struct.pack("<I", n) # vector length
603 buf += events_bytes
604 struct.pack_into("<I", buf, 0, len(buf) - 4)
605 return bytes(buf)
607def _fb_trigger_packet(triggers: 'TriggerArray') -> bytes:
608 """Hand-built size-prefixed TriggerPacket FlatBuffer (identifier TRIG)."""
609 import struct
610 n = len(triggers)
611 _V4_TRIGGER_DTYPE = np.dtype({
612 "names": ["t", "type"],
613 "formats": ["<i8", "i1"],
614 "offsets": [0, 8],
615 "itemsize": 16,
616 })
617 rec = np.zeros(n, dtype=_V4_TRIGGER_DTYPE)
618 rec["t"] = triggers["t"]
619 # 1=EXTERNAL_INPUT_RISING_EDGE, 2=EXTERNAL_INPUT_FALLING_EDGE
620 rec["type"] = np.where(triggers["p"], 1, 2)
621 triggers_bytes = rec.tobytes()
623 buf = bytearray()
624 buf += struct.pack("<I", 0) # size prefix (patched below)
625 buf += struct.pack("<I", 16) # root table offset, relative to pos 4
626 buf += b"TRIG"
627 buf += struct.pack("<HHH", 6, 8, 4) # vtable: size 6, table size 8, field0 @4
628 buf += b"\x00\x00" # pad to 4-aligned table at 20
629 buf += struct.pack("<i", 8) # soffset: table(20) - vtable(12)
630 buf += struct.pack("<I", 4) # field0: vector offset rel to pos 24
631 buf += struct.pack("<I", n) # vector length
632 buf += triggers_bytes
633 struct.pack_into("<I", buf, 0, len(buf) - 4)
634 return bytes(buf)
636class EventEncoder_Aedat(EventEncoder):
637 """Encoder for AEDAT 4.0 files."""
639 SUPPORTS_WRITE_TRIGGERS = True
641 def __init__(self, writable, width:int = 1280, height:int = 720, dt=None, compression: int = 0, **kwargs):
642 super().__init__(writable, width, height, dt)
643 self._compression = compression # 0=NONE, 1=LZ4, 2=LZ4_HIGH, 3=ZSTD, 4=ZSTD_HIGH
645 def init(self) -> None:
646 if self._is_initialized:
647 return
649 import struct
651 # Version line
652 self._fd.write(b"#!AER-DAT4.0\r\n")
654 # IOHeader XML Config
655 info_xml = f'<?xml version="1.0" encoding="UTF-8"?><node name="info"><node name="0"><attr key="typeIdentifier" type="string">EVTS</attr><attr key="sizeX" type="int">{self._width}</attr><attr key="sizeY" type="int">{self._height}</attr></node><node name="1"><attr key="typeIdentifier" type="string">TRIG</attr></node></node>'.encode("utf-8")
657 io_header = _fb_io_header(self._compression, -1, info_xml)
658 self._fd.write(struct.pack("<I", len(io_header)) + io_header)
660 self._is_initialized = True
662 def write(self, events, triggers = None) -> int:
663 if not self._is_initialized:
664 self.init()
666 if len(events) == 0 and (triggers is None or len(triggers) == 0):
667 return 0
669 import struct
671 if len(events) > 0:
672 body = _fb_event_packet(events)
673 if self._compression in (1, 2):
674 import lz4.frame
675 body = lz4.frame.compress(body)
676 elif self._compression in (3, 4):
677 import zstandard
678 ctx = zstandard.ZstdCompressor(level=3 if self._compression == 3 else 10)
679 body = ctx.compress(body)
681 # Write Packet header: StreamID (0), Size, Body
682 self._fd.write(struct.pack("<iI", 0, len(body)) + body)
683 self._n_written_events += len(events)
685 if triggers is not None and len(triggers) > 0:
686 tr_body = _fb_trigger_packet(triggers)
687 if self._compression in (1, 2):
688 import lz4.frame
689 tr_body = lz4.frame.compress(tr_body)
690 elif self._compression in (3, 4):
691 import zstandard
692 ctx = zstandard.ZstdCompressor(level=3 if self._compression == 3 else 10)
693 tr_body = ctx.compress(tr_body)
694 self._fd.write(struct.pack("<iI", 1, len(tr_body)) + tr_body)
696 return len(events)