Source code for pipeline.src.entities.alerts

from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import List

import pandas as pd
from pydantic import BaseModel, ConfigDict
from pydantic.dataclasses import dataclass as pydataclass
from sqlalchemy import Table



[docs]
class AlertType(Enum):

[docs]
    MISSING_DEP_ALERT = "MISSING_DEP_ALERT"


[docs]
    MISSING_FAR_ALERT = "MISSING_FAR_ALERT"


[docs]
    MISSING_FAR_48_HOURS_ALERT = "MISSING_FAR_48_HOURS_ALERT"


[docs]
    SUSPICION_OF_UNDER_DECLARATION_ALERT = "SUSPICION_OF_UNDER_DECLARATION_ALERT"


[docs]
    POSITION_ALERT = "POSITION_ALERT"




@pydataclass(config=ConfigDict(extra="forbid"))

[docs]
class GearSpecification:

[docs]
    gear: str


[docs]
    minMesh: float | None = None


[docs]
    maxMesh: float | None = None




@pydataclass(config=ConfigDict(extra="forbid"))

[docs]
class SpeciesSpecification:

[docs]
    species: str


[docs]
    minWeight: float | None = None





[docs]
class AdministrativeAreaType(Enum):

[docs]
    FAO_AREA = "FAO_AREA"


[docs]
    EEZ_AREA = "EEZ_AREA"


[docs]
    NEAFC_AREA = "NEAFC_AREA"


[docs]
    DISTANCE_TO_SHORE = "DISTANCE_TO_SHORE"




@dataclass

[docs]
class AreasTableMetadata:

[docs]
    table_name: str


[docs]
    geometry_column: str


[docs]
    filter_column: str





[docs]
class AdminAreasSpecification(BaseModel):

[docs]
    areaType: AdministrativeAreaType


[docs]
    areas: List




@dataclass

[docs]
class AdminAreasSpecWithTable:

[docs]
    area_type: AdministrativeAreaType


[docs]
    areas: List


[docs]
    metadata: AreasTableMetadata


[docs]
    table: Table





[docs]
class RegulatoryAreaSpecification(BaseModel):

[docs]
    lawType: str | None = None


[docs]
    topic: str | None = None


[docs]
    zone: str | None = None





[docs]
class PositionAlertSpecification(BaseModel):

[docs]
    id: int


[docs]
    name: str


[docs]
    description: str


[docs]
    natinf_code: int


[docs]
    threat: str


[docs]
    threat_characterization: str


[docs]
    is_activated: bool


[docs]
    is_in_error: bool


[docs]
    error_reason: str | None


[docs]
    validity_start_datetime_utc: datetime | None


[docs]
    validity_end_datetime_utc: datetime | None


[docs]
    repeat_each_year: bool


[docs]
    track_analysis_depth: float


[docs]
    only_fishing_positions: bool


[docs]
    gears: List[GearSpecification] | None


[docs]
    species: List[SpeciesSpecification] | None


[docs]
    species_catch_areas: List[str] | None


[docs]
    administrative_areas: List[AdminAreasSpecification] | None


[docs]
    regulatory_areas: List[RegulatoryAreaSpecification] | None


[docs]
    min_depth: float | None


[docs]
    flag_states_iso2: List[str] | None


[docs]
    vessel_ids: List[int] | None


[docs]
    district_codes: List[str] | None


[docs]
    producer_organizations: List[str] | None



[docs]
    def model_post_init(self, __context):
        if pd.isnull(self.min_depth):
            self.min_depth = None