| | |
- abc.ABC(builtins.object)
-
- runtimepy.task.basic.periodic.PeriodicTask(runtimepy.mixins.logging.LoggerMixinLevelControl, runtimepy.mixins.environment.ChannelEnvironmentMixin, abc.ABC)
- runtimepy.mixins.environment.ChannelEnvironmentMixin(builtins.object)
-
- runtimepy.task.basic.periodic.PeriodicTask(runtimepy.mixins.logging.LoggerMixinLevelControl, runtimepy.mixins.environment.ChannelEnvironmentMixin, abc.ABC)
- runtimepy.mixins.logging.LoggerMixinLevelControl(vcorelib.logging.LoggerMixin)
-
- runtimepy.task.basic.periodic.PeriodicTask(runtimepy.mixins.logging.LoggerMixinLevelControl, runtimepy.mixins.environment.ChannelEnvironmentMixin, abc.ABC)
- typing.Generic(builtins.object)
-
- runtimepy.task.basic.manager.PeriodicTaskManager
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.
| |