Back : Ajout des données vulnérabilités + commande de génération MVT

back/api/constants.py

@@ -13,3 +13,4 @@ class GeoLevel(TextChoices):
 class DataType(TextChoices):
     LCZ = "lcz", "LCZ"
     TILE = "plantability", "Plantability"
+    VULNERABILITY = "vulnerability", "Vulnerability"

back/api/management/commands/generate_mvt.py

@@ -26,9 +26,9 @@
 from django.core.management import BaseCommand
 from django.db.models import QuerySet, Model
 
-from api.constants import DEFAULT_ZOOM_LEVELS, GeoLevel
+from api.constants import DEFAULT_ZOOM_LEVELS, GeoLevel, DataType
 from api.utils.mvt_generator import MVTGenerator
-from iarbre_data.models import Tile, Lcz, MVTTile
+from iarbre_data.models import Tile, Lcz, Vulnerability, MVTTile
 
 
 class Command(BaseCommand):
@@ -48,6 +48,13 @@ def add_arguments(self, parser):
             choices=[choice for choice, _ in GeoLevel.choices],
             help=f"What geolevel to transform to MVT. Choices: {', '.join([choice for choice, _ in GeoLevel.choices])}",
         )
+        parser.add_argument(
+            "--datatype",
+            type=str,
+            required=True,
+            choices=[choice for choice, _ in DataType.choices],
+            help=f"What datatype to transform to MVT. Choices: {', '.join([choice for choice, _ in DataType.choices])}",
+        )
         parser.add_argument(
             "--keep",
             action="store_true",
@@ -95,10 +102,15 @@ def handle(self, *args, **options):
         """Handle the command."""
         number_of_thread = options["number_of_thread"]
         geolevel = options["geolevel"]
+        datatype = options["datatype"]
         if geolevel == GeoLevel.TILE.value:
             mdl = Tile
-        elif geolevel == GeoLevel.LCZ.value:
+        elif geolevel == GeoLevel.LCZ.value and datatype == DataType.LCZ.value:
             mdl = Lcz
+        elif (
+            geolevel == GeoLevel.LCZ.value and datatype == DataType.VULNERABILITY.value
+        ):
+            mdl = Vulnerability
         else:
             supported_levels = [GeoLevel.TILE.value, GeoLevel.LCZ.value]
             raise ValueError(

back/iarbre_data/management/commands/import_vulnerability.py

@@ -0,0 +1,146 @@
+"""Heat vulnerability
+
+The script assumes that the vulnerability data is stored in a geopackage file located in the 'file_data/vulnerability' directory.
+This geopackage has been produced by Maurine Di Tommaso (Service Climat & Résilience – Direction Environnement, Écologie, Énergie).
+A description of the approach can be found here : https://geoweb.grandlyon.com/portal/apps/storymaps/collections/7e7862ec92694601a7085074dcaf7481?item=3
+"""
+import geopandas
+import os
+
+from django.contrib.gis.geos import GEOSGeometry
+from django.core.management import BaseCommand
+from tqdm import tqdm
+
+
+from iarbre_data.models import Vulnerability
+from iarbre_data.settings import TARGET_MAP_PROJ, TARGET_PROJ
+from iarbre_data.management.commands.utils import log_progress
+
+
+def load_data():
+    """Open the geopackage for vulnerabilty.
+
+    Returns:
+        geopandas.GeoDataFrame: The loaded geopackage as a GeoDataFrame.
+
+    Raises:
+        FileNotFoundError: If no folder with "vulnerability" in the name is found or no .gpkg file is found in the folder.
+    """
+    vulnerability_path = "file_data/vulnerability"
+    if not os.path.isdir(vulnerability_path):
+        raise FileNotFoundError(
+            "No folder for 'vulnerability' found in 'file_data/' directory."
+        )
+    gpkg_file = None
+    for file in os.listdir(vulnerability_path):
+        if file.lower().endswith(".gpkg"):
+            gpkg_file = file
+            break
+    if not gpkg_file:
+        raise FileNotFoundError(
+            f"No geopackage file found in the folder '{vulnerability_path}'."
+        )
+
+    gpkg_path = os.path.join(vulnerability_path, gpkg_file)
+    gdf = geopandas.read_file(gpkg_path, layer="vulnérabilité_fortes_chaleurs")
+    gdf.to_crs(TARGET_PROJ, inplace=True)
+    gdf_details = geopandas.read_file(
+        gpkg_path, layer="Vulnerabilite_fortes_chaleurs_détail"
+    )
+    gdf_details.to_crs(TARGET_PROJ, inplace=True)
+
+    merged = gdf.merge(
+        gdf_details, on="ID_RSU", suffixes=("", "_details"), validate="one_to_one"
+    )
+    merged.drop(
+        columns=[
+            "id",
+            "commune",
+            "ID_RSU",
+            "Surface_RSU",
+            "NOTE_EXPO_JOUR",
+            "NOTE_EXPO_NUIT",
+            "NOTE_SENSI_JOUR",
+            "NOTE_SENSI_NUIT",
+            "NOTE_CAPAF_JOUR",
+            "NOTE_CAPAF_NUIT",
+        ],
+        inplace=True,
+    )
+    duplicate_columns = [
+        col
+        for col in merged.columns
+        if col.endswith("_details") and col[:-8] in merged.columns
+    ]
+
+    for col in duplicate_columns:
+        original_col = col[:-8]
+        if not merged[original_col].equals(merged[col]):
+            raise ValueError(f"Mismatch found in column '{original_col}' and '{col}'.")
+
+    merged.drop(columns=duplicate_columns, inplace=True)
+
+    def make_valid(geometry):
+        """Fix minor topology errors, like Polygon not closed."""
+        if geometry and not geometry.is_valid:
+            return geometry.buffer(0)
+        return geometry
+
+    merged["geometry"] = merged["geometry"].apply(make_valid)
+    merged["map_geometry"] = merged.geometry.to_crs(TARGET_MAP_PROJ)
+    merged["map_geometry"] = merged["map_geometry"].apply(make_valid)
+
+    merged.fillna(
+        0, inplace=True
+    )  # Columns ['majic_Log_av1949', 'majic_Log_1949_1989',
+    # 'majic_Log_ap1990', 'majic_nlogh', 'majic_stoth', 'majic_slocal',
+    #        'majic_nloghmais', 'majic_nloghappt', 'siret_densite_emploi'] contains NaN because
+    # these data (economical indices) are missing in water, forest, etc.
+    # For all this values 0 means absence.
+
+    return merged
+
+
+def save_geometries(vulnearbility_datas: geopandas.GeoDataFrame) -> None:
+    """Save vulnerability data to the database.
+
+    Args:
+        vulnearbility_datas (GeoDataFrame): GeoDataFrame to save to the database.
+
+    Returns:
+        None
+    """
+    batch_size = 10000
+    for start in tqdm(range(0, len(vulnearbility_datas), batch_size)):
+        end = start + batch_size
+        batch = vulnearbility_datas.iloc[start:end]
+        Vulnerability.objects.bulk_create(
+            [
+                Vulnerability(
+                    geometry=GEOSGeometry(data["geometry"].wkt),
+                    map_geometry=GEOSGeometry(data["map_geometry"].wkt),
+                    vulnerability_index_day=data["VULNERABILITE_JOUR"],
+                    vulnerability_index_night=data["VULNERABILITE_NUIT"],
+                    expo_index_day=data["EXPO_JOUR"],
+                    expo_index_night=data["EXPO_NUIT"],
+                    capaf_index_day=data["CAPAF_JOUR"],
+                    capaf_index_night=data["CAPAF_NUIT"],
+                    sensibilty_index_day=data["SENSI_JOUR"],
+                    sensibilty_index_night=data["SENSI_NUIT"],
+                )
+                for _, data in batch.iterrows()
+            ]
+        )
+
+
+class Command(BaseCommand):
+    help = "Load heat vulnerability data in the DB."
+
+    def handle(self, *args, **options):
+        """Load heat vulnerability data in the DB."""
+        log_progress("Remove existing data")
+        print(Vulnerability.objects.all().delete())
+        log_progress("Loading data")
+        lcz_data = load_data()
+        log_progress("Saving data")
+        save_geometries(lcz_data)

back/iarbre_data/management/commands/utils.py

@@ -1,3 +1,4 @@
+from datetime import datetime
 import shapely
 from django.db.models import Count
 
@@ -52,8 +53,8 @@ def remove_duplicates(Model) -> None:
     print(f"Removed duplicates for {duplicates.count()} entries.")
 
 
-def select_city(insee_code_city: str):
-    """Select a list of city based on INSEE_CODE.
+def select_city(insee_code_city: str) -> gpd.GeoDataFrame:
+    """Select a list of cities based on INSEE_CODE.
 
     Args:
         insee_code_city (str): INSEE code of the city or cities to select.
@@ -76,3 +77,14 @@ def select_city(insee_code_city: str):
             ["id", "name", "code", "tiles_generated", "tiles_computed"],
         )
     return selected_city
+
+
+def log_progress(step: str) -> None:
+    """
+    Log the progress of a step with a timestamp.
+
+    Args:
+        step (str): The description of the step being logged.
+    """
+    print("*" * 30)
+    print(f"{datetime.now().strftime('%H:%M:%S')} - {step}")

back/iarbre_data/migrations/0020_vulnerability_alter_mvttile_datatype.py

@@ -0,0 +1,59 @@
+# Generated by Django 5.1.5 on 2025-03-26 09:50
+
+import django.contrib.gis.db.models.fields
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("iarbre_data", "0019_merge_20250310_1025"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="Vulnerability",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                (
+                    "geometry",
+                    django.contrib.gis.db.models.fields.PolygonField(srid=2154),
+                ),
+                (
+                    "map_geometry",
+                    django.contrib.gis.db.models.fields.PolygonField(
+                        blank=True, null=True, srid=3857
+                    ),
+                ),
+                ("vulnerability_index_day", models.FloatField(null=True)),
+                ("vulnerability_index_night", models.FloatField(null=True)),
+                ("expo_index_day", models.FloatField(null=True)),
+                ("expo_index_night", models.FloatField(null=True)),
+                ("capaf_index_day", models.FloatField(null=True)),
+                ("capaf_index_night", models.FloatField(null=True)),
+                ("sensibilty_index_day", models.FloatField(null=True)),
+                ("sensibilty_index_night", models.FloatField(null=True)),
+            ],
+        ),
+        migrations.AlterField(
+            model_name="mvttile",
+            name="datatype",
+            field=models.CharField(
+                choices=[
+                    ("lcz", "LCZ"),
+                    ("plantability", "Plantability"),
+                    ("vulnerability", "Vulnerability"),
+                ],
+                default="plantability",
+                max_length=50,
+            ),
+        ),
+    ]

