runtimepy.mixins package

Submodules

runtimepy.mixins.async_command module

A module implementing an interface for classes with channel-command processors that need to handle commands asynchronously.

class runtimepy.mixins.async_command.AsyncCommandProcessingMixin(logger: Logger | LoggerAdapter[Any] = None, logger_name: str = None)

Bases: LoggerMixin

A class mixin for handling asynchronous commands.

command: ChannelCommandProcessor

async handle_command(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] | None) → None

Handle a command.

outgoing_commands: Queue[tuple[Namespace, BitField | Channel[Int8Primitive] | Channel[Int16Primitive] | Channel[Int32Primitive] | Channel[Int64Primitive] | Channel[Uint8Primitive] | Channel[Uint16Primitive] | Channel[Uint32Primitive] | Channel[Uint64Primitive] | Channel[FloatPrimitive] | Channel[DoublePrimitive] | Channel[BooleanPrimitive] | None]]

async process_command_queue() → None

Process any outgoing command requests.

runtimepy.mixins.enum module

A module implementing a class mixin for classes that have an optional enum-registry key.

class runtimepy.mixins.enum.EnumMixin

Bases: object

A class for working with an underlying enum attribute.

property enum: str | int

Get the enum-registry key for this channel.

property is_enum: bool

Determine if this channel has an associated enumeration.

runtimepy.mixins.environment module

A module implementing a channel-environment class mixin.

class runtimepy.mixins.environment.ChannelEnvironmentMixin(env: ChannelEnvironment = None, **kwargs)

Bases: object

A simple channel-environment mixin.

env: ChannelEnvironment

register_channel_metrics(name: str, channel: ChannelMetrics, verb: str) → None

Register individual channel metrics.

register_connection_metrics(metrics: ConnectionMetrics, *names: str, namespace: str = 'metrics') → None

Register connection metrics.

register_task_metrics(metrics: PeriodicTaskMetrics, *names: str, namespace: str = 'metrics') → None

Register periodic task metrics.

runtimepy.mixins.finalize module

A module implementing a class with a simple finalize interface.

class runtimepy.mixins.finalize.FinalizeMixin(event: Event = None)

Bases: object

A class implementing a simple finalize interface.

bypass_finalized() → Iterator[None]

Allows bypassing ‘finalized’ checks for specific situations. Ideally runtime environment entities don’t rely on this, but things like self-describing network protocols it’s difficult to resolve runtime-state storage needs during initialization alone.

finalize(strict: bool = True) → None

Finalize this instance.

property finalized: bool

Determine if this instance is finalized or not.

async wait_finalized() → None

Wait for this instance to be finalized.

runtimepy.mixins.logging module

A module implementing a logger-mixin extension.

class runtimepy.mixins.logging.LogCaptureMixin

Bases: object

A simple async file-reading interface.

async dispatch_log_capture() → None

Get the next line from this log stream.

async init_log_capture(stack: AsyncExitStack, log_paths: Iterable[tuple[LogLevel | int | str, Path | str | None]]) → None

Initialize this task with application information.

log_line(level: int, data: str) → None

Log a line for output.

logger: Logger | LoggerAdapter[Any]

safe_to_log = True

streams: list[tuple[int, Any]]

class runtimepy.mixins.logging.LogLevel(*values)

Bases: RuntimeIntEnum

A runtime enumeration for log level.

CRITICAL = 50

DEBUG = 10

ERROR = 40

INFO = 20

WARNING = 30

class runtimepy.mixins.logging.LoggerMixinLevelControl(logger: Logger | LoggerAdapter[Any] = None, logger_name: str = None)

Bases: LoggerMixin

A logger mixin that exposes a runtime-controllable level.

setup_level_channel(env: ChannelEnvironment, name: str = 'log_level', initial: str = 'info', description: str = 'Text-log level filter for this environment.') → None

Add a commandable log-level channel to the environment.

runtimepy.mixins.logging.handle_safe_log(logger: Logger | LoggerAdapter[Any], level: int, data: str, safe_to_log: bool) → None

handle a log filtering scenario.

runtimepy.mixins.psutil module

A module implementing interfaces for psutil.

class runtimepy.mixins.psutil.PsutilMixin

Bases: object

A simple psutil runtime interface.

cpu_average: WeightedAverage

cpu_percent: FloatPrimitive

init_psutil(env: ChannelEnvironment) → None

Initialize psutil-based metrics.

memory_percent: FloatPrimitive

poll_psutil(weight: float) → None

Poll psutil-based metrics.

process: Process

runtimepy.mixins.regex module

A class mixin for classes that have a regular expression that they wish to validate names with.

class runtimepy.mixins.regex.RegexMixin

Bases: object

A simple class mixin for validating names.

name_regex: Pattern | None = re.compile('^[\\w\\:.\\-_\\/]+$')

classmethod validate_name(name: str, logger: Logger | LoggerAdapter[Any] = None) → bool

Determine if a name is valid for this class.

runtimepy.mixins.trig module

A module implementing a simple trig-channel environment mixin.

class runtimepy.mixins.trig.TrigMixin(env: ChannelEnvironment, steps_controls: dict[str, int | float | bool | dict[str, int | float | bool]] | str = 'steps', amplitude_controls: dict[str, int | float | bool | dict[str, int | float | bool]] | str = 'amplitude', phase_angle_controls: dict[str, int | float | bool | dict[str, int | float | bool]] | str = 'phase')

Bases: object

A simple trig mixin class.

dispatch_trig(step: int) → None

Dispatch trig channel updates.

Module contents