Merge pull request #519 from MTES-MCT/species_maps

Géolocalisation des espèces protégées

envergo/evaluations/models.py

@@ -97,6 +97,8 @@ def params_from_url(url):
         "non_active",
         "Non disponible",
     ),  # Same message for users, but we need to separate `non_active` and `non_disponible`
+    ("derogation_inventaire", "Dérogation"),
+    ("derogation_simplifiee", "Dérogation simplifiée"),
 )
 
 

envergo/geodata/admin.py

@@ -13,13 +13,15 @@
 
 from envergo.geodata.forms import DepartmentForm
 from envergo.geodata.models import Department, Map, Zone
-from envergo.geodata.tasks import generate_map_preview, process_shapefile_map
-from envergo.geodata.utils import count_features, extract_shapefile
+from envergo.geodata.tasks import generate_map_preview, process_map
+from envergo.geodata.utils import count_features, extract_map
 
 
 class MapForm(forms.ModelForm):
     def clean_file(self):
-        """Check that the given file is a valid shapefile archive.
+        """Check that the given file is a valid map.
+
+        We handle two formats : shapefile and geopackage.
 
         The official shapefile format is just a bunch of files with
         the same name and different extensions.
@@ -29,7 +31,7 @@ def clean_file(self):
         """
         file = self.cleaned_data["file"]
         try:
-            with extract_shapefile(file):
+            with extract_map(file):
                 pass  # This file is valid, yeah \o/
         except Exception as e:
             raise ValidationError(_(f"This file does not seem valid ({e})"))
@@ -96,7 +98,7 @@ def get_search_results(self, request, queryset, search_term):
         return queryset, may_have_duplicates
 
     def save_model(self, request, obj, form, change):
-        obj.expected_zones = count_features(obj.file)
+        obj.expected_zones = count_features(obj.file.file)
         super().save_model(request, obj, form, change)
 
     def get_queryset(self, request):
@@ -195,18 +197,16 @@ def col_zones(self, obj):
 
         return f'{imported} / {obj.expected_zones or ""}'
 
-    @admin.action(description=_("Extract and import a shapefile"))
+    @admin.action(description=_("Extract and import a map (.shp / gpkg)"))
     def process(self, request, queryset):
         if queryset.count() > 1:
             error = _("Please only select one map for this action.")
             self.message_user(request, error, level=messages.ERROR)
             return
 
         map = queryset[0]
-        process_shapefile_map.delay(map.id)
-        msg = _(
-            "Your shapefile will be processed soon. It might take up to a few minutes."
-        )
+        process_map.delay(map.id)
+        msg = _("Your map will be processed soon. It might take up to a few minutes.")
         self.message_user(request, msg, level=messages.INFO)
 
     @admin.action(description=_("Generate the simplified preview geometry"))
@@ -272,7 +272,7 @@ class ZoneAdmin(gis_admin.GISModelAdmin):
         "area",
         "npoints",
     ]
-    readonly_fields = ["map", "created_at", "area", "npoints"]
+    readonly_fields = ["map", "created_at", "area", "npoints", "attributes"]
     list_filter = ["map__map_type", "map__data_type"]
 
     # Prevent an expensive count query

envergo/geodata/migrations/0017_zone_attributes.py

@@ -0,0 +1,20 @@
+# Generated by Django 4.2.13 on 2025-01-10 10:17
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("geodata", "0016_alter_department_geometry"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="zone",
+            name="attributes",
+            field=models.JSONField(
+                blank=True, null=True, verbose_name="Entity attributes"
+            ),
+        ),
+    ]

envergo/geodata/migrations/0018_alter_map_map_type.py

@@ -0,0 +1,27 @@
+# Generated by Django 4.2.13 on 2025-01-10 10:19
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("geodata", "0017_zone_attributes"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="map",
+            name="map_type",
+            field=models.CharField(
+                blank=True,
+                choices=[
+                    ("zone_humide", "Zone humide"),
+                    ("zone_inondable", "Zone inondable"),
+                    ("species", "Espèces protégées"),
+                ],
+                max_length=50,
+                verbose_name="Map type",
+            ),
+        ),
+    ]

envergo/geodata/models.py

@@ -1,8 +1,4 @@
-import glob
 import logging
-import zipfile
-from contextlib import contextmanager
-from tempfile import TemporaryDirectory
 
 from django.contrib.gis.db import models as gis_models
 from django.contrib.postgres.fields import ArrayField
@@ -24,6 +20,7 @@
 MAP_TYPES = Choices(
     ("zone_humide", _("Zone humide")),
     ("zone_inondable", _("Zone inondable")),
+    ("species", _("Espèces protégées")),
 )
 
 # Sometimes, there are map with different certainty values.
@@ -42,7 +39,7 @@
 
 
 class Map(models.Model):
-    """Holds a shapefile map."""
+    """Holds a map file (shapefile / gpkg)."""
 
     name = models.CharField(_("Name"), max_length=256)
     display_name = models.CharField(_("Display name"), max_length=256, blank=True)
@@ -103,18 +100,6 @@ class Meta:
     def __str__(self):
         return self.name
 
-    @contextmanager
-    def extract_shapefile(self):
-        with TemporaryDirectory() as tmpdir:
-            logger.info("Extracting map zip file")
-            zf = zipfile.ZipFile(self.file)
-            zf.extractall(tmpdir)
-
-            logger.info("Find .shp file path")
-            paths = glob.glob(f"{tmpdir}/*shp")  # glop glop !
-            shapefile = paths[0]
-            yield shapefile
-
 
 class Zone(gis_models.Model):
     """Stores an annotated geographic polygon(s)."""
@@ -131,6 +116,7 @@ class Zone(gis_models.Model):
     area = models.BigIntegerField(_("Area"), null=True, blank=True)
     npoints = models.BigIntegerField(_("Number of points"), null=True, blank=True)
     created_at = models.DateTimeField(_("Date created"), default=timezone.now)
+    attributes = models.JSONField(_("Entity attributes"), null=True, blank=True)
 
     class Meta:
         verbose_name = _("Zone")

envergo/geodata/tasks.py

@@ -5,14 +5,14 @@
 
 from config.celery_app import app
 from envergo.geodata.models import STATUSES, Map
-from envergo.geodata.utils import make_polygons_valid, process_shapefile, simplify_map
+from envergo.geodata.utils import make_polygons_valid, process_map_file, simplify_map
 
 logger = logging.getLogger(__name__)
 
 
 @app.task(bind=True)
 @transaction.atomic
-def process_shapefile_map(task, map_id):
+def process_map(task, map_id):
     logger.info(f"Starting import on map {map_id}")
 
     map = Map.objects.get(pk=map_id)
@@ -28,7 +28,7 @@ def process_shapefile_map(task, map_id):
     try:
         with transaction.atomic():
             map.zones.all().delete()
-            process_shapefile(map, map.file, task)
+            process_map_file(map, map.file, task)
             make_polygons_valid(map)
             map.geometry = simplify_map(map)
     except Exception as e:

envergo/geodata/utils.py

@@ -64,43 +64,78 @@ def __init__(self, *args, **kwargs):
     def feature_kwargs(self, feat):
         kwargs = super().feature_kwargs(feat)
         kwargs.update(self.extra_kwargs)
+
+        # We extract map attributes to the `attributes` json model field
+        fields = feat.fields
+        attributes = {f: self.get_attribute(feat, f) for f in fields}
+        kwargs["attributes"] = attributes
         return kwargs
 
+    def get_attribute(self, feat, field):
+        """Extract map attribute.
 
