Controller Component Manager

class ska_csp_lmc_common.controller.controller_component_manager.CSPControllerComponentManager(*args: Any, **kwargs: Any)

Bases: CSPBaseComponentManager

Component Manager for the Csp Controller.

__init__(op_state_model: ControllerOpStateModel, health_model: ControllerHealthModel, properties: ComponentManagerConfiguration, update_device_property_cbk: Callable, communication_state_changed_callback: Callable[[ska_control_model.CommunicationStatus], None] | None = None, component_state_changed_callback: Callable[[dict[str, Any]], None] | None = None, logger: Logger | None = None) → None

The Component Manager for the CSP Controller device.

Parameters:

op_state_model – The Operational State Model to evaluate the CSP Controller device state
health_model – The Health State Model to evaluate the CSP Controller health state.
properties – A class instance whose properties are the CSP Controller device properties.
update_device_property_cbk – The CSP Controller method invoked to update the properties. Defaults to None.
logger – The device or python logger if default is None.

property pstbeam_subarray_map: str: Return a JSON-encoded string representing the mapping between the PST Beam ids and the Subarray id to which the PST Beam is assigned. If PST Beam is not assigned, default value is zero.

property pstbeams_available: List[str]: Return a list of string representing the PST Beam ids available.

property quality_assurance_metrics: str: Return a JSON formatted string with quality attributes.

_create_init_callbacks() → ControllerInitCallbacks: Create controller-specific initialization callbacks.

_create_stop_callbacks() → ControllerStopCallbacks: Create controller-specific stop callbacks.

_controller_factory(logger: Logger) → ControllerComponentFactory

Configure the Factory to create the sub-systems components.

This class is specialized in Mid.CSP and Low.CSP.

Parameters:: logger – the logger for this instance.
Returns:: A factory object to create CSP Controller sub-systems observing components.

_add_subsystem_components() → None

Instantiate a Python class for each CSP sub-system component.

Each sub-system component works as an adapter and cache for the associated CSP sub-system. The sub-system component reports the information about one CSP subordinate device and provides the methods or calls to operate on the associate sub-system TANGO device.

On failure, the state and health state of the CSP Controller is forced to FAULT/FAILED.

_populate_subsystems_list(argin: List[str])

Populate a list with the Csp Subsystems components that the controller has to operate on. If argin is an empty list, the CSP Controller operates on all the online Subsystems. If any fqdn is invalid raise a CommandError.

Parameters:: argin – List of fqdn of controller’s subsystems to operate using On and Off command
Return argout:: List of components to operate on
Raises:: CommandError – in case command input argument as only invalid fqdn

_initialize_resource_manager(): Populate the CSP Resource Manager attributes with default values.

_initialize_quality_metrics() → None: Initialize the quality metrics dictionary to default values.

_on(task_name: str, argin: List[Component], task_callback: Callable | None = None, task_abort_event: Event | None = None) → None

Method submitted into thread executor for on command

Parameters:

task_name – the name of the task
argin – list with FQDNs of the CSP Sub-systems Controllers to operate on. If the list is empty operated on all of them.
task_callback – method called to update the task result related attributes
task_abort_event – flag to signal an abort request pending

_off(task_name, argin: List[Component], task_callback: Callable | None = None, task_abort_event: Event | None = None) → None

Method submitted into thread executor for off command

Parameters:

task_name – the name of the task
argin – list with FQDNs of the CSP Sub-systems Controllers to operate on. If the list is empty operated on all of them.
task_callback – method called to update the task result related attributes
task_abort_event – flag to signal an abort request pending

_standby(task_name: str, argin: List[Component], task_callback: Callable | None = None, task_abort_event: Event | None = None) → None

Method submitted into thread executor for standby command

Currently the command is rejected because not used by the sub-systems.

Parameters:

task_name – the name of the task
argin – list with FQDNs of the CSP Sub-systems Controllers to operate on. If the list is empty operated on all of them.
task_callback – method called to update the task result related attributes
task_abort_event – flag to signal an abort request pending

_init_evt_handler() → None: Instantiate and wire the CSP handler for this manager.

init() → None

Method to initialize the Controller ComponentManager. Instantiate:

the command factory
the Component factory
the event manager
a Component object for each CSP sub-system defined in the CSP Subarray Device Properties.

start_communicating() → None

Method to start the monitoring of the CSP subordinate devices, It is invoked when the adminMode of the device is ONLINE or ENGINEERING.

Establish the connection with the CSP subordinate sub-systems Controller devices and with the CSP Subarrays. On success, the main sub-systems attributes are subscribed for events.

_stop_communicating_ordered_list()

stop_communicating()

Method invoked when the CSP subarray is set OFFLINE. CSP Controller stop monitoring the subordinate sub-systems. Events on attributes are un-subscribed and proxies with sub-system devices are invalidated.

TODO: Consider to maintain the proxy and subscription points active: blocking only the aggregation and reporting of the events to the CSP device. In DISABLE state the main operations are prohibited and only ‘emergency’ actions could be allowed.

Raise:: ValueError exception if at least one subarrays is not in EMPTY.

reset(task_callback: Callable[[], None] | None = None, **kwargs: Any) → Tuple[ska_control_model.TaskStatus, str]

on(argin: List[Component], task_callback: Callable[[], None] | None = None) → Tuple[ska_control_model.TaskStatus, str]

Turn the CSP on. This operation is done powering on all subsystem in parallel way.

Parameters:

argin – list with FQDNs of the CSP Sub-systems Controllers to operate on. If the list is empty operated on all of them.
task_callback – method called to update the task result related attributes

Returns:

a tuple with TaskStatus and an informative message.

off(argin: List[Component], task_callback: Callable[[], None] | None = None) → Tuple[ska_control_model.TaskStatus, str]

Turn the CSP off.

Parameters:

argin – list with FQDNs of the CSP Sub-systems Controllers to operate on. If the list is empty operated on all of them.
task_callback – method called to update the task result related attributes

Returns:

a tuple with TaskStatus and an informative message.

standby(argin: List[Component], task_callback: Callable[[], None] | None = None) → Tuple[ska_control_model.TaskStatus, str]

Send the component to low-power mode. This operation is done on one sub-system after the other in an ordered way.

Currently the command is rejected because not used by the sub-systems.

Parameters:

argin –
list with FQDNs of the CSP Sub-systems Controllers to operate on. If the list is empty operated on all of them.

CSP Sub-system controllers are put in low-power mode, while the subarrays are switched off (observations not available in low power mode)
task_callback – method called to update the task result related attributes

Returns:

a tuple with TaskStatus and an informative message.

get_cbf_simulation_mode() → bool

Get the Cbf Simulation Mode attribute.

Returns:: Cbf Simulation mode flag.

set_cbf_simulation_mode(value) → None

Set the CBF Simulation Mode of the device.

Parameters:: value – desired Simulation Mode of the device.

update_quality_metrics(subsystem_fqdn: str, attr_name: str, attr_value: Any, valid: bool = True) → None: Callback to update the QualityAssuranceMetrics attribute.