# -*- coding: utf-8 -*
#
# This file is part of the SKA Low MCCS project
#
#
# Distributed under the terms of the BSD 3-clause new license.
# See LICENSE for more info.
"""An implementation of a health model for a controller."""
from __future__ import annotations
from typing import Callable, Optional, Sequence
from ska_control_model import HealthState
from ska_low_mccs_common.health import BaseHealthModel
from ska_low_mccs.controller.controller_health_rules import ControllerHealthRules
__all__ = ["ControllerHealthModel"]
[docs]class ControllerHealthModel(BaseHealthModel):
"""A health model for a controller."""
[docs] def __init__(
self: ControllerHealthModel,
component_state_callback: Callable[..., None],
station_trls: Sequence[str],
subarray_beam_trls: Sequence[str],
station_beam_trls: Sequence[str],
thresholds: dict[str, float],
) -> None:
"""
Initialise a new instance.
:param component_state_callback: callback to be called whenever
there is a change to this this health model's evaluated
health state.
:param station_trls: the TRLs of this controller's stations
:param subarray_beam_trls: the TRLs of this controller's
subarray beams
:param station_beam_trls: the TRLs of this controller's
station beams
:param thresholds: The thresholds for number of sub devices healthy
"""
self._station_health: dict[str, Optional[HealthState]] = {
station_trl: HealthState.UNKNOWN for station_trl in station_trls
}
self._subarray_beam_health: dict[str, Optional[HealthState]] = {
subarray_beam_trl: HealthState.UNKNOWN
for subarray_beam_trl in subarray_beam_trls
}
self._station_beam_health: dict[str, Optional[HealthState]] = {
station_beam_trl: HealthState.UNKNOWN
for station_beam_trl in station_beam_trls
}
self._health_rules = ControllerHealthRules(thresholds)
super().__init__(component_state_callback)
[docs] def station_health_changed(
self: ControllerHealthModel,
station_trl: str,
station_health: Optional[HealthState],
) -> None:
"""
Handle a change in station health.
:param station_trl: the TRL of the station whose health has
changed
:param station_health: the health state of the specified
station, or None if the station's admin mode indicates
that its health should not be rolled up.
"""
if self._station_health.get(station_trl) != station_health:
self._station_health[station_trl] = station_health
self.update_health()
[docs] def subarray_beam_health_changed(
self: ControllerHealthModel,
subarray_beam_trl: str,
subarray_beam_health: Optional[HealthState],
) -> None:
"""
Handle a change in subarray beam health.
:param subarray_beam_trl: the TRL of the subarray beam whose
health has changed
:param subarray_beam_health: the health state of the specified
subarray beam, or None if the subarray beam's admin mode
indicates that its health should not be rolled up.
"""
if self._subarray_beam_health.get(subarray_beam_trl) != subarray_beam_health:
self._subarray_beam_health[subarray_beam_trl] = subarray_beam_health
self.update_health()
[docs] def station_beam_health_changed(
self: ControllerHealthModel,
station_beam_trl: str,
station_beam_health: Optional[HealthState],
) -> None:
"""
Handle a change in station beam health.
:param station_beam_trl: the TRL of the station beam whose
health has changed
:param station_beam_health: the health state of the specified
station beam, or None if the station beam's admin mode
indicates that its health should not be rolled up.
"""
if self._station_beam_health.get(station_beam_trl) != station_beam_health:
self._station_beam_health[station_beam_trl] = station_beam_health
self.update_health()
[docs] def evaluate_health(
self: ControllerHealthModel,
) -> tuple[HealthState, str]:
"""
Compute overall health of the controller.
The overall health is based on the fault and communication
status of the controller overall, together with the health of the
stations, subarray beams and station beams.
This implementation simply sets the health of the controller to
the health of its least healthy component.
:return: an overall health of the controller
"""
controller_health, controller_report = super().evaluate_health()
for health in [
HealthState.FAILED,
HealthState.UNKNOWN,
HealthState.DEGRADED,
HealthState.OK,
]:
if health == controller_health:
return controller_health, controller_report
result, report = self._health_rules.rules[health](
self._station_health,
self._subarray_beam_health,
self._station_beam_health,
)
if result:
return health, report
return HealthState.UNKNOWN, "No rules matched"