129 lines
7.0 KiB
Python
129 lines
7.0 KiB
Python
import copy
|
|
import numpy as np
|
|
import pandas as pd
|
|
from BreCal.database.enums import StatusFlags
|
|
from BreCal.validators.validation_rule_functions import ValidationRuleFunctions
|
|
from BreCal.schemas.model import Shipcall
|
|
|
|
|
|
class ValidationRules(ValidationRuleFunctions):
|
|
"""
|
|
An object that determines the traffic light state for validation and notification. The provided feedback ('green', 'yellow', 'red')
|
|
determines, whether the state is critical. It uses ValidationRuleState enumerations.
|
|
In case of a critical validation state, the user's input prompt may be interrupted and the user may be warned.
|
|
In case of a critical notification state, the respective users will be automatically notified after n seconds. (#TODO_n_seconds_delay)
|
|
"""
|
|
def __init__(self, sql_handler): # use the entire data that is provided for this query (e.g., json input)
|
|
super().__init__(sql_handler)
|
|
|
|
self.validation_state = self.determine_validation_state()
|
|
# currently flagged: notification_state initially was based on using one ValidationRules object for each query. This is deprecated.
|
|
# self.notification_state = self.determine_notification_state() # (state:str, should_notify:bool)
|
|
return
|
|
|
|
def evaluate(self, shipcall):
|
|
"""
|
|
1.) prepare df_times, which every validation rule tends to use
|
|
calling this only once saves a lot of computational overhead
|
|
2.) apply all validation rules
|
|
returns: (evaluation_state, violations)
|
|
"""
|
|
# prepare df_times, which every validation rule tends to use
|
|
df_times = self.sql_handler.df_dict.get('times') # -> pd.DataFrame
|
|
|
|
if len(df_times)==0:
|
|
return (StatusFlags.GREEN.value, [])
|
|
|
|
# filter by shipcall id
|
|
df_times = self.sql_handler.get_times_of_shipcall(shipcall)
|
|
|
|
# apply all validation rules
|
|
# list of tuples, where each element is (state, msg)
|
|
evaluation_results = [elem(shipcall, df_times) for elem in self.get_validation_rule_functions()]
|
|
|
|
# filter out all 'None' results, which indicate that no violation occured.
|
|
evaluation_results = [evaluation_result for evaluation_result in evaluation_results if evaluation_result[1] is not None]
|
|
|
|
# 'translate' all error codes into readable, human-understandable format.
|
|
evaluation_results = [(state, self.describe_error_message(msg)) for (state, msg) in evaluation_results]
|
|
|
|
# check, what the maximum state flag is and return it
|
|
evaluation_state = np.max(np.array([result[0].value for result in evaluation_results])) if len(evaluation_results)>0 else StatusFlags.GREEN.value
|
|
evaluation_verbosity = [result[1] for result in evaluation_results]
|
|
return (evaluation_state, evaluation_verbosity)
|
|
|
|
def evaluation_verbosity(self, evaluation_state, evaluation_results):
|
|
"""This function suggestions verbosity for the evaluation results. Based on 'True'/'False' evaluation outcome, the returned string is different."""
|
|
if evaluation_state:
|
|
return f"OK! The validation was successful. There are no rule violations."
|
|
else:
|
|
verbose_string = "These are:" + "\n\t".join(evaluation_results) # every element of the list will be displayed in a new line with a tab
|
|
return f"FAILED VALIDATION. There have been {len(evaluation_results)} violations. {verbose_string}"
|
|
|
|
def evaluate_shipcall_from_df(self, x):
|
|
shipcall = Shipcall(**{**{'id':x.name}, **x.to_dict()})
|
|
evaluation_state, violations = self.evaluate(shipcall)
|
|
return evaluation_state, violations
|
|
|
|
def evaluate_shipcalls(self, shipcall_df:pd.DataFrame)->pd.DataFrame:
|
|
"""apply 'evaluate_shipcall_from_df' to each individual shipcall in {shipcall_df}. Returns shipcall_df ('evaluation' and 'evaluation_message' are updated)"""
|
|
results = shipcall_df.apply(lambda x: self.evaluate_shipcall_from_df(x), axis=1).values
|
|
|
|
# unbundle individual results. evaluation_state becomes an integer, violation
|
|
evaluation_state = [StatusFlags(res[0]).value for res in results]
|
|
violations = [",".join(res[1]) if len(res[1])>0 else None for res in results]
|
|
|
|
shipcall_df.loc[:,"evaluation"] = evaluation_state
|
|
shipcall_df.loc[:,"evaluation_message"] = violations
|
|
return shipcall_df
|
|
|
|
def determine_validation_state(self) -> str:
|
|
"""
|
|
this method determines the validation state of a shipcall. The state is either ['green', 'yellow', 'red'] and signals,
|
|
whether an entry causes issues within the workflow of users.
|
|
|
|
returns: validation_state_new (str)
|
|
"""
|
|
(validation_state_new, description) = self.undefined_method()
|
|
# should there also be notifications for critical validation states? In principle, the traffic light itself provides that notification.
|
|
self.validation_state = validation_state_new
|
|
return validation_state_new
|
|
|
|
def determine_notification_state(self) -> (str, bool):
|
|
"""
|
|
this method determines state changes in the notification state. When the state is changed to yellow or red,
|
|
a user is notified about it. The only exception for this rule is when the state was yellow or red before,
|
|
as the user has then already been notified.
|
|
|
|
returns: notification_state_new (str), should_notify (bool)
|
|
"""
|
|
(state_new, description) = self.undefined_method() # determine the successor
|
|
should_notify = self.identify_notification_state_change(state_new)
|
|
self.notification_state = state_new # overwrite the predecessor
|
|
return state_new, should_notify
|
|
|
|
def identify_notification_state_change(self, state_new) -> bool:
|
|
"""
|
|
determines, whether the observed state change should trigger a notification.
|
|
internally, this function maps a color string to an integer and determines, if the successor state is more severe than the predecessor.
|
|
|
|
state changes trigger a notification in the following cases:
|
|
green -> yellow
|
|
green -> red
|
|
yellow -> red
|
|
|
|
(none -> yellow) or (none -> red)
|
|
due to the values in the enumeration objects, the states are mapped to provide this function.
|
|
green=1, yellow=2, red=3, none=1. Hence, critical changes can be observed by simply checking with "greater than".
|
|
|
|
returns bool, whether a notification should be triggered
|
|
"""
|
|
# state_old is always considered at least 'Green' (1)
|
|
state_old = max(copy.copy(self.notification_state) if "notification_state" in list(self.__dict__.keys()) else StatusFlags.NONE, StatusFlags.GREEN.value)
|
|
return state_new.value > state_old.value
|
|
|
|
def undefined_method(self) -> str:
|
|
"""this function should apply the ValidationRules to the respective .shipcall, in regards to .times"""
|
|
# #TODO_traffic_state
|
|
return (StatusFlags.GREEN, False) # (state:str, should_notify:bool)
|