ska_tmc_subarraynode.commands package

Link to the TMC User documentation is here.

Submodules

subarray_node_command

Path: src/ska_tmc_subarraynode/commands/subarray_node_command.py

SubarrayNode command class for SubarrayNode, a subclass of TMCCommand. This class provides methods to initialize adapters and executing commands on subarray devices.

class ska_tmc_subarraynode.commands.subarray_node_command.SubarrayNodeCommand(component_manager, adapter_factory=ska_tmc_common.AdapterFactory, logger=<Logger ska_tmc_subarraynode.commands.subarray_node_command (WARNING)>)

Bases: TMCCommand

This code defines the SubarrayNodeCommand class, which is a subclass of TMCCommand. This class provides methods for initializing adapters and executing commands on subarray devices.

adapter_error_message_result(dev_name, exception)

Returns ResultCode.FAILED with failure message as a tuple

Parameters

dev_name (str) – The name of the device for which the adapter creation failed.
exception (Exception) – The exception raised during adapter creation.

Return type

Tuple[ResultCode, str]

Returns

A tuple containing the result code and the error message.

clear_command_in_progress_data(): This is a utility method used to clear command_in_progress.

clear_device_events()

Clears the device events dictionary

Return type: None

do(argin=None)

This method is called to execute the command. It sets the command timestamp and calls either do_mid or do_low based on the input parameter type. It returns the result of the command execution.

Parameters: argin (str) – The input argument for the command. Defaults to None.
Return type: Tuple[ResultCode, str]
Returns: A tuple containing the result code and a message.

do_low(argin=None)

Abstract method from TMCCommand is defined here but not utilized by this Class.

Parameters: argin (Optional[Any]) – Accepts argument if required. Defaults to None.
Return type: Tuple[ResultCode, str]
Returns: A tuple containing the result code and a message.

do_mid(argin=None)

Abstract method from TMCCommand is defined here but not utilized by this Class.

Parameters: argin (Optional[Any]) – Accepts argument if required. Defaults to None.
Return type: Tuple[ResultCode, str]
Returns: A tuple containing the result code and a message.

get_adapter_by_device_name(device_name)

The get_adapter_by_device_name method takes a device_name as input and searches for an adapter object in the adapter_factory object’s adapters attribute that matches the input device_name. If a matching adapter object is found, it is returned. If no matching adapter object is found, first adapter is returned.

Parameters: device_name (str) – The name of the device to search for.
Return type: BaseAdapter
Returns: An adapter object if a matching device is found, otherwise first adapter object is returned.

get_dish_adapter_by_device_name(device_name_list)

The get_dish_adapter_by_device_name method takes a list of device_name_list as input and searches for adapter objects in the adapter_factory object’s adapters attribute that match any of the device names in the input list. It returns a list of matching adapter objects.

Parameters: device_name_list (list) – A list of device names to search for.
Return type: List
Returns: A list of adapter objects that match the device names in the input list. If no matching adapter object is found, an empty list is returned.

init_adapters()

Initializes the adapters for the subarray node command.

Return type: Tuple[ResultCode, str]
Returns: A tuple containing the result code and a message.

init_adapters_low()

Initializes the low adapters for the subarray node command.

Return type: Tuple[ResultCode, str]
Returns: A tuple containing the result code and a message.

init_adapters_mid()

Initializes the mid adapters for the subarray node command.

Return type: Tuple[ResultCode, str]
Returns: A tuple containing the result code and a message.

reject_command(message)

Method to return task status as TaskStatus.REJECTED along with the error message

Parameters: message (str) – The error message to be logged.
Return type: Tuple[ResultCode, str]
Returns: A tuple containing the task status and the error message.

set_subarray_obs_state_to_fault_on_command_timeout(): Method to set Subarray Node ObsState to FAULT when timeout occurs on one of the leaf node.

task_callback_default(**kwargs)

Default method if the taskcallback is not passed

Parameters

kwargs – keyword arguments such as :-
status (-) – status of the task.
progress (-) – progress of the task.
result (-) – result of the task.
exception (-) – an exception raised from the task.

Return type

None

update_command_in_progress_id(command_name)

The update_command_in_progress_id method updates command_in_progress_id attributes of the component_manager object with the given command_name. It also logs information about the updated command.

