Source code for ska_low_mccs.subarray.subarray_health_rules

#  -*- 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.
"""The rules model for subarray health."""
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})


class SubarrayHealthRules(HealthRules):
    """A class to handle transition rules for station."""

    # pylint: disable=arguments-differ
    def unknown_rule(  # type: ignore[override]
        self: SubarrayHealthRules,
        station_healths: dict[str, Optional[HealthState]],
        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 beam_healths: The healths of the subarray beams.

        :return: If the subarray is in unknown state.
        """
        rule_matched = (
            HealthState.UNKNOWN in beam_healths.values()
            or HealthState.UNKNOWN in station_healths.values()
        )
        if rule_matched:
            subarray_beam_states = [
                trl
                for trl, health in beam_healths.items()
                if health is None or health == HealthState.UNKNOWN
            ]
            station_states = [
                trl
                for trl, health in station_healths.items()
                if health is None or health == HealthState.UNKNOWN
            ]
            report = (
                "Some devices are unknown: "
                f"MccsStation: {station_states}, "
                f"MccsSubarrayBeam: {subarray_beam_states}"
            )
        else:
            report = ""
        return rule_matched, report

    # pylint: disable=arguments-differ
    def failed_rule(  # type: ignore[override]
        self: SubarrayHealthRules,
        station_healths: dict[str, Optional[HealthState]],
        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 beam_healths: The healths of the subarray beams.

        :return: If the subarray is failed or not.
        """
        rule_matched = (
            self.get_fraction_in_states(beam_healths, DEGRADED_STATES)
            >= self._thresholds["subarray_beam_failed_threshold"]
            or self.get_fraction_in_states(station_healths, DEGRADED_STATES)
            >= self._thresholds["station_failed_threshold"]
        )
        if rule_matched:
            subarray_beam_states = [
                trl
                for trl, health in beam_healths.items()
                if health is None or health in DEGRADED_STATES
            ]
            station_states = [
                trl
                for trl, health in station_healths.items()
                if health is None or health in DEGRADED_STATES
            ]
            report = (
                "Too many devices in a bad state: "
                f"MccsStation: {station_states}, "
                f"MccsSubarrayBeam: {subarray_beam_states}"
            )
        else:
            report = ""
        return rule_matched, report

    # pylint: disable=arguments-differ
    def degraded_rule(  # type: ignore[override]
        self: SubarrayHealthRules,
        station_healths: dict[str, Optional[HealthState]],
        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 beam_healths: The healths of the subarray beams.

        :return: If the subarray is degraded or not.
        """
        rule_matched = (
            self.get_fraction_in_states(beam_healths, DEGRADED_STATES)
            >= self._thresholds["subarray_beam_degraded_threshold"]
            or self.get_fraction_in_states(station_healths, DEGRADED_STATES)
            >= self._thresholds["station_degraded_threshold"]
        )
        if rule_matched:
            subarray_beam_states = [
                trl
                for trl, health in beam_healths.items()
                if health is None or health in DEGRADED_STATES
            ]
            station_states = [
                trl
                for trl, health in station_healths.items()
                if health is None or health in DEGRADED_STATES
            ]
            report = (
                "Too many devices in a bad state: "
                f"MccsStation: {station_states}, "
                f"MccsSubarrayBeam: {subarray_beam_states}"
            )
        else:
            report = ""
        return rule_matched, report

    # pylint: disable=arguments-differ
    def healthy_rule(  # type: ignore[override]
        self: SubarrayHealthRules,
        station_healths: dict[str, Optional[HealthState]],
        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 beam_healths: The healths of the subarray beams.

        :return: If the subarray is healthy or not.
        """
        return True, "Health is OK."

    @property
    def default_thresholds(self: HealthRules) -> dict[str, float]:
        """
        Get the default thresholds for this device.

        :return: the default thresholds
        """
        return {
            "subarray_beam_degraded_threshold": 0.05,
            "subarray_beam_failed_threshold": 0.2,
            "station_degraded_threshold": 0.05,
            "station_failed_threshold": 0.2,
        }