ezmsg.nwb.writer#

This module provides a sink node that writes incoming data to an NWB file.

Iteratively writing data to an NWB file has a few quirks.

The main approach to iterative writing in the docs – wrapping the data source in a DataChunkIterator – does not work

well for our scenario. First, we would need to use some sort of queues to communicate between the data-receiving function and the thread / task that is yielding data to the iterator; this seems doable. Second, it’s not clear if each DataChunkIterator must exhaust itself before moving onto the next one or if they can run in parallel; this is a deal-breaker for us. Third, it seems impossible to add new TimeSeries after the io.write(nwbfile) process has begun; this is also a deal-breaker for us.

Instead, we use the “user-defined dataset write” approach, described here. With this approach, we use pynwb to construct the NWB structure, but we specify the data containers (both data and timestamps) as H5DataIO objects. After the NWB structure is written to disk, the containers expose h5py Dataset objects that we can resize and append to as data arrive.

This approach has some quirks that we have worked around and landed on the following strategies:

A - When all incoming streams are known a priori

  • Load metadata (shape, rate, channel labels) from a yaml or json file.

  • Create all H5DataIO objects for epochs and trials: id, start_time, stop_time, and custom “label” column. * Append dummy event to table. Remove this step after hdmf-dev/hdmf#1000

  • Create device, electrode group, and add all known electrodes.

  • Create all H5DataIO objects for continuous data (.data and .timestamps)

  • Attach .data and .timestamps to new TimeSeries, or new ElectricalSeries if electrodes are known.

  • Flush the NWBFile to disk – Slow!! but necessary to create h5py Datasets

  • Reopen the NWBFile in append mode – Fast; necessary to make tables appendable, but freezes the electrodes table.

  • Add any .rate attributes to the timestamps h5py datasets.

  • Get references to the .data and .timestamps datasets for each continuous data appending.

B - When streams are not known a priori

  • For the first message of a stream… * If it is continuous data, even size-zero

    • Create the H5DataIO objects for .data and .timestamps

    • If it is the FIRST continuous data stream with channel labels then create the device, elec group and electrodes

    • Attach .data and .timestamps to TimeSeries, or ElectricalSeries if electrodes are known (i.e. FIRST).

    • Flush the NWBFile to disk – Slow!! but necessary to create h5py Datasets

    • Add any .rate attributes to the timestamps h5py datasets

    • If it is an event… * Create all H5DataIO objects for epochs and trials: id, start_time, stop_time, and custom “label” column. * Append dummy event to table. Remove this step after hdmf-dev/hdmf#1000 * Flush the NWBFile to disk – Slow!! but necessary to create h5py Datasets * Reopen the NWBFile in append mode – Fast; necessary to make tables appendable, but freezes the electrodes table.

In this case, we must call io.write(nwbfile) once for each stream, which can significantly slow down performance for 1.5-4 seconds * number of unique streams. Thus, we strongly recommend running NWBSink in a separate process where possible.

C - When closing a file on quit or when splitting

  • When done appending to datasets and events tables, Flush the NWB file once more – Slow!!

  • If the file has no data other than the dummy event and recording is disabled, delete the file.

D - When opening the next file when splitting

  • Copy as much information from the old NWBFile to the new NWBFile as possible.

  • Close as in C.

  • Follow pattern A where the metadata comes from the old datasets instead of loaded from a file.

Classes

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

Bases: BaseConsumerUnit[NWBSinkSettings, AxisArray, NWBSinkConsumer]

Parameters:

settings (Settings | None)

SETTINGS#

alias of NWBSinkSettings

INPUT_ANNOTATION = InputStream:unlocated[object]()#
async on_annotation(msg)[source]#

Append one row to the per-table AnnotationSeries named by msg.table_name.

Ignores messages that don’t quack like an NWBPointRow so other traffic on this stream (if any) doesn’t crash the sink. Any exception from the writer is logged and swallowed — annotation logging is best-effort and must not break primary acquisition recording.

Return type:

None

Parameters:

msg (Any)

async shutdown()[source]#

Runs when the Unit terminates.

