441 lines
15 KiB
Python
441 lines
15 KiB
Python
from dataclasses import field, dataclass
|
|
from marshmallow import Schema, fields, post_load, INCLUDE, ValidationError
|
|
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
|
|
|
|
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):
|
|
undefined = 0
|
|
email = 1
|
|
push = 2
|
|
@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,
|
|
"operation": 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:
|
|
id: int
|
|
shipcall_id: int
|
|
level: int
|
|
type: NotificationType
|
|
message: str
|
|
created: datetime
|
|
modified: datetime
|
|
|
|
def to_json(self):
|
|
return {
|
|
"id": self.id,
|
|
"shipcall_id": self.shipcall_id,
|
|
"level": self.level,
|
|
"type": 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
|
|
flags: int
|
|
created: datetime
|
|
modified: datetime
|
|
deleted: bool
|
|
|
|
class ParticipantList(Participant):
|
|
pass
|
|
|
|
class ParticipantAssignmentSchema(Schema):
|
|
participant_id = fields.Int()
|
|
type = fields.Int()
|
|
|
|
class ShipcallSchema(Schema):
|
|
def __init__(self):
|
|
super().__init__(unknown=None)
|
|
pass
|
|
|
|
id = fields.Int()
|
|
ship_id = fields.Int()
|
|
type = fields.Enum(ShipcallType, required=True)
|
|
eta = fields.DateTime(Required = False, allow_none=True)
|
|
voyage = fields.Str(allow_none=True, metadata={'Required':False}) # Solving: RemovedInMarshmallow4Warning: Passing field metadata as keyword arguments is deprecated. Use the explicit `metadata=...` argument instead. Additional metadata: {'Required': False}
|
|
etd = fields.DateTime(Required = False, allow_none=True)
|
|
arrival_berth_id = fields.Int(Required = False, allow_none=True)
|
|
departure_berth_id = fields.Int(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.Int(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)
|
|
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.Int(Required = False, allow_none=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, metadata={'Required':False}) # Solving: RemovedInMarshmallow4Warning: Passing field metadata as keyword arguments is deprecated. Use the explicit `metadata=...` argument instead. Additional metadata: {'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.Int(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):
|
|
data['type_value'] = data['type'].value
|
|
if 'evaluation' in data:
|
|
data['evaluation_value'] = data['evaluation'].value
|
|
else:
|
|
data['evaluation_value'] = EvaluationType.undefined
|
|
return data
|
|
|
|
@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)
|
|
|
|
|
|
@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
|
|
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,
|
|
"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,
|
|
"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.Int(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)
|
|
participant_id = fields.Int(Required = True)
|
|
berth_id = fields.Int(Required = False, allow_none = True)
|
|
berth_info = fields.String(Required = False, allow_none=True)
|
|
pier_side = fields.Bool(Required = False, allow_none = True)
|
|
shipcall_id = fields.Int(Required = True)
|
|
participant_type = fields.Int(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)
|
|
|
|
# deserialize PUT object target
|
|
|
|
class UserSchema(Schema):
|
|
def __init__(self):
|
|
super().__init__(unknown=None)
|
|
pass
|
|
id = fields.Int(required=True)
|
|
first_name = fields.Str(allow_none=True, metadata={'Required':False})
|
|
last_name = fields.Str(allow_none=True, metadata={'Required':False})
|
|
user_phone = fields.Str(allow_none=True, metadata={'Required':False})
|
|
user_email = fields.Str(allow_none=True, metadata={'Required':False})
|
|
old_password = fields.Str(allow_none=True, metadata={'Required':False})
|
|
new_password = fields.Str(allow_none=True, metadata={'Required':False})
|
|
|
|
@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
|
|
|
|
@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, metadata={'Required':True})
|
|
imo = fields.Int(allow_none=False, metadata={'Required':True})
|
|
callsign = fields.String(allow_none=True, metadata={'Required':False})
|
|
participant_id = fields.Int(allow_none=True, metadata={'Required':False})
|
|
length = fields.Float(allow_none=True, metadata={'Required':False})
|
|
width = fields.Float(allow_none=True, metadata={'Required':False})
|
|
is_tug = fields.Bool(allow_none=True, metadata={'Required':False}, default=False)
|
|
bollard_pull = fields.Int(allow_none=True, metadata={'Required':False})
|
|
eni = fields.Int(allow_none=True, metadata={'Required':False})
|
|
created = fields.DateTime(allow_none=True, metadata={'Required':False})
|
|
modified = fields.DateTime(allow_none=True, metadata={'Required':False})
|
|
deleted = fields.Bool(allow_none=True, metadata={'Required':False}, default=False)
|
|
|
|
|
|
class TimesId(Schema):
|
|
pass
|
|
|
|
|
|
class BerthList(Berth):
|
|
pass
|
|
|
|
|
|
class NotificationList(Notification):
|
|
pass
|
|
|
|
|
|
class Shipcalls(Shipcall):
|
|
pass
|
|
|
|
|
|
class TimesList(Times):
|
|
pass
|