Source code for pipeline.src.entities.beacon_malfunctions

from copy import deepcopy
from dataclasses import dataclass
from datetime import datetime
from email.message import EmailMessage
from enum import Enum
from typing import List

import pandas as pd

from config import (
    CNSP_SIP_DEPARTMENT_EMAIL,
    CNSP_SIP_DEPARTMENT_FAX,
    CNSP_SIP_DEPARTMENT_MOBILE_PHONE,
)
from src.entities.communication_means import CommunicationMeans




class BeaconStatus(Enum):


    ACTIVATED = "ACTIVATED"



    DEACTIVATED = "DEACTIVATED"



    IN_TEST = "IN_TEST"



    NON_APPROVED = "NON_APPROVED"



    UNSUPERVISED = "UNSUPERVISED"


    @staticmethod


    def from_poseidon_status(poseidon_status: str):
        mapping = {
            "Activée": BeaconStatus.ACTIVATED,
            "Désactivée": BeaconStatus.DEACTIVATED,
            "En test": BeaconStatus.IN_TEST,
            "Non agréée": BeaconStatus.NON_APPROVED,
            "Non surveillée": BeaconStatus.UNSUPERVISED,
        }
        return mapping[poseidon_status]






class BeaconMalfunctionStage(Enum):


    INITIAL_ENCOUNTER = "INITIAL_ENCOUNTER"



    FOUR_HOUR_REPORT = "FOUR_HOUR_REPORT"



    AT_QUAY = "AT_QUAY"



    TARGETING_VESSEL = "TARGETING_VESSEL"



    FOLLOWING = "FOLLOWING"



    ARCHIVED = "ARCHIVED"






class BeaconMalfunctionVesselStatus(Enum):


    AT_SEA = "AT_SEA"



    AT_PORT = "AT_PORT"



    NO_NEWS = "NO_NEWS"



    ACTIVITY_DETECTED = "ACTIVITY_DETECTED"



    TECHNICAL_STOP = "TECHNICAL_STOP"



    ON_SALE = "ON_SALE"



    SUSPENDED_BECAUSE_UNPAID = "SUSPENDED_BECAUSE_UNPAID"



    IN_FOREIGN_EEZ = "IN_FOREIGN_EEZ"






class EndOfMalfunctionReason(Enum):


    RESUMED_TRANSMISSION = "RESUMED_TRANSMISSION"



    BEACON_DEACTIVATED_OR_UNEQUIPPED = "BEACON_DEACTIVATED_OR_UNEQUIPPED"






class BeaconMalfunctionNotificationType(Enum):


    MALFUNCTION_AT_SEA_INITIAL_NOTIFICATION = "MALFUNCTION_AT_SEA_INITIAL_NOTIFICATION"



    MALFUNCTION_AT_SEA_INITIAL_NOTIFICATION_UNSUPERVISED_BEACON = (
        "MALFUNCTION_AT_SEA_INITIAL_NOTIFICATION_UNSUPERVISED_BEACON"
    )



    MALFUNCTION_AT_SEA_REMINDER = "MALFUNCTION_AT_SEA_REMINDER"



    MALFUNCTION_AT_PORT_INITIAL_NOTIFICATION = (
        "MALFUNCTION_AT_PORT_INITIAL_NOTIFICATION"
    )



    MALFUNCTION_AT_PORT_INITIAL_NOTIFICATION_UNSUPERVISED_BEACON = (
        "MALFUNCTION_AT_PORT_INITIAL_NOTIFICATION_UNSUPERVISED_BEACON"
    )



    MALFUNCTION_AT_PORT_REMINDER = "MALFUNCTION_AT_PORT_REMINDER"



    END_OF_MALFUNCTION = "END_OF_MALFUNCTION"



    MALFUNCTION_NOTIFICATION_TO_FOREIGN_FMC = "MALFUNCTION_NOTIFICATION_TO_FOREIGN_FMC"




    def to_notification_subject_template(self):
        type_subject_mapping = {
            "MALFUNCTION_AT_SEA_INITIAL_NOTIFICATION": "{vessel_name} ({immat}) : interruption en mer des émissions VMS",
            "MALFUNCTION_AT_SEA_INITIAL_NOTIFICATION_UNSUPERVISED_BEACON": "{vessel_name} ({immat}) : interruption en mer des émissions VMS",
            "MALFUNCTION_AT_SEA_REMINDER": "{vessel_name} ({immat}) : RAPPEL : interruption en mer des émissions VMS",
            "MALFUNCTION_AT_PORT_INITIAL_NOTIFICATION": "{vessel_name} ({immat}) : interruption à quai des émissions VMS",
            "MALFUNCTION_AT_PORT_INITIAL_NOTIFICATION_UNSUPERVISED_BEACON": "{vessel_name} ({immat}) : interruption à quai des émissions VMS",
            "MALFUNCTION_AT_PORT_REMINDER": "{vessel_name} ({immat}) : RAPPEL : interruption à quai des émissions VMS",
            "END_OF_MALFUNCTION": "{vessel_name} ({immat}) : reprise des émissions VMS",
            "MALFUNCTION_NOTIFICATION_TO_FOREIGN_FMC": "Interruption of VMS transmissions from fishing vessel {vessel_name} ({immat})",
        }
        return type_subject_mapping[self.name]






