SubtaskResult

class ska_csp_lmc_common.commands.aggregation.task_result.SubtaskResult(name: str, status: ska_control_model.TaskStatus, result: Tuple[ska_control_model.ResultCode, str] | str | ska_control_model.ResultCode | int, device_name: str | None = None, devices: List[str] | None = None, operation: str | None = None, obs_mode: str | None = None, failed_devices: List[str] | None = None, logger: Logger | None = None)

Bases: object

Represent the latest execution update for a subtask.

A SubtaskResult stores:

  • name: stable subtask identifier

  • status: TaskStatus

  • result_code: ResultCode

  • message: human-readable info (optionally prefixed with device name)

It can also store optional context fields and device lists used for aggregation and diagnostics (device_name, devices, failed_devices, operation, obs_mode).

The incoming result field is parsed into (result_code, message) using _parse_result().

__init__(name: str, status: ska_control_model.TaskStatus, result: Tuple[ska_control_model.ResultCode, str] | str | ska_control_model.ResultCode | int, device_name: str | None = None, devices: List[str] | None = None, operation: str | None = None, obs_mode: str | None = None, failed_devices: List[str] | None = None, logger: Logger | None = None)

Create a new subtask result snapshot.

Parameters:

  • name – Subtask identifier (stable key used by the Task Tracker).

  • status – Latest subtask status.

  • result –

    Result in one of the supported formats:

    • (ResultCode, message) tuple/list

    • JSON-encoded pair (string representation of a two-element list and tuple)

    • legacy strings (e.g. "Completed")

    • ResultCode or int coercible to ResultCode

  • device_name – Originating device name (e.g. subsystem device), used to prefix the message when present.

  • devices – Optional list of devices involved in this subtask. Values are de-duplicated while preserving order.

  • operation – Optional operation context (domain-specific).

  • obs_mode – Optional observation mode context (domain-specific).

  • failed_devices – Optional list of devices that failed for this subtask. Values are de-duplicated while preserving order.

  • logger – Optional logger instance. If not provided, module_logger is used.

property device_name: str | None

Return the originating device name, if known.

property devices: List[str]

Return the list of involved devices (deduplicated, stable order).

Returns:

List of device names.

property failed_devices: List[str]

Return the list of devices that failed for this subtask.

Returns:

List of device names.

property operation: str | None

Set the operation context, if any.

Parameters:

value – Operation name or None.

property obs_mode: str | None

Return the observation mode context, if any.

Returns:

Observation mode or None.

update(status: ska_control_model.TaskStatus | None, result: Tuple[ska_control_model.ResultCode, str] | str | ska_control_model.ResultCode | int | None, device_name: str | None, devices: List[str] | None = None, operation: str | None = None, obs_mode: str | None = None, failed_devices: List[str] | None = None) → None

Update this SubtaskResult with a new status/result snapshot.

Any parameter set to None will not override the current value, with the exception of status and result which are treated as “no update” when None.

Parameters:

  • status – New status or None (no status update).

  • result – New result or None (no result update). Supported formats are the same as in __init__().

  • device_name – Originating device name. If provided, it will be stored and used to prefix the message.

  • devices – Optional list of devices involved in this subtask.

  • operation – Optional operation context update.

  • obs_mode – Optional observation mode context update.

  • failed_devices – Optional list of failed devices for this subtask.

Returns:

None.

static _normalize_message(msg: str) → str

Normalize message text for display.

Convert escaped newlines (\n) into real newlines.

_update_context_fields(*, device_name: str | None, devices: List[str] | None, operation: str | None, obs_mode: str | None, failed_devices: List[str] | None) → None

Update contextual fields only when explicitly provided.

Parameters:

  • device_name – Originating device name, or None to leave unchanged.

  • devices – Devices list, or None to leave unchanged.

  • operation – Operation, or None to leave unchanged.

  • obs_mode – Observation mode, or None to leave unchanged.

  • failed_devices – Failed devices list, or None to leave unchanged.

Returns:

None.

_prefix_device(msg: str) → str

Prefix a message with the originating device name when available.

Parameters:

msg – Message text.

Returns:

Prefixed message text if device_name is available, otherwise the original message.

static _dedup_list(values: List[str]) → List[str]

De-duplicate a list while preserving order.

Falsy entries are ignored.

Parameters:

values – Input list.

Returns:

De-duplicated list with stable ordering.

_parse_result(result: Tuple[ska_control_model.ResultCode, str] | str | ska_control_model.ResultCode | int) → Tuple[ska_control_model.ResultCode, str]

Parse an input result into (ResultCode, message).

Supported formats: - (rcode, message) tuple/list - string: JSON pair or legacy strings - ResultCode or int coercible to ResultCode

Parameters:

result – Input result payload.

Returns:

(result_code, message).

_as_pair(result: Tuple[ska_control_model.ResultCode, str] | str | ska_control_model.ResultCode | int) → Tuple[ska_control_model.ResultCode, str] | None

Return (rcode, msg) if result already looks like a pair.

Parameters:

result – Input result payload.

Returns:

(rcode, msg) when detectable, otherwise None.

_parse_result_string(result: str) → Tuple[ska_control_model.ResultCode, str]

Parse a string result: JSON pair or legacy strings.

Parameters:

result – Raw string result.

Returns:

(result_code, message).

_parse_result_codeish(result: ska_control_model.ResultCode | int) → Tuple[ska_control_model.ResultCode, str]

Parse a ResultCode or int into (rcode, message).

Parameters:

result – Result code value (enum or int).

Returns:

(result_code, message).

_coerce_resultcode(value: Any) → ska_control_model.ResultCode

Coerce a value into ResultCode.

Parameters:

value – Input value to coerce.

Returns:

A valid ResultCode when coercion succeeds, otherwise ResultCode.UNKNOWN.