TMC Base Command Class

1. BaseTMCCommand

class ska_tmc_common.tmc_command.BaseTMCCommand(component_manager, logger: <Mock id='131989266746864'>, *args, **kwargs)

Base class for managing all TMC commands.

set_command_id(command_name: str)

Sets the command id for error propagation.

Parameters:: command_name (str) – name of the command.

Method to create adapters for device.

Parameters:

device_name (str) – name of the device.
adapter_type (AdapterType) – Type of Adapter.
start_time (float) – start time.
timeout (int) – Timeout for adapter creation.

Returns:

adapter created

Raises:

ConnectionFailed – Exception is raised when connection fails
DevFailed – Exception is raised when device fails

do(argin: str | None = None) → NotImplementedError

Base method for do method for different nodes

Parameters:: argin (str) – command params.
Raises:: NotImplementedError – Not implemented error

update_task_status(**kwargs) → NotImplementedError

Method to update the task status for command.

Raises:: NotImplementedError – Not implemented error

start_tracker_thread(state_function: str, expected_state: List[IntEnum], abort_event: Event, timeout_id: str | None = None, timeout_callback: TimeoutCallback | None = None, command_id: str | None = None, lrcr_callback: LRCRCallback | None = None) → None

Creates and starts a thread that will keep track of the State/ obsState change to be monitored (For confirming command completion. Currently only supports obsState change.), the timeout callback to monitor the timeout and the longRunningCommandResult callback to keep track of LRCR events.

Parameters:

state_function (str) – The function to determine the state of the device. Should be accessible in the component_manager.
expected_state – Expected state of the device in case of successful command execution.
abort_event – threading.Event class object that is used to check if the command has been aborted.
timeout_id – Id for TimeoutCallback class object.
timeout_callback – An instance of TimeoutCallback class that acts as a callable functions to call in the event of timeout.
command_id – Id for LRCRCallback class object.
lrcr_callback – An instance of LRCRCallback class that acts as a callable functions to call when longRunningCommandResult event is received.

track_and_update_command_status(state_function: str, expected_state: List[IntEnum], abort_event: Event, timeout_id: str | None = None, timeout_callback: TimeoutCallback | None = None, command_id: str | None = None, lrcr_callback: LRCRCallback | None = None) → None

Keeps track of the obsState change and the timeout callback to determine whether timeout has occurred or the command completed successfully. Logs the result for now.

Parameters:

state_function (str) – The function to determine the state of the device. Should be accessible in the component_manager.
expected_state – Expected state of the device in case of successful command execution.
abort_event – threading.Event class object that is used to check if the command has been aborted.
timeout_id – Id for TimeoutCallback class object.
timeout_callback – An instance of TimeoutCallback class that acts as a callable function to call in the event of timeout.
command_id – Id for LRCRCallback class object.
lrcr_callback – An instance of LRCRCallback class that acts as a callable function to call when an event from the attribute longRunningCommandResult arrives.

check_abort_event(abort_event) → bool

Checks for abort event and if abort event detected, sets TaskStatus to ABORTED and stops the tracker thread

Parameters:: abort_event (bool) – threadingEvent class object that is used to check if the command has been aborted.
Returns:: if command is aborted or not
Return type:: bool

check_command_timeout(timeout_id, timeout_callback) → bool

Checks for command timeout. On timeout, it sets ResultCode to FAILED and stops the tracker thread.

Parameters:

timeout_id – Id for TimeoutCallback class object.
timeout_callback – An instance of TimeoutCallback class that acts as a callable function to call in the event of timeout.

Returns:

boolean value if timeout occurred or not

check_device_state(state_function: str, state_to_achieve: Any, expected_state: list, command_id: str) → bool

Waits for expected state with or without transitional state. On expected state occurrence, it sets ResultCode to OK and stops the tracker thread

Parameters:

state_function (str) – The function to determine the state of the device. Should be accessible in the component_manager.
state_to_achieve – A particular state to needs to be achieved for command completion.
expected_state – Expected state of the device in case of successful command execution. It’s a list contains transitional obsState if exists for a command.

Returns:

boolean value if state change occurred or not

check_command_exception(command_id, lrcr_callback) → bool

Checks if command has been failed with an exception. On exception, it sets ResultCode to FAILED and stops the tracker thread.

Parameters:

command_id – Id for LRCRCallback class object.
lrcr_callback – An instance of LRCRCallback class that acts as a callable function to call when an event from the attribute longRunningCommandResult arrives.

Returns:

boolean value if exception has occurred or not

stop_tracker_thread(timeout_id: str | None) → None

External stop method for stopping the timer thread as well as the tracker thread.

Parameters:: timeout_id (str) – Timeout id

2. TMCCommand

class ska_tmc_common.tmc_command.TMCCommand(component_manager, logger: <Mock id='131989266746864'>, *args, **kwargs)

Class to add device adapters

init_adapters()

Base method for init_adapters method for different nodes

Raises:: NotImplementedError – Not implemented error

init_adapters_mid()

Base method for init_adapters_mid method for different nodes

Raises:: NotImplementedError – Not implemented error

init_adapters_low()

Base method for init_adapters_low method for different nodes

Raises:: NotImplementedError – Not implemented error

do_mid(argin: str | None = None)

Base method for do_mid method for different nodes

Parameters:: argin (str) – Command params
Raises:: NotImplementedError – Not implemented error

do_low(argin=None)

Base method for do_low method for different nodes

Parameters:: argin (str) – Command params
Raises:: NotImplementedError – Not implemented error

3. TmcLeafNodeCommand

