ezmsg.panel.recorder#

Classes

Recorder(*args[, settings])

RecorderGUI(*args[, settings])

RecorderGUIState()

RecorderSettings(data_dir[, name, ...])

class RecorderSettings(data_dir, name='Message Recorder', msg_rate_window=2.0, write_period=0.0)[source]#

Bases: Settings

Parameters:
data_dir: Path#
name: str = 'Message Recorder'#
msg_rate_window: float = 2.0#
write_period: float = 0.0#
__init__(data_dir, name='Message Recorder', msg_rate_window=2.0, write_period=0.0)#
Parameters:
Return type:

None

class RecorderGUIState[source]#

Bases: State

message_rate: Number#
file_selector: FileSelector#
rec_dir: TextInput#
rec_name: TextInput#
rec_button: Button#
stop_button: Button#
rec_file: StaticText#
rec_msgs: Number#
msg_times: List[float]#
cur_rec: Optional[Path] = None#
start_queue: Queue[Path]#
stop_queue: Queue[Path]#
n_msgs: int = 0#
class RecorderGUI(*args, settings=None, **kwargs)[source]#

Bases: Unit

Parameters:

settings (Settings | None)

SETTINGS#

alias of RecorderSettings

STATE#

alias of RecorderGUIState

INPUT_MESSAGE = InputStream:unlocated[typing.Any]()#
OUTPUT_START = OutputStream:unlocated[<class 'pathlib.Path'>](self.num_buffers=32, self.force_tcp=False)#
INPUT_START = InputStream:unlocated[<class 'pathlib.Path'>]()#
OUTPUT_STOP = OutputStream:unlocated[<class 'pathlib.Path'>](self.num_buffers=32, self.force_tcp=False)#
INPUT_STOP = InputStream:unlocated[<class 'pathlib.Path'>]()#
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

controls()[source]#
Return type:

Viewable

content()[source]#
Return type:

Viewable

panel()[source]#
Return type:

Viewable

async start_file()[source]#
Return type:

AsyncGenerator

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

None

Parameters:

msg (Path)

async stop_file()[source]#
Return type:

AsyncGenerator

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

None

Parameters:

msg (Path)

async update_display()[source]#
Return type:

None

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

None

Parameters:

msg (Any)

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

Bases: Collection, Tab

Parameters:

settings (Settings | None)

SETTINGS#

alias of RecorderSettings

INPUT_MESSAGE = InputStream:unlocated[typing.Any]()#
GUI = <ezmsg.panel.recorder.RecorderGUI object>#
LOGGER = <ezmsg.util.messagelogger.MessageLogger 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