Health

This module implements infrastructure for health management in the MCCS subsystem.

class BaseHealthModel(health_changed_callback, **kwargs)

A simple health model.

It supports:

  • HealthState.UNKNOWN – when communication with the component is not established.

  • HealthState.FAILED – when the component has faulted

  • HealthState.OK – when neither of the above conditions holds.

This health model does not support HealthState.DEGRADED. It is up to subclasses to implement support for DEGRADED if required.

__init__(health_changed_callback, **kwargs)

Initialise a new instance.

Parameters:
  • health_changed_callback (HealthChangedCallbackProtocol) – callback to be called whenever there is a change to this this health model’s evaluated health state.

  • kwargs (Any) – additional keyword arguments specifying initial values for state.

evaluate_health()

Re-evaluate the health state.

This method contains the logic for evaluating the health. It is this method that should be extended by subclasses in order to define how health is evaluated by their particular device.

Return type:

HealthState

Returns:

the new health state.

property health_params: dict[str, float]

Get the thresholds for health rules.

Returns:

the thresholds for health rules

property health_state: HealthState

Return the health state.

Returns:

the health state.

update_health()

Update health state.

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

Return type:

None

update_state(**kwargs)

Update this health model with state relevant to evaluating health.

Parameters:

kwargs (Any) – updated state

Return type:

None

class HealthChangedCallbackProtocol(*args, **kwargs)

Specification of the callback protocol.

The callback provided to the :py:class:~`.BaseHealthModel` is a callable that can be called as callback(health_state) or callback(health=health_state), and returns nothing.

__init__(*args, **kwargs)
class HealthModel(component_state_callback)

A simple health model.

It supports:

  • HealthState.UNKNOWN – when communication with the component is not established.

  • HealthState.FAILED – when the component has faulted

  • HealthState.OK – when neither of the above conditions holds.

This health model does not support HealthState.DEGRADED. It is up to subclasses to implement support for DEGRADED if required.

__init__(component_state_callback)

Initialise a new instance.

Parameters:

component_state_callback (Callable[..., None]) – callback to be called whenever there is a change to this this health model’s evaluated health state.

component_fault(faulty)

Handle a component experiencing or recovering from a fault.

This is a callback hook that is called when the component goes into or out of FAULT state.

Parameters:

faulty (bool) – whether the component has faulted or not

Return type:

None

evaluate_health()

Re-evaluate the health state.

This method contains the logic for evaluating the health. It is this method that should be extended by subclasses in order to define how health is evaluated by their particular device.

Return type:

HealthState

Returns:

the new health state.

property health_params: dict[str, float]

Get the thresholds for health rules.

Returns:

the thresholds for health rules

property health_state: HealthState

Return the health state.

Returns:

the health state.

is_communicating(communicating)

Handle change in communication with the component.

Parameters:

communicating (bool) – whether communications with the component is established.

Return type:

None

update_health()

Update health state.

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

Return type:

None

class HealthRules(thresholds=None)

A class to store health rules.

This should be implemented by each device with health roll-up logic.

__init__(thresholds=None)

Create a new instance.

Parameters:

thresholds (Optional[dict[str, float]]) – the conditions for particular health states

property default_thresholds: dict[str, float]

Get the default thresholds for this device.

This should be overriden in a derived class to provide default thresholds.

Returns:

an empty dictionary

degraded_rule()

Rule for the DEGRADED state.

Raises:

NotImplementedError – must be implemented in derived class

Return type:

bool

failed_rule()

Rule for the FAILED state.

Raises:

NotImplementedError – must be implemented in derived class

Return type:

bool

get_count_in_states(device_dict, states)

Get the number of devices in a given list of state.

Parameters:
Return type:

int

Returns:

the number of the devices in the given states

get_fraction_in_states(device_dict, states)

Get the fraction of devices in a given list of states.

Parameters:
Return type:

float

Returns:

the fraction of the devices in the given states

healthy_rule()

Rule for the OK state.

Raises:

NotImplementedError – must be implemented in derived class

Return type:

bool

property rules: dict[ska_control_model.health_state.HealthState, Callable[..., bool]]

Get the transition rules for the station.

The rules must be implemented on a device-by-device basis.

Returns:

the transition rules for the station

unknown_rule()

Rule for the UNKNOWN state.

Raises:

NotImplementedError – must be implemented in derived class

Return type:

bool