Ajout de données depuis SeaMIS par fichiers CSV (#145)

* Work on CSV files

* Create and describe aggregates

* Ajout tables de codes

* Check mapping completeness

* Work on unique codes

* Download remote FTP folder

* Keep latest version

* Improve operations to keep

* Work on DAG download_secmar_csv_ftp

* Add other tasks

* Add process_all_days

* Work on replacement mapping

* Add create_cleaned_aggregate_files

* Resolve path

* Add base Embulk files

* Base Embulk import

* Fix filepath

* Clean timestamps

* Clean force fields

* Improve datetimes parsing

* Improve drop_duplicates for operations

* Converssion radians en degrés pour le vent

* Add pourquoi_alerte in secmar_operation_id

* Améliore les codes

* Travaille sur les colonnes cross_sitrep

* Rework Embulk order

* Rework cross_sitrep

* Update checks with new cross_sitrep format

* Add 3 new cols in operations

* Rework est_metropolitain

* Work on insert SQL scripts

* Fix duration computation

* Rewrite run SQL task

* Start to write checks

* Add missing dependency

* Download next day if available

* Replace ShortCircuit with BranchOperator

* Move to main DAG

* Update inset for secmar_csv_operations

* Mappe Polynésie française en Polynésie

* Replace more NA fields for the map

* Changement d'organisation

Author: AntoineAugusti

.circleci/config.yml

@@ -59,7 +59,7 @@ jobs:
             - ~/.cache/yarn
       - run:
           name: Pull CHANGELOG
-          command: wget https://raw.githubusercontent.com/MTES-MCT/secmar-data/master/CHANGELOG.md -O doc/CHANGELOG.md
+          command: wget https://raw.githubusercontent.com/snosan-tools/secmar-data/master/CHANGELOG.md -O doc/CHANGELOG.md
       - run:
           name: Build Markdown documentation for opendata
           command: |

.gitignore

@@ -8,3 +8,5 @@ app/node_modules
 app/platforms
 app/plugins
 app/resources/signing
+
+snosan_csv

airflow/.python-version

@@ -0,0 +1 @@
+3.5.2

airflow/dags/download_secmar_csv_ftp.py

@@ -0,0 +1,124 @@
+# -*- coding: utf-8 -*-
+"""
+# download_secmar_csv_ftp
+This DAG download CSV files from the remote FTP server.
+"""
+import os
+from datetime import datetime
+from pathlib import Path
+
+import helpers
+from airflow import DAG
+from airflow.models import Variable
+from airflow.operators.dummy_operator import DummyOperator
+from airflow.operators.python_operator import PythonOperator, BranchPythonOperator
+from transformers import secmar_csv
+
+default_args = helpers.default_args({"start_date": datetime(2022, 6, 22, 10, 0)})
+
+dag = DAG(
+    "download_secmar_csv_ftp",
+    default_args=default_args,
+    max_active_runs=1,
+    concurrency=1,
+    catchup=True,
+    schedule_interval="0 10 * * *",
+)
+dag.doc_md = __doc__
+
+
+def setup_ftp_env():
+    os.environ["FTP_PROXY"] = "false"
+    os.environ["FTP_HOST"] = Variable.get("SECMAR_CSV_FTP_HOST")
+    os.environ["FTP_USER"] = Variable.get("SECMAR_CSV_FTP_USER")
+    os.environ["FTP_PASSWORD"] = Variable.get("SECMAR_CSV_FTP_PASSWORD")
+
+
+def ftp_download_fn(**kwargs):
+    setup_ftp_env()
+    secmar_csv.ftp_download_remote_folder(kwargs["templates_dict"]["day"])
+
+
+def check_if_next_day_exists_fn(**kwargs):
+    setup_ftp_env()
+    if secmar_csv.day_exists_in_remote_ftp(kwargs["templates_dict"]["day"]):
+        return "download_next_day"
+    return "process_all_days"
+
+
+def embulk_import(dag, table):
+    script = "secmar_csv_%s" % table
+    filepath = str(secmar_csv.AGGREGATE_FOLDER / (table + ".cleaned.csv"))
+    return helpers.embulk_run(dag, script, {"EMBULK_FILEPATH": filepath})
+
+
+download_single_day = PythonOperator(
+    task_id="download_single_day",
+    python_callable=ftp_download_fn,
+    provide_context=True,
+    dag=dag,
+    templates_dict={"day": "{{ ds_nodash }}"},
+)
+
+check_if_next_day_exists = BranchPythonOperator(
+    task_id="check_if_next_day_exists",
+    python_callable=check_if_next_day_exists_fn,
+    provide_context=True,
+    dag=dag,
+    templates_dict={"day": "{{ tomorrow_ds_nodash }}"},
+)
+check_if_next_day_exists.set_upstream(download_single_day)
+
+download_next_day = PythonOperator(
+    task_id="download_next_day",
+    python_callable=ftp_download_fn,
+    provide_context=True,
+    dag=dag,
+    templates_dict={"day": "{{ tomorrow_ds_nodash }}"},
+)
+download_next_day.set_upstream(check_if_next_day_exists)
+
+process_all_days = PythonOperator(
+    task_id="process_all_days",
+    python_callable=secmar_csv.process_all_days,
+    dag=dag,
+    trigger_rule="all_done",
+)
+process_all_days.set_upstream(download_single_day)
+process_all_days.set_upstream(download_next_day)
+
+build_aggregate_files = PythonOperator(
+    task_id="build_aggregate_files",
+    python_callable=secmar_csv.build_aggregate_files,
+    dag=dag,
+)
+build_aggregate_files.set_upstream(process_all_days)
+
+check_mapping_data = PythonOperator(
+    task_id="check_mapping_data",
+    python_callable=secmar_csv.check_mapping_data,
+    dag=dag,
+)
+check_mapping_data.set_upstream(build_aggregate_files)
+
+create_cleaned_aggregate_files = PythonOperator(
+    task_id="create_cleaned_aggregate_files",
+    python_callable=secmar_csv.create_cleaned_aggregate_files,
+    dag=dag,
+)
+create_cleaned_aggregate_files.set_upstream(check_mapping_data)
+
+start_embulk = DummyOperator(task_id="start_embulk", dag=dag)
+end_embulk = DummyOperator(task_id="end_embulk", dag=dag)
+start_embulk.set_upstream(create_cleaned_aggregate_files)
+
+operation_embulk = embulk_import(dag, "operation")
+operation_embulk.set_upstream(start_embulk)
+operation_embulk.set_downstream(end_embulk)
+
+for table in [
+    Path(f).stem for f in secmar_csv.EXPECTED_FILENAMES if not f.startswith("operation")
+]:
+    t = embulk_import(dag, table)
+    t.set_upstream(operation_embulk)
+    t.set_downstream(end_embulk)

airflow/dags/extract_secmar.py

@@ -7,20 +7,23 @@
 """
 from datetime import datetime
 
+import helpers
 from airflow import DAG
-from airflow.operators.python_operator import PythonOperator
+from airflow.hooks.postgres_hook import PostgresHook
 from airflow.operators.bash_operator import BashOperator
-from airflow.operators.dummy_operator import DummyOperator
 from airflow.operators.check_operator import CheckOperator
-from airflow.hooks.postgres_hook import PostgresHook
-from operators.pg_download_operator import PgDownloadOperator
 from airflow.operators.dagrun_operator import TriggerDagRunOperator
-
-
-import helpers
-from secmar_dags import in_path, out_path, secmar_transform
-from secmar_dags import SECMAR_TABLES, secmar_transformer
-from secmar_checks import checks
+from airflow.operators.dummy_operator import DummyOperator
+from airflow.operators.python_operator import PythonOperator
+from operators.pg_download_operator import PgDownloadOperator
+from secmar_checks import checks, secmar_csv_checks
+from secmar_dags import (
+    SECMAR_TABLES,
+    in_path,
+    out_path,
+    secmar_transform,
+    secmar_transformer,
+)
 
 default_args = helpers.default_args({"start_date": datetime(2018, 4, 27, 5, 40)})
 
@@ -89,6 +92,22 @@ def execute_sql_file(filename):
     return PostgresHook("postgresql_local").run(content)
 
 
+def _execute_secmar_csv_sql_file(filename):
+    path = helpers.secmar_csv_sql_path(filename)
+    with open(path, "r", encoding="utf-8") as f:
+        content = f.read()
+    return PostgresHook("postgresql_local").run(content)
+
+
+def secmar_csv_sql_task(dag, filename):
+    return PythonOperator(
+        task_id="run_" + filename,
+        python_callable=lambda **kwargs: _execute_secmar_csv_sql_file(filename),
+        provide_context=True,
+        dag=dag,
+    )
+
+
 def make_distance_fn(distance):
     return lambda **kwargs: execute_sql_file(distance)
 
@@ -147,6 +166,33 @@ def embulk_import(dag, table):
 )
 delete_invalid_operations.set_upstream(end_import)
 
+# Insert data fetched by FTP, "secmar_csv"
+start_secmar_csv_insert = DummyOperator(task_id="start_secmar_csv_insert", dag=dag)
+end_secmar_csv_insert = DummyOperator(task_id="end_secmar_csv_insert", dag=dag)
+start_secmar_csv_insert.set_upstream(delete_invalid_operations)
+
+insert_operations = secmar_csv_sql_task(dag, "insert_operations")
+insert_operations.set_upstream(start_secmar_csv_insert)
+
+for table in ["flotteurs", "resultats_humain", "moyens"]:
+    t = secmar_csv_sql_task(dag, "insert_{table}".format(table=table))
+    t.set_upstream(insert_operations)
+    t.set_downstream(end_secmar_csv_insert)
+
+start_secmar_csv_checks = DummyOperator(task_id="start_secmar_csv_checks", dag=dag)
+end_secmar_csv_checks = DummyOperator(task_id="end_secmar_csv_checks", dag=dag)
+start_secmar_csv_checks.set_upstream(end_secmar_csv_insert)
+
+for check_name, query in secmar_csv_checks().items():
+    t = CheckOperator(
+        task_id="check_consistency_" + check_name,
+        sql=query,
+        conn_id="postgresql_local",
+        dag=dag,
+    )
+    t.set_upstream(start_secmar_csv_checks)
+    t.set_downstream(end_secmar_csv_checks)
+
 insert_operations_stats = PythonOperator(
     task_id="insert_operations_stats",
     python_callable=insert_operations_stats_fn,
@@ -161,15 +207,15 @@ def embulk_import(dag, table):
     provide_context=True,
     dag=dag,
 )
-prepare_operations_points.set_upstream(delete_invalid_operations)
+prepare_operations_points.set_upstream(end_secmar_csv_checks)
 
 insert_moyens_snsm = PythonOperator(
     task_id="insert_moyens_snsm",
     python_callable=insert_moyens_snsm_fn,
     provide_context=True,
     dag=dag,
 )
-insert_moyens_snsm.set_upstream(delete_invalid_operations)
+insert_moyens_snsm.set_upstream(end_secmar_csv_checks)
 insert_moyens_snsm.set_downstream(start_checks)
 
 distances = [
@@ -213,7 +259,7 @@ def embulk_import(dag, table):
     csv_params={"sep": ",", "index": False},
     dag=dag,
 )
-download_operations_local_time.set_upstream(delete_invalid_operations)
+download_operations_local_time.set_upstream(end_secmar_csv_checks)
 
 transform_operations_stats = PythonOperator(
     task_id="transform_operations_stats",

airflow/dags/helpers.py

@@ -16,6 +16,12 @@ def opendata_sql_path(filename):
     )
 
 
+def secmar_csv_sql_path(filename):
+    return "{base}/opendata_sql/secmar_csv/{filename}.sql".format(
+        base=base_path(), filename=filename
+    )
+
+
 def data_path(filename):
     return "{base}/data/{filename}".format(base=base_path(), filename=filename)
 

airflow/dags/secmar_checks.py

@@ -126,10 +126,10 @@ def checks():
               date_heure_reception_alerte
             from operations as op
             join (
-                select 'Corsen 2017/1305' cross_sitrep ,'2017-12-16 11:57:00+00' expected_time UNION
-                select 'Corsen 2018/1503' cross_sitrep ,'2018-10-13 08:13:00+00' expected_time UNION
-                select 'Étel 2018/3473' cross_sitrep ,'2018-12-20 08:51:00+00' expected_time UNION
-                select 'Corsen 2019/2604' cross_sitrep ,'2019-12-20 13:26:00+00' expected_time
+                select 'Corsen SAR 2017/1305' cross_sitrep ,'2017-12-16 11:57:00+00' expected_time UNION
+                select 'Corsen SAR 2018/1503' cross_sitrep ,'2018-10-13 08:13:00+00' expected_time UNION
+                select 'Étel SAR 2018/3473' cross_sitrep ,'2018-12-20 08:51:00+00' expected_time UNION
+                select 'Corsen SAR 2019/2604' cross_sitrep ,'2019-12-20 13:26:00+00' expected_time
             ) t on t.cross_sitrep = op.cross_sitrep and op.date_heure_reception_alerte::text = t.expected_time
         ) t
         """,
@@ -140,3 +140,33 @@ def checks():
         where concerne_snosan and avec_clandestins
         """,
     }
+
+
+def secmar_csv_checks():
+    return {
+        "operations_count_2021": """
+            select count(1) between 16800 and 16820
+            from operations
+            where extract(year from date_heure_reception_alerte) = 2021
+        """,
+        "operations_count_up_to_2021": """
+            select count(1) between 321500 and 321600
+            from operations
+            where extract(year from date_heure_reception_alerte) <= 2021
+        """,
+        "operations_count_cross_2021": """
+            select count(distinct "cross") = 11
+            from operations
+            where extract(year from date_heure_reception_alerte) = 2021
+        """,
+        "operations_count_2021_from_secmar_csv": """
+            select count(1) between 14670 and 14680
+            from operations
+            where extract(year from date_heure_reception_alerte) = 2021 and operation_id in (select secmar_operation_id from secmar_csv_operation)
+        """,
+        "est_metropolitain": """
+            select string_agg(distinct "cross"::varchar, '|' order by "cross"::varchar) = 'Antilles-Guyane|Gris-Nez|Guadeloupe|Guyane|La Réunion|Martinique|Mayotte|Nouvelle-Calédonie|Polynésie'
+            from operations
+            where not est_metropolitain
+        """,
+    }

embulk/secmar_csv_bilan.yml.liquid

@@ -0,0 +1,26 @@
+in:
+  type: file
+  path_prefix: {{ env.EMBULK_FILEPATH }}
+  parser:
+    charset: UTF-8
+    newline: LF
+    type: csv
+    delimiter: ','
+    quote: '"'
+    escape: '"'
+    trim_if_not_quoted: false
+    skip_header_lines: 1
+    allow_extra_columns: false
+    allow_optional_columns: false
+    columns:
+    - {name: operation_id, type: string}
+    - {name: SEC_RESULTAT_HUMAIN_resultat_humain_id, type: string}
+    - {name: SEC_RESULTAT_HUMAIN_cat_personne_id, type: string}
+    - {name: SEC_RESULTAT_HUMAIN_nb, type: long}
+    - {name: SEC_RESULTAT_HUMAIN_dont_nb_blesse, type: long}
+    - {name: operation_long_name, type: string}
+    - {name: operation_version, type: long}
+    - {name: secmar_operation_id, type: long}
+{% include 'datasources/out_postgresql' %}
+  table: secmar_csv_bilan
+  mode: truncate_insert

embulk/secmar_csv_flotteur.yml.liquid

@@ -0,0 +1,29 @@
+in:
+  type: file
+  path_prefix: {{ env.EMBULK_FILEPATH }}
+  parser:
+    charset: UTF-8
+    newline: LF
+    type: csv
+    delimiter: ','
+    quote: '"'
+    escape: '"'
+    trim_if_not_quoted: false
+    skip_header_lines: 1
+    allow_extra_columns: false
+    allow_optional_columns: false
+    columns:
+    - {name: operation_id, type: string}
+    - {name: SEC_FLOTTEUR_IMPLIQUE_mer_force, type: long}
+    - {name: SEC_FLOTTEUR_IMPLIQUE_type_flotteur_id, type: string}
+    - {name: SEC_FLOTTEUR_IMPLIQUE_pavillon_id, type: string}
+    - {name: SEC_FLOTTEUR_IMPLIQUE_num_immat_fr, type: string}
+    - {name: SEC_FLOTTEUR_IMPLIQUE_num_imo, type: double}
+    - {name: SEC_FLOTTEUR_IMPLIQUE_nom, type: string}
+    - {name: SEC_FLOTTEUR_IMPLIQUE_resultat_flotteur_id, type: string}
+    - {name: operation_long_name, type: string}
+    - {name: operation_version, type: long}
+    - {name: secmar_operation_id, type: long}
+{% include 'datasources/out_postgresql' %}
+  table: secmar_csv_flotteur
+  mode: truncate_insert