-@contextmanager
-def extract_shapefile(archive):
-    """Extract a shapefile from a zip archive."""
+        We can define custom methods for specific fields.
+        """
+        if f"get_attribute_{field}" in dir(self):
+            attr = getattr(self, f"get_attribute_{field}")(feat)
+        else:
+            attr = feat.get(field)
+        return attr
 
-    with TemporaryDirectory() as tmpdir:
-        logger.info("Extracting map zip file")
-        zf = zipfile.ZipFile(archive)
-        zf.extractall(tmpdir)
+    def get_attribute_especes(self, feat):
+        raw_especes = feat.get("especes")
+        especes = list(map(int, raw_especes.split(",")))
+        return especes
 
-        logger.info("Find .shp file path")
-        paths = glob.glob(f"{tmpdir}/*shp")  # glop glop !
 
-        try:
-            shapefile = paths[0]
-        except IndexError:
-            raise ValueError(_("No .shp file found in archive"))
+@contextmanager
+def extract_map(archive):
+    """Returns the path to the map file.
 
-        yield shapefile
+    If this is a zipped shapefile, extract it to a temporary directory.
+    """
+    if archive.name.endswith(".zip"):
+        with TemporaryDirectory() as tmpdir:
+            logger.info("Extracting map zip file")
+            zf = zipfile.ZipFile(archive)
+            zf.extractall(tmpdir)
+
+            logger.info("Find .shp file path")
+            paths = glob.glob(f"{tmpdir}/*shp")  # glop glop !
+
+            try:
+                shapefile = paths[0]
+            except IndexError:
+                raise ValueError(_("No .shp file found in archive"))
+
+            yield shapefile
+
+    elif archive.name.endswith(".gpkg"):
+        if hasattr(archive, "temporary_file_path"):
+            yield archive.temporary_file_path()
+        elif hasattr(archive, "url"):
+            yield archive.url
+        else:
+            yield archive.path
+
+    else:
+        raise ValueError(_("Unsupported file format"))
 
 
-def count_features(shapefile):
+def count_features(map_file):
     """Count the number of features from a shapefile."""
 
-    with extract_shapefile(shapefile) as file:
+    with extract_map(map_file) as file:
         ds = DataSource(file)
         layer = ds[0]
         nb_zones = len(layer)
 
     return nb_zones
 
 
-def process_shapefile(map, file, task=None):
+def process_map_file(map, file, task=None):
     logger.info("Creating temporary directory")
-    with extract_shapefile(file) as shapefile:
+    with extract_map(file) as map_file:
         if task:
             debug_stream = CeleryDebugStream(task, map.expected_zones)
         else:
@@ -111,7 +146,7 @@ def process_shapefile(map, file, task=None):
         extra = {"map": map}
         lm = CustomMapping(
             Zone,
-            shapefile,
+            map_file,
             mapping,
             transaction_mode="autocommit",
             extra_kwargs=extra,

envergo/hedges/admin.py

@@ -97,10 +97,13 @@ class SpeciesAdmin(admin.ModelAdmin):
         "common_name",
         "scientific_name",
         "group",
-        "hedge_types",
         "level_of_concern",
         "highly_sensitive",
+        "hedge_types",
+        "taxref_ids",
     ]
+    search_fields = ["group", "common_name", "scientific_name"]
     ordering = ["-common_name"]
     form = SpeciesAdminForm
-    list_filter = ["group"]
+    list_filter = ["group", "level_of_concern", "highly_sensitive"]
+    readonly_fields = ["kingdom", "taxref_ids"]

envergo/hedges/management/commands/import_taxref.py

@@ -0,0 +1,65 @@
+import csv
+import glob
+import io
+import os
+import zipfile
+from tempfile import TemporaryDirectory
+
+import requests
+from django.core.management.base import BaseCommand
+
+from envergo.hedges.models import Species
+
+# Download link can be found here
+# https://inpn.mnhn.fr/telechargement/referentielEspece/referentielTaxo
+TAXREF_URL = "https://inpn.mnhn.fr/docs-web/docs/download/454260"
+
+
+class Command(BaseCommand):
+    help = "Update species data with TaxRef identifiers (cd_nom)."
+
+    def add_arguments(self, parser):
+        parser.add_argument("taxref_url", type=str, nargs="?", default=TAXREF_URL)
+
+    def handle(self, *args, **options):
+
+        # Read the taxref file
+        taxref_url = options["taxref_url"]
+        if os.path.exists(taxref_url):
+            with open(taxref_url, "rb") as f:
+                file_content = io.BytesIO(f.read())
+        else:
+            r = requests.get(taxref_url, stream=True)
+            file_content = io.BytesIO(r.content)
+
+        with TemporaryDirectory() as tmpdir:
+            zf = zipfile.ZipFile(file_content)
+            zf.extractall(tmpdir)
+
+            paths = glob.glob(f"{tmpdir}/TAXREFv*.txt")
+            try:
+                path = paths[0]
+            except IndexError:
+                self.stderr.write(self.style.ERROR("No TAXREF file found"))
+                return
+
+            # Reset taxref ids for all species
+            Species.objects.update(taxref_ids=[])
+            species_names = Species.objects.all().values_list(
+                "scientific_name", flat=True
+            )
+
+            with open(path) as csvfile:
+                reader = csv.DictReader(csvfile, delimiter="\t")
+                for row in reader:
+                    scientific_name = row["LB_NOM"]
+                    vernacular_name_id = row["CD_NOM"]
+                    kingdom = row["REGNE"].lower()
+                    if scientific_name in species_names:
+                        # AFAIK, there is still no way to update an array field in a
+                        # sigle query. Issue open since 9 years
+                        # https://code.djangoproject.com/ticket/26355
+                        species = Species.objects.get(scientific_name=scientific_name)
+                        species.taxref_ids.append(vernacular_name_id)
+                        species.kingdom = kingdom
+                        species.save()