Parameters: command_name (str) – The name of the command being updated.
Return type: None

update_event_data_storage(device_name, unique_command_id)

This method populates the command_id in eventdatastorage.

Parameters

device_name (str) – device name
unique_command_id (str) – command id

Return type

None

update_event_data_storage_kvalue(device_name='leaf-node-dish', kvalue_validation_status=False)

This method populates the kvalue_validation_status in eventdatastorage.

Parameters

device_name (str) – device name
kvalue_validation_status (bool) – status of kvalue_validation

Return type

None

update_task_status(**kwargs)

Abstract method from TMCCommand is defined here but not utilized by this Class.

Parameters: kwargs – keyword arguments
Return type: None

assign_resources_command

Path: src/ska_tmc_subarraynode/commands/assign_resources_command.py

AssignResourcesCommand class for SubarrayNode.

class ska_tmc_subarraynode.commands.assign_resources_command.AssignResources(*args, **kwargs)

Bases: SubarrayNodeCommand

A class for SubarrayNode’s AssignResources() command.

Assigns resources to CSP, SDP via respective subarray leaf nodes. It also sets AssignedResources attribute on SubarrayNode.

assign_sdp_resources(argin)

This function accepts SDP block from input AssignResources JSON and assigns SDP resources to SDP Subarray through SDP Subarray Leaf Node.

Parameters: argin (list[str]) – JSON string containing only SDP resources.
Return type: Tuple[ResultCode, str]
Returns: A tuple containing a return code and a string message.

clear_resources()

Method for clearing resources in case of failure

Return type: None

abstract do(argin=None)

Abstract method to invoke the AssignResources command.

Parameters: argin (Optional[str]) – Optional argument
Return type: Tuple[ResultCode, str]
Returns: A tuple containing a return code and a string message indicating execution status of command.
Raises: NotImplementedError – If not implemented in derived classes.

invoke_command(method, resources): This method assign resources command. :type method: Callable[[dict], Tuple] :param method: method for assigning resources :type method: dict :type resources: dict :param resources: resource configuration :type resources: dict

load_json(argin)

This methods helps loading the string configuration into: dictionary.

Parameters: argin (str) – configure json in string format
Raises: Exception – Raises exception if json is not valid.
Returns: configuration json in dictionary data structure.
Return type: dict

update_task_status(**kwargs)

Method to update task status with result code and exception message if any.

Parameters

kwargs – keyword arguments such as :-
result (-) – Tuple containing ResultCode and unique id
status (-) – Current Task Status
message (-) – Any Failure/Exception message

Return type

None

configure_command

Path: src/ska_tmc_subarraynode/commands/configure_command.py

Configure Command class for SubarrayNode.

class ska_tmc_subarraynode.commands.configure_command.Configure(*args, **kwargs)

Bases: SubarrayNodeCommand

A class for SubarrayNode’s Configure() command.

Configures the resources assigned to the Subarray. The configuration data for SDP, CSP and Dish for mid telescope and MCCS for low telescope is extracted out of the input configuration string and relayed to the respective underlying devices (SDP Subarray Leaf Node, CSP Subarray Leaf Node and Dish Leaf Node for mid telescope and MCCS Leaf Node for low telescope ).

abstract do(argin=None)

Abstract method to invoke the AssignResources command.

Parameters: argin (Optional[str]) – Optional argument
Return type: Tuple[ResultCode, str]
Returns: A tuple containing a return code and a string message indicating execution status of command.
Raises: NotImplementedError – If not implemented in derived classes.

property is_wrap_sector_key_present

Returns True if the wrap sector key present in the configure.

Returns: True if wrap sector key present
Return type: bool

load_json(argin)

This methods helps loading the string configuration into: dictionary.

Parameters: argin (str) – configure json in string format
Raises: Exception – Raises exception if json is not valid.
Returns: configuration json in dictionary data structure.
Return type: dict

property receive_addresses

Dynamically retrieves the receive addresses with a retry mechanism.

This property tries up to self.retry_attempts times to:

Retrieve the device name via component_manager.
Get the adapter using get_adapter_by_device_name(dev_name).
Parse its receive_addresses JSON.

Returns: The parsed receive addresses if successful, otherwise None.
Return type: dict | list | None

set_scan_duration(scan_duration)

Method to set the scan duration in component manager

