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.

ska_tango_base.testing.reference.reference_base_component_manager.wait_until_done(command: Callable[[...], None]) → Callable[[...], None]

Wait until the command is done before the device may continue with other tasks.

The waited on threading event is set when the callback is called with command status equal to COMPLETED, ABORTED, FAILED or REJECTED. This is only done if the command has been passed a real task callback, and not a mock callback or no callback at all.

Parameters:

command – Command method.

Returns:

Wrapped command method.

class ska_tango_base.testing.reference.reference_base_component_manager.FakeBaseComponent

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.

__init__(time_to_return: float = 0.05, time_to_complete: float = 0.4, power: PowerState = PowerState.OFF, fault: bool | None = None, **state_kwargs: Any) → None

Initialise a new instance.

Parameters:

• time_to_return – the amount of time to delay before returning from a command method. This simulates latency in communication.

• time_to_complete – the amount of time to delay before the component calls a task callback to let it know that the task has been completed

• power – initial power state of this component

• fault – initial fault state of this component

• state_kwargs – extra keyword arguments

set_state_change_callback(state_change_callback: Callable[[...], None] | 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_command_error(task_callback: TaskCallbackType, task_abort_event: Event) → None

Simulate a command that raises a CommandError during execution.

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.

Raises:

CommandError – simulating an invalid argument.

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

Simulate a command with a is_cmd_allowed method that raises an Exception.

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.

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

Simulate a command that reports its progress as a string message.

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.

call_command_on_device(command: str, device_address: str, task_callback: TaskCallbackType, task_abort_event: Event, database: bool = True) → None

LRC that calls a LRC on another tango device to test telemetry tracing.

Parameters:

• command – name of the command to call.

• device_address – address of the tango device to connect to.

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

• database – if a tango db is available, defaults to True.

set_fault() → None

Tell the component to set a fault state.

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 ska_tango_base.testing.reference.reference_base_component_manager.GenericBaseComponentManager

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.

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

Initialise a new ComponentManager instance.

Parameters:

• component – the component that this component manager manages.

• logger – a logger for this component manager

• communication_state_callback – callback for communication state

• component_state_callback – callback for component state

• args – extra arguments

• kwargs – extra keyword arguments

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: TaskCallbackType | None = None) → tuple[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: TaskCallbackType | None = None) → tuple[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: TaskCallbackType | None = None) → tuple[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: TaskCallbackType | None = None) → tuple[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

simulate_command_error(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str]

Simulate a command that raises a CommandError during execution.

Parameters:

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

Returns:

TaskStatus and message

simulate_is_cmd_allowed_error(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str]

Simulate a command with a is_cmd_allowed method that raises an Exception.

Parameters:

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

Returns:

TaskStatus and message

report_progress_message(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str]

Simulate a command that reports its progress as a string message.

Parameters:

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

Returns:

TaskStatus and message

test_telemetry_tracing(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str]

LRC that calls a LRC on another tango device to test telemetry tracing.

Parameters:

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

Returns:

TaskStatus and message

class ska_tango_base.testing.reference.reference_base_component_manager.ReferenceBaseComponentManager

A reference base component manager for Tango devices.

__init__(logger: Logger, communication_state_callback: Callable[[CommunicationStatus], None], component_state_callback: Callable[[], None], *args: Any, _component: FakeBaseComponent | None = None, **kwargs: Any) → None

Initialise a new ComponentManager instance.

Parameters:

• logger – a logger for this component manager

• communication_state_callback – callback for communication state

• component_state_callback – callback for component state

• args – extra arguments

• _component – allows setting of the component to be managed. Note: the component will normally be a part of the external system under control, such as a piece of hardware or an external service. So there normally will not be a “component” software object to pass in here. Instead, you would pass in information needed to establish communication with your component, such as an FQDN, or an IP address/port.

• kwargs – extra keyword arguments