ezmsg.xdf.iter

Functions

labels_from_strm(strm)

Parameters:

strm (dict)

Return type:

list[str]

Classes

class XDFAxisArrayIterator(*args, **kwargs)

Bases: _XDFProducerBase, BaseStatefulProducer[XDFIteratorSettings, AxisArray, XDFIteratorState]

Loads a single stream and produces one AxisArray per chunk.

select must be a single stream name, unlike XDFIterator.

property exhausted: bool

class XDFIterator(filepath, select=None, chunk_dur=1.0, start_time=None, stop_time=None, rezero=True)

Bases: object

Parameters:

• filepath (Path | str)

• select (set[str] | None)

• chunk_dur (float)

• start_time (float | None)

• stop_time (float | None)

• rezero (bool)

__init__(filepath, select=None, chunk_dur=1.0, start_time=None, stop_time=None, rezero=True)

An Iterator that yields chunks from an XDF. A typical offline analysis might load the entire file into memory, then perform a processing step on the entire recording duration, and the next step on the entire result of the first step, and so on. This might require a tremendous amount of memory and, if one is not careful about memory layout, can be incredibly slow. An alternative procedure is to load the file into memory a chunk at a time (see Note1), then pass that chunk through the entire processing pipeline, then proceed onto the next chunk (See Note2). We create an Iterator to provide our chunks. > Note1: I have not written a true lazy-loader for XDF because it has not yet been necessary as the files are

all small. Thus, I use pyxdf.load_xdf which loads the entire raw data into memory. The processing is still done chunk-by-chunk.

> Note2: It should be possible to start on chunk[ix+1] while chunk[ix] is still going through the pipeline.

Indeed, this is (optionally) how it works online. However, the overhead of setting this up for offline analysis is not worth the gain, at least not at this stage.

Parameters:

• filepath (Path | str) – The path to the file to load and iterate over.

• select (set[str] | None) – (Optional) A set of stream names to select. If None, then all streams are selected.

• chunk_dur (float) – The duration of each chunk in seconds.

• start_time (float | None) – Start playback at this time. If rezero is True then this is relative to the file start time. If rezero is False then this is relative to the original timestamps.

• stop_time (float | None) – Truncate the playback to stop at this time. If rezero is True then this is relative to the file start time. If rezero is False then this is relative to the original timestamps.

• rezero (bool) – The absolute value of timestamps in an XDF file are useful for synchronization WITHIN file, but they are absolutely meaningless outside the exact XDF file like in an ezmsg application. Thus, by default we rezero the timestamps to start at t=0.0 for simplicity. However, there may be rare circumstances where one wants to compare the timestamps produced by ezmsg to timestamps produced by another XDF analysis tool that does not rezero. In that case, set rezero=False.

property stream_meta: list[dict] | dict

property n_chunks: int

property exhausted: bool

True once every chunk boundary has been handed out.

class XDFIteratorSettings(filepath, select='', chunk_dur=1.0, start_time=None, stop_time=None, rezero=True, playback_rate=None, self_terminating=False)

Bases: Settings

Settings shared by both AxisArray iterators.

playback_rate and self_terminating belong to the unit rather than to the reader, and are listed in NONRESET_SETTINGS_FIELDS so changing either does not reopen the file.

Parameters:

• filepath (PathLike | str)

• select (str)

• chunk_dur (float)

• start_time (float | None)

• stop_time (float | None)

• rezero (bool)

• playback_rate (float | None)

• self_terminating (bool)

filepath: PathLike | str

select: str = ''

chunk_dur: float = 1.0

start_time: float | None = None

stop_time: float | None = None

rezero: bool = True

playback_rate: float | None = None

self_terminating: bool = False

If True, the unit will raise a ez.NormalTermination exception when the file is exhausted. Note, however, that this will terminate the pipeline even if the data published by this unit are still in transit, which will lead to the pipeline output being truncated before it has finished processing the stream. self_terminating should only be used when it is not important that the pipeline finish processing data, such as during prototyping and testing.

__init__(filepath, select='', chunk_dur=1.0, start_time=None, stop_time=None, rezero=True, playback_rate=None, self_terminating=False)

Parameters:

• filepath (PathLike | str)

• select (str)