Parameters: scan_duration (int) – scan duration in seconds.
Raises: Exception – Raises exception if the scan duration is not set in component manager.

update_task_status(**kwargs)

Method to update task status with result code and exception message if any.

Parameters

kwargs – keyword arguments such as :-
result (-) – Tuple containing ResultCode and unique id
status (-) – Current Task Status
message (-) – Any Failure/Exception message

Return type

None

end_command

Path: src/ska_tmc_subarraynode/commands/end_command.py

A class for TMC SubarrayNode’s End() command (Abstract Base Class)

class ska_tmc_subarraynode.commands.end_command.End(*args, **kwargs)

Bases: SubarrayNodeCommand

A class for SubarrayNode’s End() command. This command on Subarray Node invokes End command on CSP Subarray Leaf Node and SDP Subarray Leaf Node, MCCS LeafNodes for Low and stops tracking of all the assigned dishes on Mid.

abstract do(argin=None)

Abstract method to invoke End command.

Parameters: argin – Optional argument
Return type: Tuple[ResultCode, str]
Returns: A tuple containing a return code and a string message indicating execution status of command.
Raises: NotImplementedError – If not implemented in derived classes.

end_csp()

End command on CSP Subarray Leaf Node

Return type: Tuple[ResultCode, str]
Returns: A tuple containing ResultCode and message.

end_sdp()

End command on SDP Subarray Leaf Node

Return type: Tuple[ResultCode, str]
Returns: A tuple containing ResultCode and message.

reset_scan_duration()

Helper method to reset the scan duration and handle errors.

Return type: Tuple[ResultCode, str]
Returns: A tuple containing ResultCode and message.

update_task_status(**kwargs)

Method to update task status with result code and exception message if any.

Parameters

kwargs – Keyword arguments such as :-
result (-) – Tuple containing ResultCode and unique id
status (-) – Current Task Status
message (-) – Any Failure/Exception message

Return type

None

end_scan_command

Path: src/ska_tmc_subarraynode/commands/end_scan_command.py

A class for TMC SubarrayNode’s EndScan() command (Abstract Base Class).

class ska_tmc_subarraynode.commands.end_scan_command.EndScan(*args, **kwargs)

Bases: SubarrayNodeCommand

A class for SubarrayNode’s EndScan() command.

Ends the scan. It is invoked on subarray after completion of the scan duration. It can also be invoked by an external client while a scan is in progress, Which stops the scan immediately irrespective of the provided scan duration.

abstract do(argin=None)

Abstract method to invoke EndScan command.

Parameters: argin (Optional[str]) – Optional argument
Return type: Tuple[ResultCode, str]
Returns: Tuple of ResultCode and message
Raises: NotImplementedError – If not implemented in derived classes.

endscan_csp()

Method to invoke EndScan command on CSP Subarray Leaf Node.

Return type: Tuple[ResultCode, str]
Returns: (ResultCode, message)

endscan_sdp()

Method to invoke EndScan command on SDP Subarray Leaf Node.

Return type: Tuple[ResultCode, str]
Returns: (ResultCode, message)

update_task_status(**kwargs)

Method to update task status with result code and exception message if any.

Parameters

kwargs – Keyword arguments such as :-
result (-) – Tuple containing ResultCode and unique id
status (-) – Current Task Status
message (-) – Any Failure/Exception message

Return type

None

release_all_resources_command

Path: src/ska_tmc_subarraynode/commands/release_all_resources_command.py

ReleaseAllResources Command for SubarrayNode

class ska_tmc_subarraynode.commands.release_all_resources_command.ReleaseAllResources(*args, **kwargs)

Bases: SubarrayNodeCommand

A class for TMC SubarrayNode’s ReleaseAllResources() command.

It checks whether all resources are already released. If yes then it returns code FAILED. If not it Releases all the resources from the subarray i.e. Releases resources from TMC Subarray Node, CSP Subarray and SDP Subarray. Upon successful execution, all the resources of a given subarray get released and empty array is returned. Selective release is not yet supported.

abstract clean_up_resources()

Clears the assignedResources attribute.

Cleans dictionaries of the resources across the subarraynode.

Note

Currently there are only receptors allocated so only the receptor ids details are stored.

Return type: Tuple[ResultCode, str]
Returns: Tuple of Resultcode and message

