Coverage for src/evutils/io/_event_reader.py: 87%
294 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-15 09:33 +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 Any
12import numpy as np
14from ..types import EventArray, TriggerArray
16from . import decoders as ev_decoders
17from ._prefetch import PrefetchIterator
18from ._source import ByteSource, make_source
19from .buffer import EventAccumulator
22class EventReader():
23 """Class for reading and automatically decoding and slicing events from a file or stream.
25 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.
26 The file_decoder is chosen automatically based on the file format or can be supplied explicitly.
28 Parameters
29 ----------
30 file: Path or str or io.BufferedReader or bytes or ByteSource
31 Path to the data file or a readable stream or bytes or a ByteSource
32 delta_t: int or optional
33 Time window in microseconds, by default None
34 n_events: int or None
35 Number of events to read in a chunk, by default None
36 max_events: int, default=10_000_000
37 Maximum number of events to read at once
38 mode: {'delta_t', 'n_events', 'mixed', 'all', 'auto'}, default 'auto'
39 Mode of operation ```["delta_t", "n_events", "mixed", "all", "auto"]```
40 start_ts: int, default=0
41 Start timestamp offset for the events, by default 0 (start of the file)
42 normalize_ts: bool, default=False
43 Normalize timestamps to start from zero
44 max_time: int, default=1_000_000_000_000
45 Maximum timestamp to read
46 width: int or None
47 Width of the frame, by default infered from the file
48 height: int or None
49 Height of the frame, by default infered from the file
50 ext_trigger: bool, default=False
51 Whether to read external trigger events, by default False
52 async_read: bool, default=False
53 Decode ahead in a background thread while iterating: the next window
54 is parsed while the caller processes the current one (the native
55 parsers release the GIL). Helps whenever per-chunk processing takes
56 meaningful time (numpy pipelines, GPU inference -- the read becomes
57 essentially free); can slightly hurt when the processing already
58 saturates memory bandwidth. Affects iteration only; ``read()`` /
59 ``read_all()`` stay synchronous and raise while an asynchronous
60 iterator is active.
61 real_time: bool, default=False
62 Pace :meth:`read` and iteration to the recording's own timeline: each
63 chunk is released only once the wall-clock time since the first chunk
64 matches the event time it covers, so the stream plays back as if live.
65 Pacing is anchored to an absolute start time, so time spent processing
66 a chunk is absorbed automatically (no drift); if decoding or
67 processing falls behind, chunks are released immediately with no added
68 delay. ``read_all()`` is never paced. Combine with ``async_read=True``
69 to decode ahead while waiting out the pacing delay.
70 playback_speed: float, default=1.0
71 Playback rate multiplier for ``real_time`` mode: ``2.0`` plays twice
72 as fast as the recording, ``0.5`` at half speed. Ignored when
73 ``real_time=False``.
74 max_gap: float or None, default=1.0
75 Longest idle stretch, in real seconds, that ``real_time`` pacing will
76 sleep through. If a recording goes silent for longer (e.g. a
77 late-starting stream, or a pause mid-recording), the pacer skips the
78 dead air and resumes immediately instead of blocking on one long
79 sleep. ``None`` disables the skip (strict real time). Ignored when
80 ``real_time=False``.
81 file_decoder: ev_decoders.EventDecoder or type[ev_decoders.EventDecoder] or None, default=None
82 File decoder to use, by default None - automatic
83 **kwargs
84 Additional arguments to pass to the file decoder
86 Raises
87 ------
88 ValueError
89 If the mode is not supported or if the delta_t or n_events are not specified when needed
91 Examples
92 --------
93 >>> from evutils.io import EventReader
94 >>> # Read events in chunks of 10ms (10,000 microseconds)
95 >>> reader = EventReader(b"% format EVT3;height=720;width=1280\\n% end\\n", delta_t=10000)
96 >>> for events in reader:
97 ... # Access fields directly
98 ... x, y, p, t = events.x, events.y, events.p, events.t
99 ... print(f"Processed {len(events)} events")
100 Processed 0 events
102 TODO:
103 Architectural Refactor Roadmap:
104 Decouple `EventReader`'s monolithic design by extracting the core byte-to-array decoding into
105 a native `EventStreamer`. The slicing logic (`mode="delta_t"`, `mode="n_events"`, etc.) should
106 be moved to composable pipeline generators in `evutils.chunking`. `EventReader` will then
107 become a simple Façade that dynamically assembles these pipeline generators to maintain
108 this exact backward-compatible API, while allowing power-users to compose custom pipelines
109 (e.g., slicing by external trigger boundaries).
110 """
112 READING_MODES = ["delta_t", "n_events", "mixed", "all", "auto"]
113 DEFAULT_N_EVENTS = 1_000_000
114 DEFAULT_DELTA_T = 10_000
115 def __init__(self, file: Path | str | io.BufferedReader | bytes | ByteSource,
116 delta_t:int|None=None,
117 n_events:int|None=None,
118 mode:str="auto",
119 start_ts:int=0,
120 normalize_ts: bool=False,
121 max_time:int=1_000_000_000_000,
122 max_events:int=10_000_000,
123 width:int | None=None, height:int | None=None,
124 ext_trigger: bool=False,
125 async_read: bool=False,
126 real_time: bool=False,
127 playback_speed: float=1.0,
128 max_gap: float | None=1.0,
129 file_decoder: ev_decoders.EventDecoder | type[ev_decoders.EventDecoder] | None = None,
130 **kwargs: Any) -> None:
132 # Remember the path (if any) for repr / reset semantics.
133 self._file_name: Path | None = Path(file) if isinstance(file, (str, Path)) else None
134 self._read_external_triggers = ext_trigger
136 # 1. Normalise the input into a ByteSource (path | stream | bytes |
137 # BytesIO | ByteSource -> ByteSource). Regular files are memory-mapped.
138 self._source: ByteSource = make_source(file)
140 # 2. Resolve the decoder and launch it:
141 # explicit instance > explicit class > heuristic (extension, then
142 # content sniffing of the source).
143 if isinstance(file_decoder, ev_decoders.EventDecoder):
144 self._file_decoder = file_decoder
145 else:
146 decoder_cls = file_decoder or ev_decoders.resolve_decoder_cls(self._source)
147 self._file_decoder = decoder_cls(self._source, **kwargs)
149 self._file_decoder.read_external_triggers = self._read_external_triggers
150 if self._read_external_triggers and not getattr(self._file_decoder, 'SUPPORTS_EXT_TRIGGERS', False):
151 import warnings
152 warnings.warn(f"{self._file_decoder.__class__.__name__} does not support reading external triggers.")
154 # This will now be io.BufferedReader
155 self._eof = False
157 # If not defined explicitly, the width and height are fetch from the file (not all formats support this)
158 self._width = width
159 self._height = height
160 self._start_ts = start_ts # Offset to start reading events. 0 is start of file
161 self._first_ts = 0 # First timestamp in the file, used for normalization
162 self._current_ts = self._first_ts
163 self._normalize_ts = normalize_ts # Normalize timestamps to start from zero
166 # Validate the parameters
167 if mode not in EventReader.READING_MODES:
168 raise ValueError(f"Mode {mode} not supported. Supported modes are: {EventReader.READING_MODES}")
171 self._mode = mode.lower()
173 # if mode is auto, we will try to infer the mode from the parameters
174 if self._mode == "auto":
175 # If both delta_t and n_events are specified, we will use mixed mode
176 if delta_t is not None and n_events is not None:
177 self._mode = "mixed"
179 # If only one of the parameters is specified, we will use that mode, the other will be set to the maximum
180 elif delta_t is not None:
181 self._mode = "delta_t"
182 n_events = max_events
183 elif n_events is not None:
184 self._mode = "n_events"
185 delta_t = max_time
186 else:
187 # If none of the parameters are specified, we will use the default Values
188 self._mode = "mixed"
189 delta_t = self.DEFAULT_DELTA_T
190 n_events = self.DEFAULT_N_EVENTS
192 # If the mode is not auto, we will check if the parameters are specified
193 elif self._mode == "delta_t":
194 if delta_t is None:
195 raise ValueError("delta_t must be specified")
196 n_events = max_events
197 elif self._mode == "n_events":
198 if n_events is None:
199 raise ValueError("n_events must be specified")
200 delta_t = max_time
201 elif self._mode == "mixed":
202 if delta_t is None:
203 delta_t = self.DEFAULT_DELTA_T
204 if n_events is None:
205 n_events = self.DEFAULT_N_EVENTS
207 elif self._mode == "all":
208 delta_t = max_time
209 n_events = max_events
212 # Validate the parameters
213 if delta_t is None:
214 delta_t = self.DEFAULT_DELTA_T
215 if n_events is None:
216 n_events = self.DEFAULT_N_EVENTS
218 if not isinstance(delta_t, int):
219 raise TypeError("delta_t must be an integer")
221 if not isinstance(n_events, int):
222 raise TypeError("n_events must be an integer")
225 if delta_t <= 0:
226 raise ValueError("delta_t must be positive")
228 if n_events <= 0:
229 raise ValueError("n_events must be positive")
232 # delta_t and n_events to read on each call
233 self._delta_t = delta_t
234 self._n_events = n_events
236 # Maximum number of events to read and maximum time to read in a chunk
237 self._max_events = max_events if max_events < self._n_events else self._n_events
238 self._max_time = max_time if max_time < self._delta_t else self._delta_t
240 self._is_initialized = False
242 # Windowed reads decode straight into this reused accumulator (no
243 # intermediate copy); only the returned window is copied out. Allocated
244 # lazily on first read() so the read_all() fast path never pays for it.
245 # Granularity of a single decode step, and a native fast path flag.
246 self._buffer: EventAccumulator | None = None
247 self._step = 1 << 20
248 # Hold the largest window we might be asked for, plus room for one decode
249 # step of overshoot before the front is rotated out. (np.empty is lazy, so
250 # a large capacity only costs pages actually written.)
251 self._acc_capacity = max(self._n_events, max_events) + 2 * self._step
252 self._native_fill = hasattr(self._file_decoder, "parse_step")
254 # Asynchronous iteration: when enabled, __iter__ decodes ahead in a
255 # worker thread (see io/_prefetch.py). The active iterator owns the
256 # decoder until it is exhausted or closed.
257 self._async_read = async_read
258 self._active_prefetch: PrefetchIterator | None = None
260 # Real-time playback: chunks are released against an absolute
261 # wall-clock anchor set when the first chunk is delivered, so the
262 # stream is paced like a live sensor (see _pace()).
263 if not isinstance(playback_speed, (int, float)) or playback_speed <= 0:
264 raise ValueError("playback_speed must be a positive number")
265 self._real_time = real_time
266 self._playback_speed = float(playback_speed)
267 # Longest idle stretch (real seconds) the pacer will sleep through. If a
268 # window would require a longer wait -- a silent gap in the recording,
269 # e.g. a late-starting stream -- the pacer skips the dead air and
270 # re-anchors so playback resumes immediately. None = strict real time.
271 if max_gap is not None and (not isinstance(max_gap, (int, float)) or max_gap <= 0):
272 raise ValueError("max_gap must be a positive number or None")
273 self._max_gap = float(max_gap) if max_gap is not None else None
274 self._pace_anchor: tuple[float, int] | None = None # (wall time, event ts)
276 self._n_read_events = 0 # Number of events read (not includeing events stored in buffer)
280 def init(self) -> None:
281 """Initialize the reader, can be used explicitly or implicitly by the read method."""
282 if self._is_initialized:
283 return
284 self._file_decoder.init()
285 self._is_initialized = True
287 def _check_no_active_prefetch(self) -> None:
288 """Direct reads are not allowed while a prefetching iterator owns the
289 decoder (its worker thread is reading concurrently)."""
290 if self._active_prefetch is not None:
291 raise RuntimeError(
292 "An asynchronous iterator is active on this reader: exhaust or "
293 "close() it before calling read()/read_all(), or iterate instead."
294 )
296 def _pull(self, acc: EventAccumulator, delta_t: int, n_events: int) -> int:
297 """Pull more events into the accumulator, returning the number added
298 (0 => end of stream). Native decoders decode straight into the
299 accumulator's storage (no copy); others have their ``read_chunk`` output
300 appended.
302 Parameters
303 ----------
304 acc : EventAccumulator
305 The accumulator to pull events into.
306 delta_t : int
307 The time window constraint for reading.
308 n_events : int
309 The number of events constraint for reading.
311 Returns
312 -------
313 int
314 The number of events added to the accumulator.
316 """
317 dec = self._file_decoder
318 if self._native_fill:
319 while True:
320 if dec.is_eof():
321 return 0
322 ev, tr = acc.prepare(self._step)
323 added = dec.parse_step(ev, tr) # type: ignore[attr-defined]
324 if added > 0:
325 return int(added)
326 if dec.is_eof():
327 return 0
328 # else: consumed only state/timing words; step again.
329 chunk = dec.read_chunk(delta_t, n_events)
330 if isinstance(chunk, tuple):
331 chunk, triggers = chunk
332 else:
333 triggers = None
334 if len(chunk) == 0:
335 return 0
336 acc.append(chunk, triggers)
337 return int(len(chunk))
339 def read(self, delta_t:int|None=None, n_events:int|None=None) -> 'EventArray | tuple[EventArray, TriggerArray]':
340 """Read events on the files based on the mode and the parameters.
342 Parameters
343 ----------
344 delta_t
345 Override the delta_t parameter, otherwise the default value is used from the constructor
346 n_events
347 Override the n_events parameter, otherwise the default value is used from the constructor
349 Returns
350 -------
351 EventArray
352 An array with the events
354 Examples
355 --------
356 >>> reader = EventReader(b"% format EVT3;height=720;width=1280\\n% end\\n", delta_t=10000)
357 >>> events = reader.read()
358 >>> print(len(events))
359 0
360 >>> # Override the time window for a single read
361 >>> more_events = reader.read(delta_t=5000)
363 """
364 self._check_no_active_prefetch()
365 out = self._read(delta_t, n_events)
366 if self._real_time:
367 self._pace(out)
368 return out
370 def _pace(self, out: Any) -> None:
371 """Sleep until the chunk's last event timestamp aligns with wall-clock
372 playback time (anchored at the first delivered chunk).
374 The anchor is absolute, so time the caller spends between chunks is
375 subtracted from the delay automatically; when the target time is
376 already past (slow decode or slow consumer), no delay is added.
377 """
378 ev = out[0] if isinstance(out, tuple) else out
379 if len(ev) == 0:
380 return
381 t_last = int(ev.t[-1])
382 now = time.perf_counter()
383 if self._pace_anchor is None:
384 self._pace_anchor = (now, int(ev.t[0]))
385 wall0, ts0 = self._pace_anchor
386 target = wall0 + (t_last - ts0) / (1e6 * self._playback_speed)
387 delay = target - now
388 # A wait longer than max_gap means the recording went idle (e.g. a
389 # silent lead-in before the action starts). Rather than stall on a
390 # single multi-second sleep, skip the dead air: re-anchor to now so the
391 # next window is paced relative to this one.
392 if self._max_gap is not None and delay > self._max_gap:
393 self._pace_anchor = (now, t_last)
394 return
395 if delay > 0:
396 time.sleep(delay)
398 def _read(self, delta_t:int|None=None, n_events:int|None=None) -> Any:
399 """Unguarded body of :meth:`read` (also driven by the prefetch worker)."""
400 # If not initialized, initialize
401 if not self._is_initialized:
402 self.init()
404 # Allocate the staging accumulator on first use.
405 if self._buffer is None:
406 self._buffer = EventAccumulator(self._acc_capacity)
407 acc = self._buffer
409 # "all" mode (without per-call overrides) means the whole remaining
410 # stream: skip the window cutoffs entirely and drain to EOF, so files
411 # exceeding max_time/max_events are still returned in full.
412 drain = self._mode == "all" and delta_t is None and n_events is None
414 # Override the parameters if they are specified
415 if delta_t is None:
416 delta_t = self._delta_t
417 if n_events is None:
418 n_events = self._n_events
420 # Establish the first timestamp once, at the very start of the stream.
421 if self._n_read_events == 0 and len(acc) == 0:
422 if self._pull(acc, delta_t, n_events) == 0:
423 self._eof = True
424 if self._read_external_triggers:
425 from ..types import TriggerArray
426 return EventArray.empty(), TriggerArray.empty()
427 return EventArray.empty()
428 self._first_ts = int(acc.t_window()[0])
429 self._current_ts = self._first_ts
431 start_ts: int = self._current_ts
432 end_ts: int = start_ts + delta_t # Final end_ts if we reach delta_t
433 end_idx: int = len(acc)
435 tr_end_idx: int = acc._tr.size - acc._tr_start
437 # Gather events until we hit the n_events count, the delta_t time window,
438 # or the end of the stream. Work directly on the SoA `t` column.
439 # Both cutoffs are evaluated together: whichever falls earlier in the
440 # stream wins (the buffer may hold far more than one window's worth).
441 while True:
442 t = acc.t_window()
443 time_ready = not drain and len(t) > 0 and t[-1] >= end_ts
444 count_ready = not drain and len(acc) > n_events
446 if time_ready or count_ready:
447 time_idx = int(np.searchsorted(t, end_ts)) if time_ready else len(acc) + 1
448 tr_t = acc.t_window_tr()
449 if count_ready and time_idx > n_events:
450 # n_events cutoff comes first
451 end_idx = n_events
452 self._current_ts = int(t[n_events])
453 tr_end_idx = int(np.searchsorted(tr_t, self._current_ts, side='left'))
454 else:
455 # delta_t cutoff comes first (ties go to the time window)
456 end_idx = time_idx
457 self._current_ts += delta_t
458 tr_end_idx = int(np.searchsorted(tr_t, end_ts, side='left'))
459 break
461 # Not enough buffered yet: pull more from the decoder.
462 if self._pull(acc, delta_t, n_events) == 0:
463 self._eof = True
464 # Neither cutoff met, so the whole remaining buffer is the slice.
465 end_idx = len(acc)
466 tr_end_idx = acc._tr.size - acc._tr_start
467 break
469 # Copy out the window (independent) and advance past it.
470 output, output_tr = acc.slice_copy(end_idx, tr_end_idx)
471 self._n_read_events += end_idx
473 if self._normalize_ts:
474 # Normalize the timestamps to start from zero at start_ts. slice_copy
475 # already returned an independent array, so this is safe in place.
476 output.t -= self._first_ts - self._start_ts
477 if len(output_tr) > 0:
478 output_tr.t -= self._first_ts - self._start_ts
480 if self._read_external_triggers:
481 return output, output_tr
482 return output
484 def read_all(self) -> 'EventArray | tuple[EventArray, TriggerArray]':
485 """Decode and return every remaining event at once.
487 Delegates to the decoder's :meth:`~evutils.io.common.EventDecoder.read_all`,
488 which (for the native EVT/DAT/AER decoders) decodes the whole payload
489 straight into a single output buffer -- no per-chunk copy, no final
490 ``concatenate`` -- and hands the columns back as a zero-copy
491 :class:`EventArray`. This bypasses the slicing ring buffer that
492 :meth:`read` uses for ``delta_t``/``n_events`` windowing.
494 .. note::
495 This materialises every event in memory at once. For recordings too
496 large to fit, iterate the reader (windowed :meth:`read`) instead.
498 Returns
499 -------
500 EventArray
501 All remaining events.
503 Examples
504 --------
505 >>> reader = EventReader(b"% format EVT3;height=720;width=1280\\n% end\\n")
506 >>> all_events = reader.read_all()
507 >>> print(f"Total events: {len(all_events)}")
508 Total events: 0
510 """
511 self._check_no_active_prefetch()
512 if not self._is_initialized:
513 self.init()
515 _out = self._file_decoder.read_all()
516 if self._read_external_triggers:
517 if isinstance(_out, tuple):
518 out, out_tr = _out
519 else:
520 out = _out
521 from ..types import TriggerArray
522 out_tr = TriggerArray.empty()
523 else:
524 out = _out # type: ignore
526 # Prepend anything already buffered by prior read() calls (rare: only if
527 # read() and read_all() are mixed on the same reader).
528 if self._buffer is not None and (len(self._buffer) > 0 or self._buffer._tr.size - self._buffer._tr_start > 0):
529 buffered, buffered_tr = self._buffer.slice_copy(len(self._buffer), self._buffer._tr.size - self._buffer._tr_start)
530 if len(out) == 0:
531 out = buffered
532 elif len(buffered) > 0:
533 out = EventArray(
534 np.concatenate([buffered.t, out.t]),
535 np.concatenate([buffered.x, out.x]),
536 np.concatenate([buffered.y, out.y]),
537 np.concatenate([buffered.p, out.p]),
538 )
540 if self._read_external_triggers:
541 if len(out_tr) == 0:
542 out_tr = buffered_tr
543 elif len(buffered_tr) > 0:
544 from ..types import TriggerArray
545 out_tr = TriggerArray(
546 np.concatenate([buffered_tr.t, out_tr.t]),
547 np.concatenate([buffered_tr.p, out_tr.p]),
548 np.concatenate([buffered_tr.id, out_tr.id]),
549 )
551 self._eof = True
552 self._n_read_events += len(out)
554 if self._normalize_ts and len(out) > 0:
555 # Capture the shift before modifying out.t, so triggers get the
556 # same normalization.
557 shift = int(out.t[0]) - self._start_ts
558 out.t -= shift
559 if self._read_external_triggers and len(out_tr) > 0:
560 out_tr.t -= shift
562 if self._read_external_triggers:
563 return out, out_tr
564 return out
566 def reset(self) -> None:
567 """Reset file reader back to the beginning of the file.
569 An active asynchronous iterator is cancelled first (its remaining
570 buffered chunks are discarded).
571 """
572 if self._active_prefetch is not None:
573 self._active_prefetch.close()
574 self._n_read_events = 0
575 self._eof = False
576 self._pace_anchor = None
577 if self._buffer is not None:
578 self._buffer.reset()
579 self._file_decoder.reset()
581 def __enter__(self) -> "EventReader":
582 return self
584 def is_eof(self) -> bool:
585 """Check if the end of the file is reached.
587 Returns
588 -------
589 bool
590 True if the end of the file is reached, False otherwise
592 """
593 if not self._is_initialized:
594 self.init()
595 return self._eof and (self._buffer is None or len(self._buffer) == 0)
597 def close(self) -> None:
598 """Close the reader and release resources (decoder buffer views, then the
599 underlying byte source). Cancels any active asynchronous iterator first.
600 """
601 if self._active_prefetch is not None:
602 self._active_prefetch.close()
603 # Drop decoder views (e.g. into an mmap) before closing the source.
604 self._file_decoder.close()
605 self._source.close()
607 def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
608 self.close()
610 def __repr__(self) -> str:
611 if self._is_initialized:
612 is_initialized_txt = "initialized"
613 else:
614 is_initialized_txt = "not initialized"
615 src = self._file_name if self._file_name is not None else self._source.__class__.__name__
616 return f"{self.__class__.__name__}(source={src} - {is_initialized_txt}, delta_t={self._delta_t}, n_events={self._n_events}, mode={self._mode})"
618 def __len__(self) -> int:
619 return self._n_read_events
621 def __iter__(self) -> Any:
622 """Iterate over the events in the file.
624 With ``async_read=True`` the windows are decoded ahead in a background
625 thread (bounded to a couple of windows), overlapping decode with the
626 caller's per-chunk processing. Only one asynchronous iterator can be
627 active per reader; direct :meth:`read` / :meth:`read_all` calls raise
628 until it is exhausted or closed.
630 Yields
631 ------
632 EventArray
633 An array with the events
635 Examples
636 --------
637 >>> reader = EventReader(b"% format EVT3;height=720;width=1280\\n% end\\n", n_events=5000)
638 >>> for events in reader:
639 ... print(f"Received chunk of {len(events)} events")
640 Received chunk of 0 events
642 """
643 it: Any
644 if self._async_read:
645 self._check_no_active_prefetch()
646 it = PrefetchIterator(
647 self._iter_sync(),
648 on_finish=lambda: setattr(self, "_active_prefetch", None),
649 )
650 self._active_prefetch = it
651 else:
652 it = self._iter_sync()
653 if self._real_time:
654 # Pace on the consumer side, never inside the decode path: with
655 # async_read the worker keeps decoding ahead while we sleep.
656 return self._paced_iter(it)
657 return it
659 def _paced_iter(self, it: Any) -> Any:
660 """Wrap an iterator so each chunk is released on the playback clock."""
661 try:
662 for chunk in it:
663 self._pace(chunk)
664 yield chunk
665 finally:
666 # Propagate early termination (break / close) to a prefetching
667 # iterator so its worker thread is released.
668 if hasattr(it, "close"):
669 it.close()
671 def _iter_sync(self) -> Any:
672 """The plain synchronous window generator behind :meth:`__iter__`."""
673 if not self._is_initialized:
674 self.init()
675 while not self.is_eof():
676 yield self._read()
678 def shape(self) -> tuple[int|None, int|None]:
679 """Get the shape of the frame.
681 Returns
682 -------
683 tuple[int, int]
684 The shape of the frame (width, height)
686 """
687 if not self._is_initialized:
688 self.init()
689 if self._width is not None and self._height is not None:
690 return self._width, self._height
691 else:
692 return self._file_decoder.shape()
695 def tell(self) -> int:
696 """Get the current position in the file.
698 Returns
699 -------
700 int
701 The current position in the file
703 """
704 if not self._is_initialized:
705 self.init()
706 return self._file_decoder.tell()