Python: package runtimepy.task

AsyncTask(name: str, period_s: float, env: runtimepy.channel.environment.ChannelEnvironment, average_depth: int = 10, max_iterations: int = 0) -> None

A basic implementation of a periodic task.

Method resolution order:: AsyncTask; vcorelib.logging.LoggerMixin; builtins.object

Methods defined here:

__init__(self, name: str, period_s: float, env: runtimepy.channel.environment.ChannelEnvironment, average_depth: int = 10, max_iterations: int = 0) -> None: Initialize this asynchronous task.

disable(self) -> None: Disable this task.

async dispatch(self, *_, **__) -> bool: Dispatch this task.

enable(self) -> None: Enable this task.

async init(self, *_, **__) -> bool: Initialize this task.

init_channels(self, env: runtimepy.channel.environment.ChannelEnvironment) -> None: Initialize task-specific channels.

log_metrics(self) -> None: Log information related to metrics channels.

reset_metrics(self) -> None: Reset metrics channel values.

async run(self, *args, stop_sig: asyncio.locks.Event = None, **kwargs) -> None: Run this task while it's enabled.

Readonly properties defined here:

rate_str: Get this periodic's rate as a string.

Data and other attributes defined here:

__annotations__ = {}

Methods inherited from vcorelib.logging.LoggerMixin:

log_time(self, message: str, *args, level: int = 20, reminder: bool = False, **kwargs) -> Iterator[NoneType]: A simple wrapper.

Data descriptors inherited from vcorelib.logging.LoggerMixin:

__dict__: dictionary for instance variables

__weakref__: list of weak references to the object

class PeriodicTask(runtimepy.mixins.logging.LoggerMixinLevelControl, runtimepy.mixins.environment.ChannelEnvironmentMixin, abc.ABC)

PeriodicTask(name: 'str', average_depth: 'int' = 10, metrics: 'PeriodicTaskMetrics' = None, period_s: 'float' = 1.0, env: 'ChannelEnvironment' = None) -> 'None'

A class implementing a simple periodic-task interface.

Method resolution order:: PeriodicTask; runtimepy.mixins.logging.LoggerMixinLevelControl; vcorelib.logging.LoggerMixin; runtimepy.mixins.environment.ChannelEnvironmentMixin; abc.ABC; builtins.object

Methods defined here:

__init__(self, name: 'str', average_depth: 'int' = 10, metrics: 'PeriodicTaskMetrics' = None, period_s: 'float' = 1.0, env: 'ChannelEnvironment' = None) -> 'None': Initialize this task.

disable(self) -> 'bool': Disable this task, return whether or not any action was taken.

async dispatch(self) -> 'bool': Dispatch an iteration of this task.

async run(self, period_s: 'float' = None, stop_sig: '_asyncio.Event' = None) -> 'None': Run this task by executing the dispatch method at the specified period
until a dispatch iteration fails or the task is otherwise disabled.

set_period(self, period_s: 'float' = None) -> 'bool': Attempt to set a new period for this task.

async stop(self) -> 'bool': Wait for this task to stop running (if it is).

async stop_extra(self) -> 'None': Extra actions to perform when this task is stopping.

async task(self, period_s: 'float' = None, stop_sig: '_asyncio.Event' = None) -> '_asyncio.Task[None]': Create an event-loop task for this periodic.

Data and other attributes defined here:

__abstractmethods__ = frozenset({'dispatch'})

__annotations__ = {}

auto_finalize = True

Methods inherited from runtimepy.mixins.logging.LoggerMixinLevelControl:

