adding schema_validation functions (Ship and Participant), time_logic (will be used in the validation_rules later), validation_rules (covers time-overlap handling). Further, creating simple unit tests for stubs (will be extended soon)

src/lib_brecal_utils/brecal_utils/__init__.py

@@ -1,11 +1,17 @@
 from ._version import __version__
 from brecal_utils.file_handling import get_project_root, ensure_path
 from brecal_utils.test_handling import execute_test_with_pytest, execute_coverage_test
+from brecal_utils.validators.time_logic import TimeLogic
+from brecal_utils.validators.validation_rules import ValidationRules
+from brecal_utils.validators.schema_validation import validation_state_and_validation_name
 
 __all__ = [
 	"get_project_root", 
 	"ensure_path", 
-	"execute_test_with_pytest", 
+	"execute_test_with_pytest",
 	"execute_coverage_test",
+    "TimeLogic",
+    "ValidationRules",
+    "validation_state_and_validation_name",
 ]
 

src/lib_brecal_utils/brecal_utils/validators/schema_validation.py

@@ -0,0 +1,131 @@
+# wrapper: every validation function returns a tuple of (validation_state, validation_name)
+# example: validate_ship_eni_length might return the tuple (True, 'ship_eni_length')
+# thereby, one could always know, which test causes an issue
+
+####################################### general functions #######################################
+
+def validation_state_and_validation_name(validation_name):
+    """
+    can wrap arbitrary functions, so they returned (output, validation_name)-tuples
+    usage example:
+        @validation_state_and_validation_name("ship_eni_length")
+        def validate_ship_eni_length(ship):
+            return length_matches_exactly(ship.eni,8)
+    """
+    def wrapper(validation_fct):
+        def decorated_fct(*args, **kwargs):
+            return (validation_fct(*args, **kwargs), validation_name)
+        return decorated_fct
+    return wrapper
+
+def value_in_range(query_value, start_range, end_range):
+    """determines, whether the query_value is greater than start_range, but smaller than end_range. Returns bool"""
+    return start_range<query_value<end_range
+
+def length_is_at_maximum(query_value, max_len):
+    """determines, whether the query_value's length is l<={max_len}. Returns bool"""
+    return len(str(query_value))<=max_len
+
+def length_matches_exactly(query_value, length_value):
+    """determines, whether the query_value's length is exactly l=={length_value}. Returns bool"""
+    return len(str(query_value)) == length_value
+
+
+####################################### dataclass specifics #######################################
+
+### Ship dataclass (BreCal.schema.model.Ship) ###
+
+@validation_state_and_validation_name("ship_bollard_pull")
+def ship_bollard_is_none_or_in_range(ship):
+    """a ship should either have its bollard_pull between 0 and 500, or have an undefined bollard_pull (when not a tug)"""
+    return (ship.bollard_pull is None) or (value_in_range(ship.bollard_pull, 0, 500))
+
+@validation_state_and_validation_name("ship_length")
+def ship_length_in_range(ship):
+    """ship length-values should be valid. between 0 and 500 meters is plausible. returns bool"""
+    return value_in_range(ship.length, 0, 500)
+
+@validation_state_and_validation_name("ship_width")
+def ship_width_in_range(ship):
+    """ship length-values should be valid. between 0 and 500 meters is plausible. returns bool"""
+    return value_in_range(ship.width, 0, 500)
+
+@validation_state_and_validation_name("ship_max_draft")
+def ship_max_draft_in_range(ship):
+    """ship max_draft-values should be valid. between 0 and 500 meters is plausible. returns bool"""
+    return value_in_range(ship.max_draft, 0, 20)
+
+@validation_state_and_validation_name("ship_eni_length")
+def ship_eni_len_is_eight(ship):
+    """eni-no. are standardized. They should have exactly eight characters. returns bool"""
+    return length_matches_exactly(ship.eni,8)
+
+@validation_state_and_validation_name("ship_imo_length")
+def ship_imo_len_is_seven(ship):
+    """IMO-numbers are standardized. They should have exactly seven characters. returns bool"""
+    return length_matches_exactly(ship.imo,7)
+
+@validation_state_and_validation_name("ship_callsign_length")
+def ship_callsign_len_is_seven_at_maximum(ship):
+    """the ship's callsign should have l<=7 characters. returns bool"""
+    return length_is_at_maximum(ship.callsign, 7)
+
+
+def ship_is_not_tug_or_key_is_defined(is_tug, key_):
+    """ # base function
+    function that checks, if a Ship dataclass is either
+        a) not a tug
+        b) has a defined value of {key_}
+    can be used for max_draft, participant_id and bollard_pull
+    """
+    return (not is_tug) or (key_ is not None)
+
+@validation_state_and_validation_name("ship_bollard_pull_dynamically_mandatory")
+def ship_bollard_pull_is_defined_or_is_not_tug(ship):
+    """
+    there are two valid cases for the bollard_pull:
+        a) bollard_pull is undefined (None), if the ship is not a tug
+        b) bollard_pull is defined, if the ship is a tug
+    if the ship is a tug, a separate function validates in addition, if the value is in an accepted range
+
+    returns bool
+    """
+    return ship_is_not_tug_or_key_is_defined(ship.is_tug, ship.bollard_pull)
+
+@validation_state_and_validation_name("ship_max_draft_dynamically_mandatory")
+def ship_max_draft_is_defined_or_is_not_tug(ship):
+    """
+    there are two valid cases for the max_draft:
+        a) max_draft is undefined (None), if the ship is not a tug
+        b) max_draft is defined, if the ship is a tug
+    if the ship is a tug, a separate function validates in addition, if the value is in an accepted range
+
+    returns bool
+    """
+    return ship_is_not_tug_or_key_is_defined(ship.is_tug, ship.max_draft)
+
+# #TODO_ship_tug_participant_id: is this semantically correct? Will the participant_id be entered or automatically filled?
+@validation_state_and_validation_name("ship_max_draft_dynamically_mandatory")
+def ship_participant_id_is_defined_or_is_not_tug(ship):
+    """
+    there are two valid cases for the max_draft:
+        a) participant_id is undefined (None), if the ship is not a tug
+        b) participant_id is defined, if the ship is a tug
+
+    returns bool
+    """
+    return ship_is_not_tug_or_key_is_defined(ship.is_tug, ship.participant_id)
+
+
+
+### Participant dataclass (BreCal.schema.model.Participant) ###
+
+@validation_state_and_validation_name("participant_postal_code_length")
+def participant_postal_code_len_is_five(participant):
+    """
+    validates, that a postal code has 5 characters. returns bool
+
+    # #TODO_postal_code_length_validation: might make sense to request postal_code<=5 characters
+    # is the 5-character requirement true when international ships arive?
+    """
+    return length_matches_exactly(participant.postal_code, 5)

