Operational State Model

class ska_csp_lmc_common.model.OpStateModel(op_state_init: tango.DevState, op_state_changed_callback: Callable[[tango.DevState], None], logger: Logger | None = None)

Bases: object

A simple operational state model.

  • DevState.DISABLE – when communication with the component is not established.

  • DevState.FAULT – when the component has failed

__init__(op_state_init: tango.DevState, op_state_changed_callback: Callable[[tango.DevState], None], logger: Logger | None = None) → None

Initialise a new instance.

Parameters:

  • op_state_init – the initial state of the component under control.

  • op_state_changed_callback – callback to be called whenever there is a change to evaluated operational state.

  • logger – a logger for this instance

property faulty

Return whether the component is experiencing a faulty condition or not.

Returns:

whether the component is in fault.

property disabled: bool

Return whether the component is disabled or not.

Returns:

whether the component is disabled.

property op_state: tango.DevState

Return the component operational state.

Returns:

the operational state.

_update_op_state() → None

Update operational state.

This method calls the :py:meth:evaluate_op_state method to figure out what the new operational state should be, and then updates the state attribute, calling the callback if required.

update_op_state() → None

Update operational state.

Secure the access to the internal state by using a lock.

evaluate_op_state() → tango.DevState

Re-evaluate the operational state.

This method contains the logic for evaluating the state.

This method should be extended by subclasses in order to define how state is evaluated by their particular device.

If the CSP device opState is in FAULT for an internal error (i.e not depending from the opStates of the CSP sub-systems) this state has to be maintained. The only way to exit from this state is to Reset/Reinit the CSP device. In this case the faulty flag is reset to False.

Returns:

the new state state.

component_fault(faulty: bool) → None

Handle a component experiencing or recovering from a fault.

This method is called when the component goes into or out of FAULT state.

Parameters:

faulty – whether the component has failed or not

is_disabled(disabled: bool) → None

Handle disabling the monitoring functionalities of a TANGO device.

This method is called when the communication between the TANGO device and the component under controller is disabled/enabled via the setting of the adimnistrative mode.

Parameters:

disabled – whether the communication between the component and the controlling TANGO device is disabled.

perform_action(action: str) → None

Not operative method.

This method is required by the CspSubarray InitCommand class that must inherit from the SKABaseDevice.InitCommand because all the SKA attributes and logging are initialized there.

Health State Model

class ska_csp_lmc_common.model.HealthStateModel(device_fqdn: str, init_state: ska_control_model.HealthState, health_state_callback: Callable[[ska_control_model.HealthState], None], callback: Callable[[str, Any], None], logger: Logger | None = None)

Bases: object

Manages the health state of a device, firing callbacks on change.

__init__(device_fqdn: str, init_state: ska_control_model.HealthState, health_state_callback: Callable[[ska_control_model.HealthState], None], callback: Callable[[str, Any], None], logger: Logger | None = None) → None

Initialise a new HealthStateModel instance.

Parameters:

  • device_fqdn – device fqdn, used to fill the HealthInfo attribute

  • init_state – The health state of the device at initialization.

  • health_state_callback – callback to be called whenever there is a change to this this health model’s evaluated health state.

  • callback – a callback to be called when the health or the healthInfo of the device changes.

  • logger – the logger for this object.

property health_state: ska_control_model.HealthState

Return the current health state.

property device_fqdn: str

Return the device FQDN associated with this model.

property faulty: bool

Return the flag to force the HealthState in FAILED state.

property disabled: bool

Return the flag to force the HealthState in UNKNOWN state.

update_health() → None

Public wrapper to trigger health evaluation.

get_diagnostics_context() → Dict[str, Any]

Return context consumed by health diagnostics generation.

set_signal_callback(callback: Callable[[...], None] | None) → None

Set a callback to trigger snapshot-based evaluation.

_signal_evaluation(*, immediate: bool = False) → None

Trigger evaluation via supervisor or evaluate directly.