setup_level_channel(self, env: runtimepy.channel.environment.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.

Methods inherited from vcorelib.logging.LoggerMixin:

log_time(self, message: str, *args, level: int = 20, reminder: bool = False, **kwargs) -> Iterator[NoneType]: A simple wrapper.

Data descriptors inherited from vcorelib.logging.LoggerMixin:

__dict__: dictionary for instance variables

__weakref__: list of weak references to the object

Methods inherited from runtimepy.mixins.environment.ChannelEnvironmentMixin:

register_channel_metrics(self, name: str, channel: runtimepy.metrics.channel.ChannelMetrics, verb: str) -> None: Register individual channel metrics.

register_connection_metrics(self, metrics: runtimepy.metrics.connection.ConnectionMetrics, namespace: str = 'metrics') -> None: Register connection metrics.

register_task_metrics(self, metrics: runtimepy.metrics.task.PeriodicTaskMetrics, namespace: str = 'metrics') -> None: Register periodic task metrics.

class PeriodicTaskManager(typing.Generic)

PeriodicTaskManager() -> None

A class for managing periodic tasks as a single group.

Method resolution order:: PeriodicTaskManager; typing.Generic; builtins.object

Methods defined here:

__getitem__(self, name: str) -> ~T: Get a task by name.

__init__(self) -> None: Initialize this instance.

register(self, task: ~T, period_s: float = None) -> bool: Register a periodic task.

running(self, stop_sig: asyncio.locks.Event = None) -> AsyncIterator[NoneType]: Run tasks as an async context.

async start(self, stop_sig: asyncio.locks.Event = None) -> None: Ensure tasks are started.

async stop(self) -> None: Ensure tasks are stopped.

Readonly properties defined here:

tasks: Iterate over tasks.

Data descriptors defined here:

__dict__: dictionary for instance variables

__weakref__: list of weak references to the object

Data and other attributes defined here:

__orig_bases__ = (typing.Generic[~T],)

__parameters__ = (~T,)

Class methods inherited from typing.Generic:

__class_getitem__(...) from builtins.type: Parameterizes a generic class.

At least, parameterizing a generic class is the *main* thing this
method does. For example, for some generic class `Foo`, this is called
when we do `Foo[int]` - there, with `cls=Foo` and `params=int`.

However, note that this method is also called when defining generic
classes in the first place with `class Foo[T]: ...`.

__init_subclass__(...) from builtins.type: Function to initialize subclasses.

class PeriodicTaskMetrics(builtins.tuple)

PeriodicTaskMetrics(dispatches: runtimepy.primitives.int.Uint32Primitive, rate_hz: runtimepy.primitives.float.FloatPrimitive, average_s: runtimepy.primitives.float.FloatPrimitive, max_s: runtimepy.primitives.float.FloatPrimitive, min_s: runtimepy.primitives.float.FloatPrimitive, overruns: runtimepy.primitives.int.Uint16Primitive)

Metrics for a periodic tasks.

Method resolution order:: PeriodicTaskMetrics; builtins.tuple; builtins.object

Methods defined here:

__getnewargs__(self): Return self as a plain tuple. Used by copy and pickle.

__repr__(self): Return a nicely formatted representation string

_asdict(self): Return a new dict which maps field names to their values.

_replace(self, /, **kwds): Return a new PeriodicTaskMetrics object replacing specified fields with new values

measure(self, eloop: asyncio.events.AbstractEventLoop, rate: vcorelib.math.analysis.rate.RateTracker, dispatch: vcorelib.math.analysis.average.MovingAverage, iter_time: runtimepy.primitives.float.DoublePrimitive, period_s: float) -> Iterator[NoneType]: Measure the time spent yielding and update data.

Class methods defined here:

_make(iterable) from builtins.type: Make a new PeriodicTaskMetrics object from a sequence or iterable

Static methods defined here:

__new__(_cls, dispatches: runtimepy.primitives.int.Uint32Primitive, rate_hz: runtimepy.primitives.float.FloatPrimitive, average_s: runtimepy.primitives.float.FloatPrimitive, max_s: runtimepy.primitives.float.FloatPrimitive, min_s: runtimepy.primitives.float.FloatPrimitive, overruns: runtimepy.primitives.int.Uint16Primitive): Create new instance of PeriodicTaskMetrics(dispatches, rate_hz, average_s, max_s, min_s, overruns)

create() -> 'PeriodicTaskMetrics': Create a new metrics instance.

Data descriptors defined here:

dispatches: Alias for field number 0

rate_hz: Alias for field number 1

average_s: Alias for field number 2

max_s: Alias for field number 3

min_s: Alias for field number 4

overruns: Alias for field number 5

Data and other attributes defined here:

__annotations__ = {'average_s': <class 'runtimepy.primitives.float.FloatPrimitive'>, 'dispatches': <class 'runtimepy.primitives.int.Uint32Primitive'>, 'max_s': <class 'runtimepy.primitives.float.FloatPrimitive'>, 'min_s': <class 'runtimepy.primitives.float.FloatPrimitive'>, 'overruns': <class 'runtimepy.primitives.int.Uint16Primitive'>, 'rate_hz': <class 'runtimepy.primitives.float.FloatPrimitive'>}

__match_args__ = ('dispatches', 'rate_hz', 'average_s', 'max_s', 'min_s', 'overruns')

__orig_bases__ = (<function NamedTuple>,)

_field_defaults = {}

_fields = ('dispatches', 'rate_hz', 'average_s', 'max_s', 'min_s', 'overruns')

Methods inherited from builtins.tuple:

__add__(self, value, /): Return self+value.

__contains__(self, key, /): Return bool(key in self).

__eq__(self, value, /): Return self==value.

__ge__(self, value, /): Return self>=value.

__getattribute__(self, name, /): Return getattr(self, name).

__getitem__(self, key, /): Return self[key].

__gt__(self, value, /): Return self>value.

__hash__(self, /): Return hash(self).

__iter__(self, /): Implement iter(self).

__le__(self, value, /): Return self<=value.

__len__(self, /): Return len(self).

__lt__(self, value, /): Return self<value.

__mul__(self, value, /): Return self*value.

__ne__(self, value, /): Return self!=value.

__rmul__(self, value, /): Return value*self.

count(self, value, /): Return number of occurrences of value.

index(self, value, start=0, stop=9223372036854775807, /): Return first index of value.

Raises ValueError if the value is not present.

Class methods inherited from builtins.tuple:

__class_getitem__(...) from builtins.type: See PEP 585

Data
		__all__ = ['AsyncTask', 'PeriodicTask', 'PeriodicTaskMetrics', 'PeriodicTaskManager']