Health

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

class BaseHealthModel(health_changed_callback, ignore_power_state=False, **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, ignore_power_state=False, **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.
ignore_power_state (bool) – whether the health model should ignore the power state when computing health.
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:: tuple[HealthState, str]
Returns:: the new health state and health report.

evalute_power_health()

Evalute the health state w.r.t power.

If ignore_power_state is true then return OK health If PowerState is ON or STANDBY then return OK health If PowerState is OFF or NO_SUPPLY then return last known health state

If PowerState is UNKNOWN then return UNKNOWN health state

Return type:: tuple[HealthState, str]
Returns:: The health state relating to power.

property health_params: dict[str, float]

Get the thresholds for health rules.

Returns:: the thresholds for health rules

property health_report: str

Return the health report.

This is a short string to explain why the device is in its current health state.

Returns:: a health report

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:: tuple[bool, str]

failed_rule()

Rule for the FAILED state.

Raises:: NotImplementedError – must be implemented in derived class
Return type:: tuple[bool, str]

get_count_in_states(device_dict, states)

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

Parameters:

device_dict (Mapping[str, UnionType[PowerState, HealthState, None]]) – dictionary of devices, key fqdn and value state
states (FrozenSet[UnionType[PowerState, HealthState, None]]) – the states to check

Return type:

int

Returns:

the number of the devices in the given states

get_fraction_in_states(device_dict, states, default=0)

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

Parameters:

device_dict (Mapping[str, UnionType[PowerState, HealthState, None]]) – dictionary of devices, key fqdn and value state
states (FrozenSet[UnionType[PowerState, HealthState, None]]) – the states to check
default (float) – the default if there are zero devices in the dictionary

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:: tuple[bool, str]

property rules: dict[HealthState, Callable[[...], tuple[bool, str]]]

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:: tuple[bool, str]