ska_tmc_cspsubarrayleafnode.manager package

Submodules

ska_tmc_cspsubarrayleafnode.manager.component_manager module

Component Manager class for CSP Subarray Leaf Node

class ska_tmc_cspsubarrayleafnode.manager.component_manager.CspSLNComponentManager(*args: Any, **kwargs: Any)

Bases: TmcLeafNodeComponentManager

A component manager for The CSP Subarray Leaf Node component.

It supports:

Monitoring its component, e.g. detect that it has been turned off or on

abort_commands(): Invokes Abort command on lower level Csp Subarray and changes the obsstate

assign_resources(argin: str, task_callback: ska_tango_base.base.TaskCallbackType) → Tuple[ska_tango_base.executor.TaskStatus, str]

Submit the AssignResources command in queue.

Returns:: a result code and message

configure(argin: str, task_callback: ska_tango_base.base.TaskCallbackType) → Tuple[ska_tango_base.executor.TaskStatus, str]

Submit the Configure command in queue.

Returns:: a result code and message

property delay_model: str

Returns the delayModel attribute.

Returns:: the delayModel
Return type:: str

end(task_callback: ska_tango_base.base.TaskCallbackType) → Tuple[ska_tango_base.executor.TaskStatus, str]

Submit the End command in queue.

Returns:: a result code and message

end_scan(task_callback: ska_tango_base.base.TaskCallbackType) → Tuple[ska_tango_base.executor.TaskStatus, str]

Submit the EndScan command in queue.

Returns:: a result code and message

get_device() → ska_tmc_common.device_info.SubArrayDeviceInfo

Return the device info

Parameters:: None –
Returns:: a device info
Return type:: SubArrayDeviceInfo

get_obs_state(): Get Current device obsState

property input_parameter: Return the input parameter :return: input parameter :rtype: InputParameter

is_command_allowed(command_name: str) → Union[bool, ska_tmc_common.exceptions.InvalidObsStateError, ska_tmc_common.exceptions.DeviceUnresponsive, ska_tmc_common.exceptions.CommandNotAllowed]

Checks whether this command is allowed. It checks that the device is in a state to perform this command and that all the component needed for the operation are not unresponsive.

Parameters:: command_name (str) – name of the command
Returns:: True if this command is allowed
Return type:: boolean

is_command_allowed_callable(command_name: str)

Parameters:: command_name (str) – Name for the command for which the is_allowed check need to be applied.

is_scan_completed(): A function to check whether Scan Command is completed succesfully.

off(task_callback): Empty method for resolving lint issue

on(task_callback): Empty method for resolving lint issue

raise_invalid_obsstate_error(command_name: str) → ska_tmc_common.exceptions.InvalidObsStateError: checking the InvalidObsState of Csp Subarray device :param command_name: The name of command :type command_name: str

release_all_resources(task_callback: ska_tango_base.base.TaskCallbackType) → Tuple[ska_tango_base.executor.TaskStatus, str]

Submit the ReleaseAllResources command in queue.

Returns:: a result code and message

restart(task_callback: Optional[Callable] = None) → tuple

Submit the Restart command in queue.

Returns:: a result code and message

scan(argin: str, task_callback: ska_tango_base.base.TaskCallbackType) → Tuple[ska_tango_base.executor.TaskStatus, str]

Submit the Scan command in queue.

Returns:: a result code and message

set_target(target): Set the target with KatPoint Target object

standby(task_callback): Empty method for resolving lint issue

start_communicating(): Empty method for resolving lint issue

stop() → None: Stops the event receiver

stop_communicating(): Empty method for resolving lint issue

update_command_result(device_name: str, value: str) → None: Updates the long running command result callback

update_device_admin_mode(device_name: str, admin_mode: ska_control_model.AdminMode) → None: Update a monitored device admin mode, and call the relative callbacks if available :param admin_state: admin mode of the device :type admin_mode: AdminMode

update_device_obs_state(device_name: str, obs_state: ska_tango_base.control_model.ObsState) → None: Update a monitored device obs state, and call the relative callbacks if available :param obs_state: obs state of the device :type obs_state: ObsState

update_event_failure(device_name: str) → None

update_exception_for_unresponsiveness(device_info: ska_tmc_common.device_info.SubArrayDeviceInfo, exception: str) → None: Set a device to failed and call the relative callback if available :param device_info: a device info :type device_info: DeviceInfo :param exception: an exception :type: Exception

update_responsiveness_info(device_name: str) → None

Update a device with the correct availability information.

Parameters:: dev_name (str) – name of the device

ska_tmc_cspsubarrayleafnode.manager.event_receiver module

Event Receiver for CSP Subarray Leaf Node

class ska_tmc_cspsubarrayleafnode.manager.event_receiver.CspSLNEventReceiver(*args: Any, **kwargs: Any)

Bases: EventReceiver

The CSP SLNEventReceiver class has the responsibility to receive events from the CSP Subarray managed by the CSP Subarray Leaf Node.

The ComponentManager uses the handle events methods for the attribute of interest. For each of them a callback is defined.

handle_command_result_event(event: tango.EventType.CHANGE_EVENT) → None: Handle the LRCR event.

handle_obs_state_event(event: tango.EventType.CHANGE_EVENT) → None

run() → None

subscribe_events(dev_info: ska_tmc_common.device_info.SubArrayDeviceInfo, attribute_dictionary: Optional[Dict[str, str]] = None)

Module contents

Init file for CSP Subarray Leaf Node Manager

class ska_tmc_cspsubarrayleafnode.manager.CspSLNComponentManager(*args: Any, **kwargs: Any)

Bases: TmcLeafNodeComponentManager

A component manager for The CSP Subarray Leaf Node component.

It supports:

Monitoring its component, e.g. detect that it has been turned off or on

