Reference Subarray Component Manager
This module models component management for SKA subarray devices.
- class ska_tango_base.testing.reference.reference_subarray_component_manager.FakeSubarrayComponent[source]
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. In the case of a subarray device, the “component” is likely a collection of Tango devices responsible for monitoring and controlling the various resources assigned to the subarray. The component manager should be written so that it interacts with those Tango devices. But here, we fake up a “component” object to interact with instead.
It supports the assign, release, release_all, configure, scan, end_scan, end, abort, obsreset and restart methods. For testing purposes, it can also be told to simulate a spontaneous obs_state change via simulate_power_state` and simulate_obsfault 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__(capability_types: list[str], time_to_return: float = 0.05, time_to_complete: float = 0.4, power: PowerState = PowerState.OFF, fault: bool | None = None, resourced: bool = False, configured: bool = False, scanning: bool = False, obsfault: bool = False, **kwargs: Any) None[source]
Initialise a new instance.
- Parameters:
capability_types – a list strings representing capability types.
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
resourced – initial resourced state of this component
configured – initial configured state of this component
scanning – initial scanning state of this component
obsfault – initial obsfault state of this component
kwargs – additional keyword arguments
- property configured_capabilities: list[str]
Return the configured capabilities of this component.
- Returns:
list of strings indicating number of configured instances of each capability type
- assign(resources: set[str], task_callback: TaskCallbackType, task_abort_event: Event) None[source]
Assign resources.
- Parameters:
resources – the resources to be assigned.
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.
- release(resources: set[str], task_callback: TaskCallbackType, task_abort_event: Event) None[source]
Release resources.
- Parameters:
resources – resources to be released
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.
- release_all(task_callback: TaskCallbackType, task_abort_event: Event) None[source]
Release all resources.
- 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.
- configure(blocks: int | None, channels: int | None, task_callback: TaskCallbackType, task_abort_event: Event) None[source]
Configure the component.
- Parameters:
blocks – the number of blocks. (This not meant to be particularly meaningful. The “blocks” are just MacGuffins that give us something to configure.)
channels – the number of channels. (This not meant to be particularly meaningful. The “channels” are just MacGuffins that give us something to configure.)
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.
- deconfigure(task_callback: TaskCallbackType, task_abort_event: Event) None[source]
Deconfigure this component.
- 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.
- scan(scan_id: str, task_callback: TaskCallbackType, task_abort_event: Event) None[source]
Start scanning.
- Parameters:
scan_id – ID of the scan
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.
- end_scan(task_callback: TaskCallbackType, task_abort_event: Event) None[source]
End scanning.
- 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.
- obsreset(task_callback: TaskCallbackType, task_abort_event: Event) None[source]
Reset an observation that has faulted or been aborted.
- 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.
- restart(task_callback: TaskCallbackType, task_abort_event: Event) None[source]
Restart the component after it has faulted or been aborted.
- 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.
- class ska_tango_base.testing.reference.reference_subarray_component_manager.ReferenceSubarrayComponentManager[source]
A component manager for SKA subarray Tango devices.
The current implementation is intended to * illustrate the model * enable testing of the base classes
It should not generally be used in concrete devices; instead, write a subclass specific to the component managed by the device.
- __init__(capability_types: list[str], logger: Logger, communication_state_callback: Callable[[CommunicationStatus], None], component_state_callback: Callable[[], None], _component: FakeSubarrayComponent | None = None)[source]
Initialise a new ReferenceSubarrayComponentManager instance.
- Parameters:
capability_types – types of capabiltiy supported by this subarray
logger – the logger for this component manager to log with
communication_state_callback – callback to be called when the state of communications with the component changes
component_state_callback – callback to be called when the state of the component changes
_component – an object to use as the component of this component manager; for testing purposes only.
- assign(task_callback: TaskCallbackType | None, **kwargs: Any) tuple[TaskStatus, str][source]
Assign resources to the component.
- Parameters:
task_callback – a callback to be called whenever the status of this task changes.
kwargs – keyword arguments to the command
- Returns:
task status and message
- release(task_callback: TaskCallbackType | None, **kwargs: Any) tuple[TaskStatus, str][source]
Release resources from the component.
- Parameters:
task_callback – a callback to be called whenever the status of this task changes.
kwargs – keyword arguments to the command
- Returns:
task status and message
- release_all(task_callback: TaskCallbackType | None) tuple[TaskStatus, str][source]
Release all resources.
- Parameters:
task_callback – a callback to be called whenever the status of this task changes.
- Returns:
task status and message
- configure(task_callback: TaskCallbackType | None, **kwargs: Any) tuple[TaskStatus, str][source]
Configure the component.
- Parameters:
task_callback – a callback to be called whenever the status of this task changes.
kwargs – keyword arguments to the command
- Returns:
task status and message
- deconfigure(task_callback: TaskCallbackType | None) tuple[TaskStatus, str][source]
Deconfigure this component.
- Parameters:
task_callback – a callback to be called whenever the status of this task changes.
- Returns:
task status and message
- scan(task_callback: TaskCallbackType | None, **kwargs: Any) tuple[TaskStatus, str][source]
Start scanning.
- Parameters:
task_callback – a callback to be called whenever the status of this task changes.
kwargs – keyword arguments to the command
- Returns:
task status and message
- end_scan(task_callback: TaskCallbackType | None) tuple[TaskStatus, str][source]
End scanning.
- Parameters:
task_callback – a callback to be called whenever the status of this task changes.
- Returns:
task status and message
- obsreset(task_callback: TaskCallbackType | None) tuple[TaskStatus, str][source]
Deconfigure the component but do not release resources.
- Parameters:
task_callback – a callback to be called whenever the status of this task changes.
- Returns:
task status and message
- restart(task_callback: TaskCallbackType | None) tuple[TaskStatus, str][source]
Tell the component to restart.
It will return to a state in which it is unconfigured and empty of assigned resources.
- Parameters:
task_callback – a callback to be called whenever the status of this task changes.
- Returns:
task status and message