Raise alerts for declared activity during a beacon malfunction

Author: VincentAntoine

pipeline/src/flows/beacon_malfunction_activity_detection.py

@@ -39,7 +39,7 @@ def extract_vessels_with_recent_ais() -> pd.DataFrame:
 def extract_beacon_malfunctions_with_declared_activity() -> pd.DataFrame:
     """
     Extract current beacon malfunctions during which the vessel recently declared
-    fishing activity (DEP, FAR).
+    fishing activity (DEP, FAR), along with the vessel data required to build alerts.
     This also works on vessels with paper logbook, although with some delay.
     """
     return extract(
@@ -120,6 +120,17 @@ def beacon_malfunction_activity_detection_flow():
     ais_active_reportings = extract_active_reportings.submit(
         AlertType.AIS_ACTIVITY_ON_VESSEL_NOT_EMITTING_VMS_ALERT.value
     )
+    declared_activity_silenced_alerts = extract_silenced_alerts.submit(
+        AlertType.DECLARED_FISHING_ACTIVITY_DURING_BEACON_MALFUNCTION.value,
+        # Fishing activity data may be received several months after the event,
+        # and we should be able to detect those cases - i.e. a vessel that hasn't
+        # emitted for 6 months and for which we receive today a declaration of
+        # activity that took place 3 months must trigger the alert.
+        number_of_hours=24 * 365,
+    )
+    declared_activity_active_reportings = extract_active_reportings.submit(
+        AlertType.DECLARED_FISHING_ACTIVITY_DURING_BEACON_MALFUNCTION.value
+    )
 
     # Tag is_at_port using port H3 referential, then keep only at-sea vessels
     vessels_with_recent_ais = tag_positions_at_port(vessels_with_recent_ais)
@@ -147,9 +158,22 @@ def beacon_malfunction_activity_detection_flow():
         threat="Mesures techniques et de conservation",
         threat_characterization="VMS - absence",
     )
