Main Task Composer

class ska_csp_lmc_common.commands.main_task_composer.MainTaskComposer(name: str, command_map: dict, component_manager, callback: Callable, resources: Dict[str, Any] | None = None)

Bases: object

Class that create the MainTaskComposer object, to be called in component manager in each command

compose()

Compose the MainTask starting from the command map

_clean_task(task: Task) → Task | None

Recursively clean tasks by removing tasks with empty subtasks. If a task has empty subtasks, it is removed. If subtasks of a task are cleaned and become empty, it is also removed.

Parameters:

task – The Task to be cleaned.

Returns:

The cleaned Task, or None if the task is empty.

_get_skip_subtasks(task_map: Dict) → bool

Retrieve the skip_subtasks flag if set.

Parameters:

task_map – the map describing the task.

Returns:

the skip_subtasks flag value.

_build_task(name: str, task_map: dict, allowed_attr_name, allowed_values: list, skip_subtasks: bool | None = None, is_root: bool = False, callback=None)

Recursively builds Task and Subtask objects from dictionary data.

Parameters:

  • name – Name of the task.

  • task_map – Dictionary representing a task or subtask.

  • allowed_attr_name – the name of the attribute which allow the issue of a task.

  • allowed_values – the attribute values which allow the issue of a task.

  • skip_subtasks – If set to True, all subsequent subtasks in a sequence will be skipped if a preceding task fails.

Param:

callback: the task callback.

Returns:

Task object or a SubTask object.

_get_task_handler(name: str) → Callable | None

Retrieves the handler method for a given task name.

Parameters:

name – Name of the sub-system.

Returns:

Handler method or None if not found.

_create_internal_task(task_map: Dict[str, Any], allowed_attr_name, allowed_values: list, skip_subtasks) → LeafTask

Creates an Internal LeafTask.

Parameters:

  • task_map – Dictionary containing task details.

  • allowed_attr_name – the name of the attribute which allow the issue of a task.

  • allowed_values – the attribute values which allow the issue of a task.

  • skip_subtasks – If set to True, all subsequent subtasks in a sequence will be skipped if a preceding task fails.

Returns:

Internal LeafTask object.

_create_subsystem_task(name: str, task_map: Dict[str, Any], allowed_attr_name, allowed_values: list, skip_subtasks) → LeafTask | None

Creates a Subsystem LeafTask for ‘cbf’ or ‘pss’.

Parameters:

  • name – Name of the subsystem (‘cbf’ or ‘pss’).

  • task_map – dictionary containing task details.

  • allowed_attr_name – the name of the attribute which allow the issue of a task.

  • allowed_values – the attribute values which allow the issue of a task.

  • skip_subtasks – If set to True, all subsequent subtasks in a sequence will be skipped if a preceding task fails.

Returns:

Subsystem LeafTask object or None.

_create_pst_task(task_map: dict, allowed_attr_name, allowed_values: list, skip_subtasks)

Create a parallel task on all the pst beams that are supposed to receive the command

Parameters:

  • task_map – dictionary containing task details.

  • allowed_attr_name – the name of the attribute which allow the issue of a task.

  • allowed_values – the attribute values which allow the issue of a task.

  • skip_subtasks – If set to True, all subsequent subtasks in a sequence will be skipped if a preceding task fails.

Returns:

a parallel Task to run on all the PST beams.

_create_csp_subarrays_task(task_map: Dict, allowed_attr_name, allowed_values: list, skip_subtasks)

Create a parallel task on all csp subarrays that are online and connected (case of csp controller)

Parameters:

  • task_map – dictionary containing task details.

  • allowed_attr_name – the name of the attribute which allow the issue of a task.

  • allowed_values – the attribute values which allow the issue of a task.

  • skip_subtasks – If set to True, all subsequent subtasks in a sequence will be skipped if a preceding task fails.

Returns:

a parallel Task to run on all the CSP Subarrays.