Metrics processing

.gitignore

@@ -145,9 +145,6 @@ venv.bak/
 .spyderproject
 .spyproject
 
-# VS Code
-.vscode
-
 # Rope project settings
 .ropeproject
 

backend/alembic/versions/c32d65d6e6fd_create_fct_metrics.py

@@ -0,0 +1,50 @@
+"""create fct metrics
+
+Revision ID: c32d65d6e6fd
+Revises: 06bb3c26076f
+Create Date: 2024-10-18 11:13:02.468934
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy import Inspector
+import geoalchemy2
+
+# revision identifiers, used by Alembic.
+revision = 'c32d65d6e6fd'
+down_revision = 'd9d279892b5c'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+
+    # Création de la table fct_metrics
+    op.create_table(
+        'fct_metrics',
+        sa.Column('timestamp', sa.DateTime, primary_key=True),
+        sa.Column('vessel_id', sa.Integer, primary_key=True),
+        sa.Column('type', sa.String, nullable=False), 
+        sa.Column('vessel_mmsi', sa.Integer, nullable=False),
+        sa.Column('ship_name', sa.String, nullable=False),
+        sa.Column('vessel_country_iso3', sa.String, nullable=False),
+        sa.Column('vessel_imo', sa.Integer),
+        sa.Column('duration_total', sa.FLOAT, nullable=False),
+        sa.Column('duration_fishing', sa.FLOAT, nullable=True),
+        sa.Column("zone_name", sa.String, primary_key=True),
+        sa.Column('zone_sub_category', sa.String, nullable=True),
+    )
+
+
+def downgrade() -> None:
+    # Suppression de la table fct_metrics en cas de rollback
+    conn = op.get_bind()
+    inspector = Inspector.from_engine(conn)
+    sql_tables = inspector.get_table_names()
+    tables = [
+        "fct_metrics",
+    ]
+    for t in tables:
+        if t in sql_tables:
+            op.drop_table(t)
+

backend/alembic/versions/d9d279892b5c_add_task_execution.py

@@ -0,0 +1,89 @@
+"""add task execution 
+
+Revision ID: d9d279892b5c
+Revises: 7ba4634af5ad
+Create Date: 2024-11-30 12:54:22.318425
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.sql import func
+
+
+# revision identifiers, used by Alembic.
+revision = 'd9d279892b5c'
+down_revision = '7ba4634af5ad'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    """
+    Create a new task_executions table that permit to store historical data
+    for each task execution
+    The goal is to be able to detect timeframe not covered by Spire API interrogation
+    """
+    # drop constraint from existing table to free the constraint name
+    op.drop_constraint(table_name="tasks_executions",
+                       constraint_name="tasks_executions_pkey")
+    # rename existing table to keep existing data
+    op.rename_table('tasks_executions','tasks_executions_tmp')
+    # create the new task_executions table with id and active columns in addition
+    op.create_table("tasks_executions",
+                    sa.Column("id", sa.Integer(),sa.Identity(), primary_key=True, index=True),
+                    sa.Column("task_name", sa.String),
+                    sa.Column("point_in_time", sa.DateTime(timezone=True)),
+                    sa.Column(
+                        "created_at",
+                        sa.DateTime(timezone=True),
+                        nullable=False,
+                        server_default=func.now(),
+                    ),
+                    sa.Column("updated_at", sa.DateTime(timezone=True), onupdate=func.now()),
+                    sa.Column("delta",
+                                sa.Interval,
+                                index=True,
+                                nullable=True),
+                    sa.Column("active",
+                                sa.Boolean,
+                                index=True,
+                                default=False),
+                    )
+    # copy of existing data to new table with active=True
+    op.execute("insert into tasks_executions "
+               +"(task_name,point_in_time,created_at,updated_at,active) "
+               +"select task_name,point_in_time,created_at,updated_at,true "
+               +"from tasks_executions_tmp")
+    # drop old table
+    op.drop_table('tasks_executions_tmp')
+
+    # Retrieve all historical spire api interrogation from spire_ais_data table
+    op.execute(
+        """insert into public.tasks_executions (task_name,point_in_time,created_at,delta,active)
+            select distinct
+            'load_spire_data_from_api' as "task_name",
+            T1.created_at as "point_in_time",
+            T1.created_at,
+            T1.created_at-(select distinct created_at from spire_ais_data where created_at < T1.created_at group by created_at order by created_at desc limit 1) as "delta",
+            case when T1.created_at = (select MAX(created_at) from spire_ais_data) and not EXISTS(select 1 from public.tasks_executions where task_name = 'load_spire_data_from_api' and active = True) then true else false end as "active"
+            from spire_ais_data T1
+            where T1.created_at not in (select point_in_time from public.tasks_executions where task_name = 'load_spire_data_from_api')
+            group by T1.created_at
+            order by T1.created_at desc
+        """)
+    pass
+
+
+def downgrade() -> None:
+    # delete all lines active=False as they have no equivalent in old task_executions
+    op.execute("delete from tasks_executions where active=False")
+    # drop active and id table
+    op.drop_column("tasks_executions","delta")
+    op.drop_column("tasks_executions","active")
+    op.drop_column("tasks_executions","id")
+    # recreate the primary unique key constraint of old table
+    op.create_unique_constraint("tasks_executions_pkey",
+                                "tasks_executions",
+                                ["task_name"],
+                                )
+    pass

