Base Task

class ska_csp_lmc_common.commands.task.BaseTask(name: str, task_type: TaskExecutionType | None = TaskExecutionType.SEQUENTIAL, skip_subtasks: bool | None = None, completed_callback: Callable | None = None, is_root: bool | None = False, logger: Logger | None = None)

Bases: object

Base class for all tasks in the task execution tree.

A task represents a unit of work in the command execution flow. It owns a TaskTracker used to aggregate execution state and propagate completion notifications upstream.

This class is not intended to be instantiated directly.

__init__(name: str, task_type: TaskExecutionType | None = TaskExecutionType.SEQUENTIAL, skip_subtasks: bool | None = None, completed_callback: Callable | None = None, is_root: bool | None = False, logger: Logger | None = None)

Initialise a base task.

Parameters:

  • name – Logical name identifying the task.

  • task_type – Execution strategy for the task.

  • skip_subtasks – Whether execution may stop early if a subtask fails (executor dependent).

  • completed_callback – Callback invoked when the task completes.

  • is_root – Whether this task is the root of the task tree.

  • logger – Logger used by the task and its tracker.

instantiate_tracker()

Create and configure a TaskTracker for this task.

Subclasses may override this method to customise the number of expected subtasks or tracker behaviour.

Returns:

A TaskTracker instance associated with this task.

parent_callback() → Callable

Return the callback used by subtasks to report updates upstream.

This callback is typically passed to child tasks so they can update the parent’s tracker as they progress or complete.

Returns:

Callable compatible with TaskTracker.update_task.

set_completed_callback(callback: Callable)

Set the completion callback for the task tracker.

Parameters:

callback – Callable invoked when the task completes.

Leaf Task

class ska_csp_lmc_common.commands.task.LeafTask(name: str, target: Any, command_name: str, task_type: TaskExecutionType | None = TaskExecutionType.DEVICE, skip_subtasks: bool | None = None, completed_callback: Callable | None = None, argin: Any | None = None, logger: Logger | None = None)

Bases: BaseTask

Leaf node of the task tree.

A LeafTask executes a single command on a target, such as a TANGO device or a ComponentManager method.

__init__(name: str, target: Any, command_name: str, task_type: TaskExecutionType | None = TaskExecutionType.DEVICE, skip_subtasks: bool | None = None, completed_callback: Callable | None = None, argin: Any | None = None, logger: Logger | None = None)

Initialise a leaf task.

Parameters:

  • name – Logical name of the task.

  • target – Target object on which the command is executed.

  • command_name – Name of the command to execute.

  • task_type – Execution type (DEVICE or INTERNAL).

  • skip_subtasks – Unused for leaf tasks, kept for interface consistency.

  • completed_callback – Callback invoked on task completion.

  • argin – Optional argument passed to the command.

  • logger – Logger used by this task.

Task

class ska_csp_lmc_common.commands.task.Task(name: str, subtasks: List[Task | LeafTask], task_type: TaskExecutionType | None = TaskExecutionType.SEQUENTIAL, skip_subtasks: bool | None = None, completed_callback: Callable | None = None, is_root: bool = False, logger: Logger | None = None)

Bases: BaseTask

Composite task that groups and executes multiple subtasks.

A Task may contain Task or LeafTask instances and can execute them sequentially or in parallel.

__init__(name: str, subtasks: List[Task | LeafTask], task_type: TaskExecutionType | None = TaskExecutionType.SEQUENTIAL, skip_subtasks: bool | None = None, completed_callback: Callable | None = None, is_root: bool = False, logger: Logger | None = None)

Initialise a composite task.

Parameters:

  • name – Logical name of the task.

  • subtasks – List of child Task or LeafTask objects.

  • task_type – Execution strategy (SEQUENTIAL or PARALLEL).

  • skip_subtasks – Whether execution may stop early if a subtask fails.

  • completed_callback – Callback invoked when the task completes.

  • is_root – Whether this task is the root of the task tree.

  • logger – Logger used by this task.

instantiate_tracker()

Create a TaskTracker configured for this composite task.

The tracker expects a number of subtasks equal to the length of the subtasks list.

Returns:

A TaskTracker instance.