ezmsg.panel.replay#

Classes

Replay(*args[, settings])

ReplayGUI(*args[, settings])

ReplayGUIState()

ReplaySettings(data_dir[, name])

class ReplaySettings(data_dir, name='Message Replay')[source]#

Bases: Settings

Parameters:
data_dir: Path#
name: str = 'Message Replay'#
msg_rate_window = 2.0#
__init__(data_dir, name='Message Replay')#
Parameters:
Return type:

None

class ReplayGUIState[source]#

Bases: State

message_rate: Number#
file_selector: FileSelector#
enqueue_button: Button#
pause_toggle: Toggle#
stop_button: Button#
playback_file: StaticText#
playback: Progress#
rapid: Checkbox#
rate: FloatInput#
file_queue: Queue[Path]#
stop_queue: Queue[bool]#
pause_queue: Queue[bool]#
msg_times: List[float]#
replay_status: Optional[ReplayStatusMessage] = None#
class ReplayGUI(*args, settings=None, **kwargs)[source]#

Bases: Unit

Parameters:

settings (Settings | None)

SETTINGS#

alias of ReplaySettings

STATE#

alias of ReplayGUIState

INPUT_REPLAY_STATUS = InputStream:unlocated[<class 'ezmsg.util.messagereplay.ReplayStatusMessage'>]()#
OUTPUT_FILE_REPLAY = OutputStream:unlocated[<class 'ezmsg.util.messagereplay.FileReplayMessage'>](self.num_buffers=32, self.force_tcp=False)#
OUTPUT_STOP = OutputStream:unlocated[<class 'bool'>](self.num_buffers=32, self.force_tcp=False)#
OUTPUT_PAUSE = OutputStream:unlocated[<class 'bool'>](self.num_buffers=32, self.force_tcp=False)#
async 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

content()[source]#
Return type:

Viewable

controls()[source]#
Return type:

Viewable

panel()[source]#
Return type:

Viewable

async start_file()[source]#
Return type:

AsyncGenerator

async stop_file()[source]#
Return type:

AsyncGenerator

async pause()[source]#
Return type:

AsyncGenerator

async on_replay_status(msg)[source]#
Return type:

None

Parameters:

msg (ReplayStatusMessage)

async update_display()[source]#
Return type:

None

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

Bases: Collection, Tab

Parameters:

settings (Settings | None)

SETTINGS#

alias of ReplaySettings

OUTPUT_MESSAGE = InputStream:unlocated[typing.Any]()#
OUTPUT_REPLAY_STATUS = OutputStream:unlocated[<class 'ezmsg.util.messagereplay.ReplayStatusMessage'>](self.num_buffers=32, self.force_tcp=False)#
GUI = <ezmsg.panel.replay.ReplayGUI object>#
REPLAY = <ezmsg.util.messagereplay.MessageReplay object>#
property title: str#
content()[source]#
Return type:

Viewable

sidebar()[source]#
Return type:

Viewable

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.

Return type:

None

network()[source]#

Override this method and have the definition return a NetworkDefinition which defines how InputStream and OutputStream from member Unit s will be connected.

Return type:

Iterable[Tuple[Union[Stream, str], Union[Stream, str]]]

process_components()[source]#

Override this method and have the definition return a tuple which contains Unit and Collection which should run in their own processes.

Return: the Collection.

Return type:

Tuple[Component, ...]

panel()[source]#
Return type:

Viewable