abstract do(argin=None)

Method to invoke ReleaseAllResources command.

Parameters: argin – Optional argument
Return type: Tuple[ResultCode, str]
Returns: A tuple containing a return code and “” as a string on successful release all resources.
Raises: NotImplementedError if not implemented in derived classes –

release_csp_resources()

This function invokes releaseAllResources command on CSP Subarray via CSP Subarray Leaf Node.

Return type: Tuple[ResultCode, str]
Returns: Tuple of Resultcode and message

release_sdp_resources()

This function invokes releaseAllResources command on SDP Subarray via SDP Subarray Leaf Node.

Return type: Tuple[ResultCode, str]
Returns: Tuple of Resultcode and message

update_task_status(**kwargs)

Method to update task status with result code and exception message if any.

Parameters

kwargs – Keyword arguments such as :-
result (-) – Tuple containing ResultCode and unique id
status (-) – Current Task Status
message (-) – Any Failure/Exception message

Return type

None

restart_command

Path: src/ska_tmc_subarraynode/commands/restart_command.py

Restart Command for SubarrayNode.

class ska_tmc_subarraynode.commands.restart_command.Restart(*args, **kwargs)

Bases: SubarrayNodeCommand

A class representing the Restart command for SubarrayNode.

This command restarts various leaf nodes of the Subarray, including CSP Subarray Leaf Node, SDP Subarray Leaf Node, and Mccs Subarray Leaf Node in Low and Dish Leaf Nodes in Mid, to reset ongoing activities and configurations.

Parameters

component_manager – The component manager instance.
obs_state_model – The observational state model of the Subarray.
adapter_factory – An optional adapter factory instance.
logger – An optional logger instance for logging. Default is None.

clean_up_configuration()

Cleans up the configuration and resources of the Subarray.

Return type: Tuple[ResultCode, str]
Returns: A tuple containing the result code and an empty string.

abstract do(argin=None)

Invokes the Restart command on subarray leaf nodes.

Parameters: argin (Optional[str]) – Optional argument. Default is None.
Return type: Tuple[ResultCode, str]
Returns: A tuple containing the result code and a message indicating the execution status.
Raises: NotImplementedError if not implemented in derived classes –

restart_csp()

Send Restart command to CSP devices

Return type: Tuple[ResultCode, str]
Returns: A tuple containing ResultCode and message

restart_sdp()

Send Restart command to SDP devices

Return type: Tuple[ResultCode, str]
Returns: A tuple containing ResultCode and message

update_task_status(**kwargs)

Method to update task status with result code and exception message if any.

Parameters

kwargs – Keyword Arguments such as :-
result (-) – Tuple containing ResultCode and unique id
status (-) – Current Task Status
message (-) – Any Failure/Exception message

Return type

None

scan_command

Path: src/ska_tmc_subarraynode/commands/scan_command.py

A class for TMC SubarrayNode’s Scan() command (Abstract Base Class)

class ska_tmc_subarraynode.commands.scan_command.Scan(*args, **kwargs)

Bases: SubarrayNodeCommand

A class for SubarrayNode’s Scan() command. The command accepts Scan id as an input and executes a scan on the subarray. Scan command is invoked on respective CSP and SDP subarray node for the provided interval of time. It checks whether the scan is already in progress. If yes it throws error showing duplication of command.

abstract do(argin)

Abstract method to invoke Scan command.

Parameters: argin (str) – JSON string containing id.
Return type: Tuple[ResultCode, str]
Returns: A tuple containing a return code and a string message indicating status. The message is for information purpose only.
Raises: NotImplementedError – If not implemented in derived classes.

parse_json_input(argin)

Helper method to parse JSON input string and handle errors.

Parameters: argin (str) – JSON string to parse
Return type: Tuple[ResultCode, dict]
Returns: Tuple of ResultCode and either the parsed JSON dictionary or an error message

abstract scan_csp(argin)

Abstract method to invoke Scan command on CSP Subarray Leaf Node.

Parameters: argin (str) – Additional argument
Return type: Tuple[ResultCode, str]
Returns: (ResultCode, message)
Raises: NotImplementedError – If not implemented in derived classes.

scan_sdp(argin)

Method to invoke Scan command on SDP Subarray Leaf Node.

Parameters: argin (str) – json input
Return type: Tuple[ResultCode, str]
Returns: (ResultCode, message)

start_scan_timer(scan_duration)

Method for starting scan timer

Parameters: scan_duration (int) – scan duration in second
Return type: None

submit_endscan_command()

Will submit the end scan

Return type: None

update_task_status(**kwargs)

Method to update task status with result code and exception message if any.

Parameters

kwargs – Keyword arguments such as :-
result (-) – Tuple containing ResultCode and unique id
status (-) – Current Task Status
message (-) – Any Failure/Exception message

Return type

None

off_command

Path: src/ska_tmc_subarraynode/commands/off_command.py

Off Command for SubarrayNode

class ska_tmc_subarraynode.commands.off_command.Off(*args, **kwargs)

Bases: SubarrayNodeCommand

A class for SubarrayNode’s Off() command.

abstract do(argin=None)

Invokes the Off command on subarray leaf nodes.

Parameters: argin (Optional[str]) – Optional argument. Default is None.
Return type: Tuple[ResultCode, str]
Returns: A tuple containing the result code and a message indicating the execution status.
Raises: NotImplementedError if not implemented in derived classes –

get_csp_subarray_obstate()

Return obstate of csp subarray obstate

Return type: ObsState
Returns: ObsState of the csp subarray

get_sdp_subarray_obstate()

Return obstate of sdp subarray obstate

Return type: ObsState
Returns: ObsState of the sdp subarray

get_subarray_obstate(dev_name)

Return obstate of subarray device

Parameters: dev_name (str) – Name of the subarray device
Return type: ObsState
Returns: ObsState of the subarray

invoke_sdp_off()

Invokes the Off command on the SDP Subarray Leaf Node.

Return type: Tuple[ResultCode, str]
Returns: A tuple containing a return code and a string message indicating execution status of command.

subarray_off(logger, task_callback, task_abort_event=None)

This is a long running method for Off command, it executes do hook, invokes Off command on SdpSubarrayleafnode.

Parameters

logger (Logger) – logger
task_callback (TaskCallbackType) – Update task state, defaults to None
task_abort_event (Optional[Event]) – Check for abort, defaults to None

Return type

None

update_task_status()

Method for implementing for updating task status

Return type: None

on_command

Path: src/ska_tmc_subarraynode/commands/on_command.py

On Command for SubarrayNode

class ska_tmc_subarraynode.commands.on_command.On(*args, **kwargs)

Bases: SubarrayNodeCommand

A class for the SubarrayNode’s On() command.

abstract do(argin=None)

Method to invoke On command on subarray leaf nodes.

Parameters: argin – Input json for Command, defaults to None
Return type: Tuple[ResultCode, str]
Returns: A tuple containing a return code and a string message indicating execution status of command.
Raises: NotImplementedError if not implemented in derived classes –

invoke_sdp_on()

Method to invoke On command on SdpSubarrayLeafNode.

Parameters: None –
Return type: Tuple[ResultCode, str]
Returns: A tuple containing a return code and a string message indicating execution status of command.
Raises: Exception if the command execution is not successful –

on_leaf_nodes(logger, task_callback, task_abort_event=None)

This is a long running method for On command, it executes do hook, invokes On command on CspSubarrayleafnode and SdpSubarrayleafnode.

Parameters

logger (Logger) – logger
task_callback (TaskCallbackType) – Update task state, defaults to None
task_abort_event (Optional[Event]) – Check for abort, defaults to None

Return type

None

set_admin_mode_command

Path: src/ska_tmc_subarraynode/commands/set_admin_mode_command.py

Class for setting the adminMode of lower level devices

class ska_tmc_subarraynode.commands.set_admin_mode_command.SetAdminMode(*args, **kwargs)

Bases: SubarrayNodeCommand

A class for SubarrayNode’s SetAdminMode command.

do_low(argin)

Method to invoke SetAdminMode command on the Subarray Leaf Nodes.

Parameters: argin (str) – Input JSON for the command, containing the admin mode to be set.
Return type: Tuple[ResultCode, str]
Returns: A tuple containing a return code and a string message indicating the execution status of the command.

invoke_setadminmode(argin, task_callback, task_abort_event=None)

This is a long-running method for the SetAdminMode command. It executes the do hook and invokes SetAdminMode on Subarray Leaf Nodes.

Parameters

argin (str) – Internal argument
task_callback (TaskCallbackType) – An optional callback function for task status updates. Default is None.
task_abort_event (Optional[Event]) – An optional threading event for checking task abort status. Default is None.

Return type

None

set_admin_mode_csp(admin_mode)

SetAdminMode command on CSP Subarray Leaf Node.

Parameters: admin_mode (int) – The admin mode to be set.
Return type: Tuple[ResultCode, str]
Returns: A tuple containing the result code and a message.

set_admin_mode_mccs(admin_mode)

SetAdminMode command on MCCS Subarray Leaf Node.

Parameters: admin_mode (int) – The admin mode to be set.
Return type: Tuple[ResultCode, str]
Returns: A tuple containing the result code and a message.

set_admin_mode_sdp(admin_mode)

SetAdminMode command on SDP Subarray Leaf Node.

Parameters: admin_mode (int) – The admin mode to be set.
Return type: Tuple[ResultCode, str]
Returns: A tuple containing the result code and a message.

abort_command

Path: src/ska_tmc_subarraynode/commands/abort_command.py

Abstract base class for TMC SubarrayNode’s Abort() command.

class ska_tmc_subarraynode.commands.abort_command.Abort(*args, **kwargs)

Bases: SubarrayNodeCommand

A class for SubarrayNode Abort() command. It aborts the ongoing operation on subarray node.

abort_csp()

Send abort command to CSP devices

Return type: Tuple[ResultCode, str]
Returns: A tuple containing ResultCode and message

abort_sdp()

Send abort command to SDP devices

Return type: Tuple[ResultCode, str]
Returns: A tuple containing ResultCode and message

abstract do(argin=None)

Abstract method to invoke the Abort command.

Parameters

argin (Optional[str]) – Optional argument

Return type

Tuple[ResultCode, str]

Returns

A tuple containing a return code and a string message indicating execution status of command.

Raises

NotImplementedError – If not implemented
in derived classes. –

set_subarray_obsstate_fault()

This method sets subarray obsstate to fault.

Return type: None

update_task_status(**kwargs)

Method to update task status with result code and exception message if any.

Parameters

kwargs – Keyword arguments such as :-
result (-) – Tuple containing ResultCode and unique id
status (-) – Current Task Status
message (-) – Any Failure/Exception message

Return type

None

build_up_data

Path: src/ska_tmc_subarraynode/commands/build_up_data.py

Class to generate schema.

class ska_tmc_subarraynode.commands.build_up_data.ElementDeviceData(scan_config, component_manager)

Bases: object

A class representing data for an element device.

build_up_csp_cmd_data(receive_addresses_map)

Builds the input data for CSP configuration.

Parameters: receive_addresses_map (str) – Map of receive addresses
Return type: Tuple[ResultCode, str]
Returns: Returns Resultcode and message

build_up_sdp_cmd_data()

Method to build up sdp command data

Return type: Tuple[ResultCode, str]
Returns: Returns ResultCode and sdp configuration schema

class ska_tmc_subarraynode.commands.build_up_data.ElementDeviceDataLow(scan_config, component_manager)

Bases: ElementDeviceData

A class representing data for an Low element device.

build_up_csp_cmd_data(receive_addresses_map)

Builds the input data for CSP configuration.

Parameters: receive_addresses_map (str) – Map of receive addresses
Return type: Tuple[ResultCode, str]
Returns: Result code and CSP configuration schema

class ska_tmc_subarraynode.commands.build_up_data.ElementDeviceDataMid(scan_config, component_manager)

Bases: ElementDeviceData

A class representing data for an Mid element device.

build_up_csp_cmd_data(receive_addresses_map)

Builds the input data for CSP configuration.

Parameters: receive_addresses_map (str) – Map of receive addresses
Return type: Tuple[ResultCode, str]
Returns: Result code and CSP configuration schema

build_up_dish_cmd_data()

Method to build up dish command data

Return type: Tuple[ResultCode, str]
Returns: Result code and dish configuration schema

ska_tmc_subarraynode.commands package

Submodules

subarray_node_command

assign_resources_command

configure_command

end_command

end_scan_command

release_all_resources_command

restart_command

scan_command

off_command

on_command

set_admin_mode_command

abort_command

build_up_data

Module contents