Merge pull request #734 from MTES-MCT/feat-conso-aggregation

Aggregation de la conso par toutes les combinaisons d'années

airflow/dags/ingest_population.py

@@ -4,6 +4,8 @@
 import pandas as pd
 from airflow.decorators import dag, task
 from include.container import Container
+from include.pools import DBT_POOL
+from include.utils import get_dbt_command_from_directory
 from pendulum import datetime
 
 URL = "https://www.insee.fr/fr/statistiques/fichier/3698339/base-pop-historiques-1876-2021.xlsx"
@@ -42,8 +44,14 @@ def ingest(path_on_bucket) -> int | None:
         os.remove(tmp_localpath)
         return row_count
 
+    @task.bash(pool=DBT_POOL)
+    def dbt_build() -> str:
+        return get_dbt_command_from_directory(cmd="dbt build -s +insee+")
+
     path_on_bucket = download()
-    ingest(path_on_bucket)
+    ingest_task = ingest(path_on_bucket)
+
+    path_on_bucket >> ingest_task >> dbt_build()
 
 
 ingest_population()

airflow/dags/update_app.py

@@ -106,6 +106,12 @@ def copy_table_from_dw_to_app(
                 "copy_public_data_scot_departements",
                 "copy_public_data_scot_regions",
                 "copy_public_data_cerema",
+                "copy_public_data_landconso",
+                "copy_public_data_landconsocomparison",
+                "copy_public_data_landconsostats",
+                "copy_public_data_landpop",
+                "copy_public_data_landpopcomparison",
+                "copy_public_data_landpopstats",
             ],
             type="array",
         ),
@@ -339,6 +345,72 @@ def copy_public_data_cerema(**context):
             ],
         )
 
+    @task.python
+    def copy_public_data_landconso(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_landconso",
+            to_table="public.public_data_landconso",
+            environment=context["params"]["environment"],
+            btree_index_columns=[
+                ["land_id", "land_type", "year"],
+            ],
+        )
+
+    @task.python
+    def copy_public_data_landconsocomparison(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_landconsocomparison",
+            to_table="public.public_data_landconsocomparison",
+            environment=context["params"]["environment"],
+            btree_index_columns=[
+                ["land_id", "land_type", "from_year", "to_year"],
+            ],
+        )
+
+    @task.python
+    def copy_public_data_landconsostats(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_landconsostats",
+            to_table="public.public_data_landconsostats",
+            environment=context["params"]["environment"],
+            btree_index_columns=[
+                ["land_id", "land_type", "from_year", "to_year"],
+            ],
+        )
+
+    @task.python
+    def copy_public_data_landpop(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_landpop",
+            to_table="public.public_data_landpop",
+            environment=context["params"]["environment"],
+            btree_index_columns=[
+                ["land_id", "land_type", "year"],
+            ],
+        )
+
+    @task.python
+    def copy_public_data_landpopcomparison(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_landpopcomparison",
+            to_table="public.public_data_landpopcomparison",
+            environment=context["params"]["environment"],
+            btree_index_columns=[
+                ["land_id", "land_type", "from_year", "to_year"],
+            ],
+        )
+
+    @task.python
+    def copy_public_data_landpopstats(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_landpopstats",
+            to_table="public.public_data_landpopstats",
+            environment=context["params"]["environment"],
+            btree_index_columns=[
+                ["land_id", "land_type", "from_year", "to_year"],
+            ],
+        )
+
     @task.branch
     def copy_public_data_branch(**context):
         return context["params"]["tasks"]
@@ -361,6 +433,12 @@ def copy_public_data_branch(**context):
         copy_public_data_scot_departements(),
         copy_public_data_scot_regions(),
         copy_public_data_cerema(),
+        copy_public_data_landconso(),
+        copy_public_data_landconsocomparison(),
+        copy_public_data_landconsostats(),
+        copy_public_data_landpop(),
+        copy_public_data_landpopcomparison(),
+        copy_public_data_landpopstats(),
     ]
 
 

airflow/include/sql/sparte/dbt_project.yml

@@ -13,6 +13,16 @@ clean-targets:
   - "target"
   - "dbt_packages"
 
+vars:
+  sparte:
+    for_app:
+    REGION: "REGION"
+    DEPARTEMENT: "DEPART"
+    SCOT: "SCOT"
+    EPCI: "EPCI"
+    COMMUNE: "COMM"
+    COMPOSITE: "COMP"
+    NATION: "NATION"
 
 models:
   sparte:

airflow/include/sql/sparte/macros/insee/merge_flux_population_by_admin_level.sql

@@ -6,21 +6,16 @@
     {% set code_name = group_by_column %}
 {% endif %}
 SELECT
-        {{ group_by_column }} as {{ code_name }},
-        {% call(start_year, end_year) cumulative_flux(
-            first_available_year=2009,
-            last_available_year=2020
-        ) %}
-            sum(population_{{ start_year }}_{{ end_year + 1 }})
-            as population_{{ start_year }}_{{ end_year + 1 }},
-            sum(population_{{ start_year }}_{{ end_year + 1 }}) * 100 /  sum(population_{{ start_year }})
-            as population_{{ start_year }}_{{ end_year + 1 }}_percent
-        {% endcall %}
+    {{ group_by_column }} as {{ code_name }},
+    from_year,
+    to_year,
+    {{ sum_percent_median_avg('evolution', 'start_population') }},
+    sum(start_population) as start_population
 FROM
-    {{ ref('flux_population') }} as flux_population
+    {{ ref('period_flux_population_commune') }} as population
 LEFT JOIN
-    {{ ref('commune') }}
-    ON commune.code = flux_population.code_commune
+    {{ ref('commune') }} as commune
+    ON commune.code = population.code_commune
 LEFT JOIN
     {{ ref('scot_communes') }} as scot_communes
     ON commune.code = scot_communes.commune_code
@@ -29,5 +24,7 @@ LEFT JOIN
 WHERE
     {{ group_by_column }} IS NOT NULL
 GROUP BY
-    {{ group_by_column }}
+    {{ group_by_column }},
+    from_year,
+    to_year
 {% endmacro %}