RunManager

class bapsf_motion.actors.manager_.RunManager(config: str | Dict[str, Any] | RunManagerConfig | Path, *, auto_run: bool = False, build_mode: bool = False, parent: EventActor | None = None)

Bases: EventActor

Attributes Summary

`config`	Configuration dictionary of the actor.
`is_moving`
`logger`	The `Logger` instance being used for the actor.
`loop`	The `asyncio` event loop for the actor.
`mgs`
`name`	(`str`) A unique name given for the instance of the actor.
`parent`
`tasks`	List of `asyncio.Task`s this actor has in its event loop.
`terminated`	Indicates if the actor has been terminated.
`thread`	The `Thread` the event loop is running in.

Methods Summary

`add_motion_group`(config[, identifier])
`get_motion_group`(identifier)
`pop_motion_group`(identifier)
`remove_motion_group`(identifier)
`run`([auto_run])	Activate the `asyncio` event loop.
`setup_event_loop`([loop])	Set up the `asyncio` event loop.
`terminate`([delay_loop_stop])	Stop the actor's event loop.
`validate_motion_group`(mg_config[, identifier])

Attributes Documentation

config: Configuration dictionary of the actor.

Warning

This dictionary should never be written to from outside the owning actor.

is_moving

logger: The Logger instance being used for the actor.

loop: The asyncio event loop for the actor.

mgs

name

(str) A unique name given for the instance of the actor. This name is used as an identifier in the actor logger (see logger).

If the user does not specify a name, then the Actor should auto-generate a name.

parent

tasks: List of asyncio.Tasks this actor has in its event loop.

terminated: Indicates if the actor has been terminated.

thread

The Thread the event loop is running in.

If loop was given during instantiation, then there is no way of obtaining the thread object the event loop is running in. In this case thread will be None.

The thread id can always be retrieved using _thread_id.

Methods Documentation

add_motion_group(config: Dict[str, Any] | MotionGroupConfig | MotionGroup, identifier: str | int = None)

get_motion_group(identifier: str | int)

pop_motion_group(identifier: str | int)

remove_motion_group(identifier: str | int)

run(auto_run=True)

Activate the asyncio event loop. If the event loop is running, then nothing happens. Otherwise, the event loop is placed in a separate thread and set to run_forever.

Parameters:: auto_run (bool, optional) – If False, then do NOT start the event loop. This keyword is only made available to help with subclassing. (DEFAULT: True)

setup_event_loop(loop: AbstractEventLoop | None = None)

Set up the asyncio event loop. If the given loop is not an instance of AbstractEventLoop, then a new loop will be created.

Parameters:: loop (asyncio.AbstractEventLoop) – asyncio event loop for the actor’s tasks

terminate(delay_loop_stop=False)

Stop the actor’s event loop. All actor tasks will be cancelled, the connection to the motor will be shutdown, and the event loop will be stopped.

Parameters:: delay_loop_stop (bool) – If True, then do NOT stop the event loop. In this case it is assumed the calling functionality is managing additional tasks in the event loop, and it is up to that functionality to stop the loop. (DEFAULT: False)

validate_motion_group(mg_config: Dict[str, Any] | MotionGroup | MotionGroupConfig, identifier: str | int = None) → bool