component_fault(fault: bool, reason: str) → None

Set or clear the fault flag forcing HealthState.FAILED.

When the flag fault is True, evaluation will return FAILED and healthInfo will include the provided reason. When False, the fault reason is cleared and normal evaluation rules apply.

Parameters:

  • fault – Whether to force HealthState.FAILED.

  • reason – Human-readable info stored in healthInfo when faulted.

is_disabled(disabled: bool, reason: str = '') → None

Set/Reset the flag to force the HealthState in UNKNOWN state.

Parameters:

  • disabled – whether communications with the component is established.

  • reason – message to log the reason of the modification.

set_fault_health_info(reason: str) → None

Set/Clear CSP device healthInfo entry for component fault.

set_disabled_health_info(reason: str) → None

Set/Clear CSP device healthInfo entry for disabled handling.

evaluate_from_snapshot(snap: Dict[str, ComponentSnapshot], *, now: float | None = None) → ska_control_model.HealthState

Compute aggregated HealthState from snapshot.

Parameters:

  • snap – Snapshot of subsystem states.

  • now – Optional timestamp.

Returns:

Aggregated HealthState.

apply_evaluation(new_health: ska_control_model.HealthState) → None

Apply evaluated health state using existing callbacks.

_publish_forced_healthinfo(reason: str) → None

Publish a forced healthInfo payload used by direct model paths only.

This method is intentionally limited to component_fault/is_disabled fallback mode (when no health supervisor is connected).

_evaluate_health_state() → ska_control_model.HealthState
_publish_health_state_if_changed(prev: ska_control_model.HealthState, new: ska_control_model.HealthState) → None

Call the HealthState callback only if the value changed.

_update_health_state_only() → None

Recompute healthState and update state only.

_update_health_state(new_health: ska_control_model.HealthState) → None

Update the internal state and publish via callback if changed.

Parameters:

new_health – new aggregated value of the device health state

static _has_critical(snap: Dict[str, ComponentSnapshot]) → bool

Observing State Model

class ska_csp_lmc_common.model.ObsStateModel(obs_state_init: ska_control_model.ObsState, publish_callback: Callable[[ska_control_model.ObsState], None], aggregation_policy: Callable[[], ska_control_model.ObsState] | None = None, logger: Logger | None = None, signal_callback: Callable[[ska_control_model.ObsState], None] | None = None)

Bases: object

State model for managing the observation state of a component.

This class coordinates state evaluations based on administrative status (disabled), health (faulty), or active commands (transitional). It uses an action counter to ensure that the model only reverts to automatic aggregation once all concurrent actions have completed.

Base ObsState model:

  • precedence: ABORTED > FAULT > DISABLED(EMPTY) > (policy candidate)

  • transitional (action-driven) via action counter

  • publish+dedup

  • supervisor integration via:

    • set_aggregated_candidate()

    • apply_final_state()

    • ping on exit_transitional_state()

__init__(obs_state_init: ska_control_model.ObsState, publish_callback: Callable[[ska_control_model.ObsState], None], aggregation_policy: Callable[[], ska_control_model.ObsState] | None = None, logger: Logger | None = None, signal_callback: Callable[[ska_control_model.ObsState], None] | None = None) → None

Initialize the observation state model.

Parameters:

  • obs_state_init – Initial observation state of the component.

  • publish_callback – Callback triggered on state change.

  • aggregation_policy – Logic to aggregate subordinate states.

  • logger – Logger instance for this model.

  • signal_callback – Callback to trigger the supervisor.

property obs_state: ska_control_model.ObsState

The observing state of the component under control of the CSP Subarray TANGO Device.

property faulty: bool

Return whether the component is experiencing a faulty condition or not.

Returns:

whether the component is aborted.

property aborted: bool

Return whether the component is experiencing a aborted condition or not.

Returns:

whether the component is in fault.

property fault_cause: FaultCause

