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(task_callback: ska_tango_base.base.TaskCallbackType) → Tuple

Aborting the subarray.

Returns:

a result code and message

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

build_device_attribute_map() → Dict[str, List[str]]

Builds a dictionary mapping device names to lists of attributes to be subscribed.

Returns:

A mapping from device names to list of attributes.

Return type:

Dict[str, List[str]]

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_attribute_dict() → dict

Returns:

Dictionary of attributes to be handled by the EventReceiver.

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

process_weather_data(event_data: tango.EventData) → None

Processing the weather data

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 a KatPoint Target object.

Parameters:

• target (katpoint.Target) – A KatPoint Target object representing the

• parameters. (pointing target with sky coordinates and other) –

Returns:

None

setup_event_subscription() → None

Sets up the event subscription after input parameters are updated.

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(value: str) → None

Updates the long running command result callback

update_device_admin_mode(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(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_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

update_weather_data(key: str, value: dict) → None

Called by the EventReceiver to push new sensor readings.

Parameters:

• key (str) – The name of the weather attribute to update

• value (float) – The new sensor reading for that attribute.

Raises:

KeyError – If key is not a recognized weather attribute.

Returns:

None

property weather_data: dict

Read-only access to the current weather data.

Returns:

A mapping of weather attribute names to their current value

Return type:

dict

ska_tmc_cspsubarrayleafnode.manager.event_manager module

Event manager class for CspLeafNode

class ska_tmc_cspsubarrayleafnode.manager.event_manager.CspSLNEventManager(*args: Any, **kwargs: Any)

Bases: EventManager

CspSLNEventManager class inherits from EventManager and adds event callbacks.

adminmode_event_callback(event: tango.EventData) → None

It handles the admin mode events of subsystem’s controller and subaray devices.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.

cspcontrolleradminmode_event_callback(event: tango.EventData) → None

It handles the adminMode events of csp controller device.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.

cspsubarrayobsstate_event_callback(event: tango.EventData) → None

Callback for cspSubarrayObsState attribute.

Delegates to specific handlers based on device type.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.

dishvccconfig_event_callback(event: tango.EventData) → None

Handle dishVccConfig change event.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.

dishvccmapvalidationresult_event_callback(event: tango.EventData) → None

Handle DishVccMapValidationResult change event.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.

healthstate_event_callback(event: tango.EventData) → None

It handles the health state events of subsystem’s controller and subarray devices.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.

issubsystemavailable_event_callback(event: tango.EventData) → None

Method to handle and update the latest value of isSubsystemAvailable attribute.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.

longrunningcommandresult_event_callback(event: tango.EventData) → None

Callback for longRunningCommandResult attribute.

Delegates to specific handlers based on device type.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.

obsstate_event_callback(event: tango.EventData) → None

Handles observation state event.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.

sourcedishvccconfig_event_callback(event: tango.EventData) → None

Handle sourceDishVccConfig change event.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.

state_event_callback(event: tango.EventData) → None

It handles the state events of subsystem’s controller and subaray devices.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.

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(task_callback: ska_tango_base.base.TaskCallbackType) → Tuple

Aborting the subarray.

Returns:

a result code and message

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

build_device_attribute_map() → Dict[str, List[str]]

Builds a dictionary mapping device names to lists of attributes to be subscribed.

Returns:

A mapping from device names to list of attributes.

Return type:

Dict[str, List[str]]

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_attribute_dict() → dict

Returns:

Dictionary of attributes to be handled by the EventReceiver.

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

process_weather_data(event_data: tango.EventData) → None

Processing the weather data

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 a KatPoint Target object.

Parameters:

• target (katpoint.Target) – A KatPoint Target object representing the

• parameters. (pointing target with sky coordinates and other) –

Returns:

None

setup_event_subscription() → None

Sets up the event subscription after input parameters are updated.

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(value: str) → None

Updates the long running command result callback

update_device_admin_mode(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(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_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

update_weather_data(key: str, value: dict) → None

Called by the EventReceiver to push new sensor readings.

Parameters:

• key (str) – The name of the weather attribute to update

• value (float) – The new sensor reading for that attribute.

Raises:

KeyError – If key is not a recognized weather attribute.

Returns:

None

property weather_data: dict

Read-only access to the current weather data.

Returns:

A mapping of weather attribute names to their current value

Return type:

dict

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

Returns the delayModelPSTBeam attribute.

Returns:

the delayModelPSTBeam

Return type:

str

property delay_model_pst_beam2: str

Returns the delayModelPSTBeam attribute.

Returns:

the delayModelPSTBeam

Return type:

str

property delay_model_station_beam_01: str

Get the delay model value for station beam 01.

Returns: str: JSON string representing the delay model for beam 01.

property delay_model_station_beam_02: str

Get the delay model value for station beam 02.

Returns: str: JSON string representing the delay model for beam 02.

property delay_model_station_beam_03: str

Get the delay model value for station beam 03.

Returns: str: JSON string representing the delay model for beam 03.

property delay_model_station_beam_04: str

Get the delay model value for station beam 04.

Returns: str: JSON string representing the delay model for beam 04.

get_pst_beam_tango_attributes() → list[str]

Function to fetch pst beam tango attrs from Component Manager obj Returns:

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]

get_stn_beam_tango_attributes() → list[str]

Function to fetch station beam tango attrs from Component Manager obj Returns:

set_beam_target(beam_id: int, target_object) → None

Set the target for a specific beam using a KatPoint Target object.

Parameters:

• beam_id (int) – ID of the beam to set the target for.

• target_object – KatPoint Target object representing the beam’s

• target. –

Example

set_beam_target(1, katpoint.Target(“azel, 30, 45, target1”))

Returns:

None

update_delay_model_for_station_beam(beam_id: str, model: str) → None

Update delay model value for a specific station beam.

Parameters:

• beam_id (str) – Beam ID as a string (e.g., “1”, “2”, “3”, “4”).

• model (str) – JSON string representing the delay model.

Returns:

None

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

Bases: CspSLNComponentManager

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

get_attribute_dict() → dict

Extends the common attribute dictionary with mid-specific attributes.

Returns:

Dictionary of attributes to be handled by the EventReceiver.

get_receptor_ids()

Returns the receptor IDs.

Returns:

receptor IDs

Return type:

list

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

Bases: EventManager

CspSLNEventManager class inherits from EventManager and adds event callbacks.

adminmode_event_callback(event: tango.EventData) → None

It handles the admin mode events of subsystem’s controller and subaray devices.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.

cspcontrolleradminmode_event_callback(event: tango.EventData) → None

It handles the adminMode events of csp controller device.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.

cspsubarrayobsstate_event_callback(event: tango.EventData) → None

Callback for cspSubarrayObsState attribute.

Delegates to specific handlers based on device type.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.

dishvccconfig_event_callback(event: tango.EventData) → None

Handle dishVccConfig change event.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.

dishvccmapvalidationresult_event_callback(event: tango.EventData) → None

Handle DishVccMapValidationResult change event.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.

healthstate_event_callback(event: tango.EventData) → None

It handles the health state events of subsystem’s controller and subarray devices.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.

issubsystemavailable_event_callback(event: tango.EventData) → None

Method to handle and update the latest value of isSubsystemAvailable attribute.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.

longrunningcommandresult_event_callback(event: tango.EventData) → None

Callback for longRunningCommandResult attribute.

Delegates to specific handlers based on device type.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.

obsstate_event_callback(event: tango.EventData) → None

Handles observation state event.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.

sourcedishvccconfig_event_callback(event: tango.EventData) → None

Handle sourceDishVccConfig change event.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.

state_event_callback(event: tango.EventData) → None

It handles the state events of subsystem’s controller and subaray devices.

Parameters:

event_data (tango.EventType.CHANGE_EVENT) – to flag the change in event.