# -*- 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.
"""A file to store health transition rules for station."""
from __future__ import annotations
from typing import Optional
from ska_control_model import HealthState
from ska_low_mccs_common.health import HealthRules
DEGRADED_STATES = frozenset({HealthState.DEGRADED, HealthState.FAILED, None})
[docs]class ControllerHealthRules(HealthRules):
"""A class to handle transition rules for station."""
# pylint: disable=arguments-differ
[docs] def unknown_rule( # type: ignore[override]
self: ControllerHealthRules,
station_healths: dict[str, Optional[HealthState]],
subarray_beam_healths: dict[str, Optional[HealthState]],
station_beam_healths: dict[str, Optional[HealthState]],
) -> tuple[bool, str]:
"""
Test whether UNKNOWN is valid for the station.
:param station_healths: The healths of the station subdevices.
:param subarray_beam_healths: The healths of the subarray beam subdevices.
:param station_beam_healths: The healths of the station beam subdevices.
:return: If the subarray is in unknown state.
"""
rule_matched = (
HealthState.UNKNOWN in station_healths.values()
or HealthState.UNKNOWN in subarray_beam_healths.values()
or HealthState.UNKNOWN in station_beam_healths.values()
)
if rule_matched:
station_states = [
trl
for trl, health in station_healths.items()
if health is None or health == HealthState.UNKNOWN
]
subarray_beam_states = [
trl
for trl, health in subarray_beam_healths.items()
if health is None or health == HealthState.UNKNOWN
]
station_beam_states = [
trl
for trl, health in station_beam_healths.items()
if health is None or health == HealthState.UNKNOWN
]
report = (
"Some devices are unknown: "
f"Station: {station_states}, "
f"SubarrayBeam: {subarray_beam_states}, "
f"StationBeam: {station_beam_states}"
)
else:
report = ""
return rule_matched, report
# pylint: disable=arguments-differ
[docs] def failed_rule( # type: ignore[override]
self: ControllerHealthRules,
station_healths: dict[str, Optional[HealthState]],
subarray_beam_healths: dict[str, Optional[HealthState]],
station_beam_healths: dict[str, Optional[HealthState]],
) -> tuple[bool, str]:
"""
Test whether FAILED is valid for the station.
:param station_healths: The healths of the station subdevices.
:param subarray_beam_healths: The healths of the subarray beam subdevices.
:param station_beam_healths: The healths of the station beam subdevices.
:return: If the subarray is failed or not.
"""
rule_matched = (
self.get_fraction_in_states(station_healths, DEGRADED_STATES)
>= self._thresholds["stations_failed_threshold"]
or self.get_fraction_in_states(subarray_beam_healths, DEGRADED_STATES)
>= self._thresholds["subarray_beams_failed_threshold"]
or self.get_fraction_in_states(station_beam_healths, DEGRADED_STATES)
>= self._thresholds["station_beams_failed_threshold"]
)
if rule_matched:
station_states = [
trl
for trl, health in station_healths.items()
if health is None or health in DEGRADED_STATES
]
subarray_beam_states = [
trl
for trl, health in subarray_beam_healths.items()
if health is None or health in DEGRADED_STATES
]
station_beam_states = [
trl
for trl, health in station_beam_healths.items()
if health is None or health in DEGRADED_STATES
]
report = (
"Too many devices in a bad state: "
f"Station: {station_states}, "
f"SubarrayBeam: {subarray_beam_states}, "
f"StationBeam: {station_beam_states}"
)
else:
report = ""
return rule_matched, report
# pylint: disable=arguments-differ
[docs] def degraded_rule( # type: ignore[override]
self: ControllerHealthRules,
station_healths: dict[str, Optional[HealthState]],
subarray_beam_healths: dict[str, Optional[HealthState]],
station_beam_healths: dict[str, Optional[HealthState]],
) -> tuple[bool, str]:
"""
Test whether DEGRADED is valid for the station.
:param station_healths: The healths of the station subdevices.
:param subarray_beam_healths: The healths of the subarray beam subdevices.
:param station_beam_healths: The healths of the station beam subdevices.
:return: If the subarray is degraded or not.
"""
rule_matched = (
self.get_fraction_in_states(station_healths, DEGRADED_STATES)
>= self._thresholds["stations_degraded_threshold"]
or self.get_fraction_in_states(subarray_beam_healths, DEGRADED_STATES)
>= self._thresholds["subarray_beams_degraded_threshold"]
or self.get_fraction_in_states(station_beam_healths, DEGRADED_STATES)
>= self._thresholds["station_beams_degraded_threshold"]
)
if rule_matched:
station_states = [
trl
for trl, health in station_healths.items()
if health is None or health in DEGRADED_STATES
]
subarray_beam_states = [
trl
for trl, health in subarray_beam_healths.items()
if health is None or health in DEGRADED_STATES
]
station_beam_states = [
trl
for trl, health in station_beam_healths.items()
if health is None or health in DEGRADED_STATES
]
report = (
"Too many devices in a bad state: "
f"Station: {station_states}, "
f"SubarrayBeam: {subarray_beam_states}, "
f"StationBeam: {station_beam_states}"
)
else:
report = ""
return rule_matched, report
# pylint: disable=arguments-differ
[docs] def healthy_rule( # type: ignore[override]
self: ControllerHealthRules,
station_healths: dict[str, Optional[HealthState]],
subarray_beam_healths: dict[str, Optional[HealthState]],
station_beam_healths: dict[str, Optional[HealthState]],
) -> tuple[bool, str]:
"""
Test whether OK is valid for the station.
:param station_healths: The healths of the station subdevices.
:param subarray_beam_healths: The healths of the subarray beam subdevices.
:param station_beam_healths: The healths of the station beam subdevices.
:return: If the subarray is healthy or not.
"""
return True, "Health is OK."
@property
def default_thresholds(self: ControllerHealthRules) -> dict[str, float]:
"""
Get the default thresholds for this device.
:return: default thresholds
"""
return {
"stations_failed_threshold": 0.2,
"stations_degraded_threshold": 0.05,
"subarray_beams_failed_threshold": 0.2,
"subarray_beams_degraded_threshold": 0.05,
"station_beams_failed_threshold": 0.2,
"station_beams_degraded_threshold": 0.05,
}