import inspect import types from BreCal.database.enums import ParticipantType, ShipcallType, ParticipantwiseTimeDelta import numpy as np import pandas as pd from BreCal.validators.time_logic import TimeLogic from BreCal.database.enums import StatusFlags #from BreCal.validators.schema_validation import validation_state_and_validation_name class ValidationRuleBaseFunctions(): """ Base object with individual functions, which the {ValidationRuleFunctions}-child refers to. This parent class provides base functions and helps to restructure the code in a more comprehensible way. """ def __init__(self, sql_handler): self.sql_handler = sql_handler self.time_logic = TimeLogic() def check_time_delta_violation_query_time_to_now(self, query_time:pd.Timestamp, key_time:pd.Timestamp, threshold:float)->bool: """ # base function for all validation rules in the group {0001} A-L measures the time between NOW and query_time. When the query_time lays in the past, the delta is negative when the query_time lays in the future, the delta is positive returns a violation state depending on whether the delta is Violation, if: 0 >= delta > threshold When the key time is defined (not None), there is no violation. Returns False options: query_time: will be used to measure the time difference of 'now' until the query time key_time: will be used to check, whether the respective key already has a value threshold: threshold where a time difference becomes crucial. When the delta is below the threshold, a violation might occur """ # rule is not applicable -> return 'GREEN' if key_time is not None: return False # otherwise, this rule applies and the difference between 'now' and the query time is measured delta = self.time_logic.time_delta_from_now_to_tgt(tgt_time=query_time, unit="m") # a violation occurs, when the delta (in minutes) exceeds the specified threshold of a participant # to prevent past-events from triggering violations, negative values are ignored # Violation, if 0 >= delta >= threshold violation_state = (delta >= 0) and (delta<=threshold) return violation_state def check_participants_agree_on_estimated_time(self, shipcall, query, df_times, applicable_shipcall_type)->bool: """ # base function for all validation rules in the group {0002} A-C compares, whether the participants agree on the estimated time (of arrival or departure), depending on whether the shipcall type is incoming, outgoing or shifting. No violations are observed, when - the shipcall belongs to a different type than the rule expects - there are no matching times for the provided {query} (e.g., "eta_berth") Instead of comparing each individual result, this function counts the amount of unique instances. When there is not only one unique value, there are deviating time estimates, and a violation occurs returns: violation_state (bool) """ # shipcall type filter: consider only shipcalls, where the type matches if shipcall.type != applicable_shipcall_type.value: violation_state = False return violation_state # filter by participant types of interest (agency, mooring, portauthority/administration, pilot, tug) participant_types = [ParticipantType.AGENCY.value, ParticipantType.MOORING.value, ParticipantType.PORT_ADMINISTRATION.value, ParticipantType.PILOT.value, ParticipantType.TUG.value] df_times = df_times.loc[df_times["participant_type"].isin(participant_types),:] # exclude missing entries df_times.loc[~df_times[query].isnull(),:] # when there are no entries left (no entries are provided), skip if len(df_times)==0: violation_state = False return violation_state # there should only be one eta_berth, when all participants have provided the same time # this equates to the same criteria as checking, whether # times_agency.eta_berth==times_mooring.eta_berth==times_portadministration.eta_berth==times_pilot.eta_berth==times_tug.eta_berth unique_times = len(pd.unique(df_times.loc[:,query])) violation_state = unique_times!=1 return violation_state def check_unique_shipcall_counts(self, query:str, rounding="min", maximum_threshold=3)->bool: """ # base function for all validation rules in the group {0005} A&B compares how many unique times are found for the provided {query} (e.g., "eta_berth") This function rounds the results, counts the unique values and returns a boolean state, whether the {maximum_threshold} is exceeded """ # filter the df: keep only times_agents # filter out all NaN and NaT entries times_agency = self.sql_handler.get_times_for_agency(non_null_column=query) # get values and optionally round the values (values, unique, counts) = self.sql_handler.get_unique_ship_counts(all_df_times=times_agency, query=query, rounding=rounding, maximum_threshold=maximum_threshold) # when ANY of the unique values exceeds the threshold, a violation is observed violation_state = np.any(np.greater(counts, maximum_threshold)) return violation_state class ValidationRuleFunctions(ValidationRuleBaseFunctions): """ an accumulation object that makes sure, that any validation rule is translated to a function with default naming convention and return types. Each function should return a ValidationRuleState enumeration object and a description string to which validation rule the result belongs. These are returned as tuples (ValidationRuleState, validation_name) Each rule should have the same input arguments (self, shipcall, df_times, *args, **kwargs) The object makes heavy use of calls from an SQLHandler object, which provides functions for dataframe access and filtering. each validation_name is generated by calling the function inside a method validation_name = inspect.currentframe().f_code.co_name # validation_name then returns the name of the method from where 'currentframe()' was called. # example: #def validation_rule_fct_example(self, shipcall, df_times): #validation_name = inspect.currentframe().f_code.co_name #return (ValidationRuleState.NONE, validation_name) """ def __init__(self, sql_handler): super().__init__(sql_handler) return def get_validation_rule_functions(self): """return a list of all methods in this object, which are all validation rule functions.""" return [self.__getattribute__(mthd_) for mthd_ in dir(self) if ('validation_rule_fct' in mthd_) and (isinstance(self.__getattribute__(mthd_), types.MethodType))] def validation_rule_fct_missing_time_agency_berth_eta(self, shipcall, df_times, *args, **kwargs): """ Code: #0001-A Type: Local Rule Description: this validation checks, whether there is a missing time. When the difference between an event (e.g., the shipcall eta) is below a certain threshold (e.g., 20 hours), a violation occurs 0001-A: - Checks, if times_agency.eta_berth is filled in. - Measures the difference between 'now' and 'shipcall.eta'. """ # check, if the header is filled in (agency) if len(df_times.loc[df_times["participant_type"].isin([ParticipantType.AGENCY.value])]) != 1: return (StatusFlags.GREEN, None) # preparation: obtain the correct times of the participant, define the query time and the key time times_agency = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.AGENCY.value) query_time = shipcall.eta key_time = times_agency.eta_berth threshold = ParticipantwiseTimeDelta.AGENCY violation_state = self.check_time_delta_violation_query_time_to_now(query_time=query_time, key_time=key_time, threshold=threshold) if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.YELLOW, validation_name) else: return (StatusFlags.GREEN, None) def validation_rule_fct_missing_time_agency_berth_etd(self, shipcall, df_times, *args, **kwargs): """ Code: #0001-B Type: Local Rule Description: this validation checks, whether there is a missing time. When the difference between an event (e.g., the shipcall eta) is below a certain threshold (e.g., 20 hours), a violation occurs 0001-B: - Checks, if times_agency.etd_berth is filled in. - Measures the difference between 'now' and 'shipcall.etd'. """ # check, if the header is filled in (agency) if len(df_times.loc[df_times["participant_type"].isin([ParticipantType.AGENCY.value])]) != 1: return (StatusFlags.GREEN, None) # preparation: obtain the correct times of the participant, define the query time and the key time times_agency = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.AGENCY.value) query_time = shipcall.etd key_time = times_agency.etd_berth threshold = ParticipantwiseTimeDelta.AGENCY violation_state = self.check_time_delta_violation_query_time_to_now(query_time=query_time, key_time=key_time, threshold=threshold) if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.YELLOW, validation_name) else: return (StatusFlags.GREEN, None) def validation_rule_fct_missing_time_mooring_berth_eta(self, shipcall, df_times, *args, **kwargs): """ Code: #0001-C Type: Local Rule Description: this validation checks, whether there is a missing time. When the difference between an event (e.g., the shipcall eta) is below a certain threshold (e.g., 20 hours), a violation occurs 0001-C: - Checks, if times_mooring.eta_berth is filled in. - Measures the difference between 'now' and 'times_agency.eta_berth'. """ # check, if the header is filled in (agency & MOORING) if len(df_times.loc[df_times["participant_type"].isin([ParticipantType.AGENCY.value, ParticipantType.MOORING.value])]) != 2: return (StatusFlags.GREEN, None) # preparation: obtain the correct times of the participant, define the query time and the key time times_agency = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.AGENCY.value) times_mooring = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.MOORING.value) query_time = times_agency.eta_berth key_time = times_mooring.eta_berth threshold = ParticipantwiseTimeDelta.MOORING violation_state = self.check_time_delta_violation_query_time_to_now(query_time=query_time, key_time=key_time, threshold=threshold) if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.YELLOW, validation_name) else: return (StatusFlags.GREEN, None) def validation_rule_fct_missing_time_mooring_berth_etd(self, shipcall, df_times, *args, **kwargs): """ Code: #0001-D Type: Local Rule Description: this validation checks, whether there is a missing time. When the difference between an event (e.g., the shipcall eta) is below a certain threshold (e.g., 20 hours), a violation occurs 0001-D: - Checks, if times_mooring.etd_berth is filled in. - Measures the difference between 'now' and 'times_agency.etd_berth'. """ # check, if the header is filled in (agency & MOORING) if len(df_times.loc[df_times["participant_type"].isin([ParticipantType.AGENCY.value, ParticipantType.MOORING.value])]) != 2: return (StatusFlags.GREEN, None) # preparation: obtain the correct times of the participant, define the query time and the key time times_agency = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.AGENCY.value) times_mooring = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.MOORING.value) query_time = times_agency.etd_berth key_time = times_mooring.etd_berth threshold = ParticipantwiseTimeDelta.MOORING violation_state = self.check_time_delta_violation_query_time_to_now(query_time=query_time, key_time=key_time, threshold=threshold) if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.YELLOW, validation_name) else: return (StatusFlags.GREEN, None) def validation_rule_fct_missing_time_portadministration_berth_eta(self, shipcall, df_times, *args, **kwargs): """ Code: #0001-F Type: Local Rule Description: this validation checks, whether there is a missing time. When the difference between an event (e.g., the shipcall eta) is below a certain threshold (e.g., 20 hours), a violation occurs 0001-F: - Checks, if times_port_administration.eta_berth is filled in. - Measures the difference between 'now' and 'times_agency.eta_berth'. """ # check, if the header is filled in (agency & PORT_ADMINISTRATION) if len(df_times.loc[df_times["participant_type"].isin([ParticipantType.AGENCY.value, ParticipantType.PORT_ADMINISTRATION.value])]) != 2: return (StatusFlags.GREEN, None) # preparation: obtain the correct times of the participant, define the query time and the key time times_agency = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.AGENCY.value) times_port_administration = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.PORT_ADMINISTRATION.value) query_time = times_agency.eta_berth key_time = times_port_administration.eta_berth threshold = ParticipantwiseTimeDelta.PORT_ADMINISTRATION violation_state = self.check_time_delta_violation_query_time_to_now(query_time=query_time, key_time=key_time, threshold=threshold) if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.YELLOW, validation_name) else: return (StatusFlags.GREEN, None) def validation_rule_fct_missing_time_portadministration_berth_etd(self, shipcall, df_times, *args, **kwargs): """ Code: #0001-G Type: Local Rule Description: this validation checks, whether there is a missing time. When the difference between an event (e.g., the shipcall eta) is below a certain threshold (e.g., 20 hours), a violation occurs 0001-G: - Checks, if times_port_administration.etd_berth is filled in. - Measures the difference between 'now' and 'times_agency.etd_berth'. """ # check, if the header is filled in (agency & PORT_ADMINISTRATION) if len(df_times.loc[df_times["participant_type"].isin([ParticipantType.AGENCY.value, ParticipantType.PORT_ADMINISTRATION.value])]) != 2: return (StatusFlags.GREEN, None) # preparation: obtain the correct times of the participant, define the query time and the key time times_agency = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.AGENCY.value) times_port_administration = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.PORT_ADMINISTRATION.value) query_time = times_agency.etd_berth key_time = times_port_administration.etd_berth threshold = ParticipantwiseTimeDelta.PORT_ADMINISTRATION violation_state = self.check_time_delta_violation_query_time_to_now(query_time=query_time, key_time=key_time, threshold=threshold) if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.YELLOW, validation_name) else: return (StatusFlags.GREEN, None) def validation_rule_fct_missing_time_pilot_berth_eta(self, shipcall, df_times, *args, **kwargs): """ Code: #0001-H Type: Local Rule Description: this validation checks, whether there is a missing time. When the difference between an event (e.g., the shipcall eta) is below a certain threshold (e.g., 20 hours), a violation occurs 0001-H: - Checks, if times_pilot.eta_berth is filled in. - Measures the difference between 'now' and 'times_agency.eta_berth'. """ # check, if the header is filled in (agency & PILOT) if len(df_times.loc[df_times["participant_type"].isin([ParticipantType.AGENCY.value, ParticipantType.PILOT.value])]) != 2: return (StatusFlags.GREEN, None) # preparation: obtain the correct times of the participant, define the query time and the key time times_agency = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.AGENCY.value) times_pilot = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.PILOT.value) query_time = times_agency.eta_berth key_time = times_pilot.eta_berth threshold = ParticipantwiseTimeDelta.PILOT violation_state = self.check_time_delta_violation_query_time_to_now(query_time=query_time, key_time=key_time, threshold=threshold) if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.YELLOW, validation_name) else: return (StatusFlags.GREEN, None) def validation_rule_fct_missing_time_pilot_berth_etd(self, shipcall, df_times, *args, **kwargs): """ Code: #0001-I Type: Local Rule Description: this validation checks, whether there is a missing time. When the difference between an event (e.g., the shipcall eta) is below a certain threshold (e.g., 20 hours), a violation occurs 0001-I: - Checks, if times_pilot.etd_berth is filled in. - Measures the difference between 'now' and 'times_agency.etd_berth'. """ # check, if the header is filled in (agency & PILOT) if len(df_times.loc[df_times["participant_type"].isin([ParticipantType.AGENCY.value, ParticipantType.PILOT.value])]) != 2: return (StatusFlags.GREEN, None) # preparation: obtain the correct times of the participant, define the query time and the key time times_agency = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.AGENCY.value) times_pilot = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.PILOT.value) query_time = times_agency.etd_berth key_time = times_pilot.etd_berth threshold = ParticipantwiseTimeDelta.PILOT violation_state = self.check_time_delta_violation_query_time_to_now(query_time=query_time, key_time=key_time, threshold=threshold) if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.YELLOW, validation_name) else: return (StatusFlags.GREEN, None) def validation_rule_fct_missing_time_tug_berth_eta(self, shipcall, df_times, *args, **kwargs): """ Code: #0001-J Type: Local Rule Description: this validation checks, whether there is a missing time. When the difference between an event (e.g., the shipcall eta) is below a certain threshold (e.g., 20 hours), a violation occurs 0001-J: - Checks, if times_tug.eta_berth is filled in. - Measures the difference between 'now' and 'times_agency.eta_berth'. """ # check, if the header is filled in (agency & TUG) if len(df_times.loc[df_times["participant_type"].isin([ParticipantType.AGENCY.value, ParticipantType.TUG.value])]) != 2: return (StatusFlags.GREEN, None) # preparation: obtain the correct times of the participant, define the query time and the key time times_agency = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.AGENCY.value) times_tug = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.TUG.value) query_time = times_agency.eta_berth key_time = times_tug.eta_berth threshold = ParticipantwiseTimeDelta.TUG violation_state = self.check_time_delta_violation_query_time_to_now(query_time=query_time, key_time=key_time, threshold=threshold) if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.YELLOW, validation_name) else: return (StatusFlags.GREEN, None) def validation_rule_fct_missing_time_tug_berth_etd(self, shipcall, df_times, *args, **kwargs): """ Code: #0001-K Type: Local Rule Description: this validation checks, whether there is a missing time. When the difference between an event (e.g., the shipcall eta) is below a certain threshold (e.g., 20 hours), a violation occurs 0001-K: - Checks, if times_tug.etd_berth is filled in. - Measures the difference between 'now' and 'times_agency.etd_berth'. """ # check, if the header is filled in (agency & TUG) if len(df_times.loc[df_times["participant_type"].isin([ParticipantType.AGENCY.value, ParticipantType.TUG.value])]) != 2: return (StatusFlags.GREEN, None) # preparation: obtain the correct times of the participant, define the query time and the key time times_agency = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.AGENCY.value) times_tug = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.TUG.value) query_time = times_agency.etd_berth key_time = times_tug.etd_berth threshold = ParticipantwiseTimeDelta.TUG violation_state = self.check_time_delta_violation_query_time_to_now(query_time=query_time, key_time=key_time, threshold=threshold) if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.YELLOW, validation_name) else: return (StatusFlags.GREEN, None) def validation_rule_fct_missing_time_terminal_berth_eta(self, shipcall, df_times, *args, **kwargs): """ Code: #0001-L Type: Local Rule Description: this validation checks, whether there is a missing time. When the difference between an event (e.g., the shipcall eta) is below a certain threshold (e.g., 20 hours), a violation occurs 0001-L: - Checks, if times_terminal.eta_berth is filled in. - Measures the difference between 'now' and 'times_agency.eta_berth'. """ # check, if the header is filled in (agency & terminal) if len(df_times.loc[df_times["participant_type"].isin([ParticipantType.AGENCY.value, ParticipantType.TERMINAL.value])]) != 2: return (StatusFlags.GREEN, None) # preparation: obtain the correct times of the participant, define the query time and the key time times_agency = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.AGENCY.value) times_terminal = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.TERMINAL.value) query_time = times_agency.eta_berth key_time = times_terminal.eta_berth threshold = ParticipantwiseTimeDelta.TERMINAL violation_state = self.check_time_delta_violation_query_time_to_now(query_time=query_time, key_time=key_time, threshold=threshold) if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.YELLOW, validation_name) else: return (StatusFlags.GREEN, None) def validation_rule_fct_missing_time_terminal_berth_etd(self, shipcall, df_times, *args, **kwargs): """ Code: #0001-K Type: Local Rule Description: this validation checks, whether there is a missing time. When the difference between an event (e.g., the shipcall eta) is below a certain threshold (e.g., 20 hours), a violation occurs 0001-K: - Checks, if times_terminal.etd_berth is filled in. - Measures the difference between 'now' and 'times_agency.etd_berth'. """ # check, if the header is filled in (agency & terminal) if len(df_times.loc[df_times["participant_type"].isin([ParticipantType.AGENCY.value, ParticipantType.TERMINAL.value])]) != 2: return (StatusFlags.GREEN, None) # preparation: obtain the correct times of the participant, define the query time and the key time times_agency = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.AGENCY.value) times_terminal = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.TERMINAL.value) query_time = times_agency.etd_berth key_time = times_terminal.etd_berth threshold = ParticipantwiseTimeDelta.TERMINAL violation_state = self.check_time_delta_violation_query_time_to_now(query_time=query_time, key_time=key_time, threshold=threshold) if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.YELLOW, validation_name) else: return (StatusFlags.GREEN, None) def validation_rule_fct_shipcall_incoming_participants_disagree_on_eta(self, shipcall, df_times, *args, **kwargs): """ Code: #0002-A Type: Local Rule Description: this validation checks, whether the participants expect different ETA times Filter: only applies to incoming shipcalls """ query = "eta_berth" violation_state = self.check_participants_agree_on_estimated_time( shipcall = shipcall, query=query, df_times=df_times, applicable_shipcall_type=ShipcallType.INCOMING ) if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.RED, validation_name) else: return (StatusFlags.GREEN, None) def validation_rule_fct_shipcall_outgoing_participants_disagree_on_etd(self, shipcall, df_times, *args, **kwargs): """ Code: #0002-B Type: Local Rule Description: this validation checks, whether the participants expect different ETA times Filter: only applies to outgoing shipcalls """ query = "etd_berth" violation_state = self.check_participants_agree_on_estimated_time( shipcall = shipcall, query=query, df_times=df_times, applicable_shipcall_type=ShipcallType.OUTGOING ) if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.RED, validation_name) else: return (StatusFlags.GREEN, None) def validation_rule_fct_shipcall_shifting_participants_disagree_on_eta_or_etd(self, shipcall, df_times, *args, **kwargs): """ Code: #0002-C Type: Local Rule Description: this validation checks, whether the participants expect different ETA or ETD times Filter: only applies to shifting shipcalls """ violation_state_eta = self.check_participants_agree_on_estimated_time( shipcall = shipcall, query="eta_berth", df_times=df_times, applicable_shipcall_type=ShipcallType.SHIFTING ) violation_state_etd = self.check_participants_agree_on_estimated_time( shipcall = shipcall, query="etd_berth", df_times=df_times, applicable_shipcall_type=ShipcallType.SHIFTING ) # apply 'eta_berth' check # apply 'etd_berth' # violation: if either 'eta_berth' or 'etd_berth' is violated # functionally, this is the same as individually comparing all times for the participants # times_agency.eta_berth==times_mooring.eta_berth==times_portadministration.eta_berth==times_pilot.eta_berth==times_tug.eta_berth # times_agency.etd_berth==times_mooring.etd_berth==times_portadministration.etd_berth==times_pilot.etd_berth==times_tug.etd_berth violation_state = (violation_state_eta) or (violation_state_etd) if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.RED, validation_name) else: return (StatusFlags.GREEN, None) def validation_rule_fct_eta_time_not_in_operation_window(self, shipcall, df_times, *args, **kwargs): """ Code: #0003-A Type: Local Rule Description: this validation checks, whether the ETA time is between the provided operations window of the terminal query time: eta_berth (times_agency) start_time & end_time: operations_start & operations_end (times_terminal) """ # check, if the header is filled in (agency & terminal) if len(df_times.loc[df_times["participant_type"].isin([ParticipantType.AGENCY.value, ParticipantType.TERMINAL.value])]) != 2: return (StatusFlags.GREEN, None) # get agency & terminal times times_agency = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.AGENCY.value) times_terminal = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.TERMINAL.value) if (times_terminal.operations_end is pd.NaT) or (times_agency.etd_berth is pd.NaT): return (StatusFlags.GREEN, None) # check, whether the start of operations is AFTER the estimated arrival time violation_state = times_terminal.operations_start times_agency.etd_berth if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.RED, validation_name) else: return (StatusFlags.GREEN, None) def validation_rule_fct_eta_time_not_in_tidal_window(self, shipcall, df_times, *args, **kwargs): """ Code: #0004-A Type: Local Rule Description: this validation checks, whether the ETA time is between the provided tidal window query time: eta_berth (times_agency) start_time & end_time: tidal_window_from & tidal_window_to (shipcall) """ # check, if the header is filled in (agency) if len(df_times.loc[df_times["participant_type"].isin([ParticipantType.AGENCY.value])]) != 1: return (StatusFlags.GREEN, None) times_agency = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.AGENCY.value) # requirements: tidal window (from & to) is filled in if (shipcall.tidal_window_from is pd.NaT) or (shipcall.tidal_window_to is pd.NaT) or (df_times.eta_berth is pd.NaT): return (StatusFlags.GREEN, None) # check, whether the query time is between start & end time # a violation is observed, when the is NOT between start & end violation_state = not self.time_logic.time_inbetween(query_time=times_agency.eta_berth, start_time=shipcall.tidal_window_from, end_time=shipcall.tidal_window_to) if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.RED, validation_name) else: return (StatusFlags.GREEN, None) def validation_rule_fct_etd_time_not_in_tidal_window(self, shipcall, df_times, *args, **kwargs): """ Code: #0004-B Type: Local Rule Description: this validation checks, whether the ETD time is between the provided tidal window query time: eta_berth (times_agency) start_time & end_time: tidal_window_from & tidal_window_to (shipcall) """ # check, if the header is filled in (agency) if len(df_times.loc[df_times["participant_type"].isin([ParticipantType.AGENCY.value])]) != 1: return (StatusFlags.GREEN, None) times_agency = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.AGENCY.value) # requirements: tidal window (from & to) is filled in if (shipcall.tidal_window_from is pd.NaT) or (shipcall.tidal_window_to is pd.NaT) or (df_times.eta_berth is pd.NaT): return (StatusFlags.GREEN, None) # check, whether the query time is between start & end time # a violation is observed, when the is NOT between start & end violation_state = not self.time_logic.time_inbetween(query_time=times_agency.etd_berth, start_time=shipcall.tidal_window_from, end_time=shipcall.tidal_window_to) if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.RED, validation_name) else: return (StatusFlags.GREEN, None) def validation_rule_fct_too_many_identical_eta_times(self, shipcall, df_times, rounding = "min", maximum_threshold = 3, *args, **kwargs): """ Code: #0005-A Type: Global Rule Description: this validation rule checks, whether there are too many shipcalls with identical ETA times. """ # when ANY of the unique values exceeds the threshold, a violation is observed query = "eta_berth" violation_state = self.check_unique_shipcall_counts(query, rounding=rounding, maximum_threshold=maximum_threshold) if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.YELLOW, validation_name) else: return (StatusFlags.GREEN, None) def validation_rule_fct_too_many_identical_etd_times(self, shipcall, df_times, rounding = "min", maximum_threshold = 3, *args, **kwargs): """ Code: #0005-B Type: Global Rule Description: this validation rule checks, whether there are too many shipcalls with identical ETD times. """ # when ANY of the unique values exceeds the threshold, a violation is observed query = "etd_berth" violation_state = self.check_unique_shipcall_counts(query, rounding=rounding, maximum_threshold=maximum_threshold) if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.YELLOW, validation_name) else: return (StatusFlags.GREEN, None) def validation_rule_fct_agency_and_terminal_berth_id_disagreement(self, shipcall, df_times, *args, **kwargs): """ Code: #0006-A Type: Local Rule Description: This validation rule checks, whether agency and terminal agree with their designated berth place by checking berth_id. """ # check, if the header is filled in (agency & terminal) if len(df_times.loc[df_times["participant_type"].isin([ParticipantType.AGENCY.value, ParticipantType.TERMINAL.value])]) != 2: return (StatusFlags.GREEN, None) times_agency = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.AGENCY.value) times_terminal = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.TERMINAL.value) violation_state = times_agency.berth_id!=times_terminal.berth_id if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.YELLOW, validation_name) else: return (StatusFlags.GREEN, None) def validation_rule_fct_agency_and_terminal_pier_side_disagreement(self, shipcall, df_times, *args, **kwargs): """ Code: #0006-B Type: Local Rule Description: This validation rule checks, whether agency and terminal agree with their designated pier side by checking pier_side. """ # check, if the header is filled in (agency & terminal) if len(df_times.loc[df_times["participant_type"].isin([ParticipantType.AGENCY.value, ParticipantType.TERMINAL.value])]) != 2: return (StatusFlags.GREEN, None) times_agency = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.AGENCY.value) times_terminal = self.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.TERMINAL.value) violation_state = times_agency.pier_side!=times_terminal.pier_side if violation_state: validation_name = inspect.currentframe().f_code.co_name return (StatusFlags.YELLOW, validation_name) else: return (StatusFlags.GREEN, None)