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