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.

ska_tango_base.base.base_component_manager.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

ska_tango_base.base.base_component_manager.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 ska_tango_base.base.base_component_manager.BaseComponentManager[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

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

Initialise a new ComponentManager instance.

Parameters:

  • logger – the logger to be used by this manager

  • communication_state_callback – callback to be called when the status of communications between the component manager and its component changes.

  • component_state_callback – callback to be called when the monitored state of the component changes

  • state – key/value pairs

property max_queued_tasks: int

Get the task queue size.

This property should be overridden by subclasses if they provide a task executor.

Returns:

The task queue size

property max_executing_tasks: int

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

This property should be overridden by subclasses if they provide a task executor.

Returns:

max number of simultaneously executing tasks.

cleanup() → None[source]

Cleanup any software resources in use by the component manager.

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.

_update_communication_state(communication_state: CommunicationStatus) → None[source]

Handle a change in communication status.

This is a helper method for use by subclasses.

Parameters:

communication_state – the new communication status of the component manager.

property component_state: dict[str, Any]

Return the state of this component manager’s component.

Returns:

state of the component.

_update_component_state(**kwargs: Any) → None[source]

Handle a change in component state.

This is a helper method for use by subclasses.

Parameters:

kwargs – key/values for state

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.

Deprecated:

Use abort_tasks() instead.

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.

Returns:

tuple of TaskStatus & message

Raises:

NotImplementedError – Not implemented it’s an abstract class