• chunk_dur (float)

• start_time (float | None)

• stop_time (float | None)

• rezero (bool)

• playback_rate (float | None)

• self_terminating (bool)

Return type:

None

class XDFIteratorState

Bases: object

reader: XDFIterator | None = None

template: AxisArray | None = None

class XDFMultiAxArrIterator(*args, **kwargs)

Bases: _XDFProducerBase, BaseStatefulProducer[XDFMultiIteratorSettings, AxisArray, XDFMultiIteratorState]

Loads multiple streams and produces one AxisArray per iteration.

Which stream a given message came from varies; read .key. Returns None when a chunk held nothing for any stream, and raises StopIteration only once the file is done.

force_single_sample names irregular-rate streams that may carry more than one event within chunk_dur, which AxisArray cannot represent as a single message with correct timestamps; those are split one event per message.

property exhausted: bool

class XDFMultiIteratorSettings(filepath: Union[os.PathLike, str], select: set[str] | None = None, chunk_dur: float = 1.0, start_time: float | None = None, stop_time: float | None = None, rezero: bool = True, playback_rate: float | None = None, self_terminating: bool = False, force_single_sample: set = <factory>)

Bases: XDFIteratorSettings

Parameters:

• filepath (PathLike | str)

• select (set[str] | None)

• chunk_dur (float)

• start_time (float | None)

• stop_time (float | None)

• rezero (bool)

• playback_rate (float | None)

• self_terminating (bool)

• force_single_sample (set)

select: set[str] | None = None

force_single_sample: set

__init__(filepath, select=None, chunk_dur=1.0, start_time=None, stop_time=None, rezero=True, playback_rate=None, self_terminating=False, force_single_sample=<factory>)

Parameters:

• filepath (PathLike | str)

• select (set[str] | None)

• chunk_dur (float)

• start_time (float | None)

• stop_time (float | None)

• rezero (bool)

• playback_rate (float | None)

• self_terminating (bool)

• force_single_sample (set)

Return type:

None

class XDFMultiIteratorState

Bases: object

reader: XDFIterator | None = None

templates: dict | None = None

pubqueue: SimpleQueue | None = None

class XDFIterator(filepath, select=None, chunk_dur=1.0, start_time=None, stop_time=None, rezero=True)

Bases: object

Parameters:

• filepath (Path | str)

• select (set[str] | None)

• chunk_dur (float)

• start_time (float | None)

• stop_time (float | None)

• rezero (bool)

__init__(filepath, select=None, chunk_dur=1.0, start_time=None, stop_time=None, rezero=True)

An Iterator that yields chunks from an XDF. A typical offline analysis might load the entire file into memory, then perform a processing step on the entire recording duration, and the next step on the entire result of the first step, and so on. This might require a tremendous amount of memory and, if one is not careful about memory layout, can be incredibly slow. An alternative procedure is to load the file into memory a chunk at a time (see Note1), then pass that chunk through the entire processing pipeline, then proceed onto the next chunk (See Note2). We create an Iterator to provide our chunks. > Note1: I have not written a true lazy-loader for XDF because it has not yet been necessary as the files are

all small. Thus, I use pyxdf.load_xdf which loads the entire raw data into memory. The processing is still done chunk-by-chunk.

> Note2: It should be possible to start on chunk[ix+1] while chunk[ix] is still going through the pipeline.

Indeed, this is (optionally) how it works online. However, the overhead of setting this up for offline analysis is not worth the gain, at least not at this stage.

Parameters:

• filepath (Path | str) – The path to the file to load and iterate over.

• select (set[str] | None) – (Optional) A set of stream names to select. If None, then all streams are selected.

• chunk_dur (float) – The duration of each chunk in seconds.

• start_time (float | None) – Start playback at this time. If rezero is True then this is relative to the file start time. If rezero is False then this is relative to the original timestamps.

• stop_time (float | None) – Truncate the playback to stop at this time. If rezero is True then this is relative to the file start time. If rezero is False then this is relative to the original timestamps.

• rezero (bool) – The absolute value of timestamps in an XDF file are useful for synchronization WITHIN file, but they are absolutely meaningless outside the exact XDF file like in an ezmsg application. Thus, by default we rezero the timestamps to start at t=0.0 for simplicity. However, there may be rare circumstances where one wants to compare the timestamps produced by ezmsg to timestamps produced by another XDF analysis tool that does not rezero. In that case, set rezero=False.