src/lib_brecal_utils/brecal_utils/validators/time_logic.py

@@ -0,0 +1,61 @@
+import datetime
+import numpy as np
+
+class TimeLogic():
+    def __init__(self):
+        return
+    
+    def time_delta(self, query_time, other_times):
+        return
+    
+    def time_inbetween(self, query_time:datetime.datetime, start_time:datetime.datetime, end_time:datetime.datetime) -> bool:
+        """
+        checks, whether the query time is inbetween the start & end time. Returns a bool to indicate that.
+
+        Example:
+            a = datetime.datetime(2017, 5, 16, 8, 21, 10)
+            b = datetime.datetime(2017, 5, 17, 8, 21, 10)
+            c = datetime.datetime(2017, 5, 18, 8, 21, 10)
+
+            is b between a and c? -> yes. Returns True
+            is c between a and b? -> no. Returns False
+
+        returns bool
+        """
+        assert isinstance(query_time, datetime.datetime)
+        assert isinstance(start_time, datetime.datetime)
+        assert isinstance(end_time, datetime.datetime)
+
+        return start_time <= query_time <= end_time
+    
+    def time_inbetween_absolute_delta(self, query_time:datetime.datetime, start_time:datetime.datetime, end_time:datetime.datetime) -> tuple:
+        """
+        similarly to self.time_inbetween, this function compares a query_time with the provided start and end time.
+        however, this function instead returns timedelta objects, which show the difference towards start and end
+
+        this function applies abs() to return only absolute deviations. Thereby, -23 becomes +23
+
+        returns: tuple(absolute_start_delta, absolute_end_delta)
+        """
+        return (abs(query_time-start_time), abs(query_time-end_time))
+    
+    def compare_query_is_inbetween_list(self, query_time, list_of_other_times) -> list:
+        list_of_bools = [
+            self.time_inbetween(query_time, time_elem_begin, time_elem_end)
+            for (time_elem_begin, time_elem_end) in list_of_other_times
+        ]
+        return list_of_bools
+    
+    def query_time_any_inbetween(self, query_time, list_of_other_times):
+        """
+        given a query_time element, the element will be compared to every element in a list, where each
+        element is a tuple of (start_time, end_time)
+        """
+        if len(list_of_other_times)==0:
+            # the time is not inbetween, if the provided list is empty
+            return False
+        
+        list_of_bools = self.compare_query_is_inbetween_list(query_time, list_of_other_times)
+        return np.any(list_of_bools), list_of_bools
+
+    

