ezmsg.nwb.iterator#

Classes

class NWBAxisArrayIterator(*args, **kwargs)[source]#

Bases: BaseStatefulProducer[NWBIteratorSettings, AxisArray, NWBIteratorState]

__init__(*args, **kwargs)[source]#
property exhausted: bool#
close()[source]#

Shut down the prefetch worker (if any) and close the slicer.

Safe to call repeatedly. Prefer this over relying on __del__ — in the rare cases where the iterator survives a del because some other reference is still alive, close() lets the caller drop resources deterministically rather than waiting for GC.

Return type:

None

class NWBIteratorSettings(filepath, chunk_dur=1.0, reference_clock=ReferenceClockType.SYSTEM, reref_now=False, self_terminating=True, stream_keys=None, prefetch_chunks=0, rdcc_nbytes=67108864, rdcc_nslots=10007, gap_tol=0.5, dejitter=True, clock_groups=None, dejitter_cache=True, real_gap_threshold=None, structured_ch_axis=False, storage_options=None)[source]#

Bases: Settings

Parameters:
filepath: PathLike | str#
chunk_dur: float = 1.0#
reference_clock: ReferenceClockType = 'system'#
reref_now: bool = False#
self_terminating: bool = True#
stream_keys: list[str] | None = None#
prefetch_chunks: int = 0#

Number of chunks to prefetch in a background thread.

0 (default) keeps the original synchronous behaviour: each next() call blocks on h5py I/O for the next chunk. > 0 spawns a daemon thread that fills a bounded queue with up to prefetch_chunks chunks of pre-read data, so the consumer rarely waits on disk. 2–4 is typically enough; the win is overlapping one chunk’s read with one chunk’s downstream compute.

rdcc_nbytes: int = 67108864#

HDF5 raw data chunk cache size in bytes (forwarded to NWBSlicer).

rdcc_nslots: int = 10007#

HDF5 raw data chunk cache slot count (forwarded to NWBSlicer).

gap_tol: float = 0.5#

Gap threshold for timestamped continuous streams, as a fraction of the nominal sample period: a gap is declared when an interval exceeds (1 + gap_tol) / fs. Chunks spanning a gap are split into gap-free messages so the regular LinearAxis never misplaces post-gap samples.

The default 0.5 (1.5x period) sits between neural-data jitter (<~1.05x) and the smallest real gap (one dropped sample = ~2x), so it catches every gap without splitting on jitter. Set very large to disable splitting. No effect on rate-only streams or event tables.

dejitter: bool = True#

Reconstruct smoothed monotone timestamps for streams with a paired *_device_ts sibling before chunking (forwarded to NWBSlicer). Removes the per-sample clock jitter that would otherwise gap-split such a stream into thousands of tiny messages. Set False to chunk raw timestamps unchanged.

clock_groups: list[list[str]] | None = None#

Override auto clock-group detection (forwarded to NWBSlicer). None auto-detects PTP-shared groups; a list of key-lists forces them.

dejitter_cache: bool = True#

Cache reconstructed timestamps on disk across opens (forwarded to NWBSlicer).

real_gap_threshold: float | None = None#

Real-gap guard threshold in seconds (forwarded to NWBSlicer). None auto-derives per stream; genuine gaps are preserved so chunks still split at them. float("inf") disables the guard.

structured_ch_axis: bool = False#

Build the ch axis as a structured record array from the electrodes table (forwarded to NWBSlicer), matching what a live acquisition source emits: position, label, bank, headstage, array identity. Default False – these columns are an acquisition-stack convention, not part of the NWB schema. See electrodes.

storage_options: dict | None = None#

fsspec options for a remote Zarr store (e.g. account_name for abfs://).

__init__(filepath, chunk_dur=1.0, reference_clock=ReferenceClockType.SYSTEM, reref_now=False, self_terminating=True, stream_keys=None, prefetch_chunks=0, rdcc_nbytes=67108864, rdcc_nslots=10007, gap_tol=0.5, dejitter=True, clock_groups=None, dejitter_cache=True, real_gap_threshold=None, structured_ch_axis=False, storage_options=None)#
Parameters:
Return type:

None

class NWBIteratorState[source]#

Bases: object

n_chunks: int = 0#
chunk_ix: int = 0#
slicer: NWBSlicer | None = None#
streams: dict | None = None#
deque: deque | None = None#
prefetch_queue: Any = None#
prefetch_thread: Any = None#
prefetch_stop: Any = None#