class BeaconMalfunctionNotificationRecipientFunction(Enum):


    VESSEL_CAPTAIN = "VESSEL_CAPTAIN"



    VESSEL_OPERATOR = "VESSEL_OPERATOR"



    SATELLITE_OPERATOR = "SATELLITE_OPERATOR"



    FMC = "FMC"



    FOREIGN_FMC = "FOREIGN_FMC"




@dataclass


class BeaconMalfunctionNotificationAddressee:


    function: BeaconMalfunctionNotificationRecipientFunction



    name: str



    address_or_number: str




@dataclass


class BeaconMalfunctionToNotify:


    beacon_malfunction_id: int



    vessel_cfr_or_immat_or_ircs: str



    beacon_number: str



    vessel_name: str



    malfunction_start_date_utc: datetime



    last_position_latitude: float



    last_position_longitude: float



    notification_type: BeaconMalfunctionNotificationType



    vessel_emails: List[str]



    vessel_mobile_phone: str



    vessel_fax: str



    operator_name: str



    operator_email: str



    operator_mobile_phone: str



    operator_fax: str



    satellite_operator: str



    satellite_operator_emails: List[str]



    foreign_fmc_name: str



    foreign_fmc_emails: List[str]



    previous_notification_datetime_utc: datetime



    test_mode: bool




    def __post_init__(self):
        self.notification_type = BeaconMalfunctionNotificationType(
            self.notification_type
        )




    def get_sms_addressees(self) -> List[BeaconMalfunctionNotificationAddressee]:
        if not self.test_mode:
            addressees = []

            if self.notification_type is not (
                BeaconMalfunctionNotificationType.MALFUNCTION_NOTIFICATION_TO_FOREIGN_FMC
            ):
                if self.vessel_mobile_phone:
                    addressees.append(
                        BeaconMalfunctionNotificationAddressee(
                            function=BeaconMalfunctionNotificationRecipientFunction.VESSEL_CAPTAIN,
                            name=None,
                            address_or_number=self.vessel_mobile_phone,
                        )
                    )

                if self.operator_mobile_phone:
                    addressees.append(
                        BeaconMalfunctionNotificationAddressee(
                            function=BeaconMalfunctionNotificationRecipientFunction.VESSEL_OPERATOR,
                            name=self.operator_name,
                            address_or_number=self.operator_mobile_phone,
                        )
                    )

        else:
            addressees = (
                [
                    BeaconMalfunctionNotificationAddressee(
                        function=BeaconMalfunctionNotificationRecipientFunction.FMC,
                        name="CNSP",
                        address_or_number=CNSP_SIP_DEPARTMENT_MOBILE_PHONE,
                    )
                ]
                if CNSP_SIP_DEPARTMENT_MOBILE_PHONE
                else []
            )
        return addressees




    def get_fax_addressees(self) -> List[BeaconMalfunctionNotificationAddressee]:
        if not self.test_mode:
            addressees = []

            if self.notification_type is not (
                BeaconMalfunctionNotificationType.MALFUNCTION_NOTIFICATION_TO_FOREIGN_FMC
            ):
                if self.vessel_fax:
                    addressees.append(
                        BeaconMalfunctionNotificationAddressee(
                            function=BeaconMalfunctionNotificationRecipientFunction.VESSEL_CAPTAIN,
                            name=None,
                            address_or_number=self.vessel_fax,
                        )
                    )

                if self.operator_fax:
                    addressees.append(
                        BeaconMalfunctionNotificationAddressee(
                            function=BeaconMalfunctionNotificationRecipientFunction.VESSEL_OPERATOR,
                            name=self.operator_name,
                            address_or_number=self.operator_fax,
                        )
                    )

        else:
            addressees = (
                [
                    BeaconMalfunctionNotificationAddressee(
                        function=BeaconMalfunctionNotificationRecipientFunction.FMC,
                        name="CNSP",
                        address_or_number=CNSP_SIP_DEPARTMENT_FAX,
                    )
                ]
                if CNSP_SIP_DEPARTMENT_FAX
                else []
            )
        return addressees




    def get_email_addressees(self) -> List[BeaconMalfunctionNotificationAddressee]:
        if not self.test_mode:
            vessel_emails = self.vessel_emails if self.vessel_emails else []
            satellite_operator_emails = (
                self.satellite_operator_emails if self.satellite_operator_emails else []
            )
            operator_emails = [self.operator_email] if self.operator_email else []
            foreign_fmcs_emails = (
                self.foreign_fmc_emails if self.foreign_fmc_emails else []
            )

            vessel_addressees = [
                BeaconMalfunctionNotificationAddressee(
                    function=(
                        BeaconMalfunctionNotificationRecipientFunction.VESSEL_CAPTAIN
                    ),
                    name=None,
                    address_or_number=vessel_email,
                )
                for vessel_email in vessel_emails
            ]

            satellite_operator_addressees = [
                BeaconMalfunctionNotificationAddressee(
                    function=(
                        BeaconMalfunctionNotificationRecipientFunction.SATELLITE_OPERATOR
                    ),
                    name=self.satellite_operator,
                    address_or_number=satellite_operator_email,
                )
                for satellite_operator_email in satellite_operator_emails
            ]

            operator_addressees = [
                BeaconMalfunctionNotificationAddressee(
                    function=(
                        BeaconMalfunctionNotificationRecipientFunction.VESSEL_OPERATOR
                    ),
                    name=self.operator_name,
                    address_or_number=operator_email,
                )
                for operator_email in operator_emails
            ]

            foreign_fmc_addressees = [
                BeaconMalfunctionNotificationAddressee(
                    function=BeaconMalfunctionNotificationRecipientFunction.FOREIGN_FMC,
                    name=self.foreign_fmc_name,
                    address_or_number=foreign_fmcs_email,
                )
                for foreign_fmcs_email in foreign_fmcs_emails
            ]

            if self.notification_type is (
                BeaconMalfunctionNotificationType.MALFUNCTION_NOTIFICATION_TO_FOREIGN_FMC
            ):
                addressees = foreign_fmc_addressees
            else:
                addressees = (
                    vessel_addressees
                    + operator_addressees
                    + satellite_operator_addressees
                )

        else:
            addressees = (
                [
                    BeaconMalfunctionNotificationAddressee(
                        function=BeaconMalfunctionNotificationRecipientFunction.FMC,
                        name="CNSP",
                        address_or_number=CNSP_SIP_DEPARTMENT_EMAIL,
                    )
                ]
                if CNSP_SIP_DEPARTMENT_EMAIL
                else []
            )
        return addressees




    def get_notification_subject(self):
        template = self.notification_type.to_notification_subject_template()
        return template.format(
            vessel_name=self.vessel_name, immat=self.vessel_cfr_or_immat_or_ircs
        )




    def get_formatted_malfunction_start_datetime_utc(self):
        if self.notification_type is (
            BeaconMalfunctionNotificationType.MALFUNCTION_NOTIFICATION_TO_FOREIGN_FMC
        ):
            date_format = "%d/%m/%Y at %H:%M UTC"
        else:
            date_format = "%d/%m/%Y à %Hh%M UTC"
        return self.malfunction_start_date_utc.strftime(date_format)




    def replace_pandas_nat(self):
        obj = deepcopy(self)
        if obj.malfunction_start_date_utc is pd.NaT:
            obj = obj.__replace__(malfunction_start_date_utc=None)
        if obj.previous_notification_datetime_utc is pd.NaT:
            obj = obj.__replace__(previous_notification_datetime_utc=None)
        return obj




@dataclass


class BeaconMalfunctionMessageToSend:


    message: EmailMessage



    beacon_malfunction_to_notify: BeaconMalfunctionToNotify



    communication_means: CommunicationMeans




    def get_addressees(self) -> List[BeaconMalfunctionNotificationAddressee]:
        if self.communication_means is CommunicationMeans.EMAIL:
            return self.beacon_malfunction_to_notify.get_email_addressees()
        elif self.communication_means is CommunicationMeans.SMS:
            return self.beacon_malfunction_to_notify.get_sms_addressees()
        elif self.communication_means is CommunicationMeans.FAX:
            return self.beacon_malfunction_to_notify.get_fax_addressees()
        else:
            raise ValueError(
                f"Unexpected communication_means {self.communication_means}"
            )




@dataclass


class BeaconMalfunctionNotification:


    beacon_malfunction_id: int



    date_time_utc: datetime



    notification_type: BeaconMalfunctionNotificationType



    communication_means: CommunicationMeans



    recipient_function: BeaconMalfunctionNotificationRecipientFunction



    recipient_name: str



    recipient_address_or_number: str



    success: bool



    error_message: str