"""Module that defines a class for mannaging different orthogonal conditions about a running MVP in order to aggregate the states of various
devices into a combined condition of FAULTY, OK or DEGRADED
"""
from enum import Enum, auto
from functools import reduce
from typing import (
Any,
Callable,
Dict,
ItemsView,
KeysView,
List,
NamedTuple,
Set,
Tuple,
Union,
ValuesView,
)
from ska_ser_skallop.mvp_control.describing.inspections import (
DevList,
DevName,
DevState,
Grouping,
ListInspection,
ListInspections,
)
[docs]class ConditionLabel(NamedTuple):
desc: str
label: str
dim: str = "state"
[docs]class ConditionState(Enum):
"""Used to specify the current state of a condition. OK means all the reported states are what they are supposed to be.
INCONSISTENT means some of them are in the wrong state and WRONG means all of the states are in the wrong (but consistent) state.
Lastly, UNKNOWN means the state can not be determined as it is unable to communicate with the MVP or has not yet done so.
Args:
Enum ([type]): [description]
"""
OK = auto()
INCONSISTENT = auto()
WRONG = auto()
UNKNOWN = auto()
[docs]class Getter:
_fn: Callable[..., Union[ListInspection, ListInspections]]
_args: Tuple[Any, ...]
def __init__(
self, fn: Callable[..., Union[ListInspection, ListInspections]], *args: Any
) -> None:
self._fn = fn
self._args = args
[docs] def get(self) -> Union[ListInspection, ListInspections]:
return self._fn(*self._args)
[docs]class ConditionLabels:
[docs] @classmethod
def as_labels(cls) -> Set:
attrs = [
i
for i in cls.__dict__.keys() # pylint: disable=consider-iterating-dictionary
if i[:1] != "_"
]
return {attr for attr in attrs if attr != "as_labels"}
@staticmethod
def _label(desc: str, label: str, dimension: str):
return ConditionLabel(desc, label, dimension)
[docs]class Condition:
data: Union[ListInspection, ListInspections]
state: ConditionState
get_state: Callable[[], None]
_grouping: Grouping
def __init__(self, getter: Getter, *desired_value: str) -> None:
self.state = ConditionState.UNKNOWN
self.data = ListInspection({})
self._getter = getter
self.desired_value = desired_value
self._grouping = {}
@property
def value(self) -> Union[List[DevState], DevState]:
return self.data.value
@property
def grouping(self) -> Grouping:
if not self.data:
self.update_condition()
return self._grouping
[docs] def describe_unhealthy_state(self) -> Union[str, None]:
if not self.data:
self.update_condition()
if self.is_not_ok():
if self.is_wrong():
return f" all items in {self.data.value}"
wrong_states_descriptions = [
f"{v.items} to be {k}"
for k, v in self._grouping.items()
if k not in self.desired_value
]
if wrong_states_descriptions:
if len(wrong_states_descriptions) > 1:
wrong_states_descriptions = reduce(
aggregate_string_with_and, wrong_states_descriptions
)
return f":\n{wrong_states_descriptions}"
return f" {wrong_states_descriptions[0]}"
return None
@property
def devices(self) -> DevList:
return self.data.devices
[docs] def is_ok(self) -> bool:
return self.state == ConditionState.OK
[docs] def is_not_ok(self) -> bool:
return self.state is not ConditionState.OK
[docs] def is_wrong(self) -> bool:
return self.state == ConditionState.WRONG
[docs] def is_inconsistent(self) -> bool:
return self.state == ConditionState.INCONSISTENT
def _load_state(self):
self.data = self._getter.get()
def _calc_state(self) -> bool:
assert self.data
if all(self.data.are_in_state(*self.desired_value)):
self.state = ConditionState.OK
return True
if all(self.data.are_not_in_state(*self.desired_value)) and self.data.are_all_the_same():
self.state = ConditionState.WRONG
else:
self.state = ConditionState.INCONSISTENT
return False
[docs] def in_state(self, *state: str) -> DevList:
return self.data.in_state(*state)
[docs] def are_in_state(self, *state: str) -> List[bool]:
return self.data.are_in_state(*state)
[docs] def are_not_in_state(self, *state: str) -> Dict[DevName, DevState]:
return self.data.are_not_in_state(*state)
[docs] def not_in_state(self, *state: str) -> DevList:
return self.data.not_in_state(*state)
[docs] def update_condition(self, *desired_value: str) -> bool:
if desired_value:
self.desired_value = desired_value
self._load_state()
if self.data:
self._grouping = self.data.get_grouping()
return self._calc_state()
return False
def __bool__(self) -> bool:
return self.update_condition()
[docs] def items(self) -> ItemsView[str, ListInspection]:
if isinstance(self.data, ListInspection):
return ListInspections({"1": self.data}).items()
return self.data.items()
[docs] def keys(self) -> KeysView[str]:
if isinstance(self.data, ListInspection):
return ListInspections({"1": self.data}).keys()
return self.data.keys()
[docs] def values(self) -> ValuesView[ListInspection]:
if isinstance(self.data, ListInspection):
return ListInspections({"1": self.data}).values()
return self.data.values()
[docs]class Conditional(NamedTuple):
desc: str
cond: Condition
dim: str = "state"
[docs]class ConditionDescription(NamedTuple):
label: str
desc: str
unhealthy_states: Union[str, None]
desired: Tuple
state: ConditionState
[docs]def aggregate_string_with_and(line1: str, line2: str) -> str:
return f"{line1} and,\n{line2}"
[docs]def aggregate_string(line1: str, line2: str) -> str:
return f"{line1}\n{line2}"
Label = str
Conditions = Dict[Label, Conditional]
[docs]class FilteredConditions:
def __init__(self, conditions: Conditions) -> None:
self.conditions = conditions
[docs] def includes(self, condition_label: ConditionLabel) -> bool:
label = condition_label.label
return label in self.conditions.keys()
def __getitem__(self, condition_label: ConditionLabel) -> Condition:
label = condition_label.label
return self.conditions[label].cond
[docs] def get(self, condition_label: ConditionLabel) -> Union[Condition, None]:
label = condition_label.label
conditional = self.conditions.get(label)
return self.conditions.get(label).cond if conditional else None
[docs]class Readiness:
conditions: Dict[str, Conditional]
def __init__(self, *labels: Union[Set[str], ConditionLabels]) -> None:
self.conditions: Dict[Label, Conditional] = {}
self._constructed_expection_description = ""
self._constructed_expection_getter = None
self._constructed_expectation_label = ""
self._constructed_expectation_dim = "state"
self.labels = {"general"}
for label_set in labels:
label_set = (
label_set.as_labels() if isinstance(label_set, ConditionLabels) else label_set
)
self.labels = {*label_set, *self.labels}
[docs] def get_not_ok_conditions(self) -> FilteredConditions:
return FilteredConditions(
{k: v for k, v in self.conditions.items() if v.cond.state is not ConditionState.OK}
)
[docs] def get_inconsistent_conditions(self) -> FilteredConditions:
return FilteredConditions(
{
k: v
for k, v in self.conditions.items()
if v.cond.state == ConditionState.INCONSISTENT
}
)
[docs] def get_wrong_conditions(self) -> FilteredConditions:
return FilteredConditions(
{k: v for k, v in self.conditions.items() if v.cond.state == ConditionState.WRONG}
)
def __getitem__(self, condition_label: ConditionLabel) -> Condition:
return self.conditions[condition_label.label].cond
[docs] def get(self, condition_label: ConditionLabel) -> Union[Condition, None]:
conditional = self.conditions.get(condition_label.label)
label = conditional.cond if conditional else None
return label
[docs] def update(self):
return all(conditional.cond for conditional in self.conditions.values())
@property
def aggregate_results(self) -> ListInspections:
dimensions = {cnd.dim for cnd in self.conditions.values()}
aggregate_results = ListInspections({})
for dim in dimensions:
filtered_results = [cnd.cond.data for cnd in self.conditions.values() if cnd.dim == dim]
aggregate_results[dim] = reduce(lambda x, y: x + y, filtered_results)
return aggregate_results
@property
def aggregate_condition(self) -> Dict[str, Union[List[DevState], DevState]]:
return {state: inspection.value for state, inspection in self.aggregate_results.items()}
[docs] def filter_by(self, state: str) -> ListInspection:
return self.aggregate_results[state]
[docs] def select(self, state: str, *value: str) -> DevList:
return self.filter_by(state).in_state(*value)
[docs] def set_labels(self, labels: Set[str]) -> None:
self.labels = {*labels, "general"}
[docs] def load_condition(
self, desc: str, condition: Condition, label="general", dim: str = "state"
) -> None:
assert label in self.labels, f"unrecognized label {label} can only be one of {self.labels}"
assert label not in self.conditions.keys(), "condition of type {label} already loaded"
self.conditions[label] = Conditional(desc, condition, dim)
def __bool__(self) -> bool:
return all(conditional.cond for conditional in self.conditions.values())
[docs] def expect(self, description: str, label="general", dim: str = "state"):
self._constructed_expection_description = description
self._constructed_expectation_label = label
self._constructed_expectation_dim = dim
return self
[docs] def describe_why_not_ready(self) -> str:
description = ""
descriptions = [
ConditionDescription(
label,
conditional.desc,
conditional.cond.describe_unhealthy_state(),
conditional.cond.desired_value,
conditional.cond.state,
)
for label, conditional in self.conditions.items()
if conditional.cond.describe_unhealthy_state()
]
descriptions = [
f"{item.label} - {item.state.name} state: Expected {item.desc} to be {item.desired} but instead found{item.unhealthy_states}"
for item in descriptions
]
if descriptions:
return reduce(aggregate_string, descriptions)
return description
[docs] def state(
self, get_function: Callable[..., Union[ListInspection, ListInspections]], *get_args: Any
):
if not self._constructed_expection_description:
raise Exception("malformed construction, first call expect")
self._constructed_expection_getter = Getter(get_function, *get_args)
return self
[docs] def to_be(self, *desired_values: str):
desc = self._constructed_expection_description
getter = self._constructed_expection_getter
label = self._constructed_expectation_label
dim = self._constructed_expectation_dim
if not getter:
raise Exception("malformed construction, first call state and or expect")
self.load_condition(desc, Condition(getter, *desired_values), label, dim)
[docs]class NotReady(Exception):
def __init__(self, message: str, result: Readiness) -> None:
self.result = result
super().__init__(message)