runtimepy.channel.environment.command package

runtimepy.channel.environment.command package#

Submodules#

runtimepy.channel.environment.command.parser module#

A module implementing a channel-environment command argument parser wrapper.

class runtimepy.channel.environment.command.parser.ChannelCommand(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Bases: StrEnum

An enumeration for all channel command options.

GET = 'get'#

SET = 'set'#

TOGGLE = 'toggle'#

class runtimepy.channel.environment.command.parser.CommandParser(prog=None, usage=None, description=None, epilog=None, parents=[], formatter_class=<class 'argparse.HelpFormatter'>, prefix_chars='-', fromfile_prefix_chars=None, argument_default=None, conflict_handler='error', add_help=True, allow_abbrev=True, exit_on_error=True)[source]#

Bases: ArgumentParser

An argument parser wrapper.

data: dict[str, Any]#

error(message: str)[source]#: Pass error message to error handling.

exit(status: int = 0, message: str = None)[source]#: Override exit behavior.

initialize() → None[source]#: Initialize this command parser.

runtimepy.channel.environment.command.processor module#

A module implementing a channel-command processing interface.

class runtimepy.channel.environment.command.processor.ChannelCommandProcessor(env: ChannelEnvironment, logger: Logger | LoggerAdapter[Any], **kwargs)[source]#

Bases: ChannelEnvironmentMixin

A command processing interface for channel environments.

command(value: str) → CommandResult[source]#: Process a command.

do_set(args: Namespace) → CommandResult[source]#: Attempt to set a channel value.

do_toggle(args: Namespace, channel: BitField | Channel[Int8Primitive] | Channel[Int16Primitive] | Channel[Int32Primitive] | Channel[Int64Primitive] | Channel[Uint8Primitive] | Channel[Uint16Primitive] | Channel[Uint32Primitive] | Channel[Uint64Primitive] | Channel[FloatPrimitive] | Channel[DoublePrimitive] | Channel[BooleanPrimitive]) → CommandResult[source]#: Attempt to toggle a channel.

get_suggestion(value: str) → str | None[source]#: Get an input suggestion.

handle_command(args: Namespace) → CommandResult[source]#: Handle a command from parsed arguments.

parse(value: str) → Namespace | None[source]#: Attempt to parse arguments.

class runtimepy.channel.environment.command.processor.RemoteCommandProcessor(env: ChannelEnvironment, logger: Logger | LoggerAdapter[Any], **kwargs)[source]#

Bases: ChannelCommandProcessor

A command processing interface for channel environments.

handle_command(args: Namespace) → CommandResult[source]#: Handle a command from parsed arguments.

runtimepy.channel.environment.command.result module#

A module implementing a command result interface.

class runtimepy.channel.environment.command.result.CommandResult(success: bool, reason: str | None = None)[source]#

Bases: NamedTuple

A container for command result data.

reason: str | None#: Alias for field number 1

success: bool#: Alias for field number 0

Module contents#

A module implementing UI command processing.

class runtimepy.channel.environment.command.GlobalEnvironment(root: Path = None)[source]#

Bases: UserDict[str, ChannelCommandProcessor], LoggerMixin

A class implementing a container for channel environments available globally at runtime.

clear() → None[source]#: Log environments that get cleared when clearing.

event_telemetry_output(env_pattern: str = '.*', env_exact: bool = False, channel_pattern: str = '.*', channel_exact: bool = False, event_path: str = 'event_stream.bin', text_log: bool = True) → Iterator[list[tuple[str, list[str]]]][source]#: Register file-output streams for environments based on a pattern.

export(env: str) → Path[source]#: Export an environment to a sub-directory of the root directory.

file_event_stream(env: str, pattern: str = '.*', path: str = 'event_stream.bin', exact: bool = False) → Iterator[list[str]][source]#: Enable event streaming to a file for an environment by name.

static from_root(root: Path) → GlobalEnvironment[source]#: Load a global environment from a directory.

log_file(path: str = 'log.txt') → Iterator[FileHandler][source]#: Register a logging file handler as a managed context.

static meta_path(root: Path) → Path[source]#: Get the path to the metadata file.

read_event_stream(env: str, path: str = 'event_stream.bin') → Iterator[ParsedEvent][source]#: Reade events from a specific environment.

register(name: str, env: ChannelCommandProcessor) → None[source]#: Register an environment.

static temporary() → Iterator[GlobalEnvironment][source]#: Create a temporary global environment.

property valid_root: Path#: Get the validated root directory.

write_meta(data: Dict[str, str | int | float | bool | None | Dict[str, str | int | float | bool | None] | List[str | int | float | bool | None]]) → None[source]#: Write metadata to the output directory.

runtimepy.channel.environment.command.clear_env() → None[source]#: Reset the global environment mapping.

runtimepy.channel.environment.command.register_env(name: str, env: ChannelCommandProcessor) → None[source]#: Register a channel environment globally.