backend/bloom/container.py

@@ -9,6 +9,8 @@
 from bloom.infra.repositories.repository_vessel_position import VesselPositionRepository
 from bloom.infra.repositories.repository_segment import SegmentRepository
 from bloom.infra.repositories.repository_zone import ZoneRepository
+from bloom.infra.repositories.repository_metrics import MetricsRepository
+
 from bloom.services.GetVesselsFromSpire import GetVesselsFromSpire
 from bloom.services.metrics import MetricsService
 from bloom.usecase.GenerateAlerts import GenerateAlerts
@@ -89,3 +91,8 @@ class UseCases(containers.DeclarativeContainer):
         MetricsService,
         session_factory=db.provided.session,
     )
+
+    metrics_repository = providers.Factory(
+        MetricsRepository,
+        session_factory=db.provided.session,
+    )

backend/bloom/domain/metrics.py

@@ -5,6 +5,21 @@
 from enum import Enum
 from bloom.domain.vessel import Vessel,VesselListView
 from bloom.domain.zone import Zone,ZoneListView
+from typing import Union
+
+class Metrics(BaseModel) :
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+    timestamp : datetime
+    vessel_id: int
+    type : str
+    vessel_mmsi: int
+    ship_name: str
+    vessel_country_iso3: str 
+    vessel_imo: int
+    duration_total : float
+    duration_fishing: Optional[float] = None
+    zone_name : str
+    zone_sub_category : Union[str, None]
 
 class TotalTimeActivityTypeEnum(str, Enum):
     total_time_at_sea: str = "Total Time at Sea"
@@ -31,6 +46,12 @@ class ResponseMetricsZoneVisitingTimeByVesselSchema(BaseModel):
     vessel: VesselListView
     zone_visiting_time_by_vessel: timedelta
 
+
+class ResponseMetricsVesselVisitingTimeByZoneSchema(BaseModel):
+    zone: ZoneListView
+    vessel: VesselListView
+    vessel_visiting_time_by_zone: timedelta
+
 class ResponseMetricsVesselTotalTimeActivityByActivityTypeSchema(BaseModel):
     vessel_id : int
     activity: str

backend/bloom/infra/database/sql_model.py

@@ -10,7 +10,8 @@
     Integer,
     Interval,
     String,
-    PrimaryKeyConstraint
+    PrimaryKeyConstraint,
+    Identity
 )
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.sql import func, select
@@ -224,10 +225,13 @@ class Segment(Base):
 
 class TaskExecution(Base):
     __tablename__ = "tasks_executions"
