ezmsg.nwb.clockdriven#

Clock-driven NWB producer for synchronizing NWB data to a shared clock.

Classes

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

Bases: BaseClockDrivenProducer[NWBClockDrivenSettings, NWBClockDrivenState]

Produces NWB data synchronized to clock ticks.

Three extraction strategies based on stream type:

  1. Rate-only continuous: sample-index-based using fs and counter.

  2. Timestamped continuous: time-window-based using searchsorted.

  3. Events/intervals: time-window-based, returning all events in window.

Playback can start at an arbitrary offset and run at an arbitrary rate relative to the driving clock; see start_offset and playback_rate on NWBClockDrivenSettings.

NONRESET_SETTINGS_FIELDS: ClassVar[frozenset[str]] = frozenset({'n_time', 'playback_rate', 'start_offset'})#
update_settings(new_settings)[source]#

Apply new settings to this processor, requesting a state reset if needed.

Diffs new_settings against the current self.settings. If every changed field is listed in NONRESET_SETTINGS_FIELDS, self.settings is simply rebound and the processor keeps running. Otherwise, _request_reset() is called so the next message triggers a fresh _reset_state (for stateful subclasses). Override for finer-grained control than the class-level allow-list provides.

Return type:

None

Parameters:

new_settings (NWBClockDrivenSettings)

property exhausted: bool#
seek(start_offset)[source]#

Move the playback position to start_offset seconds past the stream’s t0.

Safe to call after _reset_state has run. The file is not reopened and no slicer state is torn down, so this is cheap enough to call every time a scrub-widget emits.

Return type:

None

Parameters:

start_offset (float)

class NWBClockDrivenSettings(fs=0.0, n_time=None, filepath='', stream_key='', reference_clock=ReferenceClockType.SYSTEM, reref_now=False, start_offset=0.0, playback_rate=1.0, 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: ClockDrivenSettings

Parameters:
filepath: PathLike | str = ''#
stream_key: str = ''#
reference_clock: ReferenceClockType = 'system'#
reref_now: bool = False#
fs: float = 0.0#

Output sampling rate in Hz.

start_offset: float = 0.0#

Playback position (seconds from the stream’s t0) at which to begin reading.

playback_rate: float = 1.0#

Multiplier applied to the clock-tick duration when advancing through the file.

1.0 is real-time, 2.0 consumes two seconds of file per wall-clock second, 0.0 pauses (the producer emits nothing). Values other than 0.0/1.0 are only meaningful when n_time is None (variable chunk size); with a fixed n_time the rate acts as a pause/unpause gate since chunk size is fixed.

gap_tol: float = 0.5#

Gap threshold for timestamped continuous streams, as a fraction of the nominal sample period (forwarded to NWBSlicer.read_by_time).

The clock cadence is never altered: file_t always advances by dt per tick so sibling producers on the same clock stay aligned, and windows that fall inside a gap emit zero-length chunks. When a single window straddles a gap (interval > (1 + gap_tol) nominal periods), that chunk’s time axis becomes a CoordinateAxis carrying the true per-sample timestamps rather than a uniform LinearAxis that would misplace post-gap samples. Set very large to keep a single LinearAxis across gaps. No effect on event streams.

dejitter: bool = True#

Reconstruct smoothed monotone timestamps for streams with a paired *_device_ts sibling (forwarded to NWBSlicer). Set False to read raw timestamps unchanged.

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

Override auto clock-group detection (forwarded to NWBSlicer).

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; 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__(fs=0.0, n_time=None, filepath='', stream_key='', reference_clock=ReferenceClockType.SYSTEM, reref_now=False, start_offset=0.0, playback_rate=1.0, 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 NWBClockDrivenState[source]#

Bases: ClockDrivenState

slicer: NWBSlicer | None = None#
detected_fs: float = 0.0#
has_timestamps: bool = False#
is_event: bool = False#
template: AxisArray | None = None#
sample_idx: int = 0#
n_total_samples: int = 0#
t0: float = 0.0#
file_t: float = 0.0#
class NWBClockDrivenUnit(*args, settings=None, **kwargs)[source]#

Bases: BaseClockDrivenUnit[NWBClockDrivenSettings, NWBClockDrivenProducer]

Clock-driven NWB unit that reads one stream synchronized to a shared clock.

Parameters:

settings (Settings | None)

SETTINGS#

alias of NWBClockDrivenSettings