"""State transition computation."""
from typing import Optional
from ska_control_model import HealthState
from ska_mid_dish_manager.models.dish_enums import (
Band,
CapabilityStates,
DishMode,
PowerState,
SPFBandInFocus,
SPFHealthState,
)
from ska_mid_dish_manager.models.transition_rules import (
band_focus_rules_all_devices,
band_focus_rules_spfrx_ignored,
cap_state_rules_all_devices,
cap_state_rules_ds_only,
cap_state_rules_spf_ignored,
cap_state_rules_spfrx_ignored,
config_rules_all_devices,
config_rules_ds_only,
config_rules_spf_ignored,
config_rules_spfrx_ignored,
dish_mode_rules_all_devices,
dish_mode_rules_ds_only,
dish_mode_rules_spf_ignored,
dish_mode_rules_spfrx_ignored,
health_state_rules_all_devices,
health_state_rules_ds_only,
health_state_rules_spf_ignored,
health_state_rules_spfrx_ignored,
power_state_rules_all_devices,
power_state_rules_spf_ignored,
)
[docs]class StateTransition:
"""Computes the next state from rules based on component updates."""
[docs] def compute_dish_mode(
self,
ds_component_state: dict, # type: ignore
spfrx_component_state: Optional[dict] = None, # type: ignore
spf_component_state: Optional[dict] = None, # type: ignore
) -> DishMode:
"""Compute the dishMode based off component_states.
:param ds_component_state: DS device component state
:type ds_component_state: dict
:param spfrx_component_state: SPFRX device component state
:type spfrx_component_state: dict
:param spf_component_state: SPF device component state
:type spf_component_state: dict
:return: the calculated dishMode
:rtype: DishMode.
"""
dish_manager_states = self._collapse(
ds_component_state, spfrx_component_state, spf_component_state
)
rules_to_use = dish_mode_rules_ds_only
if spfrx_component_state and spf_component_state:
rules_to_use = dish_mode_rules_all_devices
elif spf_component_state:
rules_to_use = dish_mode_rules_spfrx_ignored
elif spfrx_component_state:
rules_to_use = dish_mode_rules_spf_ignored
for mode, rule in rules_to_use.items():
if rule.matches(dish_manager_states):
return DishMode[mode]
return DishMode.UNKNOWN
[docs] def compute_dish_health_state(
self,
ds_component_state: dict, # type: ignore
spfrx_component_state: Optional[dict] = None, # type: ignore
spf_component_state: Optional[dict] = None, # type: ignore
) -> HealthState:
"""Compute the HealthState based off component_states.
:param ds_component_state: DS device component state
:type ds_component_state: dict
:param spfrx_component_state: SPFRX device component state
:type spfrx_component_state: dict
:param spf_component_state: SPF device component state
:type spf_component_state: dict
:return: the calculated HealthState
:rtype: HealthState
"""
dish_manager_states = self._collapse(
ds_component_state, spfrx_component_state, spf_component_state
)
# Get the current enum
if ds_component_state:
dish_manager_states["DS"]["healthstate"] = ds_component_state.get(
"healthstate", HealthState.UNKNOWN
)
if spfrx_component_state:
dish_manager_states["SPFRX"]["healthstate"] = spfrx_component_state.get(
"healthstate", HealthState.UNKNOWN
)
if spf_component_state:
dish_manager_states["SPF"]["healthstate"] = spf_component_state.get(
"healthstate", SPFHealthState.UNKNOWN
)
# Build the name used on the transition rules
dish_manager_states["DS"]["healthstate"] = (
f"HealthState.{dish_manager_states['DS']['healthstate'].name}"
)
if "SPFRX" in dish_manager_states:
dish_manager_states["SPFRX"]["healthstate"] = (
f"HealthState.{dish_manager_states['SPFRX']['healthstate'].name}"
)
if "SPF" in dish_manager_states:
dish_manager_states["SPF"]["healthstate"] = (
f"SPFHealthState.{dish_manager_states['SPF']['healthstate'].name}"
)
rules_to_use = health_state_rules_ds_only
if spfrx_component_state and spf_component_state:
rules_to_use = health_state_rules_all_devices
elif spf_component_state:
rules_to_use = health_state_rules_spfrx_ignored
elif spfrx_component_state:
rules_to_use = health_state_rules_spf_ignored
for healthstate, rule in rules_to_use.items():
if rule.matches(dish_manager_states):
return HealthState[healthstate]
return HealthState.UNKNOWN
# pylint: disable=too-many-arguments
[docs] def compute_capability_state(
self,
band: str, # Literal["b1", "b2", "b3", "b4", "b5a", "b5b"],
ds_component_state: dict, # type: ignore
dish_manager_component_state: dict, # type: ignore
spfrx_component_state: Optional[dict] = None, # type: ignore
spf_component_state: Optional[dict] = None, # type: ignore
) -> CapabilityStates:
"""Compute the capabilityState based off component_states.
The same rules are used regardless of band.
This method renames b5aCapabilityState to capabilitystate to
apply the generic rules.
:param band: The band to calculate for
:type band: str
:param ds_component_state: DS device component state
:type ds_component_state: dict
:param spfrx_component_state: SPFRX device component state
:type spfrx_component_state: dict
:param spf_component_state: SPF device component state
:type spf_component_state: dict
:param dish_manager_component_state: Dish Manager device component state
:type dish_manager_component_state: dict
:return: the calculated capabilityState
:rtype: CapabilityStates
"""
# Add the generic name so the rules can be applied
# SPF
if spf_component_state:
cap_state = spf_component_state.get(f"{band}capabilitystate", None)
spf_component_state["capabilitystate"] = cap_state
# SPFRX
if spfrx_component_state:
cap_state = spfrx_component_state.get(f"{band}capabilitystate", None)
spfrx_component_state["capabilitystate"] = cap_state
dish_manager_states = self._collapse(
ds_component_state,
spfrx_component_state,
spf_component_state,
dish_manager_component_state,
)
rules_to_use = cap_state_rules_ds_only
if spfrx_component_state and spf_component_state:
rules_to_use = cap_state_rules_all_devices
elif spf_component_state:
rules_to_use = cap_state_rules_spfrx_ignored
elif spfrx_component_state:
rules_to_use = cap_state_rules_spf_ignored
new_cap_state = CapabilityStates.UNKNOWN
for capability_state, rule in rules_to_use.items():
if rule.matches(dish_manager_states):
if capability_state.startswith("STANDBY"):
new_cap_state = CapabilityStates["STANDBY"]
else:
new_cap_state = CapabilityStates[capability_state]
break
# Clean up state dicts
for state_dict in [
spfrx_component_state,
spf_component_state,
dish_manager_component_state,
]:
if state_dict and "capabilitystate" in state_dict:
del state_dict["capabilitystate"]
return new_cap_state
[docs] def compute_spf_band_in_focus(
self,
ds_component_state: dict, # type: ignore
spfrx_component_state: Optional[dict] = None, # type: ignore
) -> SPFBandInFocus:
"""Compute the bandinfocus based off component_states.
:param ds_component_state: DS device component state
:type ds_component_state: dict
:param spfrx_component_state: SPFRX device component state
:type spfrx_component_state: dict
:return: the calculated bandinfocus
:rtype: SPFBandInFocus
"""
dish_manager_states = self._collapse(ds_component_state, spfrx_component_state)
rules_to_use = band_focus_rules_all_devices
if not spfrx_component_state:
rules_to_use = band_focus_rules_spfrx_ignored
for band_number, rule in rules_to_use.items():
if rule.matches(dish_manager_states):
return SPFBandInFocus[band_number]
return SPFBandInFocus.UNKNOWN
[docs] def compute_power_state(
self,
ds_component_state: dict, # type: ignore
spf_component_state: Optional[dict] = None, # type: ignore
) -> PowerState:
"""Compute the powerstate based off component_states.
:param ds_component_state: DS device component state
:type ds_component_state: dict
:param spf_component_state: SPF device component state
:type spf_component_state: dict
:return: the calculated powerstate
:rtype: PowerState
"""
dish_manager_states = self._collapse(
ds_component_state, spf_component_state=spf_component_state
)
rules_to_use = power_state_rules_all_devices
if not spf_component_state:
rules_to_use = power_state_rules_spf_ignored
for power_state, rule in rules_to_use.items():
if rule.matches(dish_manager_states):
if power_state.startswith("UPS"):
return PowerState[power_state]
_power_state = power_state.split("_")[0] # pylint: disable=use-maxsplit-arg
return PowerState[_power_state]
return PowerState.LOW
@classmethod
def _collapse(
cls,
ds_component_state: dict, # type: ignore
spfrx_component_state: Optional[dict] = None,
spf_component_state: Optional[dict] = None, # type: ignore
dish_manager_component_state: Optional[dict] = None, # type: ignore
) -> dict: # type: ignore
"""Collapse multiple state dicts into one."""
dish_manager_states = {"DS": {}} # type: ignore
for key, val in ds_component_state.items():
dish_manager_states["DS"][key] = str(val)
if spfrx_component_state:
dish_manager_states["SPFRX"] = {}
for key, val in spfrx_component_state.items():
dish_manager_states["SPFRX"][key] = str(val)
if spf_component_state:
dish_manager_states["SPF"] = {}
for key, val in spf_component_state.items():
dish_manager_states["SPF"][key] = str(val)
if dish_manager_component_state:
dish_manager_states["DM"] = {}
for key, val in dish_manager_component_state.items():
dish_manager_states["DM"][key] = str(val)
return dish_manager_states