TMC CommandDispatcher =========================================================== Overview -------- The ``CommandDispatcher`` class and its ``dispatch`` method provide a flexible, extensible architecture for handling commands in downstream TMC repositories. This guide explains how to leverage these components to implement custom command logic while maintaining consistency with the TMC framework. .. contents:: :local: :depth: 3 Design ~~~~~~ The command execution pipeline consists of three main components: 1. **CommandDispatcher**: Routes command requests to appropriate command instances using a factory pattern 2. **BaseTMCCommand**: Base class providing the ``execute`` method that invokes the command's workflow 3. **Pre-Dispatch Hooks**: Optional per-command callables run by the dispatcher before a command is instantiated Execution Flow ~~~~~~~~~~~~~~ The following diagram shows the flow of control from the Tango command handler, through the ``CommandDispatcher``, and into command execution: .. code-block:: text +------------------------------------------------------------------+ | Tango Command Handler (Leaf Node Component Manager.dispatcher) | | args: command_name, argin, task_callback, task_abort_event | +---------------------------------+--------------------------------+ | self._dispatcher.dispatch(...) v +------------------------------------------------------------------+ | CommandDispatcher.dispatch(command_name, argin, ...) | | | | (1) hook = pre_dispatch_hooks.get(command_name) | | if hook is not None: hook() ----> Pre-Dispatch Hook | | e.g. set_abort_flag_... | | | | (2) cmd = command_factory(command_name) ----> _make_command() | | (registry lookup)| | | | | +-- cmd is None --> raise ValueError("Unknown command") | | | | (3) return cmd.execute(argin, task_callback, task_abort_event) | +---------------------------------+--------------------------------+ | v +------------------------------------------------------------------+ | BaseTMCCommand.execute() -- command workflow | | pre_process() -> initialize() -> ... -> post_process() | | (builds a CommandContext and delegates device calls) | +---------------------------------+--------------------------------+ | v +------------------------------------------------------------------+ | CommandExecutor.execute(context) | | for each DeviceCommand in context.device_commands: | | adapter = adapter_provider.get_or_create_adapter(...) | | invoke LRC on device -> collect result into context | +---------------------------------+--------------------------------+ | v ResultCode + message -> task_callback Integration in Downstream Repositories --------------------------------------- Setting Up in a Leaf Node ~~~~~~~~~~~~~~~~~~~~~~~~~~ In a leaf node component manager: .. note:: The ``v4`` suffix reflects the common library's staged version-management strategy rather than a permanent location. Versioned packages (``v1``, ``v2``, ... ``vN``) let new functionality be introduced without breaking existing consumers. The older versions currently imported by dependent repositories (for example ``v1`` and ``v2``) will be phased out gradually, while a new consolidated version supports the combined functionality of all the previous ones will reamin. These changes will be then rolled out to the dependent repositories in a staggered manner. Once the new version has proven stable, the version suffix will be removed and imports will simplify to:: from ska_tmc_common.command_dispatcher import CommandDispatcher .. code-block:: python from ska_tmc_common.v4.command_dispatcher import CommandDispatcher class MyLeafNodeManager: def __init__(self, logger): self.logger = logger self._dispatcher = CommandDispatcher( command_factory=self._make_command, pre_dispatch_hooks={"Restart": self.set_abort_flag_for_commands}, ) def _make_command(self, name: str) -> Optional: registry = { "Configure": lambda: Configure(command_runtime_context, adapter_provider,logger), "AssignResources": lambda: AssignResources(command_runtime_context, adapter_provider,logger), "ReleaseAllResources": lambda: ReleaseAllResources(command_runtime_context, adapter_provider,logger), "End": lambda: End(command_runtime_context, adapter_provider,logger), "Scan": lambda: Scan(command_runtime_context, adapter_provider,logger), "EndScan": lambda: EndScan(command_runtime_context, adapter_provider,logger), "Restart": lambda: Restart(command_runtime_context, adapter_provider,logger), } builder = registry.get(name) if builder is None: return None instance = builder() return instance def dispatcher(self, command_name, argin="", task_callback=None, task_abort_event=None): """Tango command handler""" try: return self._dispatcher.dispatch( command_name=command_name, argin=argin, task_callback=task_callback, task_abort_event=task_abort_event, ) except ValueError as e: self.logger.error(f"Invalid command: {e}") def set_abort_flag_for_commands(self): """Set abort flags for command objects""" for command_obj in self.command_in_progress_objects: if hasattr(command_obj, "set_abort_flag"): command_obj.set_abort_flag() Best Practices -------------- 1. **Use Pre-Dispatch Hooks for Global Validation** Keep pre-dispatch hooks lightweight and use them only for checks that apply to all commands of a certain class. 2. **Use pre_process() for Command-Level Setup** The ``pre_process()`` hook is called at the beginning of command execution, before ``initialize()``. Use it for: - Custom validation raw input parameters (argin) - Checking preconditions before command execution - Initializing command-specific state Example: .. code-block:: python def pre_process(self, argin: Optional[Any] = None) -> None: """Validate and parse input arguments.""" if argin: try: self.parsed_input = json.loads(argin) if "custom_key" not in self.parsed_input.keys(): self.logger.error("Key 'custom_key' missing in input") except json.JSONDecodeError as e: raise CommandRejectedError(f"Invalid JSON input: {e}") 3. **Use post_process() for Cleanup and Post-Execution Actions** The ``post_process()`` hook is called at the end of command execution, after cleanup and task status update. Use it for: - Logging final command results - Triggering dependent actions based on command success/failure Example: .. code-block:: python def post_process(self) -> None: """Publish completion event and update status.""" self.logger.info("Command %s completed with status: %s", self.context.command_id, result_code) # Publish event to notify other components self._adapter.publish_event("CommandCompleted", { "command_id": self.context.command_id, "device_results": self.context.results })