-    task_name = Column("task_name", String, primary_key=True)
+    id = Column("id", Integer, Identity(), primary_key=True)
+    task_name = Column("task_name", String)
     point_in_time = Column("point_in_time", DateTime(timezone=True))
     created_at = Column("created_at", DateTime(timezone=True), server_default=func.now())
     updated_at = Column("updated_at", DateTime(timezone=True), onupdate=func.now())
+    delta = Column("delta", Interval, nullable=False)
+    active = Column("active", Boolean, nullable=False)
 
 
 class RelSegmentZone(Base):
@@ -256,3 +260,18 @@ class MetricsVesselInActivity(Base):
     #vessel_id: Mapped[Optional[int]]
     #total_time_at_sea: Mapped[Optional[timedelta]]
 
+
+class Metrics(Base):
+    __tablename__ = "fct_metrics"
+    timestamp = Column("timestamp", DateTime(timezone=True), primary_key=True)
+    vessel_id= Column("vessel_id", Integer, ForeignKey("dim_vessel.id"), primary_key=True)
+    type = Column("type", String, nullable=False)  # Ajouté comme clé primaire
+    vessel_mmsi= Column("vessel_mmsi", Integer, ForeignKey("dim_vessel.mmsi"), nullable=False)
+    ship_name= Column("ship_name", String, ForeignKey("dim_vessel.ship_name"), nullable=False)
+    vessel_country_iso3= Column("vessel_country_iso3", String, ForeignKey("dim_vessel.country_iso3"), nullable=False)
+    vessel_imo= Column("vessel_imo", Integer, ForeignKey("dim_vessel.imo"))
+    duration_total= Column("duration_total", Double, nullable= False)
+    duration_fishing= Column("duration_fishing", Double)
+    zone_name= Column("zone_name", String, ForeignKey("dim_zone.name"),primary_key=True)
+    zone_sub_category= Column("zone_sub_category", String, ForeignKey("dim_zone.sub_category"))
+

backend/bloom/infra/repositories/repository_excursion.py

@@ -1,16 +1,23 @@
 from contextlib import AbstractContextManager
-from typing import Any, List, Union
+from typing import Any, List, Union, Optional
 
 import pandas as pd
+from bloom.routers.requests import DatetimeRangeRequest
 from dependency_injector.providers import Callable
 from geoalchemy2.shape import from_shape, to_shape
-from sqlalchemy import desc
+from sqlalchemy import desc, and_, or_
 from sqlalchemy import select
 from sqlalchemy.orm import Session
+from sqlalchemy.sql.expression import and_,or_, asc, desc
 
 from bloom.domain.excursion import Excursion
 from bloom.infra.database import sql_model
 
