# -*- coding: utf-8 -*
# pylint: disable=arguments-differ
#
# 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.
"""A file to store health transition rules for station."""
from __future__ import annotations
from ska_control_model import HealthState
from ska_low_mccs_common.health import HealthRules
DEGRADED_STATES = frozenset({HealthState.DEGRADED, HealthState.FAILED, None})
[docs]
class StationHealthRules(HealthRules):
"""A class to handle transition rules for station."""
[docs]
def unknown_rule( # type: ignore[override]
self: StationHealthRules,
field_station_health: HealthState | None,
sps_station_health: HealthState | None,
antenna_healths: dict[str, HealthState | None],
) -> tuple[bool, str]:
"""
Test whether UNKNOWN is valid for the station.
:param field_station_health: health of the FieldStation
:param sps_station_health: health of the SpsStation
:param antenna_healths: dictionary of antenna healths
:return: True if UNKNOWN is a valid state, along with a text report.
"""
rule_matched = (
(
field_station_health == HealthState.UNKNOWN
and not self._thresholds["ignore_pasd"]
)
or field_station_health is None
or (
sps_station_health == HealthState.UNKNOWN
and not self._thresholds["ignore_sps"]
)
or sps_station_health is None
or (
HealthState.UNKNOWN in antenna_healths.values()
and not self._thresholds["ignore_pasd"]
)
)
if rule_matched:
antenna_states = [
trl
for trl, health in antenna_healths.items()
if health is None or health == HealthState.UNKNOWN
]
sps_station_state = (
HealthState(sps_station_health).name
if sps_station_health is not None
else sps_station_health
)
field_station_state = (
HealthState(field_station_health).name
if field_station_health is not None
else field_station_health
)
report = (
"Some devices are unknown: "
f"FieldStation: {field_station_state} SpsStation:"
f" {sps_station_state}, Antennas: {antenna_states}"
)
else:
report = ""
return rule_matched, report
[docs]
def failed_rule( # type: ignore[override]
self: StationHealthRules,
field_station_health: HealthState | None,
sps_station_health: HealthState | None,
antenna_healths: dict[str, HealthState | None],
) -> tuple[bool, str]:
"""
Test whether FAILED is valid for the station.
:param field_station_health: health of the FieldStation
:param sps_station_health: health of the SpsStation
:param antenna_healths: dictionary of antenna healths
:return: True if FAILED is a valid state
"""
rule_matched = (
(
field_station_health == HealthState.FAILED
and not self._thresholds["ignore_pasd"]
)
or (
sps_station_health == HealthState.FAILED
and not self._thresholds["ignore_sps"]
)
or self.get_fraction_in_states(antenna_healths, DEGRADED_STATES)
>= self._thresholds["antenna_failed"]
)
if rule_matched:
antenna_states = [
trl
for trl, health in antenna_healths.items()
if health is None or health in DEGRADED_STATES
]
sps_station_state = (
HealthState(sps_station_health).name
if sps_station_health is not None
else sps_station_health
)
field_station_state = (
HealthState(field_station_health).name
if field_station_health is not None
else field_station_health
)
report = (
"Too many subdevices are in a bad state: "
f"FieldStation: {field_station_state} SpsStation:"
f" {sps_station_state}, Antennas: {antenna_states}"
)
else:
report = ""
return rule_matched, report
[docs]
def degraded_rule( # type: ignore[override]
self: StationHealthRules,
field_station_health: HealthState | None,
sps_station_health: HealthState | None,
antenna_healths: dict[str, HealthState | None],
) -> tuple[bool, str]:
"""
Test whether DEGRADED is valid for the station.
:param field_station_health: health of the FieldStation
:param sps_station_health: health of the SpsStation
:param antenna_healths: dictionary of antenna healths
:return: True if DEGRADED is a valid state
"""
rule_matched = (
(
field_station_health == HealthState.DEGRADED
and not self._thresholds["ignore_pasd"]
)
or (
sps_station_health == HealthState.DEGRADED
and not self._thresholds["ignore_sps"]
)
or self.get_fraction_in_states(antenna_healths, DEGRADED_STATES)
>= self._thresholds["antenna_degraded"]
)
if rule_matched:
antenna_states = [
trl
for trl, health in antenna_healths.items()
if health is None or health in DEGRADED_STATES
]
sps_station_state = (
HealthState(sps_station_health).name
if sps_station_health is not None
else sps_station_health
)
field_station_state = (
HealthState(field_station_health).name
if field_station_health is not None
else field_station_health
)
report = (
"Too many subdevices are in a bad state: "
f"FieldStation: {field_station_state} SpsStation:"
f" {sps_station_state}, Antennas: {antenna_states}"
)
else:
report = ""
return rule_matched, report
[docs]
def healthy_rule( # type: ignore[override]
self: StationHealthRules,
field_station_health: HealthState | None,
sps_station_health: HealthState | None,
antenna_healths: dict[str, HealthState | None],
) -> tuple[bool, str]:
"""
Test whether OK is valid for the station.
:param field_station_health: health of the FieldStation
:param sps_station_health: health of the SpsStation
:param antenna_healths: dictionary of antenna healths
:return: True if OK is a valid state or if station is empty
"""
if not isinstance(self._thresholds["ignore_pasd"], bool):
return False, "ignore_pasd parameter is not a bool"
if not isinstance(self._thresholds["ignore_sps"], bool):
return False, "ignore_sps parameter is not a bool"
rule_matched = (
(field_station_health == HealthState.OK or self._thresholds["ignore_pasd"])
and (sps_station_health == HealthState.OK or self._thresholds["ignore_sps"])
and self.get_fraction_in_states(antenna_healths, DEGRADED_STATES)
< self._thresholds["antenna_degraded"]
)
if not rule_matched:
antenna_states = [
trl
for trl, health in antenna_healths.items()
if health is None or health in DEGRADED_STATES
]
sps_station_state = (
HealthState(sps_station_health).name
if sps_station_health is not None
else sps_station_health
)
field_station_state = (
HealthState(field_station_health).name
if field_station_health is not None
else field_station_health
)
report = (
"Too many subdevices are in a bad state: "
f"FieldStation: {field_station_state} SpsStation:"
f" {sps_station_state}, Antennas: {antenna_states}"
)
else:
report = ""
return rule_matched, report
@property
def default_thresholds(self: HealthRules) -> dict[str, float | bool]:
"""
Get the default thresholds for this device.
:return: the default thresholds
"""
return {
"antenna_degraded": 0.05,
"antenna_failed": 0.2,
"ignore_sps": False,
"ignore_pasd": False,
}