ezmsg.nwb.pipeline_settings#

Typed-column pipeline-settings sink for ezmsg-nwb.

Phase 2 of the pipeline-settings story. Where the generic NWBSink lands PipelineSettingsEvent messages in a JSON-encoded AnnotationSeries (one string column, fixed schema), this subclass projects each event’s structured_value into a pynwb.epoch.TimeIntervals table named pipeline_settings with one column per dotted-key path. Columns get their values’ native dtypes, so analysis tools can read settings as a normal pandas DataFrame.

Interval semantics#

Each event opens a new interval. The PREVIOUS interval is closed (and written to disk) at the moment the next event arrives — its stop_time is the new event’s timestamp. The currently-active (open) interval is held in memory and flushed at file close.

This means: a non-graceful crash will lose the most recent open interval. That’s accepted as the cost of interval semantics — your collaborator chose this trade-off.

File rotation on schema change#

If an incoming event introduces a new column, or changes the per-cell shape of an existing column (scalar↔array, rank change, fixed-shape mismatch), the sink rotates into a fresh file segment whose table opens already populated with the new schema. Each segment’s table is internally consistent. Reading back means iterating <stem>_NN.nwb files in order.

Use the bundled PipelineSettingsTableCollection to wire a PipelineSettingsUnit (from ezmsg-baseproc) to this sink in a single component.

Classes

class NWBPipelineSettingsSink(*args, settings=None, **kwargs)[source]#

Bases: NWBSink

Specialized NWBSink that writes PipelineSettingsEvent messages into a typed-column TimeIntervals table.

Drop-in replacement for NWBSink when used inside PipelineSettingsTableCollection. Inherits all of NWBSink’s data-recording behaviour; replaces on_annotation so only PipelineSettingsEvent traffic on INPUT_ANNOTATION is consumed (other annotation-shaped messages are silently dropped — use NWBSink directly if you need the JSON-AnnotationSeries fallback).

Parameters:

settings (Settings | None)

SETTINGS#

alias of NWBPipelineSettingsSinkSettings

create_processor()[source]#

Construct the typed-column consumer rather than the generic one.

Return type:

None

async on_annotation(msg)[source]#

Dispatch PipelineSettingsEvent to the typed-column writer.

Non-PipelineSettingsEvent messages are silently ignored — keep the typed semantics clean. Wrap with another sink (or run a plain NWBSink in parallel) if you need JSON annotations alongside.

Return type:

None

Parameters:

msg (Any)

class NWBPipelineSettingsSinkConsumer(*args, settings=None, **kwargs)[source]#

Bases: NWBSinkConsumer

Adds typed-column TimeIntervals writing on top of NWBSinkConsumer.

Public method: write_settings_event(). The unit subclass calls it on each PipelineSettingsEvent. Internally the consumer detects schema-incompatible changes and rotates the file via _rotate_file().

Parameters:

settings (SettingsType)

classmethod get_state_type()[source]#
Return type:

type

write_settings_event(event)[source]#

Project a PipelineSettingsEvent into native columns and append/rotate.

Aggregation rules:

  • Per-component INITIAL events at startup are buffered in memory (pending_initial_state). The table is not registered with the file yet — registering it per component would force a schema-driven rotation per component.

  • The ``INIT_FINAL_COMPONENT_ADDRESS`` sentinel flushes the buffer as ONE merged anchor row, registers the table, and opens an interval tracking the merged state.

  • Any non-INITIAL event arriving while the buffer is populated (producer dropped the sentinel) flushes the buffer first as if the sentinel had arrived, then processes the event normally.

  • Subsequent schema-compatible event closes the prior open interval (writes a row [prev.timestamp, this.timestamp]) and opens a new one.

  • Schema-incompatible event rotates into a new file segment whose table opens with the merged schema and a fresh anchor.

  • Close flushes any pending initial buffer (merged anchor) and the still-open interval as [active_since, close_time].

