Coverage for tests/aedat_synth.py: 100%
77 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-15 09:33 +0000
1"""Spec-accurate AEDAT 1.0/2.0/3.1/4.0 file synthesizers.
3Shared between the correctness tests (tests/io/test_aedat.py) and the
4benchmarks. The builders follow the iniVation file-format documentation
5(v1-v3, big-endian jAER layouts) and the DV framework layout (v4,
6hand-built FlatBuffers), so decoders are checked against the spec rather
7than against themselves.
8"""
9import struct
11import numpy as np
12from typing import Any
14#: AEDAT 4.0 event struct layout (16 bytes: int64 t, int16 x, int16 y, uint8 p).
15_V4_EVENT_DTYPE = np.dtype({
16 "names": ["t", "x", "y", "p"],
17 "formats": ["<i8", "<i2", "<i2", "u1"],
18 "offsets": [0, 8, 10, 12],
19 "itemsize": 16,
20})
22# --------------------------------------------------------------------------- #
23# Synthesis helpers
24# --------------------------------------------------------------------------- #
25def make_aedat1(t: Any, x: Any, y: Any, p: Any, header: bool=True) -> bytes:
26 """AEDAT 1.0: '#' header + big-endian (uint16 addr, uint32 ts) records.
27 DVS128 layout: p = bit 0, x = bits 1-7, y = bits 8-14."""
28 out = bytearray()
29 if header:
30 out += b"#!AER-DAT1.0\r\n# This is a raw AE data file\r\n# sizeX 128\r\n# sizeY 128\r\n"
31 for ti, xi, yi, pi in zip(t, x, y, p):
32 addr = (int(yi) << 8) | (int(xi) << 1) | int(pi)
33 out += struct.pack(">HI", addr, int(ti))
34 return bytes(out)
37def make_aedat2(t: Any, x: Any, y: Any, p: Any, aps_every: Any=None) -> bytes:
38 """AEDAT 2.0: '#' header + big-endian (uint32 addr, uint32 ts) records.
39 DAVIS layout: p = bit 11, x = bits 12-21, y = bits 22-30
40 bit 31 = APS."""
41 out = bytearray(b"#!AER-DAT2.0\r\n# sizeX 240\r\n# sizeY 180\r\n")
42 for i, (ti, xi, yi, pi) in enumerate(zip(t, x, y, p)):
43 addr = (int(yi) << 22) | (int(xi) << 12) | (int(pi) << 11)
44 out += struct.pack(">II", addr, int(ti))
45 if aps_every and (i + 1) % aps_every == 0:
46 # Interleave an APS word (bit 31 set) that must be skipped.
47 out += struct.pack(">II", 0x8000_0000 | addr, int(ti))
48 return bytes(out)
51def _aedat3_packet(ev_type: int, records: list[bytes]) -> bytes:
52 """One AEDAT 3.1 packet: 28-byte LE header + raw records."""
53 body = b"".join(records)
54 ev_size = len(body) // max(len(records), 1) if records else 8
55 return struct.pack(
56 "<hhiiiiii", ev_type, 0, ev_size, 4,
57 getattr(_aedat3_packet, "ts_overflow", 0), len(records), len(records), len(records),
58 ) + body
61setattr(_aedat3_packet, "ts_overflow", 0)
64def make_aedat3(t: Any, x: Any, y: Any, p: Any, events_per_packet: int=4, ts_overflow: int=0) -> bytes:
65 """AEDAT 3.1: header through '#!END-HEADER' + polarity-event packets.
66 Event: uint32 data (valid bit 0, p bit 1, y bits 2-16, x bits 17-31)
67 + uint32 timestamp, little-endian."""
68 setattr(_aedat3_packet, "ts_overflow", ts_overflow)
69 out = bytearray(
70 b"#!AER-DAT3.1\r\n#Format: RAW\r\n#Source 1: Test sizeX 346 sizeY 260\r\n#!END-HEADER\r\n"
71 )
72 recs = [
73 struct.pack("<II", (int(xi) << 17) | (int(yi) << 2) | (int(pi) << 1) | 1, int(ti))
74 for ti, xi, yi, pi in zip(t, x, y, p)
75 ]
76 # An IMU packet (type 3) that must be skipped.
77 out += _aedat3_packet(3, [b"\x00" * 8] * 2)
78 for i in range(0, len(recs), events_per_packet):
79 out += _aedat3_packet(1, recs[i:i + events_per_packet])
80 return bytes(out)
83def _fb_event_packet(t: Any, x: Any, y: Any, p: Any) -> bytes:
84 """Hand-built size-prefixed EventPacket FlatBuffer (identifier EVTS)."""
85 n = len(t)
86 rec = np.zeros(n, dtype=_V4_EVENT_DTYPE)
87 rec["t"] = t
88 rec["x"] = x
89 rec["y"] = y
90 rec["p"] = p
91 events = rec.tobytes()
92 # Layout (offsets relative to buffer start):
93 # 0: u32 size prefix 12: vtable [6, 8, 4] 20: table
94 # 4: u32 root offset (rel 4) 18: padding 28: vector
95 # 8: 'EVTS'
96 buf = bytearray()
97 buf += struct.pack("<I", 0) # size prefix (patched below)
98 buf += struct.pack("<I", 16) # root table offset, relative to pos 4
99 buf += b"EVTS"
100 buf += struct.pack("<HHH", 6, 8, 4) # vtable: size 6, table size 8, field0 @4
101 buf += b"\x00\x00" # pad to 4-aligned table at 20
102 buf += struct.pack("<i", 8) # soffset: table(20) - vtable(12)
103 buf += struct.pack("<I", 4) # field0: vector offset rel to pos 24
104 buf += struct.pack("<I", n) # vector length
105 buf += events
106 struct.pack_into("<I", buf, 0, len(buf) - 4)
107 return bytes(buf)
110def _fb_io_header(compression: int, data_table_pos: int, info_node: bytes) -> bytes:
111 """Hand-built IOHeader FlatBuffer (not size-prefixed)."""
112 buf = bytearray()
113 buf += struct.pack("<I", 16) # root table offset
114 buf += struct.pack("<HHHHH", 10, 20, 4, 8, 16) # vtable @4: 3 fields
115 buf += b"\x00\x00" # pad, table at 16
116 buf += struct.pack("<i", 12) # soffset: table(16) - vtable(4)
117 buf += struct.pack("<i", compression) # field 0 @20 (voffset 4)
118 buf += struct.pack("<q", data_table_pos) # field 1 @24 (voffset 8)
119 buf += struct.pack("<I", 36 - 32) # field 2 @32: string @36
120 buf += struct.pack("<I", len(info_node)) + info_node + b"\x00"
121 return bytes(buf)
124_V4_INFO_XML = (
125 b'<dv version="2.0"><node name="outInfo" path="/outInfo/">'
126 b'<node name="0" path="/outInfo/0/">'
127 b'<attr key="typeIdentifier" type="string">EVTS</attr>'
128 b'<node name="info" path="/outInfo/0/info/">'
129 b'<attr key="sizeX" type="int">640</attr>'
130 b'<attr key="sizeY" type="int">480</attr>'
131 b"</node></node></node></dv>"
132)
135def make_aedat4(t: Any, x: Any, y: Any, p: Any, events_per_packet: int=5, compression: int=0, info: bytes=_V4_INFO_XML) -> bytes:
136 """AEDAT 4.0: version line + IOHeader FlatBuffer + (StreamID, Size)-framed
137 EventPacket FlatBuffers."""
138 io_header = _fb_io_header(compression, -1, info)
139 out = bytearray(b"#!AER-DAT4.0\r\n")
140 out += struct.pack("<I", len(io_header)) + io_header
141 for i in range(0, len(t), events_per_packet):
142 s = slice(i, i + events_per_packet)
143 body = _fb_event_packet(t[s], x[s], y[s], p[s])
144 if compression in (1, 2):
145 import lz4.frame
146 body = lz4.frame.compress(body)
147 out += struct.pack("<ii", 0, len(body)) + body
148 return bytes(out)