# -*- 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 subarrays."""
from __future__ import annotations
from typing import Callable, Optional
from ska_control_model import HealthState
from ska_low_mccs_common.health import BaseHealthModel
from .subarray_health_rules import SubarrayHealthRules
__all__ = ["SubarrayHealthModel"]
[docs]
class SubarrayHealthModel(BaseHealthModel):
"""A health model for subarrays."""
DEGRADED_CRITERIA = 0.05
FAILED_CRITERIA = 0.2
[docs]
def __init__(
self: SubarrayHealthModel,
health_changed_callback: Callable,
ignore_power_state: bool = False,
thresholds: Optional[dict[str, float]] = None,
) -> None:
"""
Initialise a new instance.
:param health_changed_callback: a callback to be called when the
health of the subarray (as evaluated by this model) changes
:param ignore_power_state: whether the health model should ignore
the power state when computing health.
:param thresholds: Thresholds for failed and degraded states.
"""
self._station_healths: dict[str, Optional[HealthState]] = {}
self._subarray_beam_healths: dict[str, Optional[HealthState]] = {}
self._station_beam_healths: dict[str, Optional[HealthState]] = {}
self._health_rules = SubarrayHealthRules(thresholds)
super().__init__(health_changed_callback, ignore_power_state=ignore_power_state)
[docs]
def evaluate_health(
self: SubarrayHealthModel,
) -> tuple[HealthState, str]:
"""
Compute overall health of the subarray.
The overall health is based on the fault and communication
status of the subarray overall, together with the health of its
stations and subarray beams.
This implementation simply sets the health of the station to the
health of its least healthy component.
:return: an overall health of the subarray
"""
subarray_health, subarray_report = super().evaluate_health()
for health in [
HealthState.FAILED,
HealthState.UNKNOWN,
HealthState.DEGRADED,
HealthState.OK,
]:
if health == subarray_health:
return subarray_health, subarray_report
result, report = self._health_rules.rules[health](
self._station_healths,
self._subarray_beam_healths,
)
if result:
return health, report
return HealthState.UNKNOWN, "No rules matched"
[docs]
def resources_changed(
self: SubarrayHealthModel,
station_trls: set[str],
subarray_beam_trls: set[str],
station_beam_trls: set[str],
) -> None:
"""
Handle change in subarray resources.
This is a callback hook, called by the component manager when
the resources of the subarray changes.
:param station_trls: the TRLs of stations assigned to this subarray
:param subarray_beam_trls: the TRLs of subarray beams assigned
to this subarray
:param station_beam_trls: the TRLs of station beams assigned
to this subarray
"""
# We are updating with sets, which are unordered. This makes for
# ugly reports, so here we convert to a list and sort before assigning.
station_trl_list = sorted(list(station_trls))
subarray_beam_trl_list = sorted(list(subarray_beam_trls))
station_beam_trl_list = sorted(list(station_beam_trls))
self._station_healths = {
trl: self._station_healths.get(trl, HealthState.UNKNOWN)
for trl in station_trl_list
}
self._subarray_beam_healths = {
trl: self._subarray_beam_healths.get(trl, HealthState.UNKNOWN)
for trl in subarray_beam_trl_list
}
self._station_beam_healths = {
trl: self._station_beam_healths.get(trl, HealthState.UNKNOWN)
for trl in station_beam_trl_list
}
self.update_health()
[docs]
def station_health_changed(
self: SubarrayHealthModel,
trl: str,
health_state: HealthState | None,
) -> None:
"""
Handle change in station health.
This is a callback hook, called by the component manager when
the health of a station changes.
:param trl: the TRL of the station whose health has
changed
:param health_state: the new health state of the station, or
None if the subarray beam's health should not be taken
into account.
"""
self._station_healths[trl] = health_state
self.update_health()
[docs]
def subarray_beam_health_changed(
self: SubarrayHealthModel,
trl: str,
health_state: HealthState | None,
) -> None:
"""
Handle change in subarray beam health.
This is a callback hook, called by the component manager when
the health of a subarray beam changes.
:param trl: the TRL of the subarray beam whose health has
changed
:param health_state: the new health state of the subarray beam,
or None if the subarray beam's health should not be
taken into account.
"""
self._subarray_beam_healths[trl] = health_state
self.update_health()
[docs]
def station_beam_health_changed(
self: SubarrayHealthModel,
trl: str,
health_state: HealthState | None,
) -> None:
"""
Handle change in station beam health.
This is a callback hook, called by the component manager when
the health of a station beam changes.
:param trl: the TRL of the station beam whose health has
changed
:param health_state: the new health state of the station beam,
or None if the station beam's health should not be taken
into account.
"""
self._station_beam_healths[trl] = health_state
self.update_health()