Base Component Manager

This module provides an abstract component manager for SKA Tango base devices.

The basic model is:

Every Tango device has a component that it monitors and/or controls. That component could be, for example:
- Hardware such as an antenna, APIU, TPM, switch, subrack, etc.
- An external software system such as a cluster manager
- A software routine, possibly implemented within the Tango device itself
- In a hierarchical system, a pool of lower-level Tango devices.
A Tango device will usually need to establish and maintain communication with its component. This connection may be deliberately broken by the device, or it may fail.
A Tango device controls its component by issuing commands that cause the component to change behaviour and/or state; and it monitors its component by keeping track of its state.

check_communicating(func: Wrapped) → Wrapped[source]

Return a function that checks component communication before calling a function.

The component manager needs to have established communications with the component, in order for the function to be called.

This function is intended to be used as a decorator:

@check_communicating
def scan(self):
    ...

Parameters:: func – the wrapped function
Returns:: the wrapped function

check_on(func: Wrapped) → Wrapped[source]

Return a function that checks the component state then calls another function.

The component needs to be turned on, and not faulty, in order for the function to be called.

This function is intended to be used as a decorator:

@check_on
def scan(self):
    ...

Parameters:: func – the wrapped function
Returns:: the wrapped function

class BaseComponentManager(logger: Logger, communication_state_callback: Callable[[CommunicationStatus], None] | None = None, component_state_callback: Callable[[...], None] | None = None, **state: Any)[source]

An abstract base class for a component manager for SKA Tango devices.

It supports:

Maintaining a connection to its component
Controlling its component via commands like Off(), Standby(), On(), etc.
Monitoring its component, e.g. detect that it has been turned off or on

property max_queued_tasks: int

Get the task queue size.

Returns:: The task queue size

property max_executing_tasks: int

Get the max number of tasks that can be executing at once.

Returns:: max number of simultaneously executing tasks.

start_communicating() → None[source]

Establish communication with the component, then start monitoring.

This is the place to do things like:

Initiate a connection to the component (if your communication is connection-oriented)
Subscribe to component events (if using “pull” model)
Start a polling loop to monitor the component (if using a “push” model)

Raises:: NotImplementedError – Not implemented it’s an abstract class

stop_communicating() → None[source]

Cease monitoring the component, and break off all communication with it.

For example,

If you are communicating over a connection, disconnect.
If you have subscribed to events, unsubscribe.
If you are running a polling loop, stop it.

Raises:: NotImplementedError – Not implemented it’s an abstract class

property communication_state: CommunicationStatus

Return the communication status of this component manager.

Returns:: status of the communication channel with the component.

property component_state: dict[str, Any]

Return the state of this component manager’s component.

Returns:: state of the component.

off(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str][source]

Turn the component off.

Parameters:: task_callback – callback to be called when the status of the command changes
Raises:: NotImplementedError – Not implemented it’s an abstract class

standby(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str][source]

Put the component into low-power standby mode.

Parameters:: task_callback – callback to be called when the status of the command changes
Raises:: NotImplementedError – Not implemented it’s an abstract class

on(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str][source]

Turn the component on.

Parameters:: task_callback – callback to be called when the status of the command changes
Raises:: NotImplementedError – Not implemented it’s an abstract class

reset(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str][source]

Reset the component (from fault state).

Parameters:: task_callback – callback to be called when the status of the command changes
Raises:: NotImplementedError – Not implemented it’s an abstract class

abort(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str][source]

Abort activities on the device.

Parameters:: task_callback – callback to be called whenever the status of the task changes.
Returns:: tuple of TaskStatus & message

abort_commands(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str][source]

Abort all tasks queued & running.

Parameters:: task_callback – callback to be called whenever the status of the task changes.
Returns:: tuple of TaskStatus & message

abort_tasks(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str][source]

Abort all tasks queued & running.

Parameters:: task_callback – callback to be called whenever the status of the task changes.
Raises:: NotImplementedError – Not implemented it’s an abstract class

class TaskCallbackType(*args, **kwargs)[source]

Structural subtyping protocol for a task_callback.

A task_callback will be called with some combination of the following arguments:

status: TaskStatus of the task.

progress: int progress of the task.

result: Any JSON serialisable result of the task.

exception: Exception raised from the task.

Each of the above arguments is optional and the callback must check which are present by testing them for None. The callback cannot assume that only one argument will be provided per call.

JSONData

A type hint for any JSON-encodable data.

JSONData = (
   None
   | bool
   | int
   | float
   | str
   | list["JSONData"]  # A list can contain more JSON-encodable data
   | dict[str, "JSONData"]  # A dict must have str keys and JSON-encodable data
   | tuple["JSONData", ...]  # A tuple can contain more JSON-encodable data
)