src/lib_brecal_utils/brecal_utils/validators/validation_rules.py

@@ -0,0 +1,66 @@
+import copy
+from brecal_utils.validators.time_logic import TimeLogic
+
+class ValidationRules():
+    """
+    An object that determines the traffic light state for validation and notification. The provided str feedback ('green', 'yellow', 'red')
+    determines, whether the state is critical. 
+        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, shipcall, times): # use the entire data that is provided for this query (e.g., json input)
+        self.time_logic = TimeLogic()
+        self.shipcall = shipcall
+        self.times = times
+
+        self.validation_state = self.determine_validation_state()
+        self.notification_state = self.determine_notification_state()
+        return
+    
+    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 = 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 = self.undefined_method() # determien 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)
+
+        returns bool, whether a notification should be triggered
+        """
+        state_old = copy.copy(self.notification_state) if "notification_state" in list(self.__dict__.keys()) else 'none'
+
+        state_mapping = {'none':0, 'green':0, 'yellow':1, 'red':2}
+        return state_mapping[state_new] > state_mapping[state_old]
+    
+    def undefined_method(self) -> str:
+        return 'green'
+

src/lib_brecal_utils/tests/stubs/test_stub_objects.py

@@ -0,0 +1,62 @@
+import pytest
+
+def test_build_stub_berth():
+    from BreCal.schemas.model import Berth
+    from brecal_utils.stubs.berth import get_berth_simple
+    berth = get_berth_simple()
+    assert isinstance(berth, Berth)
+    return
+
+def test_build_stub_participant():
+    from BreCal.schemas.model import Participant
+    from brecal_utils.stubs.participant import get_participant_simple
+    participant = get_participant_simple()
+    assert isinstance(participant, Participant)
+    return
+
+def test_build_stub_user():
+    from BreCal.schemas.model import User
+    from brecal_utils.stubs.user import get_user_simple
+    user = get_user_simple()
+    assert isinstance(user, User)
+    return
+
+def test_build_stub_ship():
+    from BreCal.schemas.model import Ship
+    from brecal_utils.stubs.ship import get_ship_simple
+    ship = get_ship_simple()
+    assert isinstance(ship, Ship)
+    return
+
+def test_build_stub_tug():
+    from brecal_utils.stubs.tug import get_tug_simple
+    with pytest.raises(ImportError):
+        from BreCal.schemas.model import Tug
+    with pytest.raises(NotImplementedError, match="unclarified"):
+        tug = get_tug_simple()
+        assert isinstance(tug, Tug)
+    return
+
+def test_build_stub_shipcall():
+    from BreCal.schemas.model import Shipcall
+    from brecal_utils.stubs.shipcall import get_shipcall_simple
+    shipcall = get_shipcall_simple()
+    assert isinstance(shipcall, Shipcall)
+    return
+
+def test_build_stub_times():
+    from BreCal.schemas.model import Times
+    from brecal_utils.stubs.times_full import get_times_full_simple
+    times = get_times_full_simple()
+    assert isinstance(times, Times)
+    return
+
+if __name__=="__main__":
+    test_build_stub_berth()
+    test_build_stub_participant()
+    test_build_stub_berth()
+    test_build_stub_user()
+    test_build_stub_ship()
+    test_build_stub_tug()
+    test_build_stub_shipcall()
+    test_build_stub_times()