Source code for ska_low_mccs.antenna.antenna_health_rules

#  -*- 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 antenna."""
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 AntennaHealthRules(HealthRules): """A class to handle transition rules for antenna."""
[docs] def unknown_rule( # type: ignore[override] self: AntennaHealthRules, smartbox_health: HealthState | None, tile_health: HealthState | None, antenna_health: HealthState, tile_adc_rms: tuple[float, float] | None, ) -> bool: """ Test whether UNKNOWN is valid for the antenna. :param smartbox_health: health of the SmartBox :param tile_health: health of the Tile :param antenna_health: the antenna's computed health from it's communication and fault states, among others. Does not include monitoring points. :param tile_adc_rms: tuple of x and y ADC RMS power for this antenna read from the tile :return: True if UNKNOWN is a valid state """ return ( HealthState.UNKNOWN in {smartbox_health, tile_health, antenna_health} or tile_adc_rms is None )
[docs] def failed_rule( # type: ignore[override] self: AntennaHealthRules, smartbox_health: HealthState | None, tile_health: HealthState | None, antenna_health: HealthState, tile_adc_rms: tuple[float, float] | None, ) -> bool: """ Test whether FAILED is valid for the antenna. :param smartbox_health: health of the SmartBox :param tile_health: health of the Tile :param antenna_health: the antenna's computed health from it's communication and fault states, among others. Does not include monitoring points. :param tile_adc_rms: tuple of x and y ADC RMS power for this antenna read from the tile :return: True if FAILED is a valid state """ return HealthState.FAILED in {smartbox_health, tile_health, antenna_health} or ( tile_adc_rms is not None and ( tile_adc_rms[0] <= self._thresholds["failed_adc_rms_x"] or tile_adc_rms[1] <= self._thresholds["failed_adc_rms_y"] ) )
[docs] def degraded_rule( # type: ignore[override] self: AntennaHealthRules, smartbox_health: HealthState | None, tile_health: HealthState | None, antenna_health: HealthState, tile_adc_rms: tuple[float, float] | None, ) -> bool: """ Test whether DEGRADED is valid for the antenna. :param smartbox_health: health of the SmartBox :param tile_health: health of the Tile :param antenna_health: the antenna's computed health from it's communication and fault states, among others. Does not include monitoring points. :param tile_adc_rms: tuple of x and y ADC RMS power for this antenna read from the tile :return: True if DEGRADED is a valid state """ return HealthState.DEGRADED in { smartbox_health, tile_health, antenna_health, } or ( tile_adc_rms is not None and ( tile_adc_rms[0] <= self._thresholds["degraded_adc_rms_x"] or tile_adc_rms[1] <= self._thresholds["degraded_adc_rms_y"] ) )
[docs] def healthy_rule( # type: ignore[override] self: AntennaHealthRules, smartbox_health: HealthState | None, tile_health: HealthState | None, antenna_health: HealthState, tile_adc_rms: tuple[float, float] | None, ) -> bool: """ Test whether OK is valid for the antenna. :param smartbox_health: health of the SmartBox :param tile_health: health of the Tile :param antenna_health: the antenna's computed health from it's communication and fault states, among others. Does not include monitoring points. :param tile_adc_rms: tuple of x and y ADC RMS power for this antenna read from the tile :return: True if OK is a valid state """ return ( all( health == HealthState.OK for health in [smartbox_health, tile_health, antenna_health] ) and tile_adc_rms is not None and tile_adc_rms[0] > self._thresholds["degraded_adc_rms_x"] and tile_adc_rms[1] > self._thresholds["degraded_adc_rms_y"] )
@property def default_thresholds(self: HealthRules) -> dict[str, float]: """ Get the default thresholds for this device. :return: the default thresholds """ return { "failed_adc_rms_x": 0.0, "failed_adc_rms_y": 0.0, "degraded_adc_rms_x": 0.0, "degraded_adc_rms_y": 0.0, }