ezmsg.neo.source#

Classes

class NeoIterator(settings)[source]#

Bases: object

Parameters:

settings (NeoIteratorSettings)

__init__(settings)[source]#
Parameters:

settings (NeoIteratorSettings)

class NeoIteratorSettings(filepath, chunk_dur=0.05, self_terminating=True, t_offset=None)[source]#

Bases: Settings

Settings for NeoIterator.

Parameters:
filepath: PathLike#
chunk_dur: float = 0.05#
self_terminating: bool = True#
t_offset: Optional[float] = None#
__init__(filepath, chunk_dur=0.05, self_terminating=True, t_offset=None)#
Parameters:
Return type:

None

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

Bases: Unit

Parameters:

settings (Settings | None)

STATE#

alias of GenState

SETTINGS#

alias of NeoIteratorSettings

OUTPUT_SIGNAL = OutputStream:unlocated[<class 'ezmsg.util.messages.axisarray.AxisArray'>](self.num_buffers=32, self.force_tcp=False)#
OUTPUT_TERM = OutputStream:unlocated[typing.Any](self.num_buffers=32, self.force_tcp=False)#
initialize()[source]#

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

Return type:

None

construct_generator()[source]#
async pub_chunk()[source]#
Return type:

AsyncGenerator

class NeoIteratorSettings(filepath, chunk_dur=0.05, self_terminating=True, t_offset=None)[source]#

Bases: Settings

Settings for NeoIterator.

Parameters:
filepath: PathLike#
chunk_dur: float = 0.05#
self_terminating: bool = True#
t_offset: Optional[float] = None#
__init__(filepath, chunk_dur=0.05, self_terminating=True, t_offset=None)#
Parameters:
Return type:

None

class NeoIterator(settings)[source]#

Bases: object

Parameters:

settings (NeoIteratorSettings)

__init__(settings)[source]#
Parameters:

settings (NeoIteratorSettings)

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

Bases: Unit

Parameters:

settings (Settings | None)

STATE#

alias of GenState

SETTINGS#

alias of NeoIteratorSettings

OUTPUT_SIGNAL = OutputStream:unlocated[<class 'ezmsg.util.messages.axisarray.AxisArray'>](self.num_buffers=32, self.force_tcp=False)#
OUTPUT_TERM = OutputStream:unlocated[typing.Any](self.num_buffers=32, self.force_tcp=False)#
initialize()[source]#

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

Return type:

None

construct_generator()[source]#
async pub_chunk()[source]#
Return type:

AsyncGenerator