Add distance to shore for each operation #42

airflow/dags/extract_secmar.py

@@ -72,8 +72,16 @@ def set_operations_stats_extra_attributes_fn(**kwargs):
     )
 
 
+def compute_operations_distances_fn(**kwargs):
+    return execute_sql_file('compute_distances')
+
+
 def insert_operations_stats_fn(**kwargs):
-    path = helpers.opendata_sql_path('insert_operations_stats')
+    return execute_sql_file('insert_operations_stats')
+
+
+def execute_sql_file(filename):
+    path = helpers.opendata_sql_path(filename)
     with open(path, 'r', encoding='utf-8') as f:
         content = f.read()
     return PostgresHook('postgresql_local').run(content)
@@ -146,6 +154,15 @@ def embulk_import(dag, table):
 insert_operations_stats.set_upstream(delete_invalid_operations)
 insert_operations_stats.set_downstream(start_checks)
 
+compute_operations_distances = PythonOperator(
+    task_id='compute_operations_distances',
+    python_callable=compute_operations_distances_fn,
+    provide_context=True,
+    dag=dag
+)
+compute_operations_distances.set_upstream(delete_invalid_operations)
+compute_operations_distances.set_downstream(insert_operations_stats)
+
 # Handle extra attributes for operations_stats
 sql = """
 select

opendata/doc.yml

@@ -399,6 +399,16 @@ components:
           type: boolean
           description: Indique si cette opération est une opération qui concerne une activité de plongée (bouteille ou apnée) ou de chasse sous-marine
           example: true
+        distance_cote_metres:
+          type: integer
+          format: int32
+          description: La distance entre la plus proche côte ou frontière terrestre en territoire français en mètres. Cette distance peut être égale à 0 si l'opération se déroule très près des côtes ou à l'intérieur des terres en cas de fausse alerte. La distance peut être très élevée lors d'un déclenchement d'une balise par erreur ou en cas de coordination d'une opération impliquant un flotteur français en eaux internationales.
+          nullable: true
+        distance_cote_milles_nautiques:
+          type: number
+          format: float
+          description: La distance entre la plus proche côte ou frontière terrestre en territoire français en milles nautiques. Cette distance peut être égale à 0 si l'opération se déroule très près des côtes ou à l'intérieur des terres en cas de fausse alerte. La distance peut être très élevée lors d'un déclenchement d'une balise par erreur ou en cas de coordination d'une opération impliquant un flotteur français en eaux internationales.
+          nullable: true
         nombre_personnes_assistees:
           type: integer
           format: int32

opendata/sql/compute_distances.sql

@@ -0,0 +1,45 @@
+-- Delete rows where position has changed
+delete from operations_points
+where operation_id in (
+  select op.operation_id
+  from operations_points op
+  join operations o on o.operation_id = op.operation_id
+  where op.latitude is distinct from o.latitude
+     or op.longitude is distinct from o.longitude
+);
+
+-- Insert new rows that were not yet in the cache
+insert into operations_points (operation_id, "point", latitude, longitude)
+select
+  operation_id,
+  ST_SetSRID(ST_MakePoint(longitude, latitude), 4326)::geography,
+  latitude,
+  longitude
+from operations
+where operation_id not in (select operation_id from operations_points);
+
+-- Compute distances when it has not been computed yet
+update operations_points set
+  distance_cote_metres = t.distance_cote_metres,
+  distance_cote_milles_nautiques = t.distance_cote_milles_nautiques
+from (
+  select
+    operation_id,
+    distance_cote_metres,
+    round((distance_cote_metres*0.000539957)::numeric, 2) distance_cote_milles_nautiques
+  from (
+    select
+      operation_id,
+      round(dst / 100.0) * 100 distance_cote_metres
+    from (
+      select
+        o.operation_id,
+        ST_Distance(fb.geography, o.point) dst
+      from operations_points o
+      join france_boundaries fb on true
+      where o.point is not null
+        and o.distance_cote_metres is null
+    ) _
+  ) _
+) t
+where t.operation_id = operations_points.operation_id;

opendata/sql/insert_operations_stats.sql

@@ -15,6 +15,8 @@ select
   null phase_journee,
   false concerne_snosan,
   false concerne_plongee,
+  op.distance_cote_metres distance_cote_metres,
+  op.distance_cote_milles_nautiques distance_cote_milles_nautiques,
   coalesce(rh.nombre_personnes_assistees, 0) nombre_personnes_assistees,
   coalesce(rh.nombre_personnes_decedees, 0) nombre_personnes_decedees,
   coalesce(rh.nombre_personnes_decedees_accidentellement, 0) nombre_personnes_decedees_accidentellement,
@@ -142,6 +144,7 @@ left join (
   from flotteurs f
   group by f.operation_id
 ) f on f.operation_id = o.operation_id
+join operations_points op on op.operation_id = o.operation_id
 ;
 
 update operations_stats set concerne_snosan = true where

opendata/sql/operations_points.sql

@@ -0,0 +1,20 @@
+create table operations_points (
+  "operation_id" bigint primary key,
+  "point" geography(point)
+);
+create index operations_points_point_idx on operations_points using gist(point);
+
+insert into operations_points
+select
+  operation_id,
+  ST_SetSRID(ST_MakePoint(longitude, latitude), 4326)::geography
+from operations;
+
+vacuum analyze operations_points;
+
+select
+  o.operation_id,
+  ST_Distance(b.geography, o.point)/1000
+from operations_points o
+join france_boundaries b on true
+where o.point is not null;

opendata/sql/schema.sql

@@ -115,6 +115,8 @@ CREATE TABLE public.operations_stats (
     "phase_journee" varchar(20),
     "concerne_snosan" boolean not null,
     "concerne_plongee" boolean not null,
+    "distance_cote_metres" int,
+    "distance_cote_milles_nautiques" numeric(6, 2),
     "nombre_personnes_assistees" smallint not null,
     "nombre_personnes_decedees" smallint not null,
     "nombre_personnes_decedees_accidentellement" smallint not null,