Coverage for src/evutils/io/common.py: 81%
129 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-18 05:24 +0000
1"""Common interface for event decoders and encoders.
3Defines the abstract base classes `EventDecoder` and `EventEncoder`.
4"""
6import io
7from abc import ABC, abstractmethod
8from datetime import datetime
9from typing import Optional, TYPE_CHECKING
11if TYPE_CHECKING:
12 from typing import Any
14from ..types import EventArray, TriggerArray
15import numpy as np
17from typing import NamedTuple
19class SeekResult(NamedTuple):
20 """Result of a seek operation: the exact landing timestamp, event index, and EOF status."""
21 ts: int
22 index: int
23 eof: bool
25class EventDecoder(ABC):
26 """ABC for reading chunks of events from a IO source object.
28 Parameters
29 ----------
30 readable
31 source to read events from
32 chunk_size
33 Size of the chunk to read
35 Raises
36 ------
37 NotImplementedError
38 If the method is not implemented in the subclass
40 """
42 # ------------------------------------------------------------------ #
43 # Capability contract
44 #
45 # These class attributes declare which optional fast paths / features a
46 # decoder supports. They all default to the conservative value here so a
47 # decoder that does not override one degrades to the safe (correct, maybe
48 # slower) path, and so the EventReader can read them directly instead of
49 # probing with getattr(..., default) -- a missing attribute then surfaces
50 # as a loud AttributeError rather than silently selecting the slow path.
51 # Subclasses override the ones they implement.
52 # ------------------------------------------------------------------ #
54 #: Decoder can decode external trigger packets alongside CD events.
55 SUPPORTS_EXT_TRIGGERS = False
57 #: Whether this decoder implements timestamp / event-index random access via
58 #: :meth:`seek`. Off by default; each seekable format opts in. The
59 #: EventReader additionally requires the underlying ByteSource to be
60 #: seekable before delegating a real (non-linear) seek here.
61 SUPPORTS_SEEK = False
63 #: Parser emits exactly one event per input record, so an output buffer
64 #: fills to precisely its capacity -- enables EventReader's zero-copy
65 #: n_events fast path. False for vectorised formats (EVT3/2.1/4).
66 _exact_window = False
68 #: read_chunk already returns slices of at most ``n_events`` events, so the
69 #: reader can hand them straight through without re-accumulating (NPZ/HDF5).
70 _independent_windows = False
72 #: The decoder buffers the entire payload in memory at init (slurping the
73 #: whole stream), so its :meth:`seek` works even when the underlying
74 #: ByteSource is not itself seekable -- e.g. reading over a compressed
75 #: stream. The EventReader uses this to allow the decoder's fast-path seek
76 #: instead of a linear scan. True only for decoders that genuinely slurp.
77 _buffers_in_memory = False
79 #: A dedicated C delta_t parser exists (one GIL-free call decodes a whole
80 #: time window). EVT3 overrides this as a property.
81 _has_delta_t_parser = False
83 #: Seek-index wiring, injected by EventReader from its ``index=`` option.
84 #: Only EVT consults them; harmless defaults for every other decoder.
85 #: ``_use_sidecar`` reads a Metavision ``.tmp_index``; ``_persist_index``
86 #: saves/loads evutils' own exact index to a ``.evidx`` sidecar.
87 _use_sidecar = False
88 _persist_index = False
89 _raw_path: "str | None" = None
91 def __init__(self, source: "io.BufferedIOBase | str | bytes", chunk_size: int = 10000, read_external_triggers: bool = False):
92 """Initialize the decoder.
94 Parameters
95 ----------
96 source : ByteSource
97 Source to read events from.
98 chunk_size : int, optional
99 Size of the chunk to read, by default 10000.
100 read_external_triggers : bool, optional
101 Whether to read external triggers, by default False.
103 """
104 # `source` is a ByteSource (see io/_source.py). `fd` is kept as a legacy
105 # alias for older decoders that still reference it.
106 self._source = source
107 self._fd = source
109 self._is_initialized = False
111 self._chunk_size = chunk_size
113 self._eof = False
115 # Corrupt-packet policy (see EventReader(strict=...)): when True, a
116 # malformed packet raises instead of being skipped with a warning.
117 self._strict: bool = False
119 self._width: int | None = None
120 self._height: int | None = None
122 self.read_external_triggers = read_external_triggers
123 if self.read_external_triggers and not self.SUPPORTS_EXT_TRIGGERS:
124 import warnings
125 warnings.warn(f"{self.__class__.__name__} does not support reading external triggers.")
127 @abstractmethod
128 def init(self) -> None:
129 """Initialize the file for reading."""
130 raise NotImplementedError
132 @abstractmethod
133 def read_chunk(self, delta_t_hint:int | None = None, n_events_hint:int | None = None) -> 'EventArray | tuple[EventArray, TriggerArray]':
134 """Read a chunk of events.
136 Parameters
137 ----------
138 delta_t_hint : int, optional
139 If not None, can be used to provide a hit about the delta_t window to be read
140 n_events_hint : int, optional
141 If not None, can be used to provide a hit about the n_events to be read
143 Returns
144 -------
145 EventArray or tuple of (EventArray, TriggerArray)
146 Chunk of events read from the source. Optionally returns triggers if `read_external_triggers` is True.
148 """
149 raise NotImplementedError
151 @abstractmethod
152 def reset(self) -> None:
153 """Reset the file pointer to the beginning of the file."""
154 raise NotImplementedError
156 def read_all(self) -> 'EventArray | tuple[EventArray, TriggerArray]':
157 """Decode and return every remaining event at once.
159 The default implementation drains :meth:`read_chunk` and concatenates the
160 chunks. SoA-native decoders (EVT/DAT/AER) override this with a
161 single-buffer decode that avoids the per-chunk copy entirely.
163 Returns
164 -------
165 EventArray
166 All remaining events.
168 """
169 if not self._is_initialized:
170 self.init()
172 # read_chunk may return a view that is invalidated by the next call, so
173 # copy each chunk before pulling the next one.
174 chunks = []
175 trigger_chunks = []
176 while True:
177 _chunk = self.read_chunk()
178 if self.read_external_triggers:
179 if isinstance(_chunk, tuple):
180 ev_chunk, tr_chunk = _chunk
181 else:
182 ev_chunk = _chunk
183 tr_chunk = TriggerArray.empty()
184 else:
185 ev_chunk = _chunk # type: ignore
186 if len(ev_chunk) == 0 and (not self.read_external_triggers or len(tr_chunk) == 0):
187 break
188 if len(ev_chunk) > 0:
189 chunks.append(ev_chunk.copy())
190 if self.read_external_triggers and len(tr_chunk) > 0:
191 trigger_chunks.append(tr_chunk.copy())
193 if not chunks:
194 res_ev = EventArray.empty()
195 elif len(chunks) == 1:
196 res_ev = chunks[0]
197 else:
198 res_ev = EventArray(
199 np.concatenate([c.t for c in chunks]),
200 np.concatenate([c.x for c in chunks]),
201 np.concatenate([c.y for c in chunks]),
202 np.concatenate([c.p for c in chunks]),
203 )
205 if self.read_external_triggers:
206 if not trigger_chunks:
207 res_tr = TriggerArray.empty()
208 elif len(trigger_chunks) == 1:
209 res_tr = trigger_chunks[0]
210 else:
211 res_tr = TriggerArray(
212 np.concatenate([c.t for c in trigger_chunks]),
213 np.concatenate([c.p for c in trigger_chunks]),
214 np.concatenate([c.id for c in trigger_chunks]),
215 )
216 return res_ev, res_tr
218 return res_ev
220 def close(self) -> None:
221 """Release any resources held by the decoder (e.g. buffer views).
223 The owning source is closed separately by the EventReader.
224 """
225 pass
227 def tell(self) -> int:
228 """Get the current position in the file.
230 Returns
231 -------
232 int
233 The current position in the file
235 """
236 return int(self._fd.tell())
238 @staticmethod
239 def _seek_axis(t: int | None, n: int | None) -> "tuple[str, int]":
240 """Validate the (t, n) pair and return the chosen axis and value.
242 Returns ``("t", value)`` or ``("n", value)``; raises ``ValueError`` if
243 neither or both were given.
244 """
245 if (t is None) == (n is None):
246 raise ValueError("seek() requires exactly one of t= or n=.")
247 if t is not None:
248 return "t", int(t)
249 return "n", int(n) # type: ignore[arg-type]
251 def seek(self, t: int | None = None, n: int | None = None) -> tuple[SeekResult, "EventArray", "TriggerArray | None"]:
252 """Reposition the decoder to an absolute timestamp or event index.
254 Exactly one of ``t`` (microseconds) or ``n`` (0-based event index) must
255 be given. After a successful seek the next :meth:`read_chunk` /
256 :meth:`read_all` yields events starting at the first event whose
257 timestamp is >= ``t`` (or the exact event at index ``n``).
259 Parameters
260 ----------
261 t : int, optional
262 Target timestamp in microseconds.
263 n : int, optional
264 False).
265 ValueError
266 If neither or both of ``t``/``n`` are provided.
267 """
268 raise NotImplementedError(
269 f"{self.__class__.__name__} does not support seek()."
270 )
272 def set_chunk_size(self, chunk_size:int) -> None:
273 """Set the chunk size.
275 Parameters
276 ----------
277 chunk_size
278 Size of the chunk to read
280 """
281 self._chunk_size = chunk_size
283 def shape(self) -> tuple[int|None, int|None]:
284 """Get the shape of the frame (width, height).
286 Returns
287 -------
288 tuple[int|None, int|None]
289 The shape of the frame (width, height), or (None, None) if the shape is not known
291 """
292 return self._width, self._height
295 def __repr__(self) -> str:
296 if self._is_initialized:
297 is_initialized_txt = "initialized"
298 else:
299 is_initialized_txt = "not initialized"
300 return f"{self.__class__} - {is_initialized_txt}"
302 def is_eof(self) -> bool:
303 """Check if the end of the file has been reached.
305 Returns
306 -------
307 bool
308 True if the end of the file has been reached
310 """
311 return self._eof
313class EventEncoder(ABC):
314 """ABC for writing chunks of events to a io object.
316 Parameters
317 ----------
318 writable
319 Destination for writing events
320 width : int, optional
321 Width of the frame, by default 1280 (not relevant for some formats)
322 height : int, optional
323 Height of the frame, by default 720 (not relevant for some formats)
324 dt : datetime, optional
325 Timestamp of the recording (default is the current time, but information is not saved in all formats)
327 Raises
328 ------
329 NotImplementedError
330 If the method is not implemented in the subclass
332 """
334 #: Whether this encoder can write external triggers. No encoder implements
335 #: trigger encoding yet; EventWriter warns (once) when triggers are passed
336 #: to an encoder without support, instead of dropping them silently.
337 SUPPORTS_WRITE_TRIGGERS = False
339 def __init__(self, writable: io.BufferedIOBase, width:int = 1280, height:int = 720, dt:Optional[datetime]=None ):
340 """Initialize the encoder.
342 Parameters
343 ----------
344 writable : io.BufferedIOBase
345 Destination for writing events.
346 width : int, optional
347 Width of the frame, by default 1280.
348 height : int, optional
349 Height of the frame, by default 720.
350 dt : datetime, optional
351 Timestamp of the recording, by default current time.
353 """
354 self._fd = writable
356 self._width = width
357 self._height = height
359 self._n_written_events = 0
360 self._is_initialized = False
362 if dt is None:
363 self._dt = datetime.now()
364 else:
365 self._dt = dt
367 @abstractmethod
368 def init(self) -> None:
369 """Initialize the file for writing."""
370 raise NotImplementedError
372 @abstractmethod
373 def write(self, events: 'np.ndarray | EventArray', triggers: 'np.ndarray | TriggerArray | None' = None) -> int:
374 """Write a chunk of events."""
375 raise NotImplementedError
377 def __len__(self) -> int:
378 return self._n_written_events
380 def __enter__(self) -> "EventEncoder":
381 return self
383 def __repr__(self) -> str:
384 if self._is_initialized:
385 is_initialized_txt = f"Written {self._n_written_events} events"
386 else:
387 is_initialized_txt = "not initialized"
388 return f"{self.__class__.__name__} - {is_initialized_txt}, {self._width}x{self._height}"
390 def flush(self) -> None:
391 """Flush any buffered data to the underlying stream."""
392 self._fd.flush()
394 def close(self) -> None:
395 """Finalize the encoder.
397 Container formats (NPZ, HDF5) override this to write the archive /
398 index before the underlying file is closed. The owning stream itself is
399 closed by the :class:`~evutils.io.EventWriter`, not here.
400 """
401 self.flush()