This is called from within the same process this unit will live in. This lifecycle hook can be overridden. It can be run as async functions by simply adding the async keyword when overriding.

This method is where you should clean up resources and perform any necessary shutdown procedures.

Return type:

None

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

Bases: BaseStatefulConsumer[NWBSinkSettings, AxisArray, NWBSinkState]

Parameters:

settings (SettingsType)

shared_start_datetime: datetime | None = None#
shared_t0: float | None = None#
shared_clock_type: ReferenceClockType | None = None#
NONRESET_SETTINGS_FIELDS: ClassVar[frozenset[str]] = frozenset({'recording', 'split_bytes'})#
__init__(*args, settings=None, **kwargs)[source]#
Parameters:

settings (NWBSinkSettings | None)

property path_on_disk: Path#
get_session_datetime(try_t0=None)[source]#
Retrieve session datetime. If it does not already exist, set it with try_t0 if provided

and clock is understood, else with datetime.now.

Parameters:

try_t0 (Optional[float]) – (Optional) The first incoming timestamp. If provided, it is used to set the session start time. The value is timezone-naive and assumed to be in the same clock as time.time().

Return type:

datetime

Returns:

Common session starttime among all instances of this class

get_session_timestamp(try_t0=None)[source]#

Retrieve session timestamp. If it does not already exist, set it with try_t0 if provided.

Parameters:

try_t0 (Optional[float]) – (Optional) The first incoming timestamp.

Return type:

float

Returns:

Common session timestamp among all instances of this class.

close(write=False, log=True)[source]#

Close the file. This will also delete the file if it is empty.

Parameters:

  • write – Set True to write the file to disk before closing.

  • log – Set True to log the closing and deletion of the file. This must be kept False when calling from __del__.

Return type:

None

toggle_recording(recording=None)[source]#

Deprecated. Send a NWBSinkSettings update with the desired recording value instead — the update is routed through update_settings() and takes effect on the next message.

Parameters:

recording (bool | None)

write_annotation(table_name, timestamp, data)[source]#

Append one row to the named AnnotationSeries, creating it if needed.

timestamp is wall-clock seconds. The value stored in the file is timestamp - start_timestamp once a data message has set start_timestamp (the file-relative convention NWB expects); before then, the wall-clock value is passed through unmodified — the sink has no baseline to subtract against, and inventing one (e.g. from datetime.now()) would silently desync from the eventual data anchor. Annotations written before any data arrives therefore carry whatever baseline their producer chose; downstream readers should treat them accordingly.

Return type:

None

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

Bases: Settings

Parameters:
filepath: str | PathLike#
overwrite_old: bool = False#
axis: str = 'time'#
recording: bool = True#
inc_clock: ReferenceClockType = 'system'#
meta_yaml: str | PathLike | None = None#
split_bytes: int = 0#
expected_series: str | PathLike | None = None#
__init__(filepath, overwrite_old=False, axis='time', recording=True, inc_clock=ReferenceClockType.SYSTEM, meta_yaml=None, split_bytes=0, expected_series=None)#
Parameters:
Return type:

None

class NWBSinkState[source]#

Bases: object

hash: int = -1#
filepath: Path | None = None#
io: NWBHDF5IO | None = None#
nwbfile: NWBFile | None = None#
series: Dict[str, SeriesState] | None = None#
start_timestamp: float = 0.0#
split_count: int = 0#
annotation_data: Dict[str, Any] | None = None#
annotation_ts: Dict[str, Any] | None = None#
class SeriesState(shape, data=None, ts=None, bytes_written=0)[source]#

Bases: object

Per-stream bookkeeping for one NWB series.

shape is the trailing (non-time) shape used by NWBSinkConsumer._check_msg_consistency(). data and ts hold the live h5py.Dataset (or pynwb container) references used by NWBSinkConsumer._process() to append rows. bytes_written tracks cumulative append size for the file-split threshold.

Parameters:
shape: Tuple[int, ...]#
data: Any = None#
ts: Any = None#
bytes_written: int = 0#
__init__(shape, data=None, ts=None, bytes_written=0)#
Parameters:
Return type:

None