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 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 # from BreCal. ... import check_if_user_is_bsmd_type 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 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): """ Any user has the attributes 'notify_email' -> NotificationType.email 'notify_popup' -> NotificationType.push 'notify_whatsapp' -> undeclared 'notify_signal' -> undeclared """ undefined = 0 email = 1 push = 2 # whatsapp = 3 # signal = 4 @classmethod def _missing_(cls, value): return cls.undefined 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' 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, "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, level, type, message, created, modified): return self(id, shipcall_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 @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) 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 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 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, "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, 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, 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)]) # #TODO: the user schema does not (yet) include the 'notify_' fields @validates("user_phone") def validate_user_phone(self, value): 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 not "@" in 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 # #TODO_clarify: should we use an IntFlag for multi-assignment? notify_whatsapp: bool # #TODO_clarify: should we use an IntFlag for multi-assignment? notify_signal: bool # #TODO_clarify: should we use an IntFlag for multi-assignment? notify_popup: bool # #TODO_clarify: should we use an IntFlag for multi-assignment? created: datetime modified: datetime @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 "", }