ska_tmc_mccssubarrayleafnode.commands package

Link to the TMC User documentation is here.

Submodules

ska_tmc_mccssubarrayleafnode.commands.mccssubarrayln_command module

This module provides the implementation of the MccsSLNCommand class, which is a common command class for MCCS Subarray Leaf Node. It contains common functionality for the commands present on this device.

class ska_tmc_mccssubarrayleafnode.commands.mccs_subarrayln_command.MccsSLNCommand(*args: Any, **kwargs: Any)

Bases: TmcLeafNodeCommand

Common Command class for MCCS Subarray Leaf Node. This class provides common functionality for the commands present on this device.

call_update_task_status(result: ska_tango_base.commands.ResultCode, message: str) → None
Call update task status and provide result, message

attribute

Parameters:

  • result (ResultCode) – Result code

  • message (str) – command success/error message

Returns:

None

do(argin: Optional[Any] = None)

Abstract Method from TmcLeafNodeCommand is defined here but not utilized by this Class.

Parameters:

argin (_type_, optional) – Accepts argument if required. Defaults to None.

do_low(argin: Optional[Any] = None)

Abstract Method from TmcLeafNodeCommand is defined here but not utilized by this Class.

Parameters:

argin (_type_, optional) – Accepts argument if required. Defaults to None.

get_obsstate_lrcr() → Tuple[ska_control_model.ObsState, Optional[ska_tango_base.commands.ResultCode]]

Retrieve observation state and long-running command result code.

Returns:

A tuple containing the

current observation state and the ResultCode.

Return type:

Tuple[ObsState, Optional[ResultCode]]

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

Initialize the adapters for communication with the MCCS Subarray device and the MCCS Master device.

This method attempts to create adapters for both the MCCS Subarray device and the MCCS Master device using the specified device names obtained from the component manager. It continues trying until both adapters are created or the specified timeout is reached.

Returns:

A tuple containing the result code and a message indicating the status of the adapter initialization.

Return type:

Tuple[ResultCode, str]

Raises:

  • ConnectionFailed – If there is an issue establishing a connection with the devices.

  • DevFailed – If there is a device-related error during adapter creation.

  • AttributeError – If there is an attribute error during adapter creation.

  • ValueError – If there is a value error during adapter creation.

  • TypeError – If there is a type error during adapter creation.

init_adapter_low()

Initialize the adapter for the low-level device.

is_aborted() → bool

Check whether the abort flag is set.

Returns:

True if the operation has been aborted, otherwise False.

Return type:

bool

set_command_id(command_name: str) → None

Sets the command id for error propagation.

Parameters:

command_name (str) – Name of the command

update_task_status(**kwargs)

Abstract Method from TmcLeafNodeCommand is defined here but not utilized by this Class.

validate_abort_completion() → bool

Check whether the abort command completion criteria is met. :return: boolean :rtype: bool

wait_for_completion(state_getter: Optional[Callable[[], bool]] = None, desired_state: bool = True, device_length: int = 1, check_command_results=True, check_abort: bool = True) → Tuple[ska_tango_base.commands.ResultCode, str]

Wait for command completion using CommandCompletionTracker.

This method blocks until the desired condition is met, optionally checking command results and abort status during execution.

Parameters:

  • state_getter (Callable[[], bool], optional) – Function used to fetch the current state. If provided, it is evaluated until it matches desired_state.

  • desired_state (bool) – Expected state to determine completion (default is True).

  • device_length (int) – Number of devices to wait for completion (default is 1).

  • check_command_results (bool) – Whether to validate command results during completion (default is True).

  • check_abort (bool) – Whether to monitor and handle abort condition (default is True).

Returns:

Result code and message indicating the outcome of the command execution.

Return type:

Tuple[ResultCode, str]

ska_tmc_mccssubarrayleafnode.commands.mccs_subarrayln_command.task_callback_default(**kwargs) → None

Default method if the taskcallback is not passed

Parameters:

  • status – status of the task.

  • progress – progress of the task.

  • result – result of the task.

  • exception – an exception raised from the task.

ska_tmcs_mccssubarrayleafnode.commands.configure_command module

Configure command class for LowTmcLeafNodeMccsSubarray.

class ska_tmc_mccssubarrayleafnode.commands.configure_command.Configure(*args: Any, **kwargs: Any)

Bases: MccsSLNCommand

This class implements the Configure command for MCCS Subarray Leaf Node device.

It provides methods to configure the Mccs Subarray device and handle the execution of the Configure command.

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

Method to invoke Configure command on Mccs Subarray.

Parameters:

argin (str) – The string in JSON format.

Example JSON:

