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, ABC

A class mixin for handling asynchronous commands.

command: ChannelCommandProcessor

abstract 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, namespace: str = 'metrics') → None

Register connection metrics.

register_task_metrics(metrics: PeriodicTaskMetrics, 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.

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.LogLevel(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)

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.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)

Bases: object

A simple trig mixin class.

dispatch_trig(step: int) → None

Dispatch trig channel updates.

Module contents