from dataclasses import field, dataclass from marshmallow import Schema, fields, INCLUDE, ValidationError, validate, validates, post_load from marshmallow.fields import Field from marshmallow_enum import EnumField from enum import IntEnum from marshmallow_dataclass import dataclass from typing import List import json import re import datetime from BreCal.validators.time_logic import validate_time_is_in_not_too_distant_future from BreCal.validators.validation_base_utils import check_if_string_has_special_characters from BreCal.database.enums import ParticipantType, ParticipantFlag def obj_dict(obj): if isinstance(obj, datetime.datetime): return obj.isoformat() if hasattr(obj, 'to_json'): return obj.to_json() return obj.__dict__ @dataclass class Berth(Schema): id: int name: str lock: bool owner_id: int authority_id: int port_id: int created: datetime modified: datetime deleted: bool @dataclass class Port(Schema): id: int name: str locode: str created: datetime modified: datetime deleted: bool class OperationType(IntEnum): undefined = 0 insert = 1 update = 2 delete = 3 class ObjectType(IntEnum): undefined = 0 shipcall = 1 times = 2 class EvaluationType(IntEnum): undefined = 0 green = 1 yellow = 2 red = 3 @classmethod def _missing_(cls, value): return cls.undefined class NotificationType(IntEnum): """ This type is not the way the user is informed but the type of the notification, e.g. time conflict, time conflict resolved, etc. It can be understood as an event type """ assignment = 1 next24h = 2 time_conflict = 3 time_conflict_resolved = 4 unassigned = 5 missing_data = 6 cancelled = 7 @classmethod def _missing_(cls, value): return cls.undefined def bitflag_to_list(bitflag: int) -> list[NotificationType]: if bitflag is None: return [] """Converts an integer bitflag to a list of NotificationType enums.""" return [nt for nt in NotificationType if bitflag & (1 << (nt.value - 1))] def list_to_bitflag(notifications: fields.List) -> int: """Converts a list of NotificationType enums to an integer bitflag.""" try: iter(notifications) return sum(1 << (nt.value - 1) for nt in notifications) except TypeError as te: return 0 class ShipcallType(IntEnum): undefined = 0 arrival = 1 departure = 2 shifting = 3 @classmethod def _missing_(cls, value): return cls.undefined @dataclass class History: def __init__(self, id, participant_id, shipcall_id, timestamp, eta, type, operation): self.id = id self.participant_id = participant_id self.shipcall_id = shipcall_id self.timestamp = timestamp self.eta = eta self.type = type self.operation = operation pass id: int participant_id: int shipcall_id: int timestamp: datetime eta: datetime type: ObjectType operation: OperationType def to_json(self): return { "id": self.id, "participant_id": self.participant_id, "shipcall_id": self.shipcall_id, "timestamp": self.timestamp.isoformat() if self.timestamp else "", "eta": self.eta.isoformat() if self.eta else "", "type": self.type.name if isinstance(self.type, IntEnum) else ObjectType(self.type).name, "operation": self.operation.name if isinstance(self.operation, IntEnum) else OperationType(self.operation).name } @classmethod def from_query_row(self, id, participant_id, shipcall_id, timestamp, eta, type, operation): return self(id, participant_id, shipcall_id, timestamp, eta, ObjectType(type), OperationType(operation)) class Error(Schema): message = fields.String(required=True) class GetVerifyInlineResp(Schema): pass @dataclass class Notification: """ Base data class for any notification. Description: 'An entry corresponds to an alarm given by a violated rule during times update' """ id: int shipcall_id: int # 'shipcall record that caused the notification' participant_id: int # 'optional participant reference that needs to be specifically notified, if null all participants are notified' level: int # 'severity of the notification' type: NotificationType # 'type of the notification' message: str # 'individual message' created: datetime modified: datetime def to_json(self): return { "id": self.id, "shipcall_id": self.shipcall_id, "participant_id": self.participant_id, "level": self.level, "type": self.type.name if isinstance(self.type, IntEnum) else NotificationType(self.type).name, "message": self.message, "created": self.created.isoformat() if self.created else "", "modified": self.modified.isoformat() if self.modified else "" } @classmethod def from_query_row(self, id, shipcall_id, participant_id, level, type, message, created, modified): return self(id, shipcall_id, participant_id, level, NotificationType(type), message, created, modified) @dataclass class Participant(Schema): id: int name: str street: str postal_code: str city: str type: int # fields.Enum(ParticipantType ...) flags: int created: datetime modified: datetime deleted: bool ports: List[int] = field(default_factory=list) @validates("type") def validate_type(self, value): # e.g., when an IntFlag has the values 1,2,4; the maximum valid value is 7 max_int = sum([int(val) for val in list(ParticipantType._value2member_map_.values())]) min_int = 0 valid_type = 0 <= value < max_int if not valid_type: raise ValidationError({"type":f"the provided integer is not supported for default behaviour of the ParticipantType IntFlag. Your choice: {value}. Supported values are: 0 <= value {max_int}"}) @validates("flags") def validate_flags(self, value): # e.g., when an IntFlag has the values 1,2,4; the maximum valid value is 7 max_int = sum([int(val) for val in list(ParticipantFlag._value2member_map_.values())]) min_int = 0 valid_type = 0 <= value < max_int if not valid_type: raise ValidationError({"flags":f"the provided integer is not supported for default behaviour of the ParticipantFlag IntFlag. Your choice: {value}. Supported values are: 0 <= value {max_int}"}) class ParticipantList(Participant): pass class ParticipantAssignmentSchema(Schema): participant_id = fields.Integer() type = fields.Integer() class ShipcallSchema(Schema): def __init__(self): super().__init__(unknown=None) pass id = fields.Integer(required=True) ship_id = fields.Integer(required=True) port_id = fields.Integer(required=True) type = fields.Enum(ShipcallType, default=ShipcallType.undefined) eta = fields.DateTime(required=False, allow_none=True) voyage = fields.String(allow_none=True, required=False, validate=[validate.Length(max=16)]) etd = fields.DateTime(required=False, allow_none=True) arrival_berth_id = fields.Integer(required=False, allow_none=True) departure_berth_id = fields.Integer(required=False, allow_none=True) tug_required = fields.Bool(required=False, allow_none=True) pilot_required = fields.Bool(required=False, allow_none=True) flags = fields.Integer(required=False, allow_none=True) pier_side = fields.Bool(required=False, allow_none=True) bunkering = fields.Bool(required=False, allow_none=True) replenishing_terminal = fields.Bool(required=False, allow_none=True) replenishing_lock = fields.Bool(required=False, allow_none=True) draft = fields.Float(required=False, allow_none=True, validate=[validate.Range(min=0, max=20, min_inclusive=False, max_inclusive=True)]) tidal_window_from = fields.DateTime(required=False, allow_none=True) tidal_window_to = fields.DateTime(required=False, allow_none=True) rain_sensitive_cargo = fields.Bool(required=False, allow_none=True) recommended_tugs = fields.Integer(required=False, allow_none=True, validate=[validate.Range(min=0, max=10, min_inclusive=True, max_inclusive=True)]) anchored = fields.Bool(required=False, allow_none=True) moored_lock = fields.Bool(required=False, allow_none=True) canceled = fields.Bool(required=False, allow_none=True) evaluation = fields.Enum(EvaluationType, required=False, allow_none=True, default=EvaluationType.undefined) evaluation_message = fields.Str(allow_none=True, required=False) evaluation_time = fields.DateTime(required=False, allow_none=True) evaluation_notifications_sent = fields.Bool(required=False, allow_none=True) time_ref_point = fields.Integer(required=False, allow_none=True) participants = fields.List(fields.Nested(ParticipantAssignmentSchema)) created = fields.DateTime(required=False, allow_none=True) modified = fields.DateTime(required=False, allow_none=True) @post_load def make_shipcall(self, data, **kwargs): if 'type' in data: data['type_value'] = int(data['type']) else: data['type_value'] = int(ShipcallType.undefined) if 'evaluation' in data: if data['evaluation']: data['evaluation_value'] = int(data['evaluation']) else: data['evaluation_value'] = int(EvaluationType.undefined) return data @validates("type") def validate_type(self, value): valid_shipcall_type = int(value) in [item.value for item in ShipcallType] if not valid_shipcall_type: raise ValidationError({"type":f"the provided type is not a valid shipcall type."}) @dataclass class Participant_Assignment: def __init__(self, participant_id, type): self.participant_id = participant_id self.type = type pass participant_id: int type: int # a variant would be to use the IntFlag type (with appropriate serialization) def to_json(self): return self.__dict__ @dataclass class Port_Assignment: def __init__(self, port_id): self.port_id = port_id pass port_id: int def to_json(self): return self.__dict__ @dataclass class Shipcall: id: int ship_id: int type: ShipcallType eta: datetime voyage: str etd: datetime arrival_berth_id: int departure_berth_id: int tug_required: bool pilot_required: bool flags: int pier_side: bool # enumerator object in database/enum/PierSide bunkering: bool replenishing_terminal: bool replenishing_lock: bool draft: float tidal_window_from: datetime tidal_window_to: datetime rain_sensitive_cargo: bool recommended_tugs: int anchored: bool moored_lock: bool canceled: bool evaluation: EvaluationType evaluation_message: str evaluation_time: datetime evaluation_notifications_sent: bool time_ref_point: int port_id: int created: datetime modified: datetime participants: List[Participant_Assignment] = field(default_factory=list) def to_json(self): return { "id": self.id, "ship_id": self.ship_id, "type": self.type.name if isinstance(self.type, IntEnum) else ShipcallType(self.type).name, "eta": self.eta.isoformat() if self.eta else "", "voyage": self.voyage, "etd": self.etd.isoformat() if self.etd else "", "arrival_berth_id": self.arrival_berth_id, "departure_berth_id": self.departure_berth_id, "tug_required": self.tug_required, "pilot_required": self.pilot_required, "flags": self.flags, "pier_side": self.pier_side, "bunkering": self.bunkering, "replenishing_terminal": self.replenishing_terminal, "replenishing_lock": self.replenishing_lock, "draft": self.draft, "tidal_window_from": self.tidal_window_from.isoformat() if self.tidal_window_from else "", "tidal_window_to": self.tidal_window_to.isoformat() if self.tidal_window_to else "", "rain_sensitive_cargo": self.rain_sensitive_cargo, "recommended_tugs": self.recommended_tugs, "anchored": self.anchored, "moored_lock": self.moored_lock, "canceled": self.canceled, "evaluation": self.evaluation.name if isinstance(self.evaluation, IntEnum) else EvaluationType(self.evaluation).name, "evaluation_message": self.evaluation_message, "evaluation_time": self.evaluation_time.isoformat() if self.evaluation_time else "", "evaluation_notifications_sent": self.evaluation_notifications_sent, "time_ref_point": self.time_ref_point, "port_id": self.port_id, "created": self.created.isoformat() if self.created else "", "modified": self.modified.isoformat() if self.modified else "", "participants": [participant.__dict__ for participant in self.participants] } @classmethod def from_query_row(self, id, ship_id, type, eta, voyage, etd, arrival_berth_id, departure_berth_id, tug_required, pilot_required, flags, pier_side, bunkering, replenishing_terminal, replenishing_lock, draft, tidal_window_from, tidal_window_to, rain_sensitive_cargo, recommended_tugs, anchored, moored_lock, canceled, evaluation, evaluation_message, evaluation_time, evaluation_notifications_sent, time_ref_point, port_id, created, modified): return self(id, ship_id, ShipcallType(type), eta, voyage, etd, arrival_berth_id, departure_berth_id, tug_required, pilot_required, flags, pier_side, bunkering, replenishing_terminal, replenishing_lock, draft, tidal_window_from, tidal_window_to, rain_sensitive_cargo, recommended_tugs, anchored, moored_lock, canceled, EvaluationType(evaluation), evaluation_message, evaluation_time, evaluation_notifications_sent, time_ref_point, port_id, created, modified) class ShipcallId(Schema): pass # this is the way! class TimesSchema(Schema): def __init__(self): super().__init__(unknown=None) pass id = fields.Integer(required=False) eta_berth = fields.DateTime(required=False, allow_none=True) eta_berth_fixed = fields.Bool(required=False, allow_none=True) etd_berth = fields.DateTime(required=False, allow_none=True) etd_berth_fixed = fields.Bool(required=False, allow_none=True) lock_time = fields.DateTime(required=False, allow_none=True) lock_time_fixed = fields.Bool(required=False, allow_none=True) zone_entry = fields.DateTime(required=False, allow_none=True) zone_entry_fixed = fields.Bool(required=False, allow_none=True) operations_start = fields.DateTime(required=False, allow_none=True) operations_end = fields.DateTime(required=False, allow_none=True) remarks = fields.String(required=False, allow_none=True, validate=[validate.Length(max=512)]) participant_id = fields.Integer(required=True) berth_id = fields.Integer(required=False, allow_none = True) berth_info = fields.String(required=False, allow_none=True, validate=[validate.Length(max=512)]) pier_side = fields.Bool(required=False, allow_none = True) shipcall_id = fields.Integer(required=True) participant_type = fields.Integer(Required = False, allow_none=True)# TODO: could become Enum. # participant_type = fields.Enum(ParticipantType, required=False, allow_none=True, default=ParticipantType.undefined) #fields.Integer(required=False, allow_none=True) ata = fields.DateTime(required=False, allow_none=True) atd = fields.DateTime(required=False, allow_none=True) eta_interval_end = fields.DateTime(required=False, allow_none=True) etd_interval_end = fields.DateTime(required=False, allow_none=True) created = fields.DateTime(required=False, allow_none=True) modified = fields.DateTime(required=False, allow_none=True) @validates("participant_type") def validate_participant_type(self, value): # #TODO: it may also make sense to block multi-assignments, whereas a value could be BSMD+AGENCY # while the validation fails when one of those multi-assignments is BSMD, it passes in cases, # such as AGENCY+PILOT # a participant type should not be .BSMD if not isinstance(value, ParticipantType): value = ParticipantType(value) if ParticipantType.BSMD in value: raise ValidationError({"participant_type":f"the participant_type must not be .BSMD"}) @validates("eta_berth") def validate_eta_berth(self, value): # violation when time is not in the future, but also does not exceed a threshold for the 'reasonable' future # when 'value' is 'None', a ValidationError is not issued. valid_time = validate_time_is_in_not_too_distant_future(raise_validation_error=True, value=value, months=12) return @validates("etd_berth") def validate_etd_berth(self, value): # violation when time is not in the future, but also does not exceed a threshold for the 'reasonable' future # when 'value' is 'None', a ValidationError is not issued. valid_time = validate_time_is_in_not_too_distant_future(raise_validation_error=True, value=value, months=12) return @validates("lock_time") def validate_lock_time(self, value): # violation when time is not in the future, but also does not exceed a threshold for the 'reasonable' future # when 'value' is 'None', a ValidationError is not issued. valid_time = validate_time_is_in_not_too_distant_future(raise_validation_error=True, value=value, months=12) return @validates("zone_entry") def validate_zone_entry(self, value): # violation when time is not in the future, but also does not exceed a threshold for the 'reasonable' future # when 'value' is 'None', a ValidationError is not issued. valid_time = validate_time_is_in_not_too_distant_future(raise_validation_error=True, value=value, months=12) return @validates("operations_start") def validate_operations_start(self, value): # violation when time is not in the future, but also does not exceed a threshold for the 'reasonable' future # when 'value' is 'None', a ValidationError is not issued. valid_time = validate_time_is_in_not_too_distant_future(raise_validation_error=True, value=value, months=12) return @validates("operations_end") def validate_operations_end(self, value): # violation when time is not in the future, but also does not exceed a threshold for the 'reasonable' future # when 'value' is 'None', a ValidationError is not issued. valid_time = validate_time_is_in_not_too_distant_future(raise_validation_error=True, value=value, months=12) return @validates("eta_interval_end") def validate_eta_interval_end(self, value): # violation when time is not in the future, but also does not exceed a threshold for the 'reasonable' future # when 'value' is 'None', a ValidationError is not issued. valid_time = validate_time_is_in_not_too_distant_future(raise_validation_error=True, value=value, months=12) return @validates("etd_interval_end") def validate_etd_interval_end(self, value): # violation when time is not in the future, but also does not exceed a threshold for the 'reasonable' future # when 'value' is 'None', a ValidationError is not issued. valid_time = validate_time_is_in_not_too_distant_future(raise_validation_error=True, value=value, months=12) return # deserialize PUT object target class UserSchema(Schema): def __init__(self): super().__init__(unknown=None) pass id = fields.Integer(required=True) first_name = fields.String(allow_none=True, required=False, validate=[validate.Length(max=64)]) last_name = fields.String(allow_none=True, required=False, validate=[validate.Length(max=64)]) user_phone = fields.String(allow_none=True, required=False) user_email = fields.String(allow_none=True, required=False, validate=[validate.Length(max=64)]) old_password = fields.String(allow_none=True, required=False, validate=[validate.Length(max=128)]) new_password = fields.String(allow_none=True, required=False, validate=[validate.Length(min=6, max=128)]) notify_email = fields.Bool(allow_none=True, required=False) notify_whatsapp = fields.Bool(allow_none=True, required=False) notify_signal = fields.Bool(allow_none=True, required=False) notify_popup = fields.Bool(allow_none=True, required=False) notify_on = fields.List(fields.Enum(NotificationType), required=False, allow_none=True) @validates("user_phone") def validate_user_phone(self, value): if value is not None: valid_characters = list(map(str,range(0,10)))+["+", " "] if not all([v in valid_characters for v in value]): raise ValidationError({"user_phone":f"one of the phone number values is not valid."}) @validates("user_email") def validate_user_email(self, value): if value and not re.match(r"[^@]+@[^@]+\.[^@]+", value): raise ValidationError({"user_email":f"invalid email address"}) @dataclass class Times: id: int eta_berth: datetime eta_berth_fixed: bool etd_berth: datetime etd_berth_fixed: bool lock_time: datetime lock_time_fixed: bool zone_entry: datetime zone_entry_fixed: bool operations_start: datetime operations_end: datetime remarks: str participant_id: int berth_id: int berth_info: str pier_side: bool participant_type: int shipcall_id: int ata: datetime atd: datetime eta_interval_end: datetime etd_interval_end: datetime created: datetime modified: datetime @dataclass class User: id: int participant_id: int first_name: str last_name: str user_name: str user_email: str user_phone: str password_hash: str api_key: str notify_email: bool notify_whatsapp: bool notify_signal: bool notify_popup: bool created: datetime modified: datetime ports: List[NotificationType] = field(default_factory=list) notify_event: List[NotificationType] = field(default_factory=list) def __hash__(self): return hash(id) def wants_notifications(self, notification_type: NotificationType): events = bitflag_to_list(self.notify_event) return notification_type in events @dataclass class Ship: id: int name: str imo: int callsign: str participant_id: int length: float width: float is_tug: bool bollard_pull: int eni: int created: datetime modified: datetime deleted: bool class ShipSchema(Schema): def __init__(self): super().__init__(unknown=None) pass id = fields.Int(required=False) name = fields.String(allow_none=False, required=True) imo = fields.Int(allow_none=False, required=True) callsign = fields.String(allow_none=True, required=False) participant_id = fields.Int(allow_none=True, required=False) length = fields.Float(allow_none=True, required=False, validate=[validate.Range(min=0, max=1000, min_inclusive=False, max_inclusive=False)]) width = fields.Float(allow_none=True, required=False, validate=[validate.Range(min=0, max=100, min_inclusive=False, max_inclusive=False)]) is_tug = fields.Bool(allow_none=True, required=False, default=False) bollard_pull = fields.Int(allow_none=True, required=False) eni = fields.Int(allow_none=True, required=False) created = fields.DateTime(allow_none=True, required=False) modified = fields.DateTime(allow_none=True, required=False) deleted = fields.Bool(allow_none=True, required=False, default=False) @validates("name") def validate_name(self, value): character_length = len(str(value)) if character_length<1: raise ValidationError({"name":f"'name' argument should have at least one character"}) elif character_length>=64: raise ValidationError({"name":f"'name' argument should have at max. 63 characters"}) if check_if_string_has_special_characters(value): raise ValidationError({"name":f"'name' argument should not have special characters."}) return @validates("imo") def validate_imo(self, value): value = str(value).zfill(7) # 1 becomes '0000001' (7 characters). 12345678 becomes '12345678' (8 characters) imo_length = len(value) if imo_length != 7: raise ValidationError({"imo":f"'imo' should be a 7-digit number"}) return @validates("callsign") def validate_callsign(self, value): if value is not None: callsign_length = len(str(value)) if callsign_length>8: raise ValidationError({"callsign":f"'callsign' argument should not have more than 8 characters"}) if check_if_string_has_special_characters(value): raise ValidationError({"callsign":f"'callsign' argument should not have special characters."}) return class TimesId(Schema): pass class BerthList(Berth): pass class NotificationList(Notification): pass class Shipcalls(Shipcall): pass class TimesList(Times): pass @dataclass class ShipcallParticipantMap: id: int shipcall_id: int participant_id: int type : ShipcallType created: datetime modified: datetime def to_json(self): return { "id": self.id, "shipcall_id": self.shipcall_id, "participant_id": self.participant_id, "type": self.type.name if isinstance(self.type, IntEnum) else ShipcallType(self.type).name, "created": self.created.isoformat() if self.created else "", "modified": self.modified.isoformat() if self.modified else "", }