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 TaskCallbackType(*args, **kwargs)[source]

Structural subtyping protocol for a TaskCallback.

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

Raises:

NotImplementedError – Not implemented it’s an abstract class