abort_commands(): Invokes Abort command on lower level Csp Subarray and changes the obsstate

assign_resources(argin: str, task_callback: ska_tango_base.base.TaskCallbackType) → Tuple[ska_tango_base.executor.TaskStatus, str]

Submit the AssignResources command in queue.

Returns:: a result code and message

configure(argin: str, task_callback: ska_tango_base.base.TaskCallbackType) → Tuple[ska_tango_base.executor.TaskStatus, str]

Submit the Configure command in queue.

Returns:: a result code and message

property delay_model: str

Returns the delayModel attribute.

Returns:: the delayModel
Return type:: str

end(task_callback: ska_tango_base.base.TaskCallbackType) → Tuple[ska_tango_base.executor.TaskStatus, str]

Submit the End command in queue.

Returns:: a result code and message

end_scan(task_callback: ska_tango_base.base.TaskCallbackType) → Tuple[ska_tango_base.executor.TaskStatus, str]

Submit the EndScan command in queue.

Returns:: a result code and message

get_device() → ska_tmc_common.device_info.SubArrayDeviceInfo

Return the device info

Parameters:: None –
Returns:: a device info
Return type:: SubArrayDeviceInfo

get_obs_state(): Get Current device obsState

property input_parameter: Return the input parameter :return: input parameter :rtype: InputParameter

is_command_allowed(command_name: str) → Union[bool, ska_tmc_common.exceptions.InvalidObsStateError, ska_tmc_common.exceptions.DeviceUnresponsive, ska_tmc_common.exceptions.CommandNotAllowed]

Checks whether this command is allowed. It checks that the device is in a state to perform this command and that all the component needed for the operation are not unresponsive.

Parameters:: command_name (str) – name of the command
Returns:: True if this command is allowed
Return type:: boolean

is_command_allowed_callable(command_name: str)

Parameters:: command_name (str) – Name for the command for which the is_allowed check need to be applied.

is_scan_completed(): A function to check whether Scan Command is completed succesfully.

off(task_callback): Empty method for resolving lint issue

on(task_callback): Empty method for resolving lint issue

raise_invalid_obsstate_error(command_name: str) → ska_tmc_common.exceptions.InvalidObsStateError: checking the InvalidObsState of Csp Subarray device :param command_name: The name of command :type command_name: str

release_all_resources(task_callback: ska_tango_base.base.TaskCallbackType) → Tuple[ska_tango_base.executor.TaskStatus, str]

Submit the ReleaseAllResources command in queue.

Returns:: a result code and message

restart(task_callback: Optional[Callable] = None) → tuple

Submit the Restart command in queue.

Returns:: a result code and message

scan(argin: str, task_callback: ska_tango_base.base.TaskCallbackType) → Tuple[ska_tango_base.executor.TaskStatus, str]

Submit the Scan command in queue.

Returns:: a result code and message

set_target(target): Set the target with KatPoint Target object

standby(task_callback): Empty method for resolving lint issue

start_communicating(): Empty method for resolving lint issue

stop() → None: Stops the event receiver

stop_communicating(): Empty method for resolving lint issue

update_command_result(device_name: str, value: str) → None: Updates the long running command result callback

update_device_admin_mode(device_name: str, admin_mode: ska_control_model.AdminMode) → None: Update a monitored device admin mode, and call the relative callbacks if available :param admin_state: admin mode of the device :type admin_mode: AdminMode

update_device_obs_state(device_name: str, obs_state: ska_tango_base.control_model.ObsState) → None: Update a monitored device obs state, and call the relative callbacks if available :param obs_state: obs state of the device :type obs_state: ObsState

update_event_failure(device_name: str) → None

update_exception_for_unresponsiveness(device_info: ska_tmc_common.device_info.SubArrayDeviceInfo, exception: str) → None: Set a device to failed and call the relative callback if available :param device_info: a device info :type device_info: DeviceInfo :param exception: an exception :type: Exception

update_responsiveness_info(device_name: str) → None

Update a device with the correct availability information.

Parameters:: dev_name (str) – name of the device

class ska_tmc_cspsubarrayleafnode.manager.CspSLNComponentManagerLow(*args: Any, **kwargs: Any)

Bases: CspSLNComponentManager

A component manager for The CSP Subarray Leaf Node Low component.

property delay_model_pst_beam: str

Returns the delayModelPSTBeam attribute.

Returns:: the delayModelPSTBeam
Return type:: str

get_station_ids() → list

Returns the station IDs received under lowcbf configurations.

Returns:: station IDs
Return type:: list

get_station_substation_ids() → list[list[Any]]

Returns the stations and sub-stations IDs received under lowcbf configurations.

Returns:: station and sub-station IDs
Return type:: list[list]

class ska_tmc_cspsubarrayleafnode.manager.CspSLNComponentManagerMid(*args: Any, **kwargs: Any)

Bases: CspSLNComponentManager

A component manager for The CSP Subarray Leaf Node Mid component.

get_receptor_ids()

Returns the receptor IDs.

Returns:: receptor IDs
Return type:: list

class ska_tmc_cspsubarrayleafnode.manager.CspSLNEventReceiver(*args: Any, **kwargs: Any)

Bases: EventReceiver

The CSP SLNEventReceiver class has the responsibility to receive events from the CSP Subarray managed by the CSP Subarray Leaf Node.

The ComponentManager uses the handle events methods for the attribute of interest. For each of them a callback is defined.

handle_command_result_event(event: tango.EventType.CHANGE_EVENT) → None: Handle the LRCR event.

handle_obs_state_event(event: tango.EventType.CHANGE_EVENT) → None

run() → None

subscribe_events(dev_info: ska_tmc_common.device_info.SubArrayDeviceInfo, attribute_dictionary: Optional[Dict[str, str]] = None)