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.

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 FakeBaseComponent(time_to_return: float = 0.05, time_to_complete: float = 0.4, power: PowerState = PowerState.OFF, fault: bool | None = 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: 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.

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: 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

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

A reference base component manager for Tango devices.