Reference Base Component Manager

This module provided reference implementations of a BaseComponentManager.

It is provided for explanatory purposes, and to support testing of this package.

class FakeBaseComponent(time_to_return: float = 0.05, time_to_complete: float = 0.4, power: PowerState = PowerState.OFF, fault: Optional[bool] = None, **state_kwargs: Any)

A fake component for the component manager to work with.

NOTE: There is usually no need to implement a component object. The “component” is an element of the external system under control, such as a piece of hardware or an external service. The component manager object communicates with the component in order to monitor and control it.

This is a very simple fake component with a power state and a fault state. When either of these aspects of state changes, it lets the component manager know by calling its state_change_callback.

It can be directly controlled via off(), standby(), on() and reset() methods. For testing purposes, it can also be told to simulate a spontaneous state change via simulate_power_state` and simulate_fault methods.

When one of these command method is invoked, the component simulates communications latency by sleeping for a short time. It then returns, but simulates any asynchronous work it needs to do by delaying updating task and component state for a short time.

set_state_change_callback(state_change_callback: Optional[Callable[[...], None]]) → None

Set a callback to be called when the state of this component changes.

Parameters:

state_change_callback – a callback to be call when the state of the component changes

off(task_callback: TaskCallbackType, task_abort_event: Event) → None

Turn the component off.

Parameters:

• task_callback – a callback to be called whenever the status of this task changes.

• task_abort_event – a threading.Event that can be checked for whether this task has been aborted.

standby(task_callback: TaskCallbackType, task_abort_event: Event) → None

Put the component into low-power standby mode.

Parameters:

• task_callback – a callback to be called whenever the status of this task changes.

• task_abort_event – a threading.Event that can be checked for whether this task has been aborted.

on(task_callback: TaskCallbackType, task_abort_event: Event) → None

Turn the component on.

Parameters:

• task_callback – a callback to be called whenever the status of this task changes.

• task_abort_event – a threading.Event that can be checked for whether this task has been aborted.

simulate_power_state(power_state: PowerState) → None

Simulate a change in component power state.

This could occur as a result of the Off command, or because of some external event/action.

Parameters:

power_state – the power state

reset(task_callback: TaskCallbackType, task_abort_event: Event) → None

Reset the component (from fault state).

Parameters:

• task_callback – a callback to be called whenever the status of this task changes.

• task_abort_event – a threading.Event that can be checked for whether this task has been aborted.

simulate_fault(fault_state: bool) → None

Tell the component to simulate (or stop simulating) a fault.

Parameters:

fault_state – whether faulty or not.

property faulty: bool

Return whether this component is faulty.

Returns:

whether this component is faulty.

property power_state: PowerState

Return the power state of this component.

Returns:

the power state of this component.

class GenericBaseComponentManager(component: ComponentT, logger: Logger, communication_state_callback: Callable[[CommunicationStatus], None], component_state_callback: Callable[[], None], *args: Any, **kwargs: Any)

A generic component manager for 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

The current implementation is intended to

• illustrate the model

• enable testing of these base classes

It should not generally be used in concrete devices; instead, write a component manager specific to the component managed by the device.

start_communicating() → None

Establish communication with the component, then start monitoring.

stop_communicating() → None

Break off communication with the component.

simulate_communication_failure(fail_communicate: bool) → None

Simulate (or stop simulating) a failure to communicate with the component.

Parameters:

fail_communicate – whether the connection to the component is failing

property power_state: PowerState

Power mode of the component.

This is just a bit of syntactic sugar for self.component_state[“power”].

Returns:

the power mode of the component

property fault_state: bool

Whether the component is currently faulting.

Returns:

whether the component is faulting

off(task_callback: ska_tango_base.base.component_manager.TaskCallbackType | None = None) → tuple[ska_control_model.task_status.TaskStatus, str]

Turn the component off.

Parameters:

task_callback – a callback to be called whenever the status of this task changes.

Returns:

TaskStatus and message

standby(task_callback: ska_tango_base.base.component_manager.TaskCallbackType | None = None) → tuple[ska_control_model.task_status.TaskStatus, str]

Put the component into low-power standby mode.

Parameters:

task_callback – a callback to be called whenever the status of this task changes.

Returns:

TaskStatus and message

on(task_callback: ska_tango_base.base.component_manager.TaskCallbackType | None = None) → tuple[ska_control_model.task_status.TaskStatus, str]

Turn the component on.

Parameters:

task_callback – a callback to be called whenever the status of this task changes.

Returns:

TaskStatus and message

reset(task_callback: ska_tango_base.base.component_manager.TaskCallbackType | None = None) → tuple[ska_control_model.task_status.TaskStatus, str]

Reset the component (from fault state).

Parameters:

task_callback – a callback to be called whenever the status of this task changes.

Returns:

TaskStatus and message

class ReferenceBaseComponentManager(logger: Logger, communication_state_callback: Callable[[CommunicationStatus], None], component_state_callback: Callable[[], None], *args: Any, _component: Optional[FakeBaseComponent] = None, **kwargs: Any)

A reference base component manager for Tango devices.