{” interface”: “https://schema.skatelescope.org/ska-low-mccs-subarray-configure/3.0”, “subarray_beams”: [{“subarray_beam_id”: 1, “update_rate”: 0.0, “logical_bands”: [{“start_channel”: 80, “number_of_channels”: 16}, {“start_channel”: 384, “number_of_channels”: 16}], “apertures”: [{“aperture_id”: “AP001.01”, “weighting_key_ref”: “aperture2”}, {“aperture_id”: “AP001.02”, “weighting_key_ref”: “aperture3”}, {“aperture_id”: “AP002.01”, “weighting_key_ref”: “aperture2”}, {“aperture_id”: “AP002.02”, “weighting_key_ref”: “aperture3”}, {“aperture_id”: “AP003.01”, “weighting_key_ref”: “aperture1”}], “sky_coordinates”:{“timestamp”: “2021-10-23T12:34:56.789Z”, “reference_frame”: “ICRS”, “c1”: 180.0, “c1_rate”: 0.0, “c2”: 45.0, “c2_rate”: 0.0}}]}

Returns:

Tuple of ResultCode and message

Return type:

Tuple[ResultCode, str]

invoke_configure(argin: str, task_callback: Optional[ska_tango_base.base.TaskCallbackType] = None, task_abort_event=None) → Optional[Tuple[ska_tango_base.commands.ResultCode, str]]

This is a long running method for Configure command, it executes the do hook of the command class.

Parameters:

argin (str) – Input json string for the configure command

Returns:

Tuple of ResultCode and message

Return type:

Tuple[ResultCode, str]

update_task_status(**kwargs) → None

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

ska_tmc_mccssubarrayleafnode.commands.end_command module

End command class for LowTmcLeafNodeMccsSubarray.

class ska_tmc_mccssubarrayleafnode.commands.end_command.End(*args: Any, **kwargs: Any)

Bases: MccsSLNCommand

This class implements the End command for MCCS Subarray Leaf Node device.

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

Method to invoke End command on Mccs Subarray.

Returns:

tuple of Resultcode and message

Return type:

Tuple[ResultCode, message]

end(task_callback: Optional[ska_tango_base.base.TaskCallbackType] = None, task_abort_event=None) → Tuple[ska_tango_base.commands.ResultCode, str]

This is a long running method for End command, it executes the do hook of the command class.

Returns:

Tuple of ResultCode and mesasge

Return type:

Tuple[ResultCode, str]

update_task_status(**kwargs) → None

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

ska_tmc_mccssubarrayleafnode.commands.abort_command module

Abort command class for LowTmcLeafNodeMccsSubarray.

class ska_tmc_mccssubarrayleafnode.commands.abort_command.Abort(*args: Any, **kwargs: Any)

Bases: MccsSLNCommand

This class implements the Abort command for Mccs Subarray. It provides methods to Abort the Mccs Subarray device and handle the execution of the Abort command.

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

Method to invoke Abort command on Mccs Subarray.

Return type:

Tuple[ResultCode, str]

invoke_abort(task_callback: ska_tango_base.base.TaskCallbackType = <function task_callback_default>, task_abort_event: ~typing.Optional[~threading.Event] = None) → Tuple[ska_tango_base.commands.ResultCode, str]

This method calls do for Abort command

Parameters:

  • task_callback (TaskCallbackType) – task_callback_default is default

  • task_abort_event (Optional[threading.Event]) – task abort event

Returns:

tuple of ResultCode and message

Return type:

Tuple[ResultCode, str]

update_task_status(**kwargs: Dict[str, Union[Tuple[ska_tango_base.commands.ResultCode, str], ska_control_model.task_status.TaskStatus, str]]) → None

Update the status of a task.

Parameters:

**kwargs – Keyword arguments for task status update.

ska_tmc_mccssubarrayleafnode.commands.restart_command module

Restart command class for LowTmcLeafNodeMccsSubarray.

This module defines the Restart command class for the MCCS Subarray Leaf Node.

class ska_tmc_mccssubarrayleafnode.commands.restart_command.Restart(*args: Any, **kwargs: Any)

Bases: MccsSLNCommand

This class implements the Restart command for Mccs Subarray Leaf Node. It provides functionality to restart the MCCS Subarray device and handles the execution of the Restart command.

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

Invoke the Restart command on the MCCS Controller device.

Returns:

A tuple containing the result code and a message indicating the status of the Restart command execution.

Return type:

Tuple[ResultCode, str]

restart(task_callback=None, task_abort_event=None) → Tuple[ska_tango_base.commands.ResultCode, str]

This is a long running method for Restart command, it executes do hook, invokes Restart command on the MCCSSubarray.

Returns:

tuple of ResultCode and message

Return type:

Tuple[ResultCode, str]

update_task_status(**kwargs: Dict[str, Union[Tuple[ska_tango_base.commands.ResultCode, str], ska_control_model.TaskStatus, str]]) → None

Update the status of a task.

Parameters:

**kwargs – Keyword arguments for task status update.

ska_tmc_mccssubarrayleafnode.commands.scan_command module

Scan command class for LowTmcLeafNodeMccsSubarray.

class ska_tmc_mccssubarrayleafnode.commands.scan_command.Scan(*args: Any, **kwargs: Any)

Bases: MccsSLNCommand

This class implements the Scan command for MCCS Subarray Leaf Node device.

It provides methods to invoke the scan command on the Mccs Subarray device and handle its execution.

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

Method to invoke Scan command on Mccs Subarray.

Parameters:

argin (str) – The string in JSON format.

Example JSON:

{ “interface”: “https://schema.skatelescope.org/ska-low-mccs-subarray-scan/3.0”, “scan_id”: 12345678, “start_time”: “2023-12-31T12:34:28Z”, “duration”: 0.0 }

Returns:

Tuple of ResultCode and message

Return type:

Tuple[ResultCode, str]

scan(argin: str, task_callback: Optional[ska_tango_base.base.TaskCallbackType] = None, task_abort_event=None) → Tuple[ska_tango_base.commands.ResultCode, str]

This method invoked the do hook of the command class. It also sets up the command tracking.

Parameters:

argin (str) – Input json string for the scan command

Returns:

tuple of resultcode and message

Return type:

Tuple[ResultCode, str]

update_task_status(**kwargs) → None

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

ska_tmc_mccssubarrayleafnode.commands.endscan_command module

EndScan command class for LowTmcLeafNodeMccsSubarray.

class ska_tmc_mccssubarrayleafnode.commands.endscan_command.EndScan(*args: Any, **kwargs: Any)

Bases: MccsSLNCommand

This class implements the EndScan command for MCCS Subarray Leaf Node device.

It provides methods to invoke endscan command on the Mccs Subarray device and handle its execution.

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

Method to invoke EndScan command on Mccs Subarray.

Returns:

tuple of result code and message.

Return type:

Tuple[ResultCode, str]

endscan(task_callback: Optional[ska_tango_base.base.TaskCallbackType] = None, task_abort_event=None) → Tuple[ska_tango_base.commands.ResultCode, str]

This method invokes the do hook for the EndScan command class. It also sets up the command tracking.

update_task_status(**kwargs) → None

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

Module contents

This is init module for MccsSubarrayleafnode commands.

class ska_tmc_mccssubarrayleafnode.commands.Abort(*args: Any, **kwargs: Any)

Bases: MccsSLNCommand

This class implements the Abort command for Mccs Subarray. It provides methods to Abort the Mccs Subarray device and handle the execution of the Abort command.

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

Method to invoke Abort command on Mccs Subarray.

Return type:

Tuple[ResultCode, str]

invoke_abort(task_callback: ska_tango_base.base.TaskCallbackType = <function task_callback_default>, task_abort_event: ~typing.Optional[~threading.Event] = None) → Tuple[ska_tango_base.commands.ResultCode, str]

This method calls do for Abort command

Parameters:

  • task_callback (TaskCallbackType) – task_callback_default is default

  • task_abort_event (Optional[threading.Event]) – task abort event

Returns:

tuple of ResultCode and message

Return type:

Tuple[ResultCode, str]

update_task_status(**kwargs: Dict[str, Union[Tuple[ska_tango_base.commands.ResultCode, str], ska_control_model.task_status.TaskStatus, str]]) → None

Update the status of a task.

Parameters:

**kwargs – Keyword arguments for task status update.

class ska_tmc_mccssubarrayleafnode.commands.Configure(*args: Any, **kwargs: Any)

Bases: MccsSLNCommand

This class implements the Configure command for MCCS Subarray Leaf Node device.

It provides methods to configure the Mccs Subarray device and handle the execution of the Configure command.

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

Method to invoke Configure command on Mccs Subarray.

Parameters:

argin (str) – The string in JSON format.

Example JSON:

{” interface”: “https://schema.skatelescope.org/ska-low-mccs-subarray-configure/3.0”, “subarray_beams”: [{“subarray_beam_id”: 1, “update_rate”: 0.0, “logical_bands”: [{“start_channel”: 80, “number_of_channels”: 16}, {“start_channel”: 384, “number_of_channels”: 16}], “apertures”: [{“aperture_id”: “AP001.01”, “weighting_key_ref”: “aperture2”}, {“aperture_id”: “AP001.02”, “weighting_key_ref”: “aperture3”}, {“aperture_id”: “AP002.01”, “weighting_key_ref”: “aperture2”}, {“aperture_id”: “AP002.02”, “weighting_key_ref”: “aperture3”}, {“aperture_id”: “AP003.01”, “weighting_key_ref”: “aperture1”}], “sky_coordinates”:{“timestamp”: “2021-10-23T12:34:56.789Z”, “reference_frame”: “ICRS”, “c1”: 180.0, “c1_rate”: 0.0, “c2”: 45.0, “c2_rate”: 0.0}}]}

Returns:

Tuple of ResultCode and message

Return type:

Tuple[ResultCode, str]

invoke_configure(argin: str, task_callback: Optional[ska_tango_base.base.TaskCallbackType] = None, task_abort_event=None) → Optional[Tuple[ska_tango_base.commands.ResultCode, str]]

This is a long running method for Configure command, it executes the do hook of the command class.

Parameters:

argin (str) – Input json string for the configure command

Returns:

Tuple of ResultCode and message

Return type:

Tuple[ResultCode, str]

update_task_status(**kwargs) → None

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

class ska_tmc_mccssubarrayleafnode.commands.End(*args: Any, **kwargs: Any)

Bases: MccsSLNCommand

This class implements the End command for MCCS Subarray Leaf Node device.

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

Method to invoke End command on Mccs Subarray.

Returns:

tuple of Resultcode and message

Return type:

Tuple[ResultCode, message]

end(task_callback: Optional[ska_tango_base.base.TaskCallbackType] = None, task_abort_event=None) → Tuple[ska_tango_base.commands.ResultCode, str]

This is a long running method for End command, it executes the do hook of the command class.

Returns:

Tuple of ResultCode and mesasge

Return type:

Tuple[ResultCode, str]

update_task_status(**kwargs) → None

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

class ska_tmc_mccssubarrayleafnode.commands.EndScan(*args: Any, **kwargs: Any)

Bases: MccsSLNCommand

This class implements the EndScan command for MCCS Subarray Leaf Node device.

It provides methods to invoke endscan command on the Mccs Subarray device and handle its execution.

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

Method to invoke EndScan command on Mccs Subarray.

Returns:

tuple of result code and message.

Return type:

Tuple[ResultCode, str]

endscan(task_callback: Optional[ska_tango_base.base.TaskCallbackType] = None, task_abort_event=None) → Tuple[ska_tango_base.commands.ResultCode, str]

This method invokes the do hook for the EndScan command class. It also sets up the command tracking.

update_task_status(**kwargs) → None

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

class ska_tmc_mccssubarrayleafnode.commands.Restart(*args: Any, **kwargs: Any)

Bases: MccsSLNCommand

This class implements the Restart command for Mccs Subarray Leaf Node. It provides functionality to restart the MCCS Subarray device and handles the execution of the Restart command.

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

Invoke the Restart command on the MCCS Controller device.

Returns:

A tuple containing the result code and a message indicating the status of the Restart command execution.

Return type:

Tuple[ResultCode, str]

restart(task_callback=None, task_abort_event=None) → Tuple[ska_tango_base.commands.ResultCode, str]

This is a long running method for Restart command, it executes do hook, invokes Restart command on the MCCSSubarray.

Returns:

tuple of ResultCode and message

Return type:

Tuple[ResultCode, str]

update_task_status(**kwargs: Dict[str, Union[Tuple[ska_tango_base.commands.ResultCode, str], ska_control_model.TaskStatus, str]]) → None

Update the status of a task.

Parameters:

**kwargs – Keyword arguments for task status update.

class ska_tmc_mccssubarrayleafnode.commands.Scan(*args: Any, **kwargs: Any)

Bases: MccsSLNCommand

This class implements the Scan command for MCCS Subarray Leaf Node device.

It provides methods to invoke the scan command on the Mccs Subarray device and handle its execution.

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

Method to invoke Scan command on Mccs Subarray.

Parameters:

argin (str) – The string in JSON format.

Example JSON:

{ “interface”: “https://schema.skatelescope.org/ska-low-mccs-subarray-scan/3.0”, “scan_id”: 12345678, “start_time”: “2023-12-31T12:34:28Z”, “duration”: 0.0 }

Returns:

Tuple of ResultCode and message

Return type:

Tuple[ResultCode, str]

scan(argin: str, task_callback: Optional[ska_tango_base.base.TaskCallbackType] = None, task_abort_event=None) → Tuple[ska_tango_base.commands.ResultCode, str]

This method invoked the do hook of the command class. It also sets up the command tracking.

Parameters:

argin (str) – Input json string for the scan command

Returns:

tuple of resultcode and message

Return type:

Tuple[ResultCode, str]

update_task_status(**kwargs) → None

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