# -*- 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 station."""
from __future__ import annotations
from typing import Optional, Sequence
from ska_control_model import HealthState
from ska_low_mccs_common.health import BaseHealthModel, HealthChangedCallbackProtocol
from ska_low_mccs.station.station_health_rules import StationHealthRules
__all__ = ["StationHealthModel"]
# pylint: disable = too-many-instance-attributes
[docs]class StationHealthModel(BaseHealthModel):
"""A health model for a station."""
[docs] def __init__(
self: StationHealthModel,
field_station_trl: str,
sps_station_trl: str,
antenna_trls: Sequence[str],
health_changed_callback: HealthChangedCallbackProtocol,
thresholds: Optional[dict[str, float]] = None,
) -> None:
"""
Initialise a new instance.
:param field_station_trl: the TRL of this station's FieldStation
:param sps_station_trl: the TRL of this MccsStation's SpsStation.
:param antenna_trls: the TRLs of this station's antennas
:param health_changed_callback: callback to be called whenever
there is a change to this this health model's evaluated
health state.
:param thresholds: the threshold parameters for the health rules
"""
self._field_station_trl = field_station_trl
self._sps_station_trl = sps_station_trl
self._field_station_health: Optional[HealthState] = HealthState.UNKNOWN
self._sps_station_health: Optional[HealthState] = HealthState.UNKNOWN
self._antenna_health: dict[str, Optional[HealthState]] = {
antenna_trl: HealthState.UNKNOWN for antenna_trl in antenna_trls
}
self._health_rules = StationHealthRules(thresholds)
super().__init__(health_changed_callback)
[docs] def field_station_health_changed(
self: StationHealthModel,
field_station_trl: Optional[str] = None,
field_station_health: Optional[HealthState] = None,
) -> None:
"""
Handle a change in FieldStation health.
:param field_station_health: the health state of the FieldStation,
or None if the FieldStation's admin mode indicates
that its health should not be rolled up.
:param field_station_trl: the TRL of the FieldStation.
"""
if field_station_trl:
assert (
self._field_station_trl == field_station_trl
), f"{self._field_station_trl} != {field_station_trl}"
if self._field_station_health != field_station_health:
self._field_station_health = field_station_health
self.update_health()
[docs] def antenna_health_changed(
self: StationHealthModel,
antenna_trl: str,
antenna_health: Optional[HealthState],
) -> None:
"""
Handle a change in antenna health.
:param antenna_trl: the TRL of the antenna whose health has
changed
:param antenna_health: the health state of the specified
antenna, or None if the antenna's admin mode indicates
that its health should not be rolled up.
"""
if self._antenna_health.get(antenna_trl) != antenna_health:
self._antenna_health[antenna_trl] = antenna_health
self.update_health()
[docs] def sps_station_health_changed(
self: StationHealthModel,
sps_station_trl: str,
sps_station_health: Optional[HealthState],
) -> None:
"""
Handle a change in SpsStation health.
:param sps_station_trl: the TRL of the SpsStation
:param sps_station_health: the health state of the
specified SpsStation, or None if the SpsStation's admin mode
indicates that its health should not be rolled up.
"""
if sps_station_trl:
assert (
self._sps_station_trl == sps_station_trl
), f"{self._sps_station_trl} != {sps_station_trl}"
if self._sps_station_health != sps_station_health:
self._sps_station_health = sps_station_health
self.update_health()
[docs] def evaluate_health(
self: StationHealthModel,
) -> tuple[HealthState, str]:
"""
Compute overall health of the station.
The overall health is based on the fault and communication
status of the station overall, together with the health of the
FieldStation, antennas and SpsStation that it manages.
This implementation simply sets the health of the station to the
health of its least healthy component.
:return: an overall health of the station
"""
if (
self._health_rules._thresholds["ignore_pasd"]
or self._health_rules._thresholds["ignore_sps"]
):
self._ignore_power_state = True
else:
self._ignore_power_state = False
station_health, station_report = super().evaluate_health()
for health in [
HealthState.FAILED,
HealthState.UNKNOWN,
HealthState.DEGRADED,
HealthState.OK,
]:
if health == station_health:
return station_health, station_report
result, report = self._health_rules.rules[health](
self._field_station_health,
self._sps_station_health,
self._antenna_health,
)
if result:
return health, report
return HealthState.UNKNOWN, "No rules matched"