Return the cause of fault.

property _is_action_driven: bool

Return whether there is an active command-driven action in progress.

This flag indicates that one or more high-level actions have been initiated and not yet completed. While an action is active, the observing state of the component must not be updated based on subsystem events or automatic aggregation logic.

During this phase, the observing state is controlled explicitly by the model (e.g. via a forced or frozen policy), regardless of whether a dedicated transitional ObsState is exposed.

The action is considered active as long as the internal action counter is greater than zero.

Note: to call with the lock.

Returns:

True if at least one action is currently in progress, False otherwise.

set_publish_callback(callback: Callable[[ska_control_model.ObsState], None]) → None

Set the callback used to publish ObsState changes.

This callback is invoked whenever the ObsStateModel determines that the effective observation state has changed (after applying precedence rules and de-duplication).

The callback is typically responsible for propagating the new ObsState to an external system (e.g. updating a TANGO attribute).

The callback is always invoked outside the internal lock to avoid deadlocks with external components.

set_signal_callback(callback: Callable[[ska_control_model.ObsState], None] | None) → None

Set the callback used to signal state-machine events to external coordinators.

This callback is used to notify an external entity (typically an observation supervisor) that the ObsStateModel has undergone a significant internal transition that may require re-evaluation of system-level logic.

Typical use cases include: - completion of a command-driven transitional state - explicit wake-up of a supervisor to trigger aggregation and consistency evaluation

The callback is not responsible for publishing ObsState values and must not assume that the provided ObsState represents a final or stable system state.

The callback is invoked outside the internal lock.

set_aggregated_candidate(candidate: ska_control_model.ObsState) → None

Cache the latest aggregated candidate from supervisor snapshot.

apply_final_state(final_state: ska_control_model.ObsState, *, notify: bool = True) → None

Apply final state decided by supervisor after consistency.

Ignored while action-driven (transitional/freeze active).

_event_driven_policy() → ska_control_model.ObsState

Return final override if present, else aggregated policy.

evaluate_obs_state() → ska_control_model.ObsState

Re-evaluate the component observing state.

This method contains the basic logic for evaluating the observing state.

This method should be extended by subclasses in order to define how observing state is evaluated by their particular device. Precedence order: 1. Aborted 2. Faulty 3. Disabled (Empty) 4. Current Policy (Transitional or Aggregated)

Returns:

the new observing state.

update_obs_state(*, notify: bool = True) → None

Coordinate state evaluation and notification.

Performs the check and application of the new state under lock, then triggers the change callback outside the lock to prevent deadlocks with external supervisors or mediators.

enter_transitional_state(obs_state: ska_control_model.ObsState) → None

Enter a transitional state when an action is initiated.

Increments the action counter and overrides the current policy with the provided transitional state.

Parameters:

obs_state – The transitional ObsState to enter.

exit_transitional_state(force_update: bool = False) → None

Exit transitional: - revert to event-driven policy (aggregated/final override) - DO NOT publish here (supervisor must recompute snapshot+consistency) - ping supervisor (signal_callback) to trigger immediate evaluation

component_fault(faulty: bool, *, cause: FaultCause = FaultCause.INTERNAL, notify: bool = True) → None

Handle a component experiencing or recovering from a fault.

This method is called when the component goes into or out of FAULT state.

Parameters:

faulty – whether the component has failed or not.

component_aborted(aborted: bool, *, notify: bool = True) → None

Handle a component experiencing or recovering from an abort.

This method is called when the component goes into or out of ABORTED state.

Parameters:

aborted – whether the component has failed or not.

component_disabled(disabled: bool, *, notify: bool = True) → None

Handle the monitoring functionalities of a TANGO Device.

This method is called when the communication between the TANGO device and the component under controller is disabled/enabled via the setting of the administrative mode.

Parameters:

disabled – whether the communication between the component and the controlling device is disabled.

_default_aggregation_policy() → ska_control_model.ObsState