feat: add management command to spatially attach trees to green sections with test coverage
This commit is contained in:
parent
958d8dbc65
commit
e201a928b8
2 changed files with 252 additions and 0 deletions
|
|
@ -0,0 +1,220 @@
|
|||
import time
|
||||
from datetime import datetime
|
||||
from django.core.management.base import BaseCommand
|
||||
from django.db import connection, transaction
|
||||
from assets.models import NatureLocation, NatureTree
|
||||
from common.models import Municipality
|
||||
|
||||
|
||||
MUNICIPALITY_PREFIXES = {
|
||||
'AND': 'Anderlecht',
|
||||
'AUD': 'Auderghem',
|
||||
'BRK': 'Berchem-Sainte-Agathe',
|
||||
'BSA': 'Berchem-Sainte-Agathe',
|
||||
'BXL': 'Bruxelles',
|
||||
'BRU': 'Bruxelles',
|
||||
'ETT': 'Etterbeek',
|
||||
'EVE': 'Evere',
|
||||
'FOR': 'Forest',
|
||||
'GAN': 'Ganshoren',
|
||||
'IXL': 'Ixelles',
|
||||
'IXE': 'Ixelles',
|
||||
'JET': 'Jette',
|
||||
'KOE': 'Koekelberg',
|
||||
'MOL': 'Molenbeek-Saint-Jean',
|
||||
'STG': 'Saint-Gilles',
|
||||
'STJ': 'Saint-Josse-ten-Noode',
|
||||
'SCH': 'Schaerbeek',
|
||||
'UCC': 'Uccle',
|
||||
'WAT': 'Watermael-Boitsfort',
|
||||
'WLU': 'Woluwe-Saint-Lambert',
|
||||
'WSL': 'Woluwe-Saint-Lambert',
|
||||
'WSP': 'Woluwe-Saint-Pierre',
|
||||
}
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
help = "Rattache spatialement les arbres (NatureTree) aux tronçons des sections vertes (NatureLocation, ex: WAT_R22_008)."
|
||||
|
||||
def add_arguments(self, parser):
|
||||
parser.add_argument(
|
||||
'--max-distance',
|
||||
type=float,
|
||||
default=25.0,
|
||||
help="Distance maximale en mètres pour le rattachement (défaut: 25.0m, 0 pour intersection stricte)"
|
||||
)
|
||||
parser.add_argument(
|
||||
'--section-code',
|
||||
type=str,
|
||||
default=None,
|
||||
help="Code d'un tronçon spécifique à cibler (ex: WAT_R22_008)"
|
||||
)
|
||||
parser.add_argument(
|
||||
'--dry-run',
|
||||
action='store_true',
|
||||
help="Simuler le rattachement sans modifier la base de données"
|
||||
)
|
||||
parser.add_argument(
|
||||
'--skip-sync-munis',
|
||||
action='store_true',
|
||||
help="Ne pas synchroniser la commune sur les sections vertes"
|
||||
)
|
||||
|
||||
def log(self, msg):
|
||||
self.stdout.write(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}")
|
||||
|
||||
def handle(self, *args, **options):
|
||||
max_dist = options['max_distance']
|
||||
target_section = options['section_code']
|
||||
dry_run = options['dry_run']
|
||||
sync_munis = not options['skip_sync_munis']
|
||||
|
||||
self.log("=== RATTACHEMENT SPATIAL DES ARBRES AUX TRONÇONS DE SECTIONS VERTES ===")
|
||||
self.log(f"Distance max : {max_dist} m {'(Intersection stricte)' if max_dist == 0 else ''}")
|
||||
if target_section:
|
||||
self.log(f"Section ciblée : {target_section}")
|
||||
if dry_run:
|
||||
self.log("[SIMULATION / DRY-RUN] Aucune modification ne sera enregistrée.")
|
||||
|
||||
# -----------------------------------------------------------------
|
||||
# 1. Synchronisation des communes sur les sections vertes
|
||||
# -----------------------------------------------------------------
|
||||
if sync_munis and not dry_run:
|
||||
self.log("1. Synchronisation des communes sur les sections vertes...")
|
||||
munis_by_name = {m.name_fr.lower(): m for m in Municipality.objects.all()}
|
||||
updated_munis = 0
|
||||
|
||||
with connection.cursor() as cursor:
|
||||
# Intersections géométriques
|
||||
cursor.execute("""
|
||||
UPDATE assets_naturelocation l
|
||||
SET municipality_id = m.id
|
||||
FROM common_municipality m
|
||||
WHERE l.code ~ '^[A-Z]{3}_'
|
||||
AND l.municipality_id IS NULL
|
||||
AND l.geom IS NOT NULL
|
||||
AND ST_Intersects(ST_PointOnSurface(l.geom), m.geom);
|
||||
""")
|
||||
updated_munis += cursor.rowcount
|
||||
|
||||
# Complément par préfixe de code pour les cas frontaliers
|
||||
locs_without_muni = NatureLocation.objects.filter(code__regex=r'^[A-Z]{3}_', municipality__isnull=True)
|
||||
for loc in locs_without_muni:
|
||||
pfx = loc.code[:3].upper()
|
||||
muni_name = MUNICIPALITY_PREFIXES.get(pfx)
|
||||
if muni_name and muni_name.lower() in munis_by_name:
|
||||
loc.municipality = munis_by_name[muni_name.lower()]
|
||||
loc.save(update_fields=['municipality'])
|
||||
updated_munis += 1
|
||||
|
||||
self.log(f" Communes renseignées sur {updated_munis} sections vertes.")
|
||||
|
||||
# -----------------------------------------------------------------
|
||||
# 2. Analyse et Rattachement des Arbres
|
||||
# -----------------------------------------------------------------
|
||||
t0 = time.time()
|
||||
self.log("2. Analyse géospatiale du rattachement arbres -> sections vertes...")
|
||||
|
||||
section_filter_sql = ""
|
||||
params = [max_dist]
|
||||
if target_section:
|
||||
section_filter_sql = "AND nearest.code = %s"
|
||||
params.append(target_section)
|
||||
|
||||
count_query = f"""
|
||||
WITH nearest AS (
|
||||
SELECT
|
||||
t.id AS tree_id,
|
||||
nearest_loc.id AS loc_id,
|
||||
nearest_loc.code AS loc_code,
|
||||
nearest_loc.name_fr AS loc_name,
|
||||
nearest_loc.dist
|
||||
FROM assets_naturetree t
|
||||
CROSS JOIN LATERAL (
|
||||
SELECT l.id, l.code, l.name_fr, ST_Distance(t.geom, l.geom) AS dist
|
||||
FROM assets_naturelocation l
|
||||
WHERE l.code ~ '^[A-Z]{{3}}_' AND l.geom IS NOT NULL
|
||||
ORDER BY t.geom <-> l.geom
|
||||
LIMIT 1
|
||||
) nearest_loc
|
||||
WHERE t.geom IS NOT NULL AND t.lat != 0 AND t.lon != 0
|
||||
AND nearest_loc.dist <= %s
|
||||
)
|
||||
SELECT
|
||||
COUNT(*) AS total_matched,
|
||||
COUNT(*) FILTER (WHERE dist = 0) AS inside,
|
||||
COUNT(*) FILTER (WHERE dist > 0 AND dist <= 5) AS within_5m,
|
||||
COUNT(*) FILTER (WHERE dist > 5 AND dist <= 15) AS within_15m,
|
||||
COUNT(*) FILTER (WHERE dist > 15 AND dist <= 25) AS within_25m,
|
||||
COUNT(*) FILTER (WHERE dist > 25) AS beyond_25m,
|
||||
COUNT(DISTINCT loc_id) AS distinct_sections
|
||||
FROM nearest
|
||||
WHERE 1=1 {section_filter_sql};
|
||||
"""
|
||||
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute(count_query, params)
|
||||
stats = cursor.fetchone()
|
||||
|
||||
total_matched = stats[0]
|
||||
inside = stats[1]
|
||||
w5m = stats[2]
|
||||
w15m = stats[3]
|
||||
w25m = stats[4]
|
||||
b25m = stats[5]
|
||||
distinct_sections = stats[6]
|
||||
|
||||
self.log(f" Arbres éligibles au rattachement : {total_matched}")
|
||||
self.log(f" - Directement à l'intérieur (0m) : {inside}")
|
||||
self.log(f" - Proximité immédiate (1 à 5m) : {w5m}")
|
||||
self.log(f" - Bordure de voirie (5 à 15m) : {w15m}")
|
||||
self.log(f" - Zone étendue (15 à 25m) : {w25m}")
|
||||
if b25m:
|
||||
self.log(f" - Au-delà (> 25m) : {b25m}")
|
||||
self.log(f" Nombre de tronçons / sections concernés : {distinct_sections}")
|
||||
|
||||
if dry_run:
|
||||
self.log("Simulation terminée avec succès (aucune écriture en base).")
|
||||
return
|
||||
|
||||
# -----------------------------------------------------------------
|
||||
# 3. Application de la mise à jour
|
||||
# -----------------------------------------------------------------
|
||||
self.log("3. Application de la mise à jour des liaisons (location_id)...")
|
||||
update_query = f"""
|
||||
WITH candidates AS (
|
||||
SELECT
|
||||
t.id AS tree_id,
|
||||
nearest.id AS loc_id
|
||||
FROM assets_naturetree t
|
||||
CROSS JOIN LATERAL (
|
||||
SELECT l.id, l.code, ST_Distance(t.geom, l.geom) AS dist
|
||||
FROM assets_naturelocation l
|
||||
WHERE l.code ~ '^[A-Z]{{3}}_' AND l.geom IS NOT NULL
|
||||
ORDER BY t.geom <-> l.geom
|
||||
LIMIT 1
|
||||
) nearest
|
||||
WHERE t.geom IS NOT NULL AND t.lat != 0 AND t.lon != 0
|
||||
AND nearest.dist <= %s
|
||||
{section_filter_sql}
|
||||
)
|
||||
UPDATE assets_naturetree t
|
||||
SET location_id = c.loc_id
|
||||
FROM candidates c
|
||||
WHERE t.id = c.tree_id
|
||||
AND (t.location_id IS DISTINCT FROM c.loc_id);
|
||||
"""
|
||||
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute(update_query, params)
|
||||
updated_count = cursor.rowcount
|
||||
|
||||
t1 = time.time()
|
||||
self.log(f"Mise à jour terminée : {updated_count} arbres rattachés en {t1 - t0:.2f} secondes.")
|
||||
|
||||
# Vérification ciblée (par exemple WAT_R22_008)
|
||||
check_code = target_section or 'WAT_R22_008'
|
||||
check_loc = NatureLocation.objects.filter(code=check_code).first()
|
||||
if check_loc:
|
||||
attached_cnt = check_loc.nature_trees.count()
|
||||
self.log(f"Vérification sur {check_code} ({check_loc.name_fr}) : {attached_cnt} arbres rattachés.")
|
||||
|
|
@ -181,4 +181,36 @@ class ImportFulcrumTreesCommandTest(TestCase):
|
|||
call_command('import_fulcrum_regional_trees', skip_trees=True, skip_gestion=True, skip_interventions=True, stdout=out)
|
||||
self.assertIn("FIN DE L'IMPORTATION", out.getvalue())
|
||||
|
||||
def test_attach_trees_to_green_sections(self):
|
||||
from assets.models import NatureLocation
|
||||
from django.contrib.gis.geos import Polygon, MultiPolygon
|
||||
# Créer une section verte polygonale
|
||||
poly = Polygon(((648000, 670000), (648100, 670000), (648100, 670100), (648000, 670100), (648000, 670000)), srid=3812)
|
||||
loc = NatureLocation.objects.create(
|
||||
code="WAT_R22_008",
|
||||
name_fr="Boulevard du Souverain",
|
||||
geom=MultiPolygon(poly, srid=3812)
|
||||
)
|
||||
# Créer un arbre à l'intérieur du polygone (sans location)
|
||||
pt = Point(648050, 670050, srid=3812)
|
||||
pt_4326 = pt.clone().transform(4326, clone=True)
|
||||
tree = NatureTree.objects.create(
|
||||
code="TR-ATTACH-TEST",
|
||||
name_fr="Arbre test Souverain",
|
||||
category=self.category,
|
||||
geom=pt,
|
||||
lon=pt_4326.x,
|
||||
lat=pt_4326.y,
|
||||
location=None
|
||||
)
|
||||
self.assertEqual(loc.nature_trees.count(), 0)
|
||||
|
||||
# Exécuter le rattachement
|
||||
call_command('attach_trees_to_green_sections', max_distance=25, skip_sync_munis=True)
|
||||
|
||||
tree.refresh_from_db()
|
||||
self.assertEqual(tree.location_id, loc.id)
|
||||
self.assertEqual(loc.nature_trees.count(), 1)
|
||||
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue