TMC Common Tango Helper Devices

TMC uses helper devices instead of real devices to test component level functionality and also functionality of overall integrated TMC in TMC integration repository. These Devices are used by multiple TMC repositories so kept in common repository to have functionality readily available across TMC repositories.

Devices and their functionality

This page contains all the helper devices used by TMC to mock behavior of sub-systems,leaf nodes or subarray device. Each helper device exposes required API’s similar to real device.

Functionality

Few Functionality provided by the helper devices are as follows:

SetDirectObsState: used to set device obsState directly.
SetDirectState: Used to set device state directly.
commandCallInfo: Used to get information of arguments used to call the command
SetDelay: Used to produce delay in execution of commands to mimic real scenario.
resetDelay: Used to reset delay to default one.
Fault Injection: To make devices faulty to check error handling in tango devices.
AddTransitions: This command will set duration for obs state such that whenrespective command for obs state is triggered then it change obs state
after provided duration.

Device List

1. Helper_CSP_Master

This module defines a helper device that acts as csp master in our testing.

class ska_tmc_common.test_helpers.helper_csp_master_device.HelperCspMasterDevice(*args: Any, **kwargs: Any)

Bases: HelperBaseDevice

A helper device class for Csp Controller device

init_device() → None

class InitCommand(*args: Any, **kwargs: Any)

Bases: InitCommand

A class for the HelperCspMasterDevice’s init_device() command.

do() → Tuple[ska_tango_base.commands.ResultCode, str]: Stateless hook for device initialisation. :return: ResultCode and message

read_sourceDishVccConfig() → str: This method reads the sourceDishVccConfig value of the dish. :return: source_dish_vcc_config value :rtype:str

read_dishVccConfig() → str: This method reads the dishVccConfig value of the dish. :return: dish_vcc_config value :rtype:str

is_On_allowed() → bool: This method checks if the On command is allowed in current state. :return: True if the command is allowed :rtype: bool :raises CommandNotAllowed: command is not allowed

On(argin: list) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes On command on CSP Master :return: ResultCode and message :rtype: Tuple

is_Off_allowed() → bool: This method checks if the Off command is allowed in current state. :return: True if the command is allowed :rtype: bool :raises CommandNotAllowed: command is not allowed

Off(argin: list) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes Off command on CSP Master :return: ResultCode and message :rtype: Tuple

is_Standby_allowed() → bool: This method checks if the Standby command is allowed in current state. :return: True if the command is allowed :rtype: bool :raises CommandNotAllowed: Standby command not allowed

Standby(argin: list) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes Standby command on CSP Master :return: ResultCode and message :rtype: Tuple

ResetSysParams() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This Command Reset Sys Param and source sys param :return: ResultCode and message

is_LoadDishCfg_allowed() → bool: This method checks if the LoadDishCfg command is allowed in current state. :return: True if the command is allowed :rtype: bool

push_dish_vcc_config_and_source_dish_vcc_config(): Push sys param and source sys param event

LoadDishCfg(argin: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]

This command updates attribute sourceDishVccConfig and dishVccConfig :return: ResultCode and message :rtype: Tuple

Parameters:: argin – json with File URI.
Dtype:: str

Example argin: { “interface”: “https://schema.skao.int/ska-mid-cbf-initial-parameters/2.2”,

“tm_data_sources”: [tm_data_sources] #The data source that is used to store the data and is accessible through the Telescope Model

“tm_data_filepath”: “path/to/file.json” #The path to the json file containing the Mid.CBF initialization parameters within the data source

}

2. Helper_CSP_Subarray

This module implements the Helper devices for subarray nodes for testing an integrated TMC

class ska_tmc_common.test_helpers.helper_csp_subarray.HelperCspSubarray(*args: Any, **kwargs: Any)

Bases: HelperSubArrayDevice

A helper CspSubarray device to mock CSP Subarray behaviour of returing the command result.

AssignResources(argin: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes AssignResources command on subarray devices :return: ResultCode and message

wait_and_update_exception(command_name): Waits for 5 secs before pushing a longRunningCommandResult event.

wait_and_update_command_result(command_name): Waits for 5 secs before pushing a longRunningCommandResult event.

ReleaseResources() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes ReleaseResources command on subarray device :return: ResultCode and message

ReleaseAllResources() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes ReleaseAllResources command on subarray device :return: ResultCode, message :rtype: tuple

Configure(argin: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes Configure command on subarray devices :return: ResultCode, message :rtype: tuple

Scan(argin: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes Scan command on subarray devices. :return: ResultCode, message :rtype: tuple

EndScan() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes EndScan command on subarray devices. :return: ResultCode, message :rtype: tuple

GoToIdle() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes GoToIdle command on subarray devices. :return: ResultCode, message :rtype: tuple

ObsReset() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]

Abort() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes Abort command on subarray devices. :return: ResultCode, message :rtype: tuple

Restart() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes Restart command on subarray devices :return: ResultCode, message :rtype: tuple

3. Helper_Dish_Device

This module implements the Helper Dish Device for testing an integrated TMC

class ska_tmc_common.test_helpers.helper_dish_device.HelperDishDevice(*args: Any, **kwargs: Any)

Bases: HelperDishLNDevice

A device exposing commands and attributes of the Dish device.

init_device()

class InitCommand(*args: Any, **kwargs: Any)

Bases: InitCommand

A class for the HelperDishDevice’s init_device() command.

do() → Tuple[ska_tango_base.commands.ResultCode, str]: Stateless hook for device initialisation. :return: ResultCode and message

read_scanID() → str: This method reads the scanID attribute of a dish. :rtype: str

write_scanID(value: str) → None: This method writes scanID attribute of dish. :param value: scan_id as given is the json :value dtype: str :rtype: None

kValue() → int: This attribute is used for storing dish kvalue into tango DB.Made this attribute memorized so that when device restart then previous set kvalue will be used validation. :return: kValue

is_SetKValue_allowed() → bool: This method checks if the SetKValue Command is allowed in current State. :rtype: bool :return: boolean value if kValue is set or not

SetKValue(kvalue: int) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]

This command invokes SetKValue command on Dish Master.

Parameters:: kvalue – k value between range 1-2222.
Returns:: ResultCode and meeage
Kvalue dtype:: int
Return type:: Tuple[List[ResultCode], List[str]]

read_pointingState() → PointingState: This method reads the pointingState of dishes. :return: pointingState of dishes :rtype: PointingState

read_configuredBand() → Band: This method reads the configuredBand of dish. :return: configure band for dishes :rtype: Band

read_offset() → str: This method reads the offset of dishes. :return: offset for dishes :rtype: str

read_programTrackTable() → list: This method reads the programTrackTable attribute of a dish. :return: programTrackTable for dishes :rtype: list

write_programTrackTable(value: list) → None: This method writes the programTrackTable attribute of dish. :param value: 50 entries of timestamp, azimuth and elevation values of the desired pointing of dishes. Example: programTrackTable = [ (timestamp1, azimuth1, elevation1), (timestamp2, azimuth2, elevation2), (timestamp3, azimuth3, elevation3),] :value dtype: list :rtype: None

read_achievedPointing() → numpy.ndarray: This method reads the achievedPointing of dishes. :return: achievedPointing of dishes :rtype: str

read_dishMode() → DishMode: This method reads the DishMode of dishes. :return: DishMode of dishes :rtype: DishMode

SetDirectDishMode(argin: DishMode) → None: Trigger a DishMode change

SetDirectPointingState(argin: PointingState) → None: Trigger a PointingState change

SetDirectConfiguredBand(argin: Band) → None: Trigger a ConfiguredBand change

AddTransition(state_duration_info: str) → None: This command will set duration for pointing state such that when respective command for pointing state is triggered then it change pointing state after provided duration

ResetTransitions() → None: This command will reset PointingState duration which is set

set_dish_mode(dishMode: DishMode) → None: This method set the Dish Mode

set_pointing_state(pointingState: PointingState) → None: This method set the Pointing State

set_configured_band(configured_band: Band) → None: This method set the Configured Band

update_dish_mode(value: DishMode, command_name: str = '') → None

Sets the dish mode back to original state.

Parameters:

value – Dish Mode to update.
command_name – Command name

Value dtype:

DishMode

Command_name dtype:

str

Return type:

None

update_pointing_state(value: PointingState, command_name: str) → None

Sets the dish mode back to original state.

Parameters:

value – Pointing state to update.
command_name – Command name

Value dtype:

PointingState

Command_name dtype:

str

Return type:

None

update_command_info(command_name: str = '', command_input: Optional[Union[str, bool]] = None) → None

This method updates the commandCallInfo attribute, with the respective command information.

Args:: command_name (str): command name command_input(str or bool): Input argin for command

push_command_status(status, command_name: str, command_id=None) → None

Push long running command result event for given command.

Params:

result: The result code to be pushed as an event dtype: ResultCode

command_name: The command name for which the event is being pushed dtype: str

exception: Exception message to be pushed as an event dtype: str

set_achieved_pointing() → None: Sets the achieved pointing for dish.

is_SetStandbyFPMode_allowed() → bool: This method checks if the is_SetStandbyFPMode_allowed Command is allowed in current State. :return: True if the command is allowed :rtype:bool :raises CommandNotAllowed: command is not allowed

SetStandbyFPMode() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes SetStandbyFPMode command on Dish Master :return: ResultCode and message :rtype: tuple

SetStandbyLPMode() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes SetStandbyLPMode command on Dish Master :return: ResultCode and message :rtype: tuple

SetOperateMode() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes SetOperateMode command on Dish Master :return: ResultCode and message :rtype: tuple

SetStowMode() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes SetStowMode command on Dish Master :return: ResultCode and message :rtype : tuple

is_Track_allowed() → bool: This method checks if the Track Command is allowed in current State. :return: True if the command is allowed :rtype: bool :raises CommandNotAllowed: command is not allowed

Track() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes Track command on Dish Master :return: ResultCode and message :rtype: tuple

is_TrackStop_allowed() → bool: This method checks if the TrackStop Command is allowed in current State. :return: True if the command is allowed :rtype: bool :raises CommandNotAllowed: command is not allowed

TrackStop() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes TrackStop command on Dish Master :return: ResultCode and message

AbortCommands() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes AbortCommands command on Dish Master :return: ResultCode and message :rtype: tuple

is_TrackLoadStaticOff_allowed() → bool: This method checks if the TrackLoadStaticOff command is allowed in current State. :return: True if the command is allowed :rtype: bool :raises CommandNotAllowed: command is not allowed

TrackLoadStaticOff(argin: List[float]) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]

This method invokes TrackLoadStaticOff command on Dish Master.

Parameters:: argin – A list containing scan_id/ time, cross elevation and elevation offsets.
Argin dtype:: List(float)
Returns:: ResultCode and message
Return type:: Tuple[List[ResultCode], List[str]]

is_ConfigureBand1_allowed() → bool: This method checks if the ConfigureBand1 command is allowed in current State. :return: True if the command is allowed :rtype: bool :raises CommandNotAllowed: command is not allowed

ConfigureBand1(argin: bool) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes ConfigureBand1 command on Dish Master :param argin: The argin is a boolean value, if it is set true it invoke ConfigureBand1 command. :argin dtype: bool :return: ResultCode and message :rtype: tuple

is_ConfigureBand2_allowed() → bool: This method checks if the ConfigureBand2 Command is allowed in current State. :return: True if the command is allowed :rtype: bool :raises CommandNotAllowed: command is not allowed

ConfigureBand2(argin: bool) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes ConfigureBand2 command on Dish Master :param argin: The argin is a boolean value, if it is set true it invoke ConfigureBand2 command. :argin dtype: bool :return: ResultCode and message :rtype: tuple

is_ConfigureBand3_allowed() → bool: This method checks if the ConfigureBand3 Command is allowed in current State. :return: True if the command is allowed :rtype:bool :raises CommandNotAllowed: command is not allowed

ConfigureBand3(argin: bool) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes ConfigureBand3 command on Dish Master :return: ResultCode and message

is_ConfigureBand4_allowed() → bool: This method checks if the ConfigureBand4 Command is allowed in current State. :rtype: bool :return: True if the command is allowed :raises CommandNotAllowed: command is not allowed

ConfigureBand4(argin: bool) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes ConfigureBand4 command on Dish Master :return: ResultCode and message

is_ConfigureBand5a_allowed() → bool: This method checks if the ConfigureBand5a Command is allowed in current State. :rtype:bool :return: True if the command is allowed :raises CommandNotAllowed: command is not allowed

ConfigureBand5a(argin: bool) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes ConfigureBand5a command on Dish Master :return: ResultCode and message

is_ConfigureBand5b_allowed() → bool: This method checks if the ConfigureBand5b Command is allowed in current State. :return: True if the command is allowed :rtype:bool :raises CommandNotAllowed: command is not allowed

ConfigureBand5b(argin: bool) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes ConfigureBand5b command on Dish Master :return: ResultCode and message

is_Scan_allowed() → Union[bool, CommandNotAllowed]: This method checks if the Scan Command is allowed in current State. :return: True if the command is allowed :rtype: Union[bool,CommandNotAllowed] :raises CommandNotAllowed: command is not allowed

Scan(argin: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method sets scanID attribute of Dish Master. :return: Tuple[List[ResultCode], List[str]]

is_EndScan_allowed() → Union[bool, CommandNotAllowed]: This method checks if the EndScan Command is allowed in current State. :rtype:bool :raises CommandNotAllowed: command is not allowed :rtype: Union[bool,CommandNotAllowed]

EndScan() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method clears the scanID attribute of Dish Master :return: ResultCode and message :rtype: Tuple[List[ResultCode], List[str]]

4. Helper_SDP_Subarray

Helper device for SdpSubarray device

class ska_tmc_common.test_helpers.helper_sdp_subarray.HelperSdpSubarray(*args: Any, **kwargs: Any)

Bases: HelperSubArrayDevice

A helper SdpSubarray device for triggering state changes with a command. It can be used to mock SdpSubarray’s bahavior to test error propagation from SdpSubarray to SdpSubarrayLeafNode in case of command failure

init_device()

class InitCommand(*args: Any, **kwargs: Any)

Bases: InitCommand

A class for the HelperSubarrayDevice’s init_device() “command”.

do() → Tuple[ska_tango_base.commands.ResultCode, str]: Stateless hook for device initialisation. :return: ResultCode and message

read_delay() → int: This method is used to read the attribute value for delay. :return: delay

SetDirectreceiveAddresses(argin: str) → None: Set the receivedAddresses

read_receiveAddresses(): Returns receive addresses. :return: receiveAddresses

read_defective() → str: Returns defective status of devices :return: defective parameters :rtype: str

push_obs_state_event(obs_state: ska_tango_base.control_model.ObsState): Place holder method. This method will be implemented in the child classes.

update_device_obsstate(obs_state: ska_tango_base.control_model.ObsState): Updates the device obsState

is_On_allowed() → bool

Check if command On is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: bool
Raises:: CommandNotAllowed – command is not allowed

On()

is_Off_allowed() → bool

Check if command Off is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: bool
Raises:: CommandNotAllowed – command is not allowed

Off()

is_AssignResources_allowed()

Check if command AssignResources is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: boolean

AssignResources(argin): This method invokes AssignResources command on SdpSubarray device. :return: None :raises throw_exception: when input json is wrong

is_ReleaseResources_allowed()

Check if command ReleaseResources is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: boolean
Raises:: CommandNotAllowed – command is not allowed

ReleaseResources(): This method invokes ReleaseResources command on SdpSubarray device.

is_ReleaseAllResources_allowed()

Check if command ReleaseAllResources is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: boolean
Raises:: CommandNotAllowed – command is not allowed

ReleaseAllResources(): This method invokes ReleaseAllResources command on SdpSubarray device.

is_Configure_allowed()

Check if command Configure is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: boolean
Raises:: CommandNotAllowed – command is not allowed

Configure(argin): This method invokes Configure command on SdpSubarray device. :raises throw_exception: when input json is wrong

is_Scan_allowed()

Check if command Scan is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: boolean
Raises:: CommandNotAllowed – command is not allowed

Scan(argin): This method invokes Scan command on SdpSubarray device. :raises throw_exception: when input json is wrong

is_EndScan_allowed()

Check if command EndScan is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: boolean
Raises:: CommandNotAllowed – command is not allowed

EndScan(): This method invokes EndScan command on SdpSubarray device.

is_End_allowed()

Check if command End is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: boolean
Raises:: CommandNotAllowed – command is not allowed

End(): This method invokes End command on SdpSubarray device.

is_Abort_allowed()

Check if command Abort is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: boolean
Raises:: CommandNotAllowed – command is not allowed

Abort(): This method invokes Abort command on SdpSubarray device.

is_Restart_allowed()

Check if command Restart is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: boolean
Raises:: CommandNotAllowed – command is not allowed

Restart(): This method invokes Restart command on SdpSubarray device.

5. Helper_SDP_Subarray_Leaf_Device

This module implements the Helper devices for subarray leaf nodes for testing an integrated TMC

class ska_tmc_common.test_helpers.helper_sdp_subarray_leaf_device.HelperSdpSubarrayLeafDevice(*args: Any, **kwargs: Any)

Bases: HelperSubarrayLeafDevice

A device exposing commands and attributes of the CSP Subarray Leaf Nodes devices.

init_device() → None

class InitCommand(*args: Any, **kwargs: Any)

Bases: InitCommand

A class for the HelperSubarrayDevice’s init_device() “command”.

do() → Tuple[ska_tango_base.commands.ResultCode, str]: Stateless hook for device initialisation. :return: ResultCode and message

read_sdpSubarrayObsState(): Reads the current observation state of the SDP subarray :return: obsstate value

SetSdpSubarrayLeafNodeObsState(argin: ska_tango_base.control_model.ObsState) → None: Trigger a ObsState change

push_obs_state_event(obs_state: ska_tango_base.control_model.ObsState) → None: Push Obs State Change Event

6. Helper_MCCS_Controller

This module implements the Helper MCCS controller devices for testing an integrated TMC

class ska_tmc_common.test_helpers.helper_mccs_controller_device.HelperMCCSController(*args: Any, **kwargs: Any)

Bases: HelperBaseDevice

A helper MCCS controller device for triggering state changes with a command

init_device() → None

class InitCommand(*args: Any, **kwargs: Any)

Bases: InitCommand

A class for the HelperMCCSController’s init_device() “command”.

do() → Tuple[ska_tango_base.commands.ResultCode, str]: Stateless hook for device initialisation. :return: ResultCode, message :rtype:tuple

induce_fault(command_name: str, command_id: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]

Induces a fault into the device based on the given parameters.

Parameters:

command_name (str) – The name of the command for which a fault is being induced.
command_id (str) – command id

Returns:

ResultCode and command id

Return type:

Tuple[List[ResultCode], List[str]]

Example: defective = json.dumps( { “enabled”: False, “fault_type”: FaultType.FAILED_RESULT, “error_message”: “Default exception.”, “result”: ResultCode.FAILED, } ) defective_params = json.loads(defective)

Detailed Explanation: This method simulates inducing various types of faults into a device to test its robustness and error-handling capabilities.

FAILED_RESULT: A fault type that triggers a failed result code for the command. The device will return a result code of ‘FAILED’ along with a custom error message, indicating that the command execution has failed.
LONG_RUNNING_EXCEPTION: A fault type that simulates a command getting stuck in an intermediate state for an extended period. This could simulate a situation where a command execution hangs due to some internal processing issue.
STUCK_IN_INTERMEDIATE_STATE: This fault type represents a scenario where the device gets stuck in an intermediate state between two well-defined states. This can help test the device’s state recovery and error handling mechanisms.
COMMAND_NOT_ALLOWED: This fault type represents a situation where the given command is not allowed to be executed due to some authorization or permission issues. The device should respond with an appropriate error code and message.

New method is explicitly introduced since MCCS Master Leaf Node does not have Obs State.

Raises:: None

SetDefective(values: str) → None: Trigger defective change :param: values :type: str

SetRaiseException(value: bool) → None: Set Raise Exception

wait_and_update_exception(command_id): Waits for 5 secs before pushing a longRunningCommandResult event.

update_lrcr(command_name: str = '', command_id: str = '') → None: Updates the longrunningcommandresult after a delay.

is_Allocate_allowed() → bool

Check if command Allocate is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: bool
Raises:: CommandNotAllowed – command is not allowed

Allocate(argin: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]

This method invokes Allocate command on MCCS controller device

Returns:: a tuple containing ResultCode and Message
Return type:: Tuple

is_Release_allowed() → bool

Check if command Release is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: bool
Raises:: CommandNotAllowed – command is not allowed

Release(argin: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]

This method invokes Release command on MCCS controller device

Returns:: a tuple containing Resultcode and Message
Return type:: Tuple

is_RestartSubarray_allowed() → bool: This method checks if the RestartSubarray command is allowed in the current device state. :return: True if the command is allowed :rtype:bool :raises CommandNotAllowed: command is not allowed

RestartSubarray(argin: int) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This is the method to invoke RestartSubarray command. :param argin: an integer subarray_id. :return: ResultCode, command id :rtype: tuple

push_command_result(result: ska_tango_base.commands.ResultCode, command_id: str, exception: str = '') → None

Push long running command result event for given command.

Params:

result: The result code to be pushed as an event dtype: ResultCode

command_id: The command_id for which the event is being pushed dtype: str

exception: Exception message to be pushed as an event dtype: str

7. Helper_Subarray_Device

This module implements the Helper devices for subarray nodes for testing an integrated TMC

class ska_tmc_common.test_helpers.helper_subarray_device.HelperSubArrayDevice(*args: Any, **kwargs: Any)

Bases: SKASubarray

A generic subarray device for triggering state changes with a command. It can be used as helper device for element subarray node

init_device()

class InitCommand(*args: Any, **kwargs: Any)

Bases: InitCommand

A class for the HelperSubarrayDevice’s init_device() “command”.

do() → Tuple[ska_tango_base.commands.ResultCode, str]: Stateless hook for device initialization. :return: ResultCode

assignedResources() → list

read_scanId() → int: This method is used to read the attribute value for scanId. :return: scan_id

read_obsStateTransitionDuration(): Read transition :return: state dureation info

read_isSubsystemAvailable() → bool: Returns avalability status for the leaf nodes devices :return: avalability status for the leaf nodes devices :rtype: bool

SetSubsystemAvailable(value: bool) → None: Sets Availability of the device :rtype: bool

AddTransition(state_duration_info: str) → None: This command will set duration for obs state such that when respective command for obs state is triggered then it change obs state after provided duration

ResetTransitions() → None: This command will reset ObsState duration which is set

read_commandCallInfo(): This method is used to read the attribute value for commandCallInfo. :return: attribute value for commandCallInfo

read_commandDelayInfo(): This method is used to read the attribute value for delay. :return: attribute value for delay

read_raiseException() → bool

This method is used to read the attribute value for raise exception :return: attribute value for raise exception

Return type:: bool

read_commandInProgress() → str: This method is used to read, which command is in progress :return: command in progress :rtype:str

read_defective() → str: This method is used to read the value of the attribute defective :return: attribute value defective :rtype: str

read_receiveAddresses() → str: This method is used to read receiveAddresses attribute :return: attribute receiveAddresses :rtype:str

push_command_result(result: ska_tango_base.commands.ResultCode, command_name: str, exception: str = '') → None: Push long running command result event for given command. :param result: The result code to be pushed as an event :type: ResultCode :param command_name: The command name for which event is being pushed :type: str :param exception: Exception message to be pushed as an event :type: str

update_device_obsstate(value: ska_tango_base.control_model.ObsState, command_name: str = '') → None: Updates the given data after a delay.

update_command_info(command_name: str = '', command_input: str = '') → None

This method updates the commandCallInfo attribute, with the respective command information.

Args:: command_name (str): command name command_input (str): Input argin for command

create_component_manager() → EmptySubArrayComponentManager: This method is used to create an instance of EmptySubarrayComponentManager :return: component manager :rtype: EmptySubArrayComponentManager

SetRaiseException(value: bool) → None: Set Raise Exception

SetDelay(command_delay_info: str) → None: Update delay value

ResetDelay() → None: Reset Delay to it’s default values

ClearCommandCallInfo() → None: Clears commandCallInfo to empty list

SetDirectObsState(argin: ska_tango_base.control_model.ObsState) → None: Trigger a ObsState change

SetDirectState(argin: tango.DevState) → None

Trigger a DevState if self.dev_state() != argin:

self.set_state(argin): change

SetDirectHealthState(argin: ska_tango_base.control_model.HealthState) → None: Trigger a HealthState change

SetDirectCommandInProgress(argin: str) → None: Trigger a CommandInProgress change

SetDirectassignedResources(argin: str) → None: Triggers an assignedResources attribute change. Note: This method should be used only in case of low.

is_On_allowed() → bool

Check if command On is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: bool
Raises:: CommandNotAllowed – command is not allowed

On() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes On command on Subarray Device :return: ResultCode :rtype: Tuple

is_Off_allowed() → bool

Check if command Off is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: bool
Raises:: CommandNotAllowed – command is not allowed

induce_fault(command_name: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]

Induces a fault into the device based on the given parameters.

Parameters:: command_name (str) – The name of the command for which a fault is being induced.
Returns:: ResultCode and message
Return type:: Tuple[List[ResultCode], List[str]]

Example: defective = json.dumps( { “enabled”: False, “fault_type”: FaultType.FAILED_RESULT, “error_message”: “Default exception.”, “result”: ResultCode.FAILED, “target_obsstates”: [ObsState.RESOURCING, ObsState.EMPTY], } ) defective_params = json.loads(defective)

Detailed Explanation: This method simulates inducing various types of faults into a device to test its robustness and error-handling capabilities.

FAILED_RESULT: A fault type that triggers a failed result code for the command. The device will return a result code of ‘FAILED’ along with a custom error message, indicating that the command execution has failed.
LONG_RUNNING_EXCEPTION: A fault type that simulates a command getting stuck in an intermediate state for an extended period. This could simulate a situation where a command execution hangs due to some internal processing issue.
STUCK_IN_INTERMEDIATE_STATE: This fault type represents a scenario where the device gets stuck in an intermediate state between two well-defined states. This can help test the device’s state recovery and error handling mechanisms.
STUCK_IN_OBS_STATE: This fault type represents a scenario where the device gets stuck in a transitional obsstate as there is some failure. It also raise exception of the same.
COMMAND_NOT_ALLOWED: This fault type represents a situation where the given command is not allowed to be executed due to some authorization or permission issues. The device should respond with an appropriate error code and message.

Raises:: None

SetDefective(values: str) → None: Trigger defective change :param: values :type: str

Off() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes Off command on Subarray Device :return: ResultCode and message :rtype: Tuple

is_Standby_allowed() → bool

Check if command Standby is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: bool
Raises:: CommandNotAllowed – command is not allowed

Standby() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes Standby command on subarray devices :return: ResultCode, message :rtype: tuple

is_AssignResources_allowed() → bool

Check if command AssignResources is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: bool
Raises:: CommandNotAllowed – command is not allowed

AssignResources(argin: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes AssignResources command on subarray devices :return: ResultCode and message

wait_and_update_exception(command_name): Waits for 5 secs before pushing a longRunningCommandResult event.

is_ReleaseResources_allowed() → bool

Check if command ReleaseResources is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: bool
Raises:: CommandNotAllowed – command is not allowed

ReleaseResources(argin) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes ReleaseResources command on subarray device :return: ResultCode and message

is_ReleaseAllResources_allowed() → bool

Check if command ReleaseAllResources is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: bool
Raises:: CommandNotAllowed – command is not allowed

ReleaseAllResources() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes ReleaseAllResources command on subarray device :return: ResultCode, message :rtype: tuple

is_Configure_allowed() → bool

Check if command Configure is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: bool
Raises:: CommandNotAllowed – command is not allowed

Configure(argin: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes Configure command on subarray devices :return: ResultCode, message :rtype: tuple

is_Scan_allowed() → bool

Check if command Scan is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: bool
Raises:: CommandNotAllowed – command is not allowed

Scan(argin: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes Scan command on subarray devices. :return: ResultCode, message :rtype: tuple

is_EndScan_allowed() → bool

Check if command EndScan is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: bool
Raises:: CommandNotAllowed – command is not allowed

EndScan() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes EndScan command on subarray devices. :return: ResultCode, message :rtype: tuple

is_End_allowed() → bool

Check if command End is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: bool
Raises:: CommandNotAllowed – command is not allowed

End() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes End command on subarray devices. :return: ResultCode, message :rtype: tuple

is_GoToIdle_allowed() → bool

Check if command GoToIdle is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: bool
Raises:: CommandNotAllowed – command is not allowed

GoToIdle() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes GoToIdle command on subarray devices. :return: ResultCode, message :rtype: tuple

is_ObsReset_allowed() → bool

Check if command ObsReset is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: bool
Raises:: CommandNotAllowed – command is not allowed

ObsReset() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: ObsReset Command :return: ResultCode and message

is_Abort_allowed() → bool

Check if command Abort is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: bool
Raises:: CommandNotAllowed – command is not allowed

Abort() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes Abort command on subarray devices. :return: ResultCode, message :rtype: tuple

is_Restart_allowed() → bool

Check if command Restart is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: bool
Raises:: CommandNotAllowed – command is not allowed

Restart() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes Restart command on subarray devices :return: ResultCode, message :rtype: tuple

8. Helper_Subarray_Leaf_Device

This module implements the Helper devices for subarray leaf nodes for testing an integrated TMC

class ska_tmc_common.test_helpers.helper_subarray_leaf_device.HelperSubarrayLeafDevice(*args: Any, **kwargs: Any)

Bases: HelperBaseDevice

A device exposing commands and attributes of the Subarray Leaf Nodes devices.

init_device() → None

class InitCommand(*args: Any, **kwargs: Any)

Bases: InitCommand

A class for the HelperSubarrayDevice’s init_device() “command”.

do() → Tuple[ska_tango_base.commands.ResultCode, str]: Stateless hook for device initialisation. :return: ResultCode and message

read_obsStateTransitionDuration(): Read transition :return: state duration info

read_commandCallInfo(): This method is used to read the attribute value for commandCallInfo. :return: attribute value for commandCallInfo

read_defective() → str: Returns defective status of devices :return: attribute value defective :rtype: str

read_delay() → int: This method is used to read the attribute value for delay. :return: attribute value for delay

read_obsState() → ska_tango_base.control_model.ObsState: This method is used to read the attribute value for obsState. :return: attribute value for obsstate

update_command_info(command_name: str = '', command_input: str = '') → None

This method updates the commandCallInfo attribute, with the respective command information.

Args:: command_name (str): command name command_input (str): Input argin for command

ClearCommandCallInfo() → None: Clears commandCallInfo to empty list

AddTransition(state_duration_info: str) → None: This command will establish a duration for the observation state, so that when the corresponding command for the observation state is triggered,it will transition to a provided observation state after the specified duration

ResetTransitions() → None: This command will reset ObsState duration which is set

SetDirectObsState(value: ska_tango_base.control_model.ObsState) → None: Trigger defective change :param: values :type: str

SetDelay(value: int) → None: Update delay value

push_obs_state_event(obs_state: ska_tango_base.control_model.ObsState) → None: Push Obs State Change Event

update_device_obsstate(obs_state: ska_tango_base.control_model.ObsState): Updates the device obsState

is_On_allowed() → bool: This method checks if the On command is allowed or not :return: True if the command is allowed :raises CommandNotAllowed: command is not allowed

On() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]

is_Off_allowed() → bool: This method checks if the Off command is allowed or not :return: True if the command is allowed :raises CommandNotAllowed: command is not allowed

Off() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]

is_Standby_allowed() → bool: This method checks if the Standby command is allowed or not :return: True if the command is allowed :raises CommandNotAllowed: command is not allowed

Standby() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]

is_AssignResources_allowed() → bool: This method checks if the AssignResources command is allowed or not :return: True if the command is allowed :raises CommandNotAllowed: command is not allowed

AssignResources(argin: str = '') → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This is the method to invoke AssignResources command. :return: ResultCode, message :rtype: tuple

is_Configure_allowed() → bool: This method checks the Configure is allowed in the current device state. :return: True if the command is allowed :rtype:bool :raises CommandNotAllowed: command is not allowed

Configure(argin: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This is the method to invoke Configure command. :return: ResultCode, message :rtype: tuple

is_Scan_allowed() → bool: This method checks if the Scan command is allowed in the current device state. :return: True if the command is allowed :rtype:bool :raises CommandNotAllowed: command is not allowed

Scan(argin: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This is the method to invoke Scan command. :return: ResultCode, message :rtype: tuple

is_EndScan_allowed() → bool: This method checks if the EndScan command is allowed in the current device state. :return: True if the command is allowed :rtype:bool :raises CommandNotAllowed: command is not allowed

EndScan() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This is the method to invoke EndScan command. :return: ResultCode, message :rtype: tuple

is_End_allowed() → bool: This method checks if the End command is allowed in the current device state. :return: True if the command is allowed :rtype:bool :raises CommandNotAllowed: command is not allowed

End() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This is the method to invoke End command. :return: ResultCode, message :rtype: tuple

is_GoToIdle_allowed() → bool: This method checks if the GoToIdle command is allowed in the current device state. :return: True if the command is allowed :rtype:bool :raises CommandNotAllowed: command is not allowed

GoToIdle() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This is the method to invoke GoToIdle command. :return: ResultCode, message :rtype: tuple

is_Abort_allowed() → bool: This method checks if the Abort command is allowed in the current device state. :return: True if the command is allowed :rtype:bool :raises CommandNotAllowed: command is not allowed

Abort() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This is the method to invoke Abort command. :return: ResultCode, message :rtype: tuple

is_Restart_allowed() → bool: This method checks if the Restart command is allowed in the current device state. :return: True if the command is allowed :rtype:bool :raises CommandNotAllowed: command is not allowed

Restart() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This is the method to invoke Restart command. :return: ResultCode, message :rtype: tuple

is_ReleaseAllResources_allowed() → bool: This method checks if the ReleaseAllResources command is allowed in the current device state. :return: True if the command is allowed :raises CommandNotAllowed: command is not allowed :rtype: bool

ReleaseAllResources() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This is the method to invoke ReleaseAllResources command. :return: ResultCode, message :rtype: tuple

is_ReleaseResources_allowed() → bool: This method checks if the ReleaseResources command is allowed in the current device state. :return: True if the command is allowed :raises CommandNotAllowed: command is not allowed :rtype: bool

ReleaseResources(argin: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This is the method to invoke ReleaseResources command. :return: ResultCode, message :rtype: tuple

9. Helper_Dish_LN_Device

This module implements the Helper Dish Leaf Node Device for testing an integrated TMC.

class ska_tmc_common.test_helpers.helper_dish_ln_device.HelperDishLNDevice(*args: Any, **kwargs: Any)

Bases: HelperBaseDevice

A device exposing commands and attributes of the Dish Leaf Node device.

init_device() → None

class InitCommand(*args: Any, **kwargs: Any)

Bases: InitCommand

A class for the HelperDishLNDevice’s init_device() command.

do() → Tuple[ska_tango_base.commands.ResultCode, str]: Stateless hook for device initialisation. :return: ResultCode and message

read_kValueValidationResult() → str: Get the k-value validation result. :return: kValue validation result :rtype:str

kValue(kvalue: str) → None: Set memorized dish vcc map :param kvalue: dish vcc config json string :type str

read_delay() → int: This method is used to read the attribute value for delay. :return: delay

read_defective() → str: Returns defective status of devices :return: defective status of devices :rtype: str

read_actualPointing() → str: Read method for actual pointing. :return: actual pointing value

read_isSubsystemAvailable() → bool: Returns avalability status for the leaf nodes devices :return: boolean value for attribute isSubsystemAvailable :rtype: bool

SetDirectkValueValidationResult(result_code: str) → None: Set the kValuValidationResult :argin dtype: str :rtype:None

set_offset(cross_elevation: float, elevation: float) → None: Sets the offset for Dish.

read_commandCallInfo(): This method is used to read the attribute value for commandCallInfo. :return: command_call_info value

read_commandDelayInfo() → str: This method is used to read the attribute value for delay. :return: command_delay_infp value

push_dish_kvalue_validation_result(): Push Dish k-value Validation result event If memorized k-value already set then push Result Code as OK else push result code event as UNKNOWN

push_dish_kvalue_val_result_after_initialization(): This method gets invoked only once after initialization and push the k-value validation result.

is_SetKValue_allowed() → bool: This method checks if the SetKValue Command is allowed in current State. :return: boolean value if command is allowed or not :rtype: bool

SetKValue(kvalue: int) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]

This command invokes SetKValue command on Dish Master.

Parameters:: kvalue – k value between range 1-2222.
Kvalue dtype:: int
Returns:: ResultCode and message
Return type:: Tuple[List[ResultCode], List[str]]

SetDelay(command_delay_info: str) → None: Update delay value

ResetDelay() → None: Reset Delay to it’s default values

ClearCommandCallInfo() → None: Clears commandCallInfo to empty list

is_Off_allowed() → bool: This method checks if the Off Command is allowed in current State. :return: True if the command is allowed :rtype:bool :raises CommandNotAllowed: command is not allowed

Off()

is_SetStandbyFPMode_allowed() → bool: This method checks if the is_SetStandbyFPMode_allowed Command is allowed in current State. :return: True if the command is allowed :rtype:bool :raises CommandNotAllowed: command is not allowed

SetStandbyFPMode() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes SetStandbyFPMode command on Dish Master :return: ResultCode and message :rtype: tuple

is_SetStandbyLPMode_allowed() → bool: This method checks if the is_SetStandbyLPMode_allowed Command is allowed in current State. :return: True if the command is allowed :rtype: bool :raises CommandNotAllowed: command is not allowed

SetStandbyLPMode() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes SetStandbyLPMode command on Dish Master :return: ResultCode and message :rtype: tuple

is_SetOperateMode_allowed() → bool: This method checks if the SetOperateMode Command is allowed in current State. :return: True if the command is allowed :rtype:bool :raises CommandNotAllowed: command is not allowed

SetOperateMode() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes SetOperateMode command on Dish Master :return: ResultCode and message :rtype: tuple

is_SetStowMode_allowed() → bool: This method checks if the SetStowMode Command is allowed in current State. :return: True if the command is allowed :rtype: bool :raises CommandNotAllowed: command is not allowed

SetStowMode() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes SetStowMode command on Dish Master :return: ResultCode and message :rtype : tuple

is_Track_allowed() → bool: This method checks if the Track Command is allowed in current State. :return: True if the command is allowed :rtype: bool :raises CommandNotAllowed: command is not allowed

Track() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes Track command on Dish Master :return: ResultCode and message :rtype: tuple

is_TrackStop_allowed() → bool: This method checks if the TrackStop Command is allowed in current State. :return: True if the command is allowed :rtype: bool :raises CommandNotAllowed: command is not allowed

TrackStop() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes TrackStop command on Dish Master :return: ResultCode and message :rtype: tuple

is_AbortCommands_allowed() → bool: This method checks if the AbortCommands command is allowed in current State. :return: True if the command is allowed :rtype: bool :raises CommandNotAllowed: command is not allowed

AbortCommands() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes AbortCommands command on Dish Master :return: ResultCode and message :rtype: tuple

is_Configure_allowed() → bool: This method checks if the Configure Command is allowed in current State. :return: True if the command is allowed :rtype: bool :raises CommandNotAllowed: command is not allowed

Configure(argin: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes Configure command on Dish Master :return: ResultCode and message :rtype: tuple

is_TrackLoadStaticOff_allowed() → bool: This method checks if the TrackLoadStaticOff command is allowed in current State. :return: True if the command is allowed :rtype: bool :raises CommandNotAllowed: command is not allowed

TrackLoadStaticOff(argin: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]

This method invokes TrackLoadStaticOff command on Dish Master.

Parameters:: argin – A list containing scan_id/ time, cross elevation and elevation offsets.
Argin dtype:: str(List)
Returns:: ResultCode and message
Return type:: Tuple[List[ResultCode], List[str]]

update_command_info(command_name: str = '', command_input: str = '') → None

This method updates the commandCallInfo attribute, with the respective command information.

Args:: command_name (str): command name command_input (str): Input argin for command

is_Scan_allowed() → Union[bool, CommandNotAllowed]: This method checks if the Scan Command is allowed in current State. :return: True if the command is allowed :raises CommandNotAllowed: command is not allowed :rtype: Union[bool,CommandNotAllowed]

Scan(argin: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes Scan command on Dish Master :param argin: scan_id as and string. :return: ResultCode and message :rtype: Tuple[List[ResultCode], List[str]]

is_EndScan_allowed() → Union[bool, CommandNotAllowed]: This method checks if the EndScan Command is allowed in current State. :rtype:bool :raises CommandNotAllowed: command is not allowed :rtype: Union[bool,CommandNotAllowed]

EndScan() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes EndScan command on Dish Master :return: ResultCode and message :rtype: Tuple[List[ResultCode], List[str]]

10. Helper_Mccs_Subarray_Device

This module implements the Helper Mccs subarray device

class ska_tmc_common.test_helpers.helper_mccs_subarray_device.HelperMccsSubarrayDevice(*args: Any, **kwargs: Any)

Bases: HelperSubArrayDevice

A device exposing commands and attributes of the Mccs Subarray Device.

push_command_result(result: ska_tango_base.commands.ResultCode, command_name: str, exception: str = '') → None: Push long running command result event for given command. :param result: The result code to be pushed as an event :type: ResultCode :param command_name: The command name for which event is being pushed :type: str :param exception: Exception message to be pushed as an event :type: str

11. Helper_MCCS_Master_Leaf_Node

This module implements the Helper MCCS master leaf node devices for testing an integrated TMC

class ska_tmc_common.test_helpers.helper_mccs_master_leaf_node_device.HelperMCCSMasterLeafNode(*args: Any, **kwargs: Any)

Bases: HelperBaseDevice

A helper MCCS master leafnode device for triggering state changes with a command

init_device() → None

class InitCommand(*args: Any, **kwargs: Any)

Bases: InitCommand

A class for the HelperMccsStateDevice’s init_device() “command”.

do() → Tuple[ska_tango_base.commands.ResultCode, str]: Stateless hook for device initialisation. :return: ResultCode, message :rtype:tuple

push_command_result(result: ska_tango_base.commands.ResultCode, command_id: str, exception: str = '') → None

Push long running command result event for given command.

Params:

result: The result code to be pushed as an event dtype: ResultCode

command_id: The command_id for which the event is being pushed dtype: str

exception: Exception message to be pushed as an event dtype: str

is_AssignResources_allowed() → bool

Check if command AssignResources is allowed in the current device state.

Returns:: True if the command is allowed
Return type:: bool
Raises:: CommandNotAllowed – command is not allowed

AssignResources(argin: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]

This method invokes AssignResources command on MCCS master leaf node device.

Returns:: a tuple containing ResultCode and Message
Return type:: Tuple

is_ReleaseAllResources_allowed() → bool

Check if command ReleaseAllResources is allowed in the current device state.

Returns:: True if the command is allowed
Raises:: CommandNotAllowed – command is not allowed
Return type:: bool

ReleaseAllResources(argin: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]

This method invokes ReleaseAllResources command on MCCS master leaf node device.

Returns:: a tuple containing ResultCode and Message
Return type:: Tuple

12. Helper_Base_Device

A common module for different helper devices(mock devices)”

class ska_tmc_common.test_helpers.helper_base_device.HelperBaseDevice(*args: Any, **kwargs: Any)

Bases: SKABaseDevice

A common base device for helper devices.

init_device() → None

class InitCommand(*args: Any, **kwargs: Any)

Bases: InitCommand

A class for the HelperBaseDevice’s init_device() command.

do() → Tuple[ska_tango_base.commands.ResultCode, str]

create_component_manager() → EmptyComponentManager: Creates an instance of EmptyComponentManager :return: component manager instance :rtype: EmptyComponentManager

read_raiseException() → bool: This method is used to read the attribute value for raise exception :return: attribute value for raise exception :rtype: bool

read_defective() → str: Returns defective status of devices :return: attribute value for defective params :rtype: str

read_isSubsystemAvailable() → bool: Returns availability status for the leaf nodes devices :return: availabitlity status :rtype: bool

always_executed_hook() → None

delete_device() → None

SetDefective(values: str) → None: Trigger defective change :param: values :type: str

induce_fault(command_name: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]

Induces a fault into the device based on the given parameters.

Parameters:: command_name (str) – The name of the command for which a fault is being induced.
Returns:: ResultCode and message
Return type:: Tuple[List[ResultCode], List[str]]

Example: defective = json.dumps( { “enabled”: False, “fault_type”: FaultType.FAILED_RESULT, “error_message”: “Default exception.”, “result”: ResultCode.FAILED, } ) defective_params = json.loads(defective)

Detailed Explanation: This method simulates inducing various types of faults into a device to test its robustness and error-handling capabilities.

FAILED_RESULT: A fault type that triggers a failed result code for the command. The device will return a result code of ‘FAILED’ along with a custom error message, indicating that the command execution has failed.
LONG_RUNNING_EXCEPTION: A fault type that simulates a command getting stuck in an intermediate state for an extended period. This could simulate a situation where a command execution hangs due to some internal processing issue.
STUCK_IN_INTERMEDIATE_STATE: This fault type represents a scenario where the device gets stuck in an intermediate state between two well-defined states. This can help test the device’s state recovery and error handling mechanisms.
COMMAND_NOT_ALLOWED: This fault type represents a situation where the given command is not allowed to be executed due to some authorization or permission issues. The device should respond with an appropriate error code and message.

Raises:: None

push_command_result(result_code: ska_tango_base.commands.ResultCode, command_name: str, **kwargs) → None: Push long running command result event for given command. :param result_code: The result code to be pushed as an event :type: ResultCode :param command_name: The command name for which event is being pushed :type: str :param kwargs: Additional key word arguments :type kwargs: dict

SetDirectState(argin: tango.DevState) → None

Trigger a DevState change

:param tango.DevState

SetSubsystemAvailable(value: bool) → None: Sets Availability of the device :rtype: bool

SetDirectHealthState(argin: ska_tango_base.control_model.HealthState) → None: Trigger a HealthState change :param tango.DevState

SetRaiseException(value: bool) → None: Set Raise Exception

is_On_allowed() → bool: This method checks if the On command is allowed in current state. :rtype: bool :return: True if the command is allowed :raises CommandNotAllowed: command is not allowed

On() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes On command :return: ResultCode and message :rtype: Tuple

is_Off_allowed() → bool: This method checks if the Off command is allowed in current state. :return: True if the command is allowed :rtype: bool :raises CommandNotAllowed: command is not allowed

Off() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes Off command :return: ResultCode and message :rtype: Tuple

is_Standby_allowed() → bool: This method checks if the Standby command is allowed in current state. :return: True if the command is allowed :rtype: bool :raises CommandNotAllowed: command is not allowed

Standby() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes Standby command :return: ResultCode and message :rtype: Tuple

is_disable_allowed() → bool: This method checks if the Disable command is allowed in current state. :return: True if the command is allowed :rtype: bool :raises CommandNotAllowed: command is not allowed

Disable() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This method invokes Disable command :return: ResultCode and message :rtype: Tuple

13. Helper_Csp_Master_Leaf_Device

This module defines a helper device that acts as csp master in our testing.

class ska_tmc_common.test_helpers.helper_csp_master_leaf_node.HelperCspMasterLeafDevice(*args: Any, **kwargs: Any)

Bases: HelperBaseDevice

A helper device class for Csp Master Leaf device

init_device() → None

memorizedDishVccMap(value: str): Set memorized dish vcc map :param value: dish vcc config json string :type str

class InitCommand(*args: Any, **kwargs: Any)

Bases: InitCommand

A class for the HelperCspMasterDevice’s init_device() command.

do() → Tuple[ska_tango_base.commands.ResultCode, str]: Stateless hook for device initialisation. :return: ResultCode and message

read_sourceDishVccConfig() → str: This method reads the sourceDishVccConfig value of the dish. :return: sourceDishVccConfig value of the dish. :rtype:str

read_dishVccConfig() → str: This method reads the dishVccConfig value of the dish. :return: dishVccConfig value of the dish :rtype:str

read_DishVccMapValidationResult() → str

Returns:: DishVccMapValidationResult value of the dish
Return type:: str

ResetSysParams() → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: This Command Reset dishVccConfig and sourceDishVccConfig attribute :return: ResultCode and message

is_LoadDishCfg_allowed() → bool: This method checks if the LoadDishCfg command is allowed in current state. :return: True if the command is allowed :rtype: bool

LoadDishCfg(argin: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]

This command updates attribute sourceDishVccConfig and dishVccConfig :return: ResultCode and message :rtype: Tuple

Parameters:: argin – json with File URI.
Dtype:: str

Example argin: { “interface”: “https://schema.skao.int/ska-mid-cbf-initial-parameters/2.2”,

“tm_data_sources”: [tm_data_sources] #The data source that is used to store the data and is accessible through the Telescope Model

“tm_data_filepath”: “path/to/file.json” #The path to the json file containing the Mid.CBF initialization parameters within the data source

}

SetDishVccValidationResult(result: str) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]: Set DishVccValidationResult and push event for same :return: ResultCode and message

push_dish_vcc_validation_result(): Push Dish Vcc Validation result event If memorized dish vcc already set then push Result Code as OK else push result code event as UNKNOWN

start_dish_vcc_validation(): Push Dish Vcc Validation result after Initialization

14. Helper_Csp_Subarray_Leaf_Device

This module implements the Helper devices for subarray leaf nodes for testing an integrated TMC

class ska_tmc_common.test_helpers.helper_csp_subarray_leaf_device.HelperCspSubarrayLeafDevice(*args: Any, **kwargs: Any)

Bases: HelperSubarrayLeafDevice

A device exposing commands and attributes of the CSP Subarray Leaf Node devices.

init_device() → None

class InitCommand(*args: Any, **kwargs: Any)

Bases: InitCommand

A class for the HelperSubarrayDevice’s init_device() “command”.

do() → Tuple[ska_tango_base.commands.ResultCode, str]: Stateless hook for device initialisation. :return: ResultCode and message

read_cspSubarrayObsState(): Reads the current observation state of the CSP subarray :return: obs state

SetCspSubarrayLeafNodeObsState(argin: int) → None: Trigger a ObsState change

push_obs_state_event(obs_state: ska_tango_base.control_model.ObsState) → None: Push Obs State Change Event

15. Helper_Sdp_Queue_Connector

This module defines a helper device that acts as SDP Queue Connector in testing.

class ska_tmc_common.test_helpers.helper_sdp_queue_connector_device.HelperSdpQueueConnector(*args: Any, **kwargs: Any)

Bases: Device

A helper device that emulates the behavior of pointing_offsets attribute from SdpQueueConnector device for testing. Queue Connector is a tango device of SDP.

init_device()

pointing_offsets() → tuple[str, bytes]: Returns the attribute value for pointing_offsets from QueueConnector SDP device. The returned tuple contains a string literal msgpack_numpy and a numpy ndarray in encoded byte form with below values in each array: [ “Antenna_Name,” “CrossElevation_Offset,CrossElevation_Offset_Std,” “Elevation_Offset,Elevation_Offset_Std,” “Expected_Width_H,Expected_Width_V,” “Fitted_Width_H,Fitted_Width_H_Std,” “Fitted_Width_V,Fitted_Width_V_Std,” “Fitted_Height,Fitted_Height_Std” ] :return: the attribute value for pointing_offsets from QueueConnector SDP device

SetDirectPointingOffsets(pointing_offsets_data) → tango.CmdArgType.DevVoid: Sets the attribute value for pointing_offsets for testing purposes. :param pointing_offsets_data: The variable contains a string literal msgpack_numpy and a numpy ndarray in encoded byte form with below values in each array: [ Antenna_Name, CrossElevation_Offset,CrossElevation_Offset_Std, Elevation_Offset,Elevation_Offset_Std, Expected_Width_H,Expected_Width_VFitted_Width_H, Fitted_Width_H_Std,Fitted_Width_V,Fitted_Width_V_Std, Fitted_Height,Fitted_Height_Std ]

Conclusion

This module contain all helper devices required in TMC for testing and debugging