class ska_tmc_common.tmc_command.TmcLeafNodeCommand(component_manager, logger: <Mock id='131989266746864'>, *args, **kwargs)

Class to add adapters for LeafNode devices

init_adapter()

Base method for init_adapter method for different nodes

Raises:: NotImplementedError – Not implemented error

do_mid(argin: str | None = None)

Base method for do_mid method for different nodes

Parameters:: argin (str) – Command params
Raises:: NotImplementedError – Not implemented error

do_low(argin: str | None = None)

Base method for do_low method for different nodes

Parameters:: argin (str) – Command params
Raises:: NotImplementedError – Not implemented error

init_adapter_mid()

Base method for init_adapter_mid method for different nodes

Raises:: NotImplementedError – Not implemented error

init_adapter_low()

Base method for init_adapter_low method for different nodes

Raises:: NotImplementedError – Not implemented error

invoke_command_lrc_cb(device_name: str, update_event_callback: Callable | None = None) → Callable: LRC callback for command result tracking.

invoke_command_and_track(adapter, command_name: str, command_input=None, update_event_callback: Callable | None = None) → Tuple[ska_tango_base.commands.ResultCode, str]: Invoke command using LRC and track execution.

call_adapter_method(device: str, adapter, command_name: str, argin=None) → Tuple[List[ska_tango_base.commands.ResultCode], List[str]]

Method to invoke commands on device adapters.

Parameters:

device (str) – Device name
adapter (Adapter) – Adapter to use
command_name (str) – Command name
argin (str) – Command params

Returns:

ResultCode and message

call_update_task_status(result, message) → None

Call update task status and provide result, message attributes.

Parameters:

result (ResultCode) – Result code.
message (str) – Result message.

set_abort_flag() → None: Set abort flag to stop command completion tracker.

is_aborted() → bool: Check if command is aborted.

get_obs_state() → ObsState

Return the current observation state.

This method acquires the component manager lock before retrieving the observation state to ensure thread-safe access.

Returns:: The current observation state.
Return type:: ObsState

4. BaseTMCCommand

class ska_tmc_common.v4.tmc_command.BaseTMCCommand(command_runtime_context: ~ska_tmc_common.v4.command_context.CommandRuntimeContext, adapter_provider: ~ska_tmc_common.adapters.AdapterFactory, logger: <Mock id='131989247099040'>)

Base class implementing the Template Method Pattern for TMC commands.

execute()
initialize()
prepare_command()
build_device_commands()
executor.execute()
wait_for_completion()
evaluate_result()
post_process()
cleanup()

Subclasses must implement the abstract methods and can override the hook methods to customize behavior.

command_name: str = ''

execute(argin: Any | None = None, task_callback=None, task_abort_event=None) → Tuple[ska_tango_base.commands.ResultCode, str]

Execute the command following the template method pattern.

Parameters:

argin (Optional[Any]) – Optional input arguments for the command
task_callback – Callback function to update task status
task_abort_event – Event to signal command abortion

pre_process(argin: Any | None = None) → None: Optional hook for pre-processing before command execution.

map_exception_to_resultcode(exception: Exception) → tuple[ska_tango_base.commands.ResultCode, str]: Handle exceptions raised during command execution and map them to appropriate ResultCode and message. :param exception: The exception raised during command execution :return: A tuple containing the ResultCode and message

abstract build_device_commands() → None

Build the list of device commands to be executed.

This method must be implemented by subclasses to populate self.context.device_commands with the appropriate commands.

initialize(argin: Any | None = None, task_callback=None, task_abort_event=None) → None

Initialize the command context.

Parameters:

argin (str) – Optional input arguments for the command
task_callback – Callback function to update task status
task_abort_event – Event to signal command abortion

cleanup() → None: Clean up command resources and reset context.

wait_for_completion(): Wait for command completion using the completion tracker.

prepare_command() → None

Optional hook for command preparation.

Examples of use: - Parse JSON input - Validate input parameters - Download Telescope Model - Validate ObsState

post_process() → None

Optional hook for post-processing after command completion.

Examples of use: - Update task status - Publish events - Update attributes

should_stop_waiting() → bool

Returns True when the command no longer needs to wait for additional events.

The default implementation stops waiting when either: - the command has failed, or - the command has completed successfully.

has_failed() → bool: Returns True when the command has failed.

is_complete() → bool: Override this method in subclass and check command specific completion criteria.

are_all_results_received() → bool: Returns True when results have been received from every device on which the command was invoked.

is_state_complete() → bool: Override this method in subclasses if command completion depends on a specific state.

create_completion_context() → CommandCompletionContext: create a CommandCompletionContext for the current command execution.

allowed_result_codes() → set[ska_tango_base.commands.ResultCode]

Return the set of command result codes that are considered successful for this command.

Subclasses may override this method if additional result codes should be treated as successful.

Returns:: Set of allowed ResultCode values.

evaluate_result() → tuple[ska_tango_base.commands.ResultCode, str]

Evaluate the final outcome of the command.

The default implementation considers the command successful if every device returned one of the allowed result codes.

Returns:: Tuple containing the final ResultCode and message.

call_update_task_status(result: ska_tango_base.commands.ResultCode, message: str) → None

Call update task status and provide result, message attributes.

Parameters:

result (ResultCode) – Result code.
message (str) – Result message.

set_abort_flag() → None: Set abort flag to stop command completion tracker.

is_aborted() → bool: Check if command is aborted.

abstract update_task_status(**kwargs) → None

Update the task status with result code and exception message if any.

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