+from bloom.routers.requests import ( DatetimeRangeRequest,
+                                     OrderByRequest,
+                                        PageParams,
+                                        OrderByEnum)
+
 
 class ExcursionRepository:
     def __init__(
@@ -34,9 +41,29 @@ def get_param_from_last_excursion(self, session: Session, vessel_id: int) -> Uni
             return None
         return {"arrival_port_id": result.arrival_port_id, "arrival_position": result.arrival_position}
 
-    def get_excursions_by_vessel_id(self, session: Session, vessel_id: int) -> List[Excursion]:
+    def get_excursions_by_vessel_id(self,
+                                    session: Session,
+                                    vessel_id: int,
+                                    datetime_range: DatetimeRangeRequest,
+                                    order: OrderByRequest,
+                                    pagination: PageParams
+                                    ) -> List[Excursion]:
         """Recheche l'excursion en cours d'un bateau, c'est-à-dire l'excursion qui n'a pas de date d'arrivée"""
-        stmt = select(sql_model.Excursion).where(sql_model.Excursion.vessel_id == vessel_id)
+        stmt = select(sql_model.Excursion).where(
+            and_(
+                sql_model.Excursion.vessel_id == vessel_id,
+                sql_model.Excursion.departure_at < datetime_range.end_at,
+                or_(
+                    sql_model.Excursion.arrival_at > datetime_range.start_at,
+                    sql_model.Excursion.arrival_at == None
+                ),
+            )
+        )
+        stmt =  stmt.order_by(asc(sql_model.Excursion.departure_at))\
+                if  order.order == OrderByEnum.ascending \
+                else stmt.order_by(desc(sql_model.Excursion.departure_at))
+        stmt = stmt.offset(pagination.offset) if pagination.offset != None else stmt
+        stmt = stmt.limit(pagination.limit) if pagination.limit != None else stmt
         result = session.execute(stmt).scalars().all()
         if not result:
             return []

backend/bloom/infra/repositories/repository_metrics.py

@@ -0,0 +1,91 @@
+from bloom.domain.metrics import Metrics
+from contextlib import AbstractContextManager
+from typing import Any, List, Union
+
+import pandas as pd
+from dependency_injector.providers import Callable
+from sqlalchemy.orm import Session
+
+from bloom.infra.database import sql_model
+
+from sqlalchemy import and_, or_, select, update, text, join
+
+
+
+
+class MetricsRepository:
+
+    def __init__(
+            self,
+            session_factory: Callable,
+    ) -> Callable[..., AbstractContextManager]:
+        self.session_factory = session_factory
+
+  #  def get_vessel_excursion_segment_by_id(self, session, segment_id: int) -> pd.DataFrame:
+  #      stmt = select(
+  #          sql_model.Vessel.id,
+  #          sql_model.Vessel.mmsi,
+  #          sql_model.Vessel.ship_name
+  #          sql_model.Vessel.country_iso3
+  #          sql_model.Vessel.imo
+  #      ).join(
+  #          sql_model.Excursion,
+  #          sql_model.Excursion.vessel_id == sql_model.Vessel.id
+  #      ).join(
+  #          sql_model.Segment,
+  #          sql_model.Segment.excursion_id == sql_model.Excursion.id
+  #      ).where(
+  #          sql_model.Segment.id == segment_id
+  #      )
+
+  #      result = session.execute(stmt)
+  #      if not result:
+  #          return None
+  #      df = pd.DataFrame(result, columns=["vessel_id", "vessel_mmsi", "ship_name", "vessel_country_iso3","vessel_imo"])
+  #      return df
+
+    def batch_create_metrics(
+            self, session: Session, metricss: list[Metrics]
+    ) -> list[Metrics]:
+        orm_list = [MetricsRepository.map_to_orm(metrics) for metrics in metricss]
+        session.add_all(orm_list)
+        return [MetricsRepository.map_to_domain(orm) for orm in orm_list]
+
+
+    @staticmethod
+    def map_to_orm(metrics: Metrics) -> sql_model.Metrics:
+        return sql_model.Metrics(
+            timestamp=metrics.timestamp,
+            vessel_id=metrics.vessel_id,
+            type=metrics.type,
+            vessel_mmsi=metrics.vessel_mmsi,
+            ship_name=metrics.ship_name,
+            vessel_country_iso3=metrics.vessel_country_iso3,
+            vessel_imo=metrics.vessel_imo,
+            duration_total=metrics.duration_total,
+            duration_fishing=metrics.duration_fishing,
+            zone_name=metrics.zone_name,
+            zone_sub_category=metrics.zone_sub_category,
+        )            
+
+
+    @staticmethod
+    def map_to_domain(metrics: sql_model.Metrics) -> Metrics:
+            return Metrics(
+                timestamp=metrics.timestamp,
+                vessel_id=metrics.vessel_id,
+                type=metrics.type,
+                vessel_mmsi=metrics.vessel_mmsi,
+                ship_name=metrics.ship_name,
+                vessel_country_iso3=metrics.vessel_country_iso3,
+                vessel_imo=metrics.vessel_imo,
+                duration_total=metrics.duration_total,
+                duration_fishing=metrics.duration_fishing,
+                zone_name=metrics.zone_name,
+                zone_sub_category=metrics.zone_sub_category,
+            )            
+
+
+
+
+