Coverage for src/evutils/io/_event_reader.py: 89%
654 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-18 05:24 +0000
1"""Event reader module.
3Provides the `EventReader` class for reading and decoding event data
4from a file or byte stream.
5"""
7import io
8import time
9from pathlib import Path
10from typing import TYPE_CHECKING, Any, Optional
12import numpy as np
14from ..types import EventArray, TriggerArray
16_EMPTY_EVENTS = EventArray.empty()
17from . import decoders as ev_decoders
18from ._native_core import (EVUTILS_PARSE_OUTPUT_FULL,
19 EVUTILS_PARSE_WINDOW_DONE, EventSoABuffers,
20 TriggerSoABuffers, events_view)
21from ._prefetch import PrefetchIterator
22from ._source import ByteSource, make_source
23from .buffer import EventAccumulator
24from .common import SeekResult
27class EventReader():
28 """Class for reading and automatically decoding and slicing events from a file or stream.
30 The reader supports different modes of operation, including reading a fixed number of events, reading events within a time window, or a combination of both.
31 The file_decoder is chosen automatically based on the file format or can be supplied explicitly.
33 Parameters
34 ----------
35 file: Path or str or io.BufferedReader or bytes or ByteSource
36 Path to the data file or a readable stream or bytes or a ByteSource.
37 Compressed paths (``.gz`` / ``.zst`` / ``.xz`` / ``.bz2``, e.g.
38 ``foo.raw.zst``) are decompressed transparently; the inner extension
39 selects the format. An already-open compressed file object
40 (``gzip.GzipFile`` / ``lzma.LZMAFile`` / ``compression.zstd.ZstdFile``)
41 may also be passed directly.
42 delta_t: int or optional
43 Time window in microseconds, by default None
44 n_events: int or None
45 Number of events to read in a chunk, by default None
46 max_events: int, default=10_000_000
47 Maximum number of events to read at once
48 mode: {'delta_t', 'n_events', 'mixed', 'all', 'auto'}, default 'auto'
49 Mode of operation ```["delta_t", "n_events", "mixed", "all", "auto"]```
50 start_ts: int, default=0
51 Start timestamp offset for the events, by default 0 (start of the file)
52 normalize_ts: bool, default=False
53 Normalize timestamps to start from zero
54 max_time: int, default=1_000_000_000_000
55 Maximum timestamp to read
56 width: int or None
57 Width of the frame, by default inferred from the file
58 height: int or None
59 Height of the frame, by default inferred from the file
60 ext_trigger: bool, default=False
61 Whether to read external trigger events, by default False
62 async_read: bool, default=False
63 Decode ahead in a background thread while iterating: the next window
64 is parsed while the caller processes the current one (the native
65 parsers release the GIL). Helps whenever per-chunk processing takes
66 meaningful time (numpy pipelines, GPU inference -- the read becomes
67 essentially free); can slightly hurt when the processing already
68 saturates memory bandwidth. Affects iteration only; ``read()`` /
69 ``read_all()`` stay synchronous and raise while an asynchronous
70 iterator is active.
71 prefetch_depth: int or None, default=None
72 How many decoded windows the ``async_read`` worker may buffer ahead of
73 the consumer (default 2 when ``None``). Deeper queues cost memory
74 (depth x window size) but ride out I/O stalls: for real-time playback
75 from a cold file, a depth of ~6 absorbs multi-window disk page-fault
76 hitches that a depth of 2 cannot. Ignored without ``async_read``.
77 reuse_buffers: bool, default=False
78 Recycle the decode buffers of ``delta_t`` windows (openeb-style ring)
79 instead of allocating a fresh buffer per window. Large windows make the
80 per-window allocation the dominant cost (page-faulting hundreds of MB
81 every window); with recycling the buffer pages stay warm and delta_t
82 reads run at ~the raw decoder ceiling. The returned array then
83 **aliases** a recycled buffer: it is valid until ``prefetch_depth``
84 further windows have been read (or one window, without ``async_read``)
85 -- copy it if you keep it longer, exactly like ``EventStreamer``
86 chunks. Off by default: every window is an independent array.
87 real_time: bool, default=False
88 Pace :meth:`read` and iteration to the recording's own timeline: each
89 chunk is released only once the wall-clock time since the first chunk
90 matches the event time it covers, so the stream plays back as if live.
91 Pacing is anchored to an absolute start time, so time spent processing
92 a chunk is absorbed automatically (no drift); if decoding or
93 processing falls behind, chunks are released immediately with no added
94 delay. ``read_all()`` is never paced. Combine with ``async_read=True``
95 to decode ahead while waiting out the pacing delay.
96 playback_speed: float, default=1.0
97 Playback rate multiplier for ``real_time`` mode: ``2.0`` plays twice
98 as fast as the recording, ``0.5`` at half speed. Ignored when
99 ``real_time=False``.
100 max_gap: float or None, default=1.0
101 Longest idle stretch, in real seconds, that ``real_time`` pacing will
102 sleep through. If a recording goes silent for longer (e.g. a
103 late-starting stream, or a pause mid-recording), the pacer skips the
104 dead air and resumes immediately instead of blocking on one long
105 sleep. ``None`` disables the skip (strict real time). Ignored when
106 ``real_time=False``.
107 index: str or bool, default="auto"
108 Seek-index policy for random access (:meth:`seek`). ``"auto"``/``True``/
109 ``False`` all build an exact in-memory index lazily on the first seek
110 (``False`` additionally never touches a sidecar). ``"metavision"`` reads
111 a Metavision ``.tmp_index`` sidecar instead -- fast, but approximate near
112 large event gaps. ``"persist"`` builds evutils' own exact index on the
113 first seek and writes it to a ``<raw>.evidx`` sidecar, then loads that
114 sidecar on subsequent opens instead of rebuilding (exact and reused
115 across runs). Only EVT formats use an index; others seek by record math
116 or ``searchsorted``.
117 strict: bool, default=False
118 Corrupt-packet policy. When False (default), a malformed packet is
119 skipped with a :class:`UserWarning` and decoding resumes -- a robust
120 decoder (like Metavision's UNRELIABLE mode). When True, a malformed
121 packet raises instead (a SAFE decoder). Warnings can be captured with
122 :func:`warnings.catch_warnings`.
123 file_decoder: ev_decoders.EventDecoder or type[ev_decoders.EventDecoder] or None, default=None
124 File decoder to use, by default None - automatic
125 **kwargs
126 Additional arguments to pass to the file decoder
128 Raises
129 ------
130 ValueError
131 If the mode is not supported or if the delta_t or n_events are not specified when needed
133 Examples
134 --------
135 >>> from evutils.io import EventReader
136 >>> # Read events in chunks of 10ms (10,000 microseconds)
137 >>> reader = EventReader(b"% format EVT3;height=720;width=1280\\n% end\\n", delta_t=10000)
138 >>> for events in reader:
139 ... # Access fields directly
140 ... x, y, p, t = events.x, events.y, events.p, events.t
141 ... print(f"Processed {len(events)} events")
142 Processed 0 events
143 """
145 READING_MODES = ["delta_t", "n_events", "mixed", "all", "auto"]
146 DEFAULT_N_EVENTS = 1_000_000
147 DEFAULT_DELTA_T = 10_000
148 def __init__(self, file: Path | str | io.BufferedReader | bytes | ByteSource,
149 delta_t:int|None=None,
150 n_events:int|None=None,
151 mode:str="auto",
152 start_ts:int=0,
153 normalize_ts: bool=False,
154 max_time:int=1_000_000_000_000,
155 max_events:int=10_000_000,
156 width:int | None=None, height:int | None=None,
157 ext_trigger: bool=False,
158 async_read: bool=False,
159 prefetch_depth: int | None=None,
160 reuse_buffers: bool=False,
161 real_time: bool=False,
162 playback_speed: float=1.0,
163 max_gap: float | None=1.0,
164 index: "str | bool" = "auto",
165 strict: bool=False,
166 batch_mode: bool=False,
167 file_decoder: ev_decoders.EventDecoder | type[ev_decoders.EventDecoder] | None = None,
168 **kwargs) -> None:
170 # Remember the path (if any) for repr / reset semantics.
171 self._file_name: Path | None = Path(file) if isinstance(file, (str, Path)) else None
172 self._read_external_triggers = ext_trigger
173 self._batch_mode = batch_mode
175 # 1. Normalise the input into a ByteSource (path | stream | bytes |
176 # BytesIO | ByteSource -> ByteSource). Regular files are memory-mapped.
177 self._source: ByteSource = make_source(file)
179 # 2. Resolve the decoder and launch it:
180 # explicit instance > explicit class > heuristic (extension, then
181 # content sniffing of the source).
182 if isinstance(file_decoder, ev_decoders.EventDecoder):
183 self._file_decoder = file_decoder
184 else:
185 decoder_cls = file_decoder or ev_decoders.resolve_decoder_cls(self._source)
186 self._file_decoder = decoder_cls(self._source, **kwargs)
188 self._file_decoder.read_external_triggers = self._read_external_triggers
189 # Corrupt-packet policy: strict=True re-raises the decoder's WARNING
190 # (a malformed packet) instead of skipping it. Default is robust
191 # (warn + skip + resume), like Metavision's UNRELIABLE decoder.
192 self._file_decoder._strict = strict
193 if self._read_external_triggers and not self._file_decoder.SUPPORTS_EXT_TRIGGERS:
194 import warnings
195 warnings.warn(f"{self._file_decoder.__class__.__name__} does not support reading external triggers.")
197 # This will now be io.BufferedReader
198 self._eof = False
200 # If not defined explicitly, the width and height are fetch from the file (not all formats support this)
201 self._width = width
202 self._height = height
203 self._start_ts = start_ts # Normalization origin: with normalize_ts=True the earliest event's timestamp becomes start_ts. NOT a seek -- use seek() to skip to a time.
204 self._first_ts = 0 # First timestamp in the file, used for normalization
205 self._current_ts = self._first_ts
206 self._normalize_ts = normalize_ts # Normalize timestamps to start from zero
208 # Validate the parameters
209 if mode not in EventReader.READING_MODES:
210 raise ValueError(f"Mode {mode} not supported. Supported modes are: {EventReader.READING_MODES}")
212 self._mode = mode.lower()
214 # if mode is auto, we will try to infer the mode from the parameters
215 if self._mode == "auto":
216 # If both delta_t and n_events are specified, we will use mixed mode
217 if delta_t is not None and n_events is not None:
218 self._mode = "mixed"
220 # If only one of the parameters is specified, we will use that mode, the other will be set to the maximum
221 elif delta_t is not None:
222 self._mode = "delta_t"
223 n_events = max_events
224 elif n_events is not None:
225 self._mode = "n_events"
226 delta_t = max_time
227 else:
228 # If none of the parameters are specified, we will use the default Values
229 self._mode = "mixed"
230 delta_t = self.DEFAULT_DELTA_T
231 n_events = self.DEFAULT_N_EVENTS
233 # If the mode is not auto, we will check if the parameters are specified
234 elif self._mode == "delta_t":
235 if delta_t is None:
236 raise ValueError("delta_t must be specified")
237 if n_events is None:
238 n_events = max_events
239 elif self._mode == "n_events":
240 if n_events is None:
241 raise ValueError("n_events must be specified")
242 if delta_t is None:
243 delta_t = max_time
244 elif self._mode == "mixed":
245 if delta_t is None:
246 delta_t = self.DEFAULT_DELTA_T
247 if n_events is None:
248 n_events = self.DEFAULT_N_EVENTS
250 elif self._mode == "all":
251 delta_t = max_time
252 n_events = max_events
254 # Validate the parameters
255 if delta_t is None:
256 delta_t = self.DEFAULT_DELTA_T
257 if n_events is None:
258 n_events = self.DEFAULT_N_EVENTS
260 if not isinstance(delta_t, int):
261 raise TypeError("delta_t must be an integer")
263 if not isinstance(n_events, int):
264 raise TypeError("n_events must be an integer")
266 if delta_t <= 0:
267 raise ValueError("delta_t must be positive")
269 if n_events <= 0:
270 raise ValueError("n_events must be positive")
272 # delta_t and n_events to read on each call
273 self._delta_t = delta_t
274 self._n_events = n_events
276 # Maximum number of events to read and maximum time to read in a chunk
277 self._max_events = max_events if max_events < self._n_events else self._n_events
278 self._max_time = max_time if max_time < self._delta_t else self._delta_t
280 self._is_initialized = False
282 # Windowed reads decode straight into this reused accumulator (no
283 # intermediate copy); only the returned window is copied out. Allocated
284 # lazily on first read() so the read_all() fast path never pays for it.
285 # Granularity of a single decode step, and a native fast path flag.
286 self._buffer: EventAccumulator | None = None
287 self._step = 1 << 20
288 # Staging capacity: one window plus a decode step of overshoot. Kept
289 # deliberately small (and capped at a few steps) so the parser writes
290 # into a cache/TLB-warm working set -- sizing it to the full max_events
291 # up front inflated the buffer ~4x for the common 1M-event window and
292 # roughly halved decode throughput. It grows on demand (EventAccumulator)
293 # if an actual window -- or drain-to-EOF mode -- needs more, so this only
294 # sets the starting point.
295 self._acc_capacity = min(self._n_events, 4 * self._step) + 2 * self._step
296 self._native_fill = hasattr(self._file_decoder, "parse_step")
298 # delta_t fast-path state (see _read_delta_t_fast): the overshoot carried
299 # from the previous window (events past its time boundary), a rolling
300 # window-size estimate to size the next buffer, and a throwaway trigger
301 # sink. Only used in pure delta_t mode with a native-fill decoder.
302 self._dt_carry: EventSoABuffers | None = None
303 self._dt_est = self._step
304 self._dt_tr: TriggerSoABuffers | None = None
305 # The fast path decodes in smaller steps than the accumulator so the
306 # window's trailing overshoot (the only part copied) stays small; too
307 # large and the carry copy rivals the full-window copy it replaces.
308 self._dt_step = 1 << 17
309 # Buffer recycling for delta_t windows (reuse_buffers=True): a small
310 # ring of persistent decode buffers, cycled per window so their pages
311 # stay warm and no per-window allocation/page-fault cost enters the hot
312 # path. Ring size covers every window that can be alive at once: the
313 # consumer's current one plus everything buffered by an async prefetch.
314 self._reuse_buffers = bool(reuse_buffers)
315 self._dt_slots: list[EventSoABuffers] = []
316 self._dt_slot_i = 0
318 # Asynchronous iteration: when enabled, __iter__ decodes ahead in a
319 # worker thread (see io/_prefetch.py). The active iterator owns the
320 # decoder until it is exhausted or closed.
321 self._async_read = async_read
322 if prefetch_depth is not None and prefetch_depth < 1:
323 raise ValueError("prefetch_depth must be >= 1")
324 self._prefetch_depth = prefetch_depth
325 self._active_prefetch: PrefetchIterator | None = None
327 # Real-time playback: chunks are released against an absolute
328 # wall-clock anchor set when the first chunk is delivered, so the
329 # stream is paced like a live sensor (see _pace()).
330 if not isinstance(playback_speed, (int, float)) or playback_speed <= 0:
331 raise ValueError("playback_speed must be a positive number")
332 self._real_time = real_time
333 self._playback_speed = float(playback_speed)
334 # Longest idle stretch (real seconds) the pacer will sleep through. If a
335 # window would require a longer wait -- a silent gap in the recording,
336 # e.g. a late-starting stream -- the pacer skips the dead air and
337 # re-anchors so playback resumes immediately. None = strict real time.
338 if max_gap is not None and (not isinstance(max_gap, (int, float)) or max_gap <= 0):
339 raise ValueError("max_gap must be a positive number or None")
340 self._max_gap = float(max_gap) if max_gap is not None else None
341 self._pace_anchor: tuple[float, int] | None = None # (wall time, event ts)
343 self._n_read_events = 0 # Number of events read (not includeing events stored in buffer)
344 self._last_seek: "SeekResult | None" = None # result of the most recent seek()
346 # Seeking (see seek()). ``_anchored`` decouples the "first timestamp"
347 # anchoring from ``_n_read_events == 0`` so a post-seek read keeps the
348 # sought ``_current_ts`` instead of re-anchoring to the stream start.
349 self._anchored = False
350 self._index_opt = index
351 # ``index`` selects the seek index source for formats that use one (EVT):
352 # "auto"/True/False -> build an exact index in memory, lazily on the
353 # first seek() (never on open);
354 # "metavision" -> read a Metavision `.tmp_index` sidecar when
355 # present (fast, but approximate near large event
356 # gaps), else fall back to the built index.
357 # "persist" -> build evutils' own exact index on the first
358 # seek() and save it to a `<raw>.evidx` sidecar;
359 # on open, load a fresh sidecar instead of
360 # rebuilding (exact, and reused across runs).
361 self._file_decoder._use_sidecar = (index == "metavision")
362 self._file_decoder._persist_index = (index == "persist")
363 if self._file_name is not None:
364 self._file_decoder._raw_path = str(self._file_name)
366 def init(self) -> None:
367 """Initialize the reader, can be used explicitly or implicitly by the read method."""
368 if self._is_initialized:
369 return
370 self._file_decoder.init()
371 self._is_initialized = True
373 def _check_no_active_prefetch(self) -> None:
374 """Direct reads are not allowed while a prefetching iterator owns the
375 decoder (its worker thread is reading concurrently)."""
376 if self._active_prefetch is not None:
377 raise RuntimeError(
378 "An asynchronous iterator is active on this reader: exhaust or "
379 "close() it before calling read()/read_all(), or iterate instead."
380 )
382 def _pull(self, acc: EventAccumulator, delta_t: int, n_events: int) -> int:
383 """Pull more events into the accumulator, returning the number added
384 (0 => end of stream). Native decoders decode straight into the
385 accumulator's storage (no copy); others have their ``read_chunk`` output
386 appended.
388 Parameters
389 ----------
390 acc : EventAccumulator
391 The accumulator to pull events into.
392 delta_t : int
393 The time window constraint for reading.
394 n_events : int
395 The number of events constraint for reading.
397 Returns
398 -------
399 int
400 The number of events added to the accumulator.
402 """
403 dec = self._file_decoder
404 if self._native_fill:
405 while True:
406 if dec.is_eof():
407 return 0
408 ev, tr = acc.prepare(self._step)
409 added = dec.parse_step(ev, tr) # type: ignore[attr-defined]
410 if added > 0:
411 return int(added)
412 if dec.is_eof():
413 return 0
414 # else: consumed only state/timing words; step again.
415 chunk = dec.read_chunk(delta_t, n_events)
416 if isinstance(chunk, tuple):
417 chunk, triggers = chunk
418 else:
419 triggers = None
420 if len(chunk) == 0 and (triggers is None or len(triggers) == 0):
421 return 0
422 acc.append(chunk, triggers)
423 return int(len(chunk)) if len(chunk) > 0 else 1
425 def read(self, delta_t:int|None=None, n_events:int|None=None) -> 'EventArray | tuple[EventArray, TriggerArray]':
426 """Read events on the files based on the mode and the parameters.
428 Parameters
429 ----------
430 delta_t
431 Override the delta_t parameter, otherwise the default value is used from the constructor
432 n_events
433 Override the n_events parameter, otherwise the default value is used from the constructor
435 Returns
436 -------
437 EventArray
438 An array with the events
440 Examples
441 --------
442 >>> reader = EventReader(b"% format EVT3;height=720;width=1280\\n% end\\n", delta_t=10000)
443 >>> events = reader.read()
444 >>> print(len(events))
445 0
446 >>> # Override the time window for a single read
447 >>> more_events = reader.read(delta_t=5000)
449 """
450 self._check_no_active_prefetch()
451 out = self._read(delta_t, n_events)
452 if self._real_time:
453 self._pace(out)
454 if self._batch_mode:
455 from ..types import DataBatch, TriggerArray
456 if isinstance(out, tuple):
457 return DataBatch(events=out[0], triggers=out[1])
458 else:
459 return DataBatch(events=out, triggers=TriggerArray.empty())
460 return out
462 def _pace(self, out: "np.ndarray | EventArray") -> None:
463 """Sleep until the chunk's last event timestamp aligns with wall-clock
464 playback time (anchored at the first delivered chunk).
466 The anchor is absolute, so time the caller spends between chunks is
467 subtracted from the delay automatically; when the target time is
468 already past (slow decode or slow consumer), no delay is added.
469 """
470 ev = out[0] if isinstance(out, tuple) else out
471 if len(ev) == 0:
472 return
473 t_last = int(ev.t[-1])
474 now = time.perf_counter()
475 if self._pace_anchor is None:
476 self._pace_anchor = (now, int(ev.t[0]))
477 wall0, ts0 = self._pace_anchor
478 target = wall0 + (t_last - ts0) / (1e6 * self._playback_speed)
479 delay = target - now
480 # A wait longer than max_gap means the recording went idle (e.g. a
481 # silent lead-in before the action starts). Rather than stall on a
482 # single multi-second sleep, skip the dead air: re-anchor to now so the
483 # next window is paced relative to this one.
484 if self._max_gap is not None and delay > self._max_gap:
485 self._pace_anchor = (now, t_last)
486 return
487 if delay > 0:
488 time.sleep(delay)
490 def _read_n_events_fast(self, n_events: int) -> "np.ndarray | EventArray":
491 """Decode ~``n_events`` straight into a fresh output buffer and hand it
492 out with no staging copy.
494 Valid only in pure ``n_events`` mode: the count cutoff is enforced by the
495 output buffer's capacity (the parser stops when it is full), so there is
496 no time boundary to ``searchsorted`` and no overshoot to carry in a
497 staging accumulator. This turns the windowed read from two passes over
498 the data (decode into the reused accumulator, then copy the window out)
499 into one, roughly matching :class:`EventStreamer` while still returning
500 an independent array (the buffer is fresh per call, never reused).
501 """
502 dec = self._file_decoder
503 # Fresh per-call output; handed to the caller as zero-copy views, so it
504 # must not be reused. This path is only taken for "exact-capacity"
505 # decoders (one event per record: EVT2/EVT4/DAT), so the parser fills
506 # to *exactly* n_events and stops -- a single parse_step completes the
507 # window with no overshoot and no remainder to carry. (Vector formats
508 # would stop a few short and a second parse_step on the full buffer is
509 # misread as EOF, so they stay on the accumulator path.)
510 out = EventSoABuffers(n_events)
511 out.c.capacity = n_events
512 # Throwaway trigger sink (this path is only taken when the caller did not
513 # request triggers); reset each step so it never fills and stalls the
514 # parser on a trigger-dense region.
515 tr = TriggerSoABuffers(max(n_events // 16, 1024))
516 while out.size < n_events and not dec.is_eof():
517 tr.reset()
518 before = out.size
519 dec.parse_step(out, tr)
520 if out.size == before and dec.is_eof():
521 break
522 self._eof = dec.is_eof()
524 if out.size == 0:
525 return EventArray.empty()
526 if not self._anchored:
527 self._first_ts = int(out.t[0])
528 self._current_ts = self._first_ts
529 self._anchored = True
530 self._n_read_events += out.size
532 output = events_view(out)
533 if self._normalize_ts:
534 output.t -= self._first_ts - self._start_ts
535 return output
537 def _read_n_events_readchunk(self, n_events: int) -> "np.ndarray | EventArray":
538 """Hand out one ``read_chunk(n_events)`` result directly, no accumulator.
540 For decoders whose ``read_chunk`` already returns an *independent* array
541 of at most ``n_events`` events (``_independent_windows``): NPZ slices its
542 loaded columns into fresh arrays, CSV parses into fresh arrays. Skipping
543 the staging accumulator drops the append + slice_copy double copy.
544 """
545 dec = self._file_decoder
546 chunk = dec.read_chunk(n_events_hint=n_events)
547 if isinstance(chunk, tuple):
548 chunk = chunk[0]
549 # A full window is exactly n_events; anything short means the stream is
550 # drained. (Do not use dec.is_eof(): for buffered decoders like CSV it
551 # signals "input exhausted" while read_chunk still has buffered events.)
552 self._eof = len(chunk) < n_events
553 if len(chunk) == 0:
554 return EventArray.empty()
555 if self._n_read_events == 0:
556 self._first_ts = int(chunk.t[0])
557 self._current_ts = self._first_ts
558 self._n_read_events += len(chunk)
559 if self._normalize_ts:
560 chunk.t -= self._first_ts - self._start_ts # chunk is independent
561 return chunk
563 def _dt_trigger_sink(self) -> TriggerSoABuffers:
564 """Lazily-allocated throwaway trigger sink for the delta_t fast path.
566 Reset before every ``parse_step`` so it never fills and stalls the parser
567 on a trigger-dense region. Only used when the caller did not request
568 external triggers, so its contents are always discarded.
569 """
570 if self._dt_tr is None:
571 self._dt_tr = TriggerSoABuffers(max(self._step // 16, 1024))
572 return self._dt_tr
574 @staticmethod
575 def _grow_out(out: EventSoABuffers, step: int) -> None:
576 """Ensure ``out`` has room for one more ``step`` events, then cap the SoA
577 capacity the parser sees to ``size + step`` so a single ``parse_step``
578 overshoots the time window by at most one step's worth."""
579 if out.capacity - out.size < step:
580 out.grow(max(out.size + step, int(out.capacity * 1.5) + 1))
581 out.c.capacity = out.size + step
583 def _finalize_delta_t_window(self, out: EventSoABuffers, delta_t: int,
584 count_cut: bool, resume_ts: int | None = None
585 ) -> "EventArray":
586 """Shared bookkeeping after a delta_t window's events are staged in ``out``.
588 Both delta_t fast paths (the generic ``searchsorted`` one and the
589 dedicated C one) end identically: advance the window clock -- or, on a
590 count cutoff, stay in / resume the same time window -- refresh the
591 window-size estimate and the event counter, then hand out a normalized,
592 zero-copy view of the window. ``out.c.size`` must already be the window's
593 final event count, and ``self._eof`` must already be set by the caller
594 (it differs: the carry-based path also requires an empty overshoot).
596 ``count_cut`` marks a window cut short by the ``n_events`` safety cap.
597 ``resume_ts`` (fast path only) is the timestamp to resume mid-window from
598 on such a cut; ``None`` leaves the clock untouched (C path) or, on a
599 normal window, advances it by ``delta_t``.
600 """
601 size = out.size
602 # Track the largest window seen so the next buffer is pre-sized for it.
603 self._dt_est = max(self._dt_est, size)
604 if count_cut:
605 # Count cutoff: continue the same time window on the next call.
606 if resume_ts is not None:
607 self._current_ts = resume_ts
608 else:
609 self._current_ts += delta_t
610 self._n_read_events += size
612 output = events_view(out)
613 if self._normalize_ts:
614 output.t -= self._first_ts - self._start_ts
615 return output
617 def _read_delta_t_fast(self, delta_t: int) -> "np.ndarray | EventArray":
618 """One ``delta_t`` window, fast path: decode into a fresh buffer, hand out
619 a zero-copy view of the window, and carry only the small overshoot
620 (events past the boundary, at most one decode step) to the next call.
622 Mirrors :meth:`_read_n_events_fast` for time-windowed reads. The staging
623 accumulator path copies the *entire* window out on every call
624 (:meth:`EventAccumulator.slice_copy`) -- millions of events per window on
625 a dense recording. Here the window is a view into a per-call buffer (never
626 reused, so the view stays valid), and only the overshoot is copied.
628 Valid only for native-fill decoders in pure ``delta_t`` mode with no
629 external triggers (the guard in :meth:`_read` enforces this). The
630 ``n_events`` safety cap (``max_events`` in delta_t mode) is honoured: a
631 window larger than the cap is cut at the count boundary, exactly like the
632 accumulator path.
633 """
634 dec = self._file_decoder
635 step = self._dt_step
636 n_cap = self._n_events
637 tr = self._dt_trigger_sink()
639 # Decode buffer (pooled or fresh), pre-sized to ~one window plus 25%
640 # headroom so a normal window never triggers a mid-decode grow (a grow
641 # reallocates + copies the whole buffer -- a large per-frame hitch).
642 # Seed it with the previous call's overshoot.
643 carry = self._dt_carry
644 k = carry.size if carry is not None else 0
645 need = max(self._dt_est * 5 // 4 + step, k + step)
646 out = self._acquire_window_buffer(need)
647 if k:
648 out.t[:k] = carry.t[:k]
649 out.x[:k] = carry.x[:k]
650 out.y[:k] = carry.y[:k]
651 out.p[:k] = carry.p[:k]
652 out.c.size = k
653 self._dt_carry = None
655 # Anchor the stream's first timestamp / window origin on the very first
656 # decoded event (only reachable on the first call, when there is no carry).
657 if not self._anchored and out.size == 0:
658 while out.size == 0 and not dec.is_eof():
659 self._grow_out(out, step)
660 tr.reset()
661 dec.parse_step(out, tr)
662 if out.size == 0:
663 self._eof = True
664 return EventArray.empty()
665 self._first_ts = int(out.t[0])
666 self._current_ts = self._first_ts
667 self._anchored = True
669 end_ts = self._current_ts + delta_t
671 # Decode until the buffer holds an event at/after the time boundary, the
672 # count cap is exceeded, or the stream ends.
673 while not dec.is_eof():
674 n = out.size
675 if n and (int(out.t[n - 1]) >= end_ts or n > n_cap):
676 break
677 self._grow_out(out, step)
678 tr.reset()
679 added = dec.parse_step(out, tr)
680 if added == 0 and dec.is_eof():
681 break
683 size = out.size
684 t = out.t[:size].view(np.int64)
685 idx = int(np.searchsorted(t, end_ts)) if size else 0
686 count_cut = idx > n_cap
687 if count_cut:
688 idx = n_cap
690 # Everything from idx on is overshoot: copy it (small) for the next call.
691 if idx < size:
692 rem = size - idx
693 c = EventSoABuffers(rem)
694 c.t[:rem] = out.t[idx:size]
695 c.x[:rem] = out.x[idx:size]
696 c.y[:rem] = out.y[idx:size]
697 c.p[:rem] = out.p[idx:size]
698 c.c.size = rem
699 self._dt_carry = c
701 out.c.size = idx
702 # EOF only once the stream is drained *and* no overshoot remains, else the
703 # final carried events would never be emitted (is_eof() stops iteration).
704 self._eof = dec.is_eof() and self._dt_carry is None
705 return self._finalize_delta_t_window(
706 out, delta_t, count_cut,
707 resume_ts=int(t[idx]) if count_cut else None)
709 def _peek_first_ts(self) -> int | None:
710 """Timestamp of the stream's first event, without disturbing later reads.
712 Decodes a small scratch batch to learn the first timestamp; the caller
713 then :meth:`~decoder.reset`\\ s the decoder so the real window-0 decode
714 starts cleanly from the beginning. Returns ``None`` if the stream is
715 empty.
716 """
717 dec = self._file_decoder
718 scratch = EventSoABuffers(4096)
719 tr = self._dt_trigger_sink()
720 while scratch.size == 0 and not dec.is_eof():
721 tr.reset()
722 dec.parse_step(scratch, tr)
723 return int(scratch.t[0]) if scratch.size else None
725 def _dt_ring_size(self) -> int:
726 """Number of recycled window buffers needed so no live window is
727 overwritten: the consumer's current window plus one being decoded
728 (sync), plus everything an async prefetch may hold queued."""
729 if self._async_read:
730 from ._prefetch import DEFAULT_DEPTH
731 depth = self._prefetch_depth if self._prefetch_depth is not None else DEFAULT_DEPTH
732 return depth + 2
733 return 2
735 def _acquire_window_buffer(self, need: int) -> EventSoABuffers:
736 """Decode buffer for one delta_t window, sized for ``need`` events.
738 Default: a fresh buffer per window (independent result, safe to keep).
739 With ``reuse_buffers``: cycle the persistent ring -- pages stay warm, no
740 per-window allocation -- and the returned window aliases the slot until
741 the ring wraps back to it.
742 """
743 if not self._reuse_buffers:
744 return EventSoABuffers(need)
745 slots = self._dt_slots
746 i = self._dt_slot_i
747 if i >= len(slots):
748 slots.append(EventSoABuffers(need))
749 self._dt_slot_i = (i + 1) % self._dt_ring_size()
750 buf = slots[i]
751 if buf.capacity < need:
752 buf.grow(need)
753 buf.c.size = 0
754 buf.c.capacity = buf.capacity
755 return buf
757 def _read_delta_t_c(self, delta_t: int) -> "EventArray":
758 """One delta_t window via the dedicated C parser.
760 The C parser stops exactly when an event's timestamp reaches ``end_ts``,
761 so a whole window is decoded in one (GIL-free) call straight into the
762 output buffer -- no ``searchsorted`` boundary hunt and no overshoot carry
763 (every returned event already has ``ts < end_ts``). This is the openeb
764 approach: the slicing lives in the decoder, not in Python.
765 """
766 dec = self._file_decoder
767 n_cap = self._n_events
768 tr = self._dt_trigger_sink()
770 # Anchor the window origin on the first event once, then rewind so the C
771 # parser redoes window 0 from the start with the right end_ts.
772 if not self._anchored:
773 first = self._peek_first_ts()
774 if first is None:
775 self._eof = True
776 return EventArray.empty()
777 self._first_ts = first
778 self._current_ts = first
779 self._anchored = True
780 dec.reset()
781 self._eof = False
783 end_ts = self._current_ts + delta_t
785 out = self._acquire_window_buffer(
786 max(self._dt_est * 5 // 4 + self._dt_step, self._dt_step))
787 # The n_events safety cap (max_events in delta_t mode) is enforced by
788 # clamping the capacity the C parser sees, so a single C call can never
789 # decode past it. The parser stops within its reserved vector headroom
790 # below the cap; the decoder offset stays exact, so the next call
791 # continues the same window with no loss.
792 out.c.capacity = min(out.capacity, n_cap)
793 count_cut = False
794 while True:
795 tr.reset()
796 _, status = dec.parse_step_delta_t(out, tr, end_ts)
797 if status == EVUTILS_PARSE_WINDOW_DONE:
798 break
799 if dec.is_eof():
800 break
801 if status == EVUTILS_PARSE_OUTPUT_FULL:
802 if out.c.capacity >= n_cap:
803 # Hit the max_events cap: emit what we have and continue
804 # this same time window on the next call.
805 count_cut = True
806 break
807 # Window not finished but the buffer filled: grow and continue.
808 new_cap = min(max(out.size + self._dt_step,
809 int(out.c.capacity * 1.5) + 1), n_cap)
810 out.grow(new_cap) # realloc only if backing too small
811 out.c.capacity = new_cap # parser-visible cap (recycled slots)
812 # else EVUTILS_PARSE_OK: input drained this call; loop (tail -> EOF).
814 # On a normal window advance the clock; on a count-cut stay in the same
815 # time window so the next call continues it (matches the accumulator path).
816 self._eof = dec.is_eof()
817 return self._finalize_delta_t_window(out, delta_t, count_cut)
819 def _flush_dt_carry(self) -> None:
820 """Fold a pending delta_t fast-path overshoot into the accumulator.
822 The fast paths keep up to one decode step of overshoot in
823 ``_dt_carry``; any code path that reads via the accumulator (mixed-mode
824 overrides, ``read_all``) must fold it in first or those events are
825 silently lost. Allocates the accumulator if needed.
826 """
827 carry = self._dt_carry
828 if carry is None or carry.size == 0:
829 self._dt_carry = None
830 return
831 if self._buffer is None:
832 self._buffer = EventAccumulator(self._acc_capacity)
833 self._buffer.append(events_view(carry))
834 self._dt_carry = None
835 # Carried events exist => stream is not exhausted for the reader even
836 # if the decoder itself hit EOF.
837 self._eof = False
839 def _attach_sensor_size(self, out: "EventArray") -> "EventArray":
840 """Stamp ``sensor_size=(width, height)`` onto returned events, when known.
842 Applied at the single internal read chokepoint so every delivery path
843 (``read``, ``read_all``, sync and async iteration) carries the sensor
844 geometry. Triggers are left untouched. Formats that do not expose a
845 geometry (``shape()`` returns ``None``) leave ``sensor_size`` as ``None``.
846 """
847 w, h = self._file_decoder.shape()
848 if w is None or h is None:
849 return out
850 ev = out[0] if isinstance(out, tuple) else out
851 try:
852 ev.sensor_size = (int(w), int(h))
853 except AttributeError:
854 pass
855 return out
857 def _read(self, delta_t:int|None=None, n_events:int|None=None) -> "EventArray":
858 """Thin wrapper stamping sensor_size onto the decoded window."""
859 return self._attach_sensor_size(self._read_impl(delta_t, n_events))
861 def _read_impl(self, delta_t:int|None=None, n_events:int|None=None) -> "EventArray":
862 """Unguarded body of :meth:`read` (also driven by the prefetch worker)."""
863 # If not initialized, initialize
864 if not self._is_initialized:
865 self.init()
867 # Fast path: pure n_events streaming with no time window, no triggers,
868 # and nothing left buffered from an earlier windowed read. Skips the
869 # staging accumulator (and its per-window copy).
870 if (self._mode == "n_events" and delta_t is None
871 and not self._read_external_triggers
872 and (self._buffer is None or len(self._buffer) == 0)):
873 dec = self._file_decoder
874 n = n_events if n_events is not None else self._n_events
875 # Exact-capacity native decoders (EVT2/DAT/AER) decode straight into
876 # a fresh output buffer via parse_step.
877 if self._native_fill and dec._exact_window:
878 return self._read_n_events_fast(n)
879 # Decoders whose read_chunk already returns independent, bounded
880 # chunks (NPZ/CSV) can hand that out directly.
881 if dec._independent_windows:
882 return self._read_n_events_readchunk(n)
884 # Fast path: pure delta_t streaming with a native-fill decoder, no
885 # triggers, and nothing left in the staging accumulator. Hands out a
886 # zero-copy view of the window instead of copying it out (slice_copy).
887 if (self._mode == "delta_t" and n_events is None
888 and self._native_fill
889 and not self._read_external_triggers
890 and (self._buffer is None or len(self._buffer) == 0)):
891 dt = delta_t if delta_t is not None else self._delta_t
892 # Prefer the dedicated C delta_t parser (one GIL-free call per window,
893 # no boundary search or overshoot carry) when the format has one.
894 if self._file_decoder._has_delta_t_parser:
895 return self._read_delta_t_c(dt)
896 return self._read_delta_t_fast(dt)
898 # Allocate the staging accumulator on first use.
899 if self._buffer is None:
900 self._buffer = EventAccumulator(self._acc_capacity)
901 acc = self._buffer
903 # If a delta_t fast path ran earlier and left an overshoot carry, fold
904 # it in first: those events precede anything the decoder emits next.
905 # (Mixed-path reads -- per-call overrides after windowed fast-path
906 # iteration -- would otherwise silently lose them.)
907 self._flush_dt_carry()
909 # "all" mode (without per-call overrides) means the whole remaining
910 # stream: skip the window cutoffs entirely and drain to EOF, so files
911 # exceeding max_time/max_events are still returned in full.
912 drain = self._mode == "all" and delta_t is None and n_events is None
914 # Override the parameters if they are specified
915 if delta_t is None:
916 delta_t = self._delta_t
917 if n_events is None:
918 n_events = self._n_events
920 # Establish the first timestamp once, at the very start of the stream.
921 if not self._anchored and len(acc) == 0:
922 if self._pull(acc, delta_t, n_events) == 0:
923 self._eof = True
924 if self._read_external_triggers:
925 from ..types import TriggerArray
926 return EventArray.empty(), TriggerArray.empty()
927 return EventArray.empty()
928 self._first_ts = int(acc.t_window()[0])
929 self._current_ts = self._first_ts
930 self._anchored = True
932 start_ts: int = self._current_ts
933 end_ts: int = start_ts + delta_t # Final end_ts if we reach delta_t
934 end_idx: int = len(acc)
936 tr_end_idx: int = acc._tr.size - acc._tr_start
938 # Gather events until we hit the n_events count, the delta_t time window,
939 # or the end of the stream. Work directly on the SoA `t` column.
940 # Both cutoffs are evaluated together: whichever falls earlier in the
941 # stream wins (the buffer may hold far more than one window's worth).
942 while True:
943 t = acc.t_window()
944 time_ready = not drain and len(t) > 0 and t[-1] >= end_ts
945 count_ready = not drain and len(acc) > n_events
947 if time_ready or count_ready:
948 time_idx = int(np.searchsorted(t, end_ts)) if time_ready else len(acc) + 1
949 tr_t = acc.t_window_tr()
950 if count_ready and time_idx > n_events:
951 # n_events cutoff comes first
952 end_idx = n_events
953 self._current_ts = int(t[n_events])
954 tr_end_idx = int(np.searchsorted(tr_t, self._current_ts, side='left'))
955 else:
956 # delta_t cutoff comes first (ties go to the time window)
957 end_idx = time_idx
958 self._current_ts += delta_t
959 tr_end_idx = int(np.searchsorted(tr_t, end_ts, side='left'))
960 break
962 # Not enough buffered yet: pull more from the decoder.
963 if self._pull(acc, delta_t, n_events) == 0:
964 self._eof = True
965 # Neither cutoff met, so the whole remaining buffer is the slice.
966 end_idx = len(acc)
967 tr_end_idx = acc._tr.size - acc._tr_start
968 break
970 # Copy out the window (independent) and advance past it.
971 output, output_tr = acc.slice_copy(end_idx, tr_end_idx)
972 self._n_read_events += end_idx
974 if self._normalize_ts:
975 # Normalize the timestamps to start from zero at start_ts. slice_copy
976 # already returned an independent array, so this is safe in place.
977 output.t -= self._first_ts - self._start_ts
978 if len(output_tr) > 0:
979 output_tr.t -= self._first_ts - self._start_ts
981 if self._read_external_triggers:
982 return output, output_tr
983 return output
985 def read_all(self) -> 'EventArray | tuple[EventArray, TriggerArray]':
986 """Decode and return every remaining event at once.
988 Delegates to the decoder's :meth:`~evutils.io.common.EventDecoder.read_all`,
989 which (for the native EVT/DAT/AER decoders) decodes the whole payload
990 straight into a single output buffer -- no per-chunk copy, no final
991 ``concatenate`` -- and hands the columns back as a zero-copy
992 :class:`EventArray`. This bypasses the slicing ring buffer that
993 :meth:`read` uses for ``delta_t``/``n_events`` windowing.
995 .. note::
996 This materialises every event in memory at once. For recordings too
997 large to fit, iterate the reader (windowed :meth:`read`) instead.
999 Returns
1000 -------
1001 EventArray
1002 All remaining events.
1004 Examples
1005 --------
1006 >>> reader = EventReader(b"% format EVT3;height=720;width=1280\\n% end\\n")
1007 >>> all_events = reader.read_all()
1008 >>> print(f"Total events: {len(all_events)}")
1009 Total events: 0
1011 """
1012 self._check_no_active_prefetch()
1013 if not self._is_initialized:
1014 self.init()
1016 # Fold in any delta_t fast-path overshoot so it is prepended below
1017 # along with the rest of the staging buffer.
1018 self._flush_dt_carry()
1020 _out = self._file_decoder.read_all()
1021 if self._read_external_triggers:
1022 if isinstance(_out, tuple):
1023 out, out_tr = _out
1024 else:
1025 out = _out
1026 from ..types import TriggerArray
1027 out_tr = TriggerArray.empty()
1028 else:
1029 out = _out # type: ignore
1031 # Prepend anything already buffered by prior read() calls (rare: only if
1032 # read() and read_all() are mixed on the same reader).
1033 if self._buffer is not None and (len(self._buffer) > 0 or self._buffer._tr.size - self._buffer._tr_start > 0):
1034 buffered, buffered_tr = self._buffer.slice_copy(len(self._buffer), self._buffer._tr.size - self._buffer._tr_start)
1035 if len(out) == 0:
1036 out = buffered
1037 elif len(buffered) > 0:
1038 out = EventArray(
1039 np.concatenate([buffered.t, out.t]),
1040 np.concatenate([buffered.x, out.x]),
1041 np.concatenate([buffered.y, out.y]),
1042 np.concatenate([buffered.p, out.p]),
1043 )
1045 if self._read_external_triggers:
1046 if len(out_tr) == 0:
1047 out_tr = buffered_tr
1048 elif len(buffered_tr) > 0:
1049 from ..types import TriggerArray
1050 out_tr = TriggerArray(
1051 np.concatenate([buffered_tr.t, out_tr.t]),
1052 np.concatenate([buffered_tr.p, out_tr.p]),
1053 np.concatenate([buffered_tr.id, out_tr.id]),
1054 )
1056 self._eof = True
1057 self._n_read_events += len(out)
1058 self._attach_sensor_size(out)
1060 if self._normalize_ts and len(out) > 0:
1061 # Capture the shift before modifying out.t, so triggers get the
1062 # same normalization.
1063 shift = int(out.t[0]) - self._start_ts
1064 out.t -= shift
1065 if self._read_external_triggers and len(out_tr) > 0:
1066 out_tr.t -= shift
1068 if self._batch_mode:
1069 from ..types import DataBatch, TriggerArray
1070 if self._read_external_triggers:
1071 return DataBatch(events=out, triggers=out_tr)
1072 return DataBatch(events=out, triggers=TriggerArray.empty())
1073 if self._read_external_triggers:
1074 return out, out_tr
1075 return out
1077 def reset(self) -> None:
1078 """Reset file reader back to the beginning of the file.
1080 An active asynchronous iterator is cancelled first (its remaining
1081 buffered chunks are discarded).
1082 """
1083 if self._active_prefetch is not None:
1084 self._active_prefetch.close()
1085 self._n_read_events = 0
1086 self._eof = False
1087 self._anchored = False
1088 self._pace_anchor = None
1089 self._dt_carry = None
1090 self._dt_est = self._step
1091 self._dt_slot_i = 0 # keep the recycled slots themselves (warm pages)
1092 if self._buffer is not None:
1093 self._buffer.reset()
1094 self._file_decoder.reset()
1096 @property
1097 def last_seek(self) -> "SeekResult | None":
1098 """The :class:`~evutils.io.common.SeekResult` of the most recent
1099 :meth:`seek`, or ``None`` if no seek has been performed."""
1100 return self._last_seek
1102 @property
1103 def event_index(self) -> int:
1104 """Current 0-based event index: the number of events already read (and,
1105 after a :meth:`seek`, the index the cursor landed on). Complements
1106 :meth:`tell` (byte position) with an unambiguous event-coordinate."""
1107 return self._n_read_events
1109 def seek(self, t: int | None = None, n: int | None = None,
1110 relative: bool = False) -> SeekResult:
1111 """Reposition the read cursor by timestamp or event index.
1113 Random access in *event coordinates*: give exactly one of ``t``
1114 (absolute microseconds) or ``n`` (absolute 0-based event index). With
1115 ``relative=True`` the value is added to the current cursor instead
1116 (``whence=CUR``). Seeks work both forward and backward. After a seek the
1117 next :meth:`read` / iteration continues in the reader's configured
1118 ``delta_t`` / ``n_events`` mode, re-anchored at the new point.
1120 A seekable source uses the decoder's index / binary search (fast); a
1121 non-seekable one falls back to iterating and dropping events up to the
1122 target (backward requires restarting the stream).
1124 Parameters
1125 ----------
1126 t : int, optional
1127 Target timestamp in microseconds.
1128 n : int, optional
1129 Target event index (0-based).
1130 relative : bool, optional
1131 Interpret ``t`` / ``n`` relative to the current cursor.
1133 Returns
1134 -------
1135 SeekResult
1136 A named tuple ``(ts, index, eof)``: the absolute timestamp of the
1137 first event that will be read next (``ts``, also the tuple's first
1138 field for backward compatibility), the 0-based event index it landed
1139 on (``index``), and whether the seek ran off the end of the stream
1140 (``eof``). Also stored on :attr:`last_seek`.
1142 Raises
1143 ------
1144 NotImplementedError
1145 If the decoder does not support seeking.
1146 ValueError
1147 If neither or both of ``t`` / ``n`` are given.
1149 Examples
1150 --------
1151 >>> reader = EventReader(b"% format EVT3;height=720;width=1280\\n% end\\n", delta_t=10000)
1152 >>> _ = reader.seek(t=0)
1153 """
1154 if not self._is_initialized:
1155 self.init()
1156 if (t is None) == (n is None):
1157 raise ValueError("seek() requires exactly one of t= or n=.")
1159 dec = self._file_decoder
1160 if not dec.SUPPORTS_SEEK:
1161 raise NotImplementedError(
1162 f"{dec.__class__.__name__} does not support seeking."
1163 )
1165 if self._active_prefetch is not None:
1166 self._active_prefetch.close()
1168 if relative:
1169 if t is not None:
1170 t = self._current_ts + t
1171 else:
1172 n = self._n_read_events + n
1174 # Anchor the normalization origin (_first_ts) on the stream's *first*
1175 # event before the cursor moves, so normalize_ts is call-order
1176 # independent (seek-then-read == read-then-seek). Peeking rewinds the
1177 # decoder, so it is skipped on non-seekable sources (and when
1178 # normalization is off) -- there the origin falls back to the seek
1179 # landing point below.
1180 first_ts: int | None = None
1181 seekable = getattr(self._source, "seekable", lambda: True)()
1182 # Slurp-based decoders (EVT/DAT/...) buffer the whole payload in memory,
1183 # so their seek() -- and the pre-seek first-ts peek -- work even when the
1184 # underlying source is not itself seekable (e.g. a compressed stream).
1185 can_fast_seek = seekable or dec._buffers_in_memory
1186 if not self._anchored and self._normalize_ts and can_fast_seek:
1187 first_ts = self._peek_stream_first_ts()
1189 self._eof = False
1190 self._pace_anchor = None
1191 self._dt_carry = None
1192 self._dt_est = self._step
1193 self._dt_slot_i = 0
1194 if self._buffer is not None:
1195 self._buffer.reset()
1197 if can_fast_seek:
1198 try:
1199 res, rem_ev, rem_tr = dec.seek(t=t, n=n)
1200 except (io.UnsupportedOperation, OSError):
1201 res, rem_ev, rem_tr = self._seek_linear(t, n)
1202 else:
1203 res, rem_ev, rem_tr = self._seek_linear(t, n)
1205 if len(rem_ev) > 0 or (rem_tr is not None and len(rem_tr) > 0):
1206 if self._buffer is None:
1207 self._buffer = EventAccumulator(self._acc_capacity)
1208 self._buffer.append(rem_ev, rem_tr)
1210 if not self._anchored:
1211 self._first_ts = first_ts if first_ts is not None else res.ts
1212 self._n_read_events = res.index if res.index >= 0 else (int(n) if n is not None else 0)
1213 self._current_ts = res.ts
1214 self._anchored = True
1215 # Store the full result and hand it back. SeekResult is a NamedTuple with
1216 # ``ts`` first, so ``== ts``-style int comparisons still need ``.ts`` but
1217 # positional / ``.ts`` access stays source-compatible.
1218 self._last_seek = SeekResult(ts=res.ts, index=self._n_read_events, eof=res.eof)
1219 return self._last_seek
1221 def _peek_stream_first_ts(self) -> int | None:
1222 """Timestamp of the stream's very first event, leaving the decoder
1223 rewound to the start.
1225 Used to anchor the ``normalize_ts`` origin before a seek moves the
1226 cursor. Only called before any read on a seekable source (rewinding a
1227 non-seekable decoder would lose data). Returns ``None`` for an empty
1228 stream.
1229 """
1230 dec = self._file_decoder
1231 dec.reset()
1232 first: int | None
1233 if self._native_fill:
1234 first = self._peek_first_ts()
1235 else:
1236 chunk = dec.read_chunk()
1237 if isinstance(chunk, tuple):
1238 chunk = chunk[0]
1239 first = int(chunk.t[0]) if len(chunk) > 0 else None
1240 dec.reset()
1241 return first
1243 def _seek_linear(self, t: int | None, n: int | None) -> tuple["SeekResult", "EventArray", "TriggerArray | None"]:
1244 """Fallback seek for non-seekable sources: iterate and drop to target."""
1245 from .common import SeekResult
1246 dec = self._file_decoder
1247 axis = "t" if t is not None else "n"
1248 target = t if t is not None else n
1250 behind = ((axis == "t" and target < self._current_ts)
1251 or (axis == "n" and target < self._n_read_events))
1252 seen = 0
1253 if behind:
1254 dec.reset()
1255 else:
1256 seen = self._n_read_events
1258 landed_ts = target
1259 idx = target if axis == "n" else -1
1260 rem_ev = _EMPTY_EVENTS
1261 rem_tr = None
1263 while True:
1264 chunk = dec.read_chunk()
1265 if isinstance(chunk, tuple):
1266 chunk, tr_chunk = chunk
1267 else:
1268 tr_chunk = None
1270 if len(chunk) == 0:
1271 self._eof = True
1272 break
1273 if axis == "t":
1274 k = int(np.searchsorted(chunk.t, target, side="left"))
1275 hit = k < len(chunk)
1276 else:
1277 k = max(0, min(target - seen, len(chunk)))
1278 hit = target < seen + len(chunk)
1279 if hit:
1280 rem_ev = chunk[k:].copy()
1281 if tr_chunk is not None:
1282 if axis == "t":
1283 tr_k = int(np.searchsorted(tr_chunk.t, target, side="left"))
1284 else:
1285 tr_k = 0
1286 rem_tr = tr_chunk[tr_k:].copy()
1288 landed_ts = int(rem_ev.t[0]) if len(rem_ev) > 0 else target
1289 idx = seen + k
1290 break
1291 seen += len(chunk)
1292 return SeekResult(ts=landed_ts, index=idx, eof=self._eof), rem_ev, rem_tr
1294 def __enter__(self) -> "EventReader":
1295 return self
1297 def is_eof(self) -> bool:
1298 """Check if the end of the file is reached.
1300 Returns
1301 -------
1302 bool
1303 True if the end of the file is reached, False otherwise
1305 """
1306 if not self._is_initialized:
1307 self.init()
1308 return self._eof and (self._buffer is None or len(self._buffer) == 0)
1310 def close(self) -> None:
1311 """Close the reader and release resources (decoder buffer views, then the
1312 underlying byte source). Cancels any active asynchronous iterator first.
1313 """
1314 if self._active_prefetch is not None:
1315 self._active_prefetch.close()
1316 # Drop decoder views (e.g. into an mmap) before closing the source.
1317 self._file_decoder.close()
1318 self._source.close()
1320 def __exit__(self, exc_type: "type[BaseException] | None", exc_value: "BaseException | None", traceback: "types.TracebackType | None") -> None:
1321 self.close()
1323 def __repr__(self) -> str:
1324 if self._is_initialized:
1325 is_initialized_txt = "initialized"
1326 else:
1327 is_initialized_txt = "not initialized"
1328 src = self._file_name if self._file_name is not None else self._source.__class__.__name__
1329 return f"{self.__class__.__name__}(source={src} - {is_initialized_txt}, delta_t={self._delta_t}, n_events={self._n_events}, mode={self._mode})"
1331 def __len__(self) -> int:
1332 return self._n_read_events
1334 def __iter__(self) -> "Iterator[EventArray]":
1335 """Iterate over the events in the file.
1337 With ``async_read=True`` the windows are decoded ahead in a background
1338 thread (bounded to a couple of windows), overlapping decode with the
1339 caller's per-chunk processing. Only one asynchronous iterator can be
1340 active per reader; direct :meth:`read` / :meth:`read_all` calls raise
1341 until it is exhausted or closed.
1343 Yields
1344 ------
1345 EventArray
1346 An array with the events
1348 Examples
1349 --------
1350 >>> reader = EventReader(b"% format EVT3;height=720;width=1280\\n% end\\n", n_events=5000)
1351 >>> for events in reader:
1352 ... print(f"Received chunk of {len(events)} events")
1353 Received chunk of 0 events
1355 """
1356 it: Any
1357 if self._async_read:
1358 self._check_no_active_prefetch()
1359 kwargs: dict[str, str | int | float | bool] = {}
1360 if self._prefetch_depth is not None:
1361 kwargs["depth"] = self._prefetch_depth
1362 it = PrefetchIterator(
1363 self._iter_sync(),
1364 on_finish=lambda: setattr(self, "_active_prefetch", None),
1365 **kwargs,
1366 )
1367 self._active_prefetch = it
1368 else:
1369 it = self._iter_sync()
1370 if self._real_time:
1371 # Pace on the consumer side, never inside the decode path: with
1372 # async_read the worker keeps decoding ahead while we sleep.
1373 return self._paced_iter(it)
1374 return it
1376 def _paced_iter(self, it: "Iterator[EventArray]") -> "Iterator[EventArray]":
1377 """Wrap an iterator so each chunk is released on the playback clock."""
1378 try:
1379 for chunk in it:
1380 self._pace(chunk)
1381 yield chunk
1382 finally:
1383 # Propagate early termination (break / close) to a prefetching
1384 # iterator so its worker thread is released.
1385 if hasattr(it, "close"):
1386 it.close()
1388 def _iter_sync(self) -> "Iterator[Any]":
1389 """The plain synchronous window generator behind :meth:`__iter__`."""
1390 if not self._is_initialized:
1391 self.init()
1392 while not self.is_eof():
1393 res = self._read()
1394 if self._batch_mode:
1395 from ..types import DataBatch, TriggerArray
1396 if isinstance(res, tuple):
1397 yield DataBatch(events=res[0], triggers=res[1])
1398 else:
1399 yield DataBatch(events=res, triggers=TriggerArray.empty())
1400 else:
1401 yield res
1402 def shape(self) -> tuple[int|None, int|None]:
1403 """Get the shape of the frame.
1405 Returns
1406 -------
1407 tuple[int, int]
1408 The shape of the frame (width, height)
1410 """
1411 if not self._is_initialized:
1412 self.init()
1413 if self._width is not None and self._height is not None:
1414 return self._width, self._height
1415 else:
1416 return self._file_decoder.shape()
1418 def tell(self) -> int:
1419 """Get the current position in the file.
1421 Returns
1422 -------
1423 int
1424 The current position in the file
1426 """
1427 if not self._is_initialized:
1428 self.init()
1429 return self._file_decoder.tell()