property stream_meta: list[dict] | dict

property n_chunks: int

property exhausted: bool

True once every chunk boundary has been handed out.

labels_from_strm(strm)

Parameters:

strm (dict)

Return type:

list[str]

class XDFIteratorSettings(filepath, select='', chunk_dur=1.0, start_time=None, stop_time=None, rezero=True, playback_rate=None, self_terminating=False)

Bases: Settings

Settings shared by both AxisArray iterators.

playback_rate and self_terminating belong to the unit rather than to the reader, and are listed in NONRESET_SETTINGS_FIELDS so changing either does not reopen the file.

Parameters:

• filepath (PathLike | str)

• select (str)

• chunk_dur (float)

• start_time (float | None)

• stop_time (float | None)

• rezero (bool)

• playback_rate (float | None)

• self_terminating (bool)

filepath: PathLike | str

select: str = ''

chunk_dur: float = 1.0

start_time: float | None = None

stop_time: float | None = None

rezero: bool = True

playback_rate: float | None = None

self_terminating: bool = False

If True, the unit will raise a ez.NormalTermination exception when the file is exhausted. Note, however, that this will terminate the pipeline even if the data published by this unit are still in transit, which will lead to the pipeline output being truncated before it has finished processing the stream. self_terminating should only be used when it is not important that the pipeline finish processing data, such as during prototyping and testing.

__init__(filepath, select='', chunk_dur=1.0, start_time=None, stop_time=None, rezero=True, playback_rate=None, self_terminating=False)

Parameters:

• filepath (PathLike | str)

• select (str)

• chunk_dur (float)

• start_time (float | None)

• stop_time (float | None)

• rezero (bool)

• playback_rate (float | None)

• self_terminating (bool)

Return type:

None

class XDFMultiIteratorSettings(filepath: Union[os.PathLike, str], select: set[str] | None = None, chunk_dur: float = 1.0, start_time: float | None = None, stop_time: float | None = None, rezero: bool = True, playback_rate: float | None = None, self_terminating: bool = False, force_single_sample: set = <factory>)

Bases: XDFIteratorSettings

Parameters:

• filepath (PathLike | str)

• select (set[str] | None)

• chunk_dur (float)

• start_time (float | None)

• stop_time (float | None)

• rezero (bool)

• playback_rate (float | None)

• self_terminating (bool)

• force_single_sample (set)

select: set[str] | None = None

force_single_sample: set

__init__(filepath, select=None, chunk_dur=1.0, start_time=None, stop_time=None, rezero=True, playback_rate=None, self_terminating=False, force_single_sample=<factory>)

Parameters:

• filepath (PathLike | str)

• select (set[str] | None)

• chunk_dur (float)

• start_time (float | None)

• stop_time (float | None)

• rezero (bool)

• playback_rate (float | None)

• self_terminating (bool)

• force_single_sample (set)

Return type:

None

class XDFIteratorState

Bases: object

reader: XDFIterator | None = None

template: AxisArray | None = None

class XDFMultiIteratorState

Bases: object

reader: XDFIterator | None = None

templates: dict | None = None

pubqueue: SimpleQueue | None = None

class XDFAxisArrayIterator(*args, **kwargs)

Bases: _XDFProducerBase, BaseStatefulProducer[XDFIteratorSettings, AxisArray, XDFIteratorState]

Loads a single stream and produces one AxisArray per chunk.

select must be a single stream name, unlike XDFIterator.

property exhausted: bool

class XDFMultiAxArrIterator(*args, **kwargs)

Bases: _XDFProducerBase, BaseStatefulProducer[XDFMultiIteratorSettings, AxisArray, XDFMultiIteratorState]

Loads multiple streams and produces one AxisArray per iteration.

Which stream a given message came from varies; read .key. Returns None when a chunk held nothing for any stream, and raises StopIteration only once the file is done.

force_single_sample names irregular-rate streams that may carry more than one event within chunk_dur, which AxisArray cannot represent as a single message with correct timestamps; those are split one event per message.

property exhausted: bool