Return type:

None

Parameters:

event (PipelineSettingsEvent)

class NWBPipelineSettingsSinkSettings(filepath, overwrite_old=False, axis='time', recording=True, inc_clock=ReferenceClockType.SYSTEM, meta_yaml=None, split_bytes=0, expected_series=None, pipeline_settings_table_name='pipeline_settings')[source]#

Bases: NWBSinkSettings

Settings for NWBPipelineSettingsSink.

Inherits all file-level fields from NWBSinkSettings (filepath, overwrite_old, inc_clock, recording, split_bytes, meta_yaml, expected_series, axis).

Parameters:
pipeline_settings_table_name: str = 'pipeline_settings'#

Name of the per-file TimeIntervals table to write into. Lives under nwbfile.intervals[<name>].

__init__(filepath, overwrite_old=False, axis='time', recording=True, inc_clock=ReferenceClockType.SYSTEM, meta_yaml=None, split_bytes=0, expected_series=None, pipeline_settings_table_name='pipeline_settings')#
Parameters:
Return type:

None

class NWBPipelineSettingsSinkState[source]#

Bases: NWBSinkState

State for NWBPipelineSettingsSinkConsumer.

Adds in-memory bookkeeping for the open interval. settings_columns locks the table schema for the current file segment; settings_state mirrors the most-recent flat settings dict so a closed interval can be written without re-flattening.

settings_columns: List[str] | None = None#
settings_state: Dict[str, Any] | None = None#
settings_active_since: float | None = None#
settings_prev_component: str = '__init__'#
pending_initial_state: Dict[str, Any] | None = None#

Per-component INITIAL events accumulate here while the startup snapshot is in flight. None means we’re not currently buffering; the dict is populated by per-component INITIALs and flushed to one merged anchor row on the INIT_FINAL_COMPONENT_ADDRESS sentinel (or on the first non-INITIAL event, if the producer dropped the sentinel).

pending_initial_first_seen: float | None = None#

Timestamp of the first INITIAL event in the current buffer — used as the merged anchor row’s start_time when we eventually flush.

class PipelineSettingsTableCollection(*args, settings=None, **kwargs)[source]#

Bases: Collection

Bundle a PipelineSettingsUnit with an NWBPipelineSettingsSink.

Accepts an external INPUT_SIGNAL (relayed to the sink) so the Collection drops into a pipeline as a normal NWB sink, with settings logging happening transparently alongside acquisition recording. Also relays INPUT_SETTINGS so users can push NWBSinkSettings updates from outside the Collection.

Parameters:

settings (Settings | None)

SETTINGS#

alias of PipelineSettingsTableCollectionSettings

PUB = <ezmsg.baseproc.util.pipeline_settings.PipelineSettingsUnit object>#
SINK = <ezmsg.nwb.pipeline_settings.NWBPipelineSettingsSink object>#
INPUT_SIGNAL = InputTopicStream:unlocated[AxisArray]()#
INPUT_SETTINGS = InputTopicStream:unlocated[NWBPipelineSettingsSinkSettings]()#
configure()[source]#

A lifecycle hook that runs when the Collection is instantiated.

This is the best place to call Unit.apply_settings() on each member Unit of the Collection. Override this method to perform collection-specific configuration of child components.

Return type:

None

network()[source]#

Override this method and have the definition return a NetworkDefinition which defines how InputStreams and OutputStreams from member Units will be connected.

The NetworkDefinition specifies the message routing between components by connecting output streams to input streams.

Returns:

Network definition specifying stream connections

Return type:

Iterable[tuple[Stream | str, Stream | str]]

class PipelineSettingsTableCollectionSettings(producer, sink)[source]#

Bases: Settings

Parameters:
producer: PipelineSettingsProducerSettings#
sink: NWBPipelineSettingsSinkSettings#
__init__(producer, sink)#
Parameters:
Return type:

None