back/iarbre_data/models.py

@@ -9,6 +9,12 @@
 from api.constants import GeoLevel, DataType
 
 
+def create_mapgeometry(instance):
+    """Transform the geometry to the map geometry."""
+    if instance.map_geometry is None:
+        instance.map_geometry = instance.geometry.transform(TARGET_MAP_PROJ, clone=True)
+
+
 class TileAggregateBase(models.Model):
     """Abstract base class for aggregating Tiles at IRIS and city level."""
 
@@ -95,13 +101,6 @@ def get_layer_properties(self):
         }
 
 
-@receiver(pre_save, sender=Tile)
-def before_save_tile(sender, instance, **kwargs):
-    """Transform the geometry to the map geometry."""
-    if instance.map_geometry is None:
-        instance.map_geometry = instance.geometry.transform(3857, clone=True)
-
-
 class Data(models.Model):
     """Land occupancy data"""
 
@@ -192,11 +191,56 @@ def get_layer_properties(self):
         }
 
 
+class Vulnerability(models.Model):
+    """Elementary element on the map with the value of the vulnerability description."""
+
+    geometry = PolygonField(srid=2154)
+    map_geometry = PolygonField(srid=TARGET_MAP_PROJ, null=True, blank=True)
+    vulnerability_index_day = models.FloatField(null=True)
+    vulnerability_index_night = models.FloatField(null=True)
+    expo_index_day = models.FloatField(null=True)
+    expo_index_night = models.FloatField(null=True)
+    capaf_index_day = models.FloatField(null=True)
+    capaf_index_night = models.FloatField(null=True)
+    sensibilty_index_day = models.FloatField(null=True)
+    sensibilty_index_night = models.FloatField(null=True)
+
+    geolevel = GeoLevel.LCZ.value
+    datatype = DataType.VULNERABILITY.value
+
+    @property
+    def color(self):
+        """Return the color of the ICU based on the vulnerability_index_day."""
+        if self.vulnerability_index_day is None:
+            return "purple"
+        elif self.vulnerability_index_day < 3:
+            return "#006837"
+        elif self.vulnerability_index_day < 6:
+            return "#E0E0E0"
+        else:
+            return "red"
+
+    def get_layer_properties(self):
+        """Return the properties of the tile for the MVT datatype."""
+        return {
+            "id": self.id,
+            "indice_day": self.vulnerability_index_day,
+            "indice_night": self.vulnerability_index_night,
+            "expo_index_day": self.expo_index_day,
+            "expo_index_night": self.expo_index_night,
+            "capaf_index_day": self.capaf_index_day,
+            "capaf_index_night": self.capaf_index_night,
+            "sensibilty_index_day": self.sensibilty_index_day,
+            "sensibilty_index_night": self.sensibilty_index_night,
+            "color": self.color,
+        }
+
+
 @receiver(pre_save, sender=Lcz)
-def before_save_lcz(sender, instance, **kwargs):
-    """Transform the geometry to the map geometry."""
-    if instance.map_geometry is None:
-        instance.map_geometry = instance.geometry.transform(3857, clone=True)
+@receiver(pre_save, sender=Vulnerability)
+@receiver(pre_save, sender=Tile)
+def before_save(sender, instance, **kwargs):
+    create_mapgeometry(instance)
 
 
 class Feedback(models.Model):