Coverage for src/evutils/io/_aer.py: 83%
127 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-18 05:24 +0000
1"""Prophesee AER CD-event decoder/encoder.
3AER is a raw 32-bit-per-event encoding with **no header and no timestamps**:
4``y[0:8]`` (9 bits), ``x[9:17]`` (9 bits), ``p[18]``. The 9-bit fields cap
5coordinates at 512 (e.g. GenX320). Decoding uses the native
6``AER_parse_chunk_soa``; encoding is vectorised numpy.
8Since the format carries no time information, the decoder's ``timestamps``
9parameter selects how the ``t`` column is generated:
11* ``"zero"`` (default) -- every event gets ``t = 0``;
12* ``"sequential"`` -- ``t = t_start + i * t_step``, generated in the native
13 parser and carried across chunks;
14* an array -- user-provided timestamps, assigned positionally (event ``i`` in
15 the stream gets ``timestamps[i]``).
16"""
17from __future__ import annotations
19import io
20from datetime import datetime
22import numpy as np
24from ..types import EventArray, TriggerArray
25from .common import EventDecoder, EventEncoder
26from ._native_core import (
27 EventSoABuffers,
28 TriggerSoABuffers,
29 decode_all_soa,
30 events_view,
31 parse_step,
32)
33from ._native_aer import (
34 AER_TS_SEQUENTIAL,
35 AER_TS_ZERO,
36 AerInput,
37 AerParser,
38)
39from ._source import ByteSource
41_EMPTY_EVENTS = EventArray.empty()
43class EventDecoder_AER(EventDecoder):
44 """Decode raw AER streams into ``EventArray`` chunks. Since AER is designed
45 for real-time streaming, it has no header and no timestamps; see the
46 ``timestamps`` parameter for how the ``t`` column is generated.
48 Parameters
49 ----------
50 source
51 Byte source to read from.
52 chunk_size
53 Maximum number of events produced per :meth:`read_chunk` call (the
54 native output-buffer capacity). Does not bound the file size.
55 timestamps : {"zero", "sequential"} or array_like, default "zero"
56 Timestamp generation mode: ``"zero"`` fills ``t = 0``,
57 ``"sequential"`` fills ``t = t_start + i * t_step``, and an integer
58 array assigns user-provided timestamps positionally (its length must
59 cover every decoded event).
60 t_start, t_step : int
61 Start value and per-event increment for ``"sequential"`` mode.
63 References
64 ----------
65 [1] Prophesee AER format: https://docs.prophesee.ai/stable/data/encoding_formats/aer.html
67 """
69 #: AER is exactly one event per uint32 word, so the parser fills an output
70 #: buffer to precisely its capacity -> eligible for EventReader's zero-copy
71 #: n_events fast path (via parse_step).
72 _exact_window = True
74 #: init() slurps the whole payload into memory (or mmaps it).
75 _buffers_in_memory = True
77 def __init__(self, source: ByteSource, chunk_size: int = 1_000_000,
78 timestamps: 'str | np.ndarray' = "zero",
79 t_start: int = 0, t_step: int = 1):
80 super().__init__(source, chunk_size)
81 self._buf: bytes | bytearray | None = None
82 self._words: "np.ndarray | None" = None # uint32 view of the payload (1 word / event)
83 self._offset: int = 0
84 self._parser: "Callable | None" = None
85 self._events: "EventArray | None" = None
86 self._triggers: "TriggerArray | None" = None
88 self._custom_ts: np.ndarray | None = None
89 if isinstance(timestamps, str):
90 modes = {"zero": AER_TS_ZERO, "sequential": AER_TS_SEQUENTIAL}
91 if timestamps not in modes:
92 raise ValueError(
93 f"timestamps must be 'zero', 'sequential' or an array, got {timestamps!r}"
94 )
95 self._ts_mode = modes[timestamps]
96 else:
97 ts = np.ascontiguousarray(timestamps, dtype=np.int64)
98 if ts.ndim != 1:
99 raise ValueError("custom timestamps must be a 1-D array")
100 self._custom_ts = ts
101 self._ts_mode = AER_TS_ZERO # parser fills 0; overwritten below
102 self._t_start = int(t_start)
103 self._t_step = int(t_step)
104 self._n_decoded = 0 # stream position, for indexing the custom array
106 def init(self) -> None:
107 """Initialize the AER reader.
109 Returns
110 -------
111 None
113 """
114 if self._is_initialized:
115 return
117 if self._source.mappable():
118 self._buf = self._source.buffer()
119 else:
120 self._buf = memoryview(self._source.read(-1))
122 n_events = len(self._buf) // 4 # AER has no header, 4 bytes / event
123 if n_events > 0:
124 self._words = np.frombuffer(self._buf, dtype=np.uint32, count=n_events)
125 else:
126 self._words = np.empty(0, dtype=np.uint32)
128 if self._custom_ts is not None and len(self._custom_ts) < n_events:
129 raise ValueError(
130 f"custom timestamps array has {len(self._custom_ts)} entries, "
131 f"but the stream contains {n_events} events"
132 )
134 self._offset = 0
135 self._parser = AerParser(self._ts_mode, self._t_start, self._t_step)
136 self._input_cls = AerInput
137 self._word_dtype = np.uint32
138 cap = int(self._chunk_size)
139 self._events = EventSoABuffers(cap)
140 self._triggers = TriggerSoABuffers(1) # AER has no triggers
141 self._is_initialized = True
143 def _apply_custom_ts(self, t_out: np.ndarray) -> None:
144 """Overwrite ``t_out`` (the decoded slice's t column, int64/uint64 view)
145 with the next ``len(t_out)`` user-provided timestamps.
146 """
147 assert self._custom_ts is not None
148 n = len(t_out)
149 t_out.view(np.int64)[:] = self._custom_ts[self._n_decoded:self._n_decoded + n]
151 def parse_step(self, events: EventSoABuffers, triggers: TriggerSoABuffers) -> int:
152 """Run the parser once, appending into ``events``; advance the offset.
153 See :meth:`EventDecoder_EVT.parse_step`.
154 """
155 if not self._is_initialized:
156 self.init()
157 if self._words is None or self._offset >= len(self._words):
158 self._eof = True
159 return 0
160 appended, self._offset = parse_step(
161 self._words, self._offset, AerInput, self._parser, events, triggers,
162 )
163 if appended and self._custom_ts is not None:
164 self._apply_custom_ts(events.t[events.size - appended:events.size])
165 self._n_decoded += appended
166 if self._offset >= len(self._words):
167 self._eof = True
168 return appended
170 def read_chunk(self, delta_t_hint: int | None = None,
171 n_events_hint: int | None = None) -> EventArray:
172 if not self._is_initialized:
173 self.init()
175 if self._words is None or self._offset >= len(self._words):
176 self._eof = True
177 return _EMPTY_EVENTS
179 ev, tr = self._events, self._triggers
180 ev.reset()
181 tr.reset()
182 appended = 0
183 while appended == 0 and self._offset < len(self._words):
184 appended = self.parse_step(ev, tr)
186 n = ev.size
187 if n == 0:
188 return _EMPTY_EVENTS
189 # Zero-copy view (valid until the next read_chunk); see EVT decoder.
190 return events_view(ev)
192 def read_all(self) -> EventArray:
193 """Decode the whole remaining payload into one buffer (no per-chunk copy)."""
194 if not self._is_initialized:
195 self.init()
196 if self._words is None or self._offset >= len(self._words):
197 self._eof = True
198 return _EMPTY_EVENTS
199 start = self._n_decoded
200 # Exactly one event per uint32 word.
201 out, self._offset = decode_all_soa(
202 self._words, self._offset, AerInput, self._parser,
203 est_events_per_word=1.0,
204 )
205 self._n_decoded = start + len(out)
206 if self._custom_ts is not None and len(out) > 0:
207 out.t[:] = self._custom_ts[start:start + len(out)]
208 self._eof = True
209 return out
211 def reset(self) -> None:
212 """Reset the AER reader to the beginning.
214 Returns
215 -------
216 None
218 """
219 self._offset = 0
220 self._n_decoded = 0
221 self._eof = False
222 if self._parser is not None:
223 self._parser.reset()
225 def tell(self) -> int:
226 """Get the current byte offset.
228 Returns
229 -------
230 int
231 Current byte offset.
233 """
234 return self._offset * 4
236 def close(self) -> None:
237 """Close the AER reader.
239 Returns
240 -------
241 None
243 """
244 self._words = None
245 self._buf = None
247class EventEncoder_AER(EventEncoder):
248 """Encode events into a raw AER stream. Since AER is designed for real-time
249 streaming, it has no header and no timestamps.
250 Timestamps are dropped and coordinates are masked to 9 bits (values >= 512
251 are truncated), per the AER encoding.
253 Parameters
254 ----------
255 writable
256 Destination stream to write to.
257 width, height : int
258 Frame geometry written into the header.
259 dt : datetime, optional
260 No effect, since AER has no timestamps.
262 References
263 ----------
264 [1] Prophesee AER format: https://docs.prophesee.ai/stable/data/encoding_formats/aer.html
266 """
268 def __init__(self, writable: io.BufferedWriter, width: int = 512, height: int = 512, dt: datetime | None = None):
269 super().__init__(writable, width, height, dt)
271 def init(self) -> None:
272 """Initialize the AER writer.
274 Returns
275 -------
276 None
278 """
279 self._is_initialized = True # AER has no header
281 def write(self, events: 'np.ndarray | EventArray', triggers: 'np.ndarray | TriggerArray | None' = None) -> int:
282 """Write events to the AER file.
284 Parameters
285 ----------
286 events : np.ndarray or EventArray
287 Array of events to write.
289 Returns
290 -------
291 int
292 Number of written events.
294 """
295 if not self._is_initialized:
296 self.init()
298 if isinstance(events, EventArray):
299 x, y, p = events.x, events.y, events.p
300 else:
301 x, y, p = events["x"], events["y"], events["p"]
303 out = (
304 (y.astype(np.uint32) & np.uint32(0x1FF))
305 | ((x.astype(np.uint32) & np.uint32(0x1FF)) << np.uint32(9))
306 | ((p.astype(np.uint32) & np.uint32(0x1)) << np.uint32(18))
307 )
308 self._fd.write(out.astype(np.uint32).tobytes())
309 self._n_written_events += len(out)
310 return len(out)