-    filtered_alerts = filter_alerts(
+    filtered_ais_alerts = filter_alerts(
         ais_alerts, ais_silenced_alerts, ais_active_reportings
     )
+    declared_activity_alerts = make_alerts(
+        current_malfunctions_with_declared_activity,
+        AlertType.DECLARED_FISHING_ACTIVITY_DURING_BEACON_MALFUNCTION.value,
+        "Activité de pêche déclarée pendant une avarie VMS",
+        natinf_code=27688,
+        threat="Mesures techniques et de conservation",
+        threat_characterization="VMS - absence",
+    )
+    filtered_declared_activity_alerts = filter_alerts(
+        declared_activity_alerts,
+        declared_activity_silenced_alerts,
+        declared_activity_active_reportings,
+    )
 
     # Load
     update_beacon_malfunction_is_followed.map(
@@ -162,6 +186,12 @@ def beacon_malfunction_activity_detection_flow():
     )
     load_new_beacon_malfunctions(new_malfunctions)
     load_alerts(
-        filtered_alerts,
+        filtered_ais_alerts,
         alert_config_name=AlertType.AIS_ACTIVITY_ON_VESSEL_NOT_EMITTING_VMS_ALERT.value,
     )
+    load_alerts(
+        filtered_declared_activity_alerts,
+        alert_config_name=(
+            AlertType.DECLARED_FISHING_ACTIVITY_DURING_BEACON_MALFUNCTION.value
+        ),
+    )

pipeline/src/queries/monitorfish/beacon_malfunctions_with_declared_activity.sql

@@ -1,29 +1,42 @@
-WITH current_malfunctions AS (
-    SELECT
-        id,
-        internal_reference_number AS cfr,
-        vessel_status_last_modification_date_utc,
-        is_followed
-    FROM beacon_malfunctions bm
-    WHERE stage IN ('INITIAL_ENCOUNTER', 'AT_QUAY', 'FOLLOWING', 'TARGETING_VESSEL') AND internal_reference_number IS NOT NULL
-),
-
-latest_activities AS (
+WITH latest_activities AS (
     SELECT
         cfr,
         MAX(activity_datetime_utc) AS latest_activity_datetime_utc
     FROM logbook_reports
     WHERE
-        operation_datetime_utc >= :utcnow - INTERVAL '24h'
+        operation_datetime_utc >= :utcnow - INTERVAL '24 hours'
         AND operation_type = 'DAT'
         AND log_type IN ('DEP', 'FAR')
     GROUP BY cfr
 )
 
-SELECT
-    cm.id,
-    is_followed
-FROM latest_activities la
-JOIN current_malfunctions cm
-ON la.cfr = cm.cfr
-WHERE latest_activity_datetime_utc > cm.vessel_status_last_modification_date_utc
+SELECT DISTINCT ON (bm.id)
+    bm.id,
+    bm.is_followed,
+    bm.internal_reference_number AS cfr,
+    COALESCE(v.external_immatriculation, lp.external_immatriculation) AS external_immatriculation,
+    COALESCE(v.ircs, lp.ircs) AS ircs,
+    COALESCE(v.vessel_name, lp.vessel_name) AS vessel_name,
+    COALESCE(v.flag_state, lp.flag_state) AS flag_state,
+    'INTERNAL_REFERENCE_NUMBER' AS vessel_identifier,
+    v.id AS vessel_id,
+    d.dml,
+    COALESCE(f.facade::text, d.facade) AS facade,
+    lp.risk_factor,
+    la.latest_activity_datetime_utc AS triggering_behaviour_datetime_utc
+FROM beacon_malfunctions bm
+JOIN latest_activities la ON la.cfr = bm.internal_reference_number
+LEFT JOIN vessels v ON v.cfr = bm.internal_reference_number
+LEFT JOIN last_positions lp ON lp.cfr = bm.internal_reference_number
+LEFT JOIN districts d ON d.district_code = v.district_code
+-- The sea front is derived from the vessel's last known position, falling back to
+-- its district's sea front for vessels that never emitted (no last position). The
+-- beacon malfunction's own coordinates are the last VMS emission, not where the
+-- declared fishing activity took place, so they are not used here.
+LEFT JOIN facade_areas_subdivided f
+    ON ST_Intersects(ST_SetSRID(ST_Point(lp.longitude, lp.latitude), 4326), f.geometry)
+WHERE
+    bm.stage IN ('INITIAL_ENCOUNTER', 'AT_QUAY', 'FOLLOWING', 'TARGETING_VESSEL')
+    AND bm.internal_reference_number IS NOT NULL
+    AND la.latest_activity_datetime_utc > bm.vessel_status_last_modification_date_utc
+ORDER BY bm.id

pipeline/tests/test_flows/test_beacon_malfunction_activity_detection.py

@@ -150,7 +150,7 @@ def reset_test_data_with_declared_activity(reset_test_data):
     Malfunction id=5 (CFR 'ABC000306959', stage INITIAL_ENCOUNTER,
     vessel_status_last_modification_date_utc 2h10 ago) is followed in the test data.
     Unfollow it and add a recent FAR declaration for that vessel so the flow
-    re-follows it based on the declared fishing activity.
+    re-follows it and raises an alert based on the declared fishing activity.
     """
     e = create_engine(db="monitorfish_remote")
     with e.begin() as con:
@@ -186,20 +186,34 @@ def reset_test_data_with_declared_activity(reset_test_data):
         )
 
 
+DECLARED_ACTIVITY_ALERT = "DECLARED_FISHING_ACTIVITY_DURING_BEACON_MALFUNCTION"
+
+
 def test_flow_follows_malfunction_with_declared_activity(
     reset_test_data_with_declared_activity,
 ):
     """
-    On top of the AIS path (malfunction id=6, see `test_flow`), the flow follows
-    non-archived, not-yet-followed malfunctions during which the vessel recently
-    declared fishing activity (malfunction id=5).
+    On top of the AIS path (malfunction id=6, see `test_flow`), for every
+    non-archived malfunction during which the vessel recently declared fishing
+    activity (malfunction id=5), the flow:
+
+      - follows the malfunction if it is not already followed
+      - raises a DECLARED_FISHING_ACTIVITY_DURING_BEACON_MALFUNCTION pending alert
+        for the vessel
     """
     headers = {
         "Accept": "application/json, text/plain",
         "Content-Type": "application/json;charset=UTF-8",
         "X-API-KEY": "backend_api_key",
     }
 
+    initial_pending_alerts = read_query(
+        "SELECT * FROM pending_alerts", db="monitorfish_remote"
+    )
+    assert (
+        DECLARED_ACTIVITY_ALERT not in initial_pending_alerts.alert_config_name.values
+    )
+
     with patch("src.shared_tasks.beacon_malfunctions.requests") as mock_requests:
         state = beacon_malfunction_activity_detection_flow(return_state=True)
 
@@ -221,3 +235,14 @@ def test_flow_follows_malfunction_with_declared_activity(
         any_order=True,
     )
     assert mock_requests.patch.call_count == 2
+
+    final_pending_alerts = read_query(
+        "SELECT * FROM pending_alerts", db="monitorfish_remote"
+    )
+    declared_activity_alerts = final_pending_alerts.loc[
+        final_pending_alerts.alert_config_name == DECLARED_ACTIVITY_ALERT
+    ]
+    assert declared_activity_alerts.internal_reference_number.tolist() == [
+        "ABC000306959"
+    ]
+    assert declared_activity_alerts.iloc[0].value == DECLARED_ACTIVITY_ALERT