Valogeo/loko
Valogeo/loko
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: add management command to import Fulcrum tree inventories and interventions

Browse source
This commit is contained in:
kdeterme 2026-10-10 12:17:57 +02:00
parent b953b1f984
commit 958d8dbc65
3 changed files with 1126 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
loko/assets/management/commands/import_fulcrum_regional_trees.py Normal file
View file

@ -0,0 +1,5 @@
from assets.management.commands.import_fulcrum_trees import Command as BaseFulcrumCommand
class Command(BaseFulcrumCommand):
help = "Importe l'inventaire Fulcrum des arbres régionaux (arbres, commandes/gestion, interventions)."

937
loko/assets/management/commands/import_fulcrum_trees.py Normal file
View file

@ -0,0 +1,937 @@
import os
import csv
import json
import re
from datetime import datetime, date
from collections import defaultdict
from django.core.management.base import BaseCommand
from django.db import transaction
from django.contrib.contenttypes.models import ContentType
from django.contrib.gis.geos import Point, MultiPolygon
from common.models import Thematic, Municipality
from assets.models import AssetCategory, NatureTree, NatureAssetModel, NatureLocation
from interventions.models import (
Intervention, InterventionAsset, SourceCategory, Symptom, InterventionSequence
)
from contracts.models import Contract, ContractOrder, Company
try:
from assets.utils.tree_ecoservices import compute_tree_ecoservices
except Exception:
compute_tree_ecoservices = None
class Command(BaseCommand):
help = "Importe ou met à jour l'inventaire Fulcrum des arbres, commandes/gestion et interventions."
def add_arguments(self, parser):
parser.add_argument(
'--arbes-file',
default='/tmp/fulcrum_arbes.geojson',
help="Chemin vers fulcrum_arbes.geojson"
)
parser.add_argument(
'--gestion-file',
default='/tmp/fulcrum_gestion.csv',
help="Chemin vers fulcrum_gestion.csv"
)
parser.add_argument(
'--interventions-file',
default='/tmp/fulcrum_interventions.geojson',
help="Chemin vers fulcrum_interventions.geojson"
)
parser.add_argument(
'--managing-authority',
default='Bruxelles Mobilité',
help="Autorité gestionnaire (défaut: Bruxelles Mobilité)"
)
parser.add_argument(
'--management-type',
default='regional',
choices=['municipal', 'regional', 'community', 'private'],
help="Type de gestionnaire (municipal, regional, community, private)"
)
parser.add_argument(
'--code-prefix',
default='BM-ARB-',
help="Préfixe pour les codes d'arbres générés (défaut: BM-ARB-)"
)
parser.add_argument(
'--company-name',
default='Krinkels',
help="Nom de l'entreprise prestataire par défaut (défaut: Krinkels)"
)
parser.add_argument(
'--fallback-contract',
default='X00.007',
help="Numéro de contrat de repli (défaut: X00.007)"
)
parser.add_argument(
'--skip-trees',
action='store_true',
help="Ignorer l'importation des arbres"
)
parser.add_argument(
'--skip-gestion',
action='store_true',
help="Ignorer l'importation de la gestion/commandes"
)
parser.add_argument(
'--skip-interventions',
action='store_true',
help="Ignorer l'importation des interventions"
)
parser.add_argument(
'--limit',
type=int,
default=0,
help="Nombre maximal d'éléments à traiter par étape (0 = tout)"
)
def log(self, msg):
self.stdout.write(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}")
def handle(self, *args, **options):
arbes_file = options['arbes_file']
gestion_file = options['gestion_file']
interventions_file = options['interventions_file']
managing_authority = options['managing_authority']
management_type = options['management_type']
code_prefix = options['code_prefix']
company_name = options['company_name']
fallback_contract_num = options['fallback_contract']
limit = options['limit']
# Fallbacks locaux si chemins /tmp n'existent pas
if not os.path.exists(arbes_file):
alt = os.path.expanduser('~/Downloads/fulcrum_arbes.geojson')
if os.path.exists(alt): arbes_file = alt
if not os.path.exists(gestion_file):
alt = os.path.expanduser('~/Downloads/fulcrum_gestion.csv')
if os.path.exists(alt): gestion_file = alt
if not os.path.exists(interventions_file):
alt = os.path.expanduser('~/Downloads/fulcrum_interventions.geojson')
if os.path.exists(alt): interventions_file = alt
self.log("=== DÉBUT IMPORTATION DONNÉES FULCRUM (ARBRES & INTERVENTIONS) ===")
self.log(f"Fichier arbres : {arbes_file}")
self.log(f"Fichier gestion : {gestion_file}")
self.log(f"Fichier interventions : {interventions_file}")
self.log(f"Gestionnaire : {managing_authority} ({management_type})")
self.log(f"Préfixe code : {code_prefix}")
thematic_nature, _ = Thematic.objects.get_or_create(
name_fr="Nature",
defaults={"name_nl": "Natuur"}
)
cat_arbres = AssetCategory.objects.filter(id=60).first() or AssetCategory.objects.filter(name_fr="Arbres").first()
if not cat_arbres:
cat_arbres = AssetCategory.objects.create(id=60, name_fr="Arbres", name_nl="Bomen")
# -------------------------------------------------------------
# ÉTAPE 1 : COMMANDES / GESTION
# -------------------------------------------------------------
order_map = {} # order_code -> ContractOrder.id
order_contract_map = {} # order_code -> Contract.id
if not options['skip_gestion'] and os.path.exists(gestion_file):
self.log("--- ÉTAPE 1 : Importation de gestion.csv ---")
self.import_gestion(
gestion_file, thematic_nature, company_name, fallback_contract_num,
order_map, order_contract_map, limit
)
else:
self.log("Étape 1 sautée. Chargement des commandes existantes...")
# Toujours s'assurer que toutes les commandes existantes sont en mémoire pour les étapes suivantes
for o in ContractOrder.objects.filter(contract__thematics=thematic_nature).values('id', 'order_code', 'contract_id'):
order_map[o['order_code']] = o['id']
order_contract_map[o['order_code']] = o['contract_id']
# -------------------------------------------------------------
# ÉTAPE 2 : ARBRES
# -------------------------------------------------------------
tree_uuid_to_id = {} # Fulcrum _record_id -> NatureTree.id
sit_pos_to_tree_id = {} # f"{sit_id}.{pos_id}" -> NatureTree.id
if not options['skip_trees'] and os.path.exists(arbes_file):
self.log("--- ÉTAPE 2 : Importation de arbes.geojson ---")
self.import_trees(
arbes_file, cat_arbres, managing_authority, management_type,
code_prefix, tree_uuid_to_id, sit_pos_to_tree_id, limit
)
else:
self.log("Étape 2 sautée. Chargement des arbres existants...")
# Toujours s'assurer que les arbres référencés sont en mémoire
for t in NatureTree.objects.exclude(external_reference__isnull=True).exclude(external_reference="").values('id', 'external_reference', 'tag_number'):
tree_uuid_to_id[t['external_reference']] = t['id']
# -------------------------------------------------------------
# ÉTAPE 3 : INTERVENTIONS
# -------------------------------------------------------------
if not options['skip_interventions'] and os.path.exists(interventions_file):
self.log("--- ÉTAPE 3 : Importation de interventions.geojson ---")
self.import_interventions(
interventions_file, thematic_nature, cat_arbres, company_name, fallback_contract_num,
order_map, order_contract_map, tree_uuid_to_id, sit_pos_to_tree_id, limit
)
self.log("=== FIN DE L'IMPORTATION AVEC SUCCÈS ===")
# -----------------------------------------------------------------
# Gestion des contrats & Commandes
# -----------------------------------------------------------------
def get_or_create_contract(self, number, title, thematic, company_name):
company = Company.objects.filter(name__icontains=company_name).first()
if not company:
company = Company.objects.create(name=company_name)
contract = Contract.objects.filter(contract_number=number).first()
if not contract:
contract = Contract.objects.create(
company=company,
contract_number=number,
description=title,
start_date=date(2020, 1, 1),
end_date=date(2030, 12, 31)
)
contract.thematics.add(thematic)
return contract
def import_gestion(self, gestion_file, thematic_nature, company_name, fallback_contract_num, order_map, order_contract_map, limit=0):
# Contrats connus fréquents
c_e21_088 = self.get_or_create_contract("E21.088", "Marché E21.088 - Soins Nature et Ville (SNV)", thematic_nature, company_name)
c_e21_089 = self.get_or_create_contract("E21.089", "Marché E21.089 - Laanbomen / Arbres d'alignement", thematic_nature, company_name)
c_e19_051 = self.get_or_create_contract("E19.051", "Marché E19.051 - Diagnostic arboricole", thematic_nature, company_name)
c_e20_086 = self.get_or_create_contract("E20.086", "Marché E20.086 - Stock d'arbres", thematic_nature, company_name)
c_fallback = Contract.objects.filter(contract_number=fallback_contract_num).first() or self.get_or_create_contract(
fallback_contract_num, f"Contrat {fallback_contract_num} - {company_name}", thematic_nature, company_name
)
regie = Company.objects.filter(name__icontains="Régie").first()
c_r00_001 = Contract.objects.filter(contract_number="R00.001").first()
if not c_r00_001 and regie:
c_r00_001 = Contract.objects.create(
company=regie,
contract_number="R00.001",
description="Régie",
start_date=date(2020, 1, 1),
end_date=date(2030, 12, 31)
)
c_r00_001.thematics.add(thematic_nature)
with open(gestion_file, mode='r', encoding='utf-8') as f:
reader = csv.DictReader(f)
rows = list(reader)
self.log(f"Nombre total de lignes dans gestion.csv : {len(rows)}")
# Agréger par (contract, com_id)
orders_data = {}
for r in rows:
com_id = (r.get('com_id') or '').strip()
if not com_id:
continue
com_typ = r.get('com_typ') or ''
if 'E21.088' in com_typ:
contract = c_e21_088
elif 'E21.089' in com_typ:
contract = c_e21_089
elif 'E19.051' in com_typ:
contract = c_e19_051
elif 'E20.086' in com_typ:
contract = c_e20_086
elif ('INTERNE' in com_typ or 'REGIE' in com_typ) and c_r00_001:
contract = c_r00_001
else:
contract_match = re.search(r'\b([A-Z]\d{2}\.\d{3})\b', com_typ)
if contract_match:
c_num = contract_match.group(1)
contract = Contract.objects.filter(contract_number=c_num).first() or self.get_or_create_contract(
c_num, f"Marché {c_num}", thematic_nature, company_name
)
else:
contract = c_fallback
key = (contract.id, com_id)
if key not in orders_data:
orders_data[key] = {
'contract': contract,
'order_code': com_id,
'description': r.get('com_nom') or r.get('_title') or com_id,
'status_raw': r.get('_status') or '',
'date_raw': r.get('com_dat') or r.get('_created_at') or '',
'ech_raw': r.get('com_ech') or r.get('pla_fin') or ''
}
self.log(f"Nombre de bons de commande uniques identifiés : {len(orders_data)}")
existing_orders = {
(o.contract_id, o.order_code): o
for o in ContractOrder.objects.filter(contract__thematics=thematic_nature)
}
created_cnt = 0
updated_cnt = 0
for key, odata in orders_data.items():
if limit and (created_cnt + updated_cnt) >= limit:
break
status_raw = odata['status_raw'].lower()
if status_raw in ('execute', 'controle', 'reception provisoire', 'valide'):
order_status = 'completed'
elif status_raw in ('commande', 'en cours'):
order_status = 'sent'
elif status_raw in ('annulee', 'annulée'):
order_status = 'cancelled'
else:
order_status = 'pending'
# Parse dates
order_date = date(2020, 1, 1)
try:
d_str = odata['date_raw'][:10]
order_date = datetime.strptime(d_str, '%Y-%m-%d').date()
except Exception:
pass
delivery_date = order_date
try:
e_str = odata['ech_raw'][:10]
delivery_date = datetime.strptime(e_str, '%Y-%m-%d').date()
except Exception:
delivery_date = order_date
contract = odata['contract']
order_code = odata['order_code']
existing = existing_orders.get(key)
if existing:
existing.description = existing.description or odata['description']
existing.order_status = order_status
existing.order_date = order_date
existing.delivery_date = delivery_date
existing.save(update_fields=['description', 'order_status', 'order_date', 'delivery_date'])
order_map[order_code] = existing.id
order_contract_map[order_code] = contract.id
updated_cnt += 1
else:
new_order = ContractOrder.objects.create(
contract=contract,
order_code=order_code,
description=odata['description'],
order_status=order_status,
order_date=order_date,
delivery_date=delivery_date
)
order_map[order_code] = new_order.id
order_contract_map[order_code] = contract.id
existing_orders[key] = new_order
created_cnt += 1
self.log(f"Gestion terminée : {created_cnt} créés, {updated_cnt} mis à jour.")
# -----------------------------------------------------------------
# Importation des Arbres
# -----------------------------------------------------------------
def import_trees(
self, arbes_file, cat_arbres, managing_authority, management_type,
code_prefix, tree_uuid_to_id, sit_pos_to_tree_id, limit=0
):
self.log("Lecture du fichier GeoJSON des arbres...")
with open(arbes_file, 'r', encoding='utf-8') as f:
data = json.load(f)
features = data.get('features', [])
total_feats = len(features)
self.log(f"Nombre total d'arbres dans GeoJSON : {total_feats}")
# Indexer les modèles botaniques par genre
models_by_genus = {}
for m in NatureAssetModel.objects.all():
first_word = m.name_fr.strip().split()[0] if m.name_fr else ''
if first_word and first_word not in models_by_genus:
models_by_genus[first_word.lower()] = m
# Indexer les NatureLocation existantes
locations_by_code = {loc.code: loc for loc in NatureLocation.objects.exclude(code__isnull=True).exclude(code="")}
locations_by_name = {loc.name_fr.lower(): loc for loc in NatureLocation.objects.exclude(name_fr__isnull=True)}
# Indexer les Communes
munis = {m.name_fr.lower(): m for m in Municipality.objects.all()}
# Pré-charger tous les NatureTree existants
self.log("Indexation des arbres existants en base de données...")
existing_by_ext_ref = {}
existing_by_code = {}
spatial_grid = defaultdict(list)
db_trees_qs = NatureTree.objects.all().values('id', 'code', 'external_reference', 'tag_number', 'lon', 'lat')
for t in db_trees_qs:
tid = t['id']
code = t['code']
ext_ref = t['external_reference']
lon = t['lon']
lat = t['lat']
existing_by_code[code] = tid
if ext_ref:
existing_by_ext_ref[ext_ref] = tid
tree_uuid_to_id[ext_ref] = tid
if lon is not None and lat is not None:
gx = int(lon * 10000)
gy = int(lat * 10000)
spatial_grid[(gx, gy)].append((lon, lat, tid, code))
self.log(f"Arbres existants indexés : {len(existing_by_code)}")
# Traitement des entités Fulcrum
trees_to_update = []
trees_to_create = []
processed = 0
for feat in features:
if limit and processed >= limit:
break
processed += 1
props = feat.get('properties') or {}
geom = feat.get('geometry') or {}
coords = geom.get('coordinates')
if not coords or len(coords) < 2:
continue
lon, lat = float(coords[0]), float(coords[1])
rec_id = props.get('_record_id')
id_complet = (props.get('id_complet') or '').strip()
id_sit_pos = (props.get('id_sit_pos') or '').strip()
id_visuel = (props.get('id_visuel') or '').strip()
pos_id = str(props.get('pos_id') or '').strip()
sit_id = (props.get('sit_id') or '').strip()
sit_nom = (props.get('sit_nom') or '').strip()
comm_commune = (props.get('comm_commune') or '').strip()
arb_ess = (props.get('arb_ess') or '').strip()
arb_vern = (props.get('arb_vern') or '').strip()
arb_tax = (props.get('arb_tax') or '').strip()
arb_cult = (props.get('arb_cult') or '').strip()
genus = ''
if arb_tax:
genus = arb_tax.split()[0].capitalize()
elif arb_ess:
genus = arb_ess.split()[0].capitalize()
# Modèle botanique
asset_model = None
if genus and genus.lower() in models_by_genus:
asset_model = models_by_genus[genus.lower()]
# Emplacement (NatureLocation)
location = None
if sit_id and sit_id in locations_by_code:
location = locations_by_code[sit_id]
elif sit_nom and sit_nom.lower() in locations_by_name:
location = locations_by_name[sit_nom.lower()]
if not location and sit_nom:
muni = munis.get(comm_commune.lower()) if comm_commune else None
loc_pt_3812 = Point(lon, lat, srid=4326).transform(3812, clone=True)
loc_poly = MultiPolygon(loc_pt_3812.buffer(5))
location = NatureLocation.objects.create(
code=sit_id or f"LOC-{comm_commune[:3].upper()}-{processed}",
name_fr=sit_nom,
name_nl=sit_nom,
municipality=muni,
geom=loc_poly,
geojson=f'{{"type":"Point","coordinates":[{lon},{lat}]}}'
)
if sit_id:
locations_by_code[sit_id] = location
locations_by_name[sit_nom.lower()] = location
# Statut biologique et sanitaire
st_raw = props.get('arb_st')
asset_status = 'active'
is_dead = False
is_stump = False
vitality = 'normal'
vit_map = {'Bo': 'normal', 'Mo': 'declining', 'De': 'weak', 'TB': 'strong'}
if props.get('arb_vital') in vit_map:
vitality = vit_map[props.get('arb_vital')]
if st_raw == 'V':
asset_status = 'active'
elif st_raw == 'M':
asset_status = 'active'
is_dead = True
vitality = 'dead'
elif st_raw == 'S':
asset_status = 'active'
is_dead = True
is_stump = True
elif st_raw == 'E':
asset_status = 'to_replace'
elif st_raw in ('A', 'H'):
asset_status = 'archived'
elif st_raw == 'T':
asset_status = 'removed'
# Stade ontogénique
ont_map = {'Je': 'young', 'Sm': 'semi_mature', 'Ad': 'mature', 'Se': 'senescent', 'Re': 'remarkable'}
ont_raw = props.get('arb_ont')
dev_stage = ont_map.get(ont_raw, 'mature')
is_remarkable = (ont_raw == 'Re')
is_chronoxyle = (ont_raw == 'Se')
# Conduite
prun_map = {'SE-LI': 'semi_free', 'LI': 'free_crown', 'AR-MA': 'architectured_head', 'AR-RI': 'architectured_curtain', 'FO': 'formation', 'NE': 'none'}
pruning = prun_map.get(props.get('ges_conduite'), 'free_crown')
# Dendrométrie
haut = float(props['arb_haut']) if props.get('arb_haut') is not None else None
circ = float(props['arb_circ']) if props.get('arb_circ') is not None else None
diam = float(props['arb_diam_tmp']) if props.get('arb_diam_tmp') is not None else None
hsh = float(props['arb_hsh']) if props.get('arb_hsh') is not None else None
brin = int(props['arb_brin_nb']) if props.get('arb_brin_nb') is not None else 1
cf_sit = float(props['cf_sit']) if props.get('cf_sit') is not None else None
# Date de plantation
anpla = str(props.get('arb_anpla') or '')
anpla_match = re.search(r'\b(18\d\d|19\d\d|20[0-2]\d)\b', anpla)
planting_date = None
if anpla_match:
planting_date = date(int(anpla_match.group(1)), 1, 1)
# Géométrie EPSG:3812
pt_3812 = Point(lon, lat, srid=4326).transform(3812, clone=True)
geojson_str = f'{{"type":"Point","coordinates":[{lon},{lat}]}}'
tag_no = id_visuel or pos_id
name_label = arb_vern or arb_ess or id_visuel or id_complet
# ---------------------------------------------------------
# DÉTECTION D'ARBRE EXISTANT (Pas de suppression, mise à jour)
# ---------------------------------------------------------
matched_id = None
# 1. Par UUID Fulcrum
if rec_id and rec_id in existing_by_ext_ref:
matched_id = existing_by_ext_ref[rec_id]
# 2. Par code exact
if not matched_id and id_sit_pos:
cand_code = f"{code_prefix}{id_sit_pos}"
if cand_code in existing_by_code:
matched_id = existing_by_code[cand_code]
# 3. Par numéro de visuel
if not matched_id and id_visuel:
if id_visuel in existing_by_code:
matched_id = existing_by_code[id_visuel]
else:
cand_visuel = f"{code_prefix}{id_visuel}"
if cand_visuel in existing_by_code:
matched_id = existing_by_code[cand_visuel]
# 4. Par proximité spatiale fine (< 1.2 mètre)
if not matched_id:
gx = int(lon * 10000)
gy = int(lat * 10000)
for dx in (-1, 0, 1):
for dy in (-1, 0, 1):
for elon, elat, etid, ecode in spatial_grid.get((gx + dx, gy + dy), []):
d_m = (((lon - elon) * 70000) ** 2 + ((lat - elat) * 111000) ** 2) ** 0.5
if d_m <= 1.2:
matched_id = etid
break
if matched_id: break
if matched_id: break
sit_pos_key = f"{sit_id}.{pos_id}" if sit_id and pos_id else None
if matched_id:
tree_uuid_to_id[rec_id] = matched_id
if sit_pos_key:
sit_pos_to_tree_id[sit_pos_key] = matched_id
trees_to_update.append({
'id': matched_id,
'external_reference': rec_id,
'tag_number': tag_no,
'scientific_name': arb_tax or arb_ess,
'vernacular_name': arb_vern or arb_ess,
'genus': genus,
'height': haut,
'circumference': circ,
'crown_diameter': diam,
'crown_clearance': hsh,
'number_of_trunks': brin,
'situation_coefficient': cf_sit,
'managing_authority': managing_authority,
'management_type': management_type,
'status': asset_status,
'vitality': vitality,
'is_dead': is_dead,
'is_stump': is_stump,
'is_remarkable': is_remarkable,
'is_chronoxyle': is_chronoxyle,
'development_stage': dev_stage,
'pruning_type': pruning,
'planting_date': planting_date,
'model_id': asset_model.id if asset_model else None,
'location_id': location.id if location else None,
'lon': lon,
'lat': lat,
'geom': pt_3812,
'geojson': geojson_str,
})
else:
code = f"{code_prefix}{id_sit_pos}" if id_sit_pos else (
f"{code_prefix}{id_visuel}" if id_visuel else f"{code_prefix}{processed}"
)
existing_by_code[code] = None
new_tree = NatureTree(
code=code,
name_fr=name_label or f"Arbre {code}",
name_nl=name_label or f"Boom {code}",
external_reference=rec_id,
tag_number=tag_no,
scientific_name=arb_tax or arb_ess,
vernacular_name=arb_vern or arb_ess,
genus=genus,
cultivar=arb_cult,
height=haut,
circumference=circ,
crown_diameter=diam,
crown_clearance=hsh,
number_of_trunks=brin,
situation_coefficient=cf_sit,
managing_authority=managing_authority,
management_type=management_type,
category=cat_arbres,
status=asset_status,
vitality=vitality,
is_dead=is_dead,
is_stump=is_stump,
is_remarkable=is_remarkable,
is_chronoxyle=is_chronoxyle,
development_stage=dev_stage,
pruning_type=pruning,
planting_date=planting_date,
model=asset_model,
location=location,
lon=lon,
lat=lat,
geom=pt_3812,
geojson=geojson_str,
)
if compute_tree_ecoservices:
compute_tree_ecoservices(new_tree, overwrite_existing=False)
trees_to_create.append((rec_id, sit_pos_key, new_tree))
self.log(f"Arbres identifiés pour mise à jour : {len(trees_to_update)}")
self.log(f"Arbres identifiés pour création : {len(trees_to_create)}")
# Exécution des mises à jour en lots
BATCH_SIZE = 2000
for i in range(0, len(trees_to_update), BATCH_SIZE):
chunk = trees_to_update[i:i + BATCH_SIZE]
chunk_ids = [c['id'] for c in chunk]
trees_objs = {t.id: t for t in NatureTree.objects.filter(id__in=chunk_ids)}
to_save = []
for c in chunk:
obj = trees_objs.get(c['id'])
if obj:
for k, v in c.items():
if k != 'id':
setattr(obj, k, v)
if compute_tree_ecoservices:
compute_tree_ecoservices(obj, overwrite_existing=False)
to_save.append(obj)
NatureTree.objects.bulk_update(
to_save,
fields=[
'external_reference', 'tag_number', 'scientific_name', 'vernacular_name',
'genus', 'height', 'circumference', 'crown_diameter', 'crown_clearance',
'number_of_trunks', 'situation_coefficient', 'plantation_coefficient',
'amenity_value', 'carbon_stock', 'cooling_indicator', 'cooling_energy_indicator', 'biodiversity_index',
'managing_authority', 'management_type', 'status', 'vitality', 'is_dead', 'is_stump',
'is_remarkable', 'is_chronoxyle', 'development_stage', 'pruning_type',
'planting_date', 'model_id', 'location_id', 'lon', 'lat', 'geom', 'geojson'
],
batch_size=BATCH_SIZE
)
self.log(f" Mise à jour arbres : {min(i + BATCH_SIZE, len(trees_to_update))} / {len(trees_to_update)}")
# Exécution des créations en lots
for i in range(0, len(trees_to_create), BATCH_SIZE):
chunk = trees_to_create[i:i + BATCH_SIZE]
objs = [item[2] for item in chunk]
created_objs = NatureTree.objects.bulk_create(objs, batch_size=BATCH_SIZE)
for (rec_id, sit_pos_key, _), obj in zip(chunk, created_objs):
if rec_id: tree_uuid_to_id[rec_id] = obj.id
if sit_pos_key: sit_pos_to_tree_id[sit_pos_key] = obj.id
self.log(f" Création arbres : {min(i + BATCH_SIZE, len(trees_to_create))} / {len(trees_to_create)}")
self.log(f"Importation des arbres terminée. Total en base : {NatureTree.objects.count()}")
# -----------------------------------------------------------------
# Importation des Interventions
# -----------------------------------------------------------------
def import_interventions(
self, interventions_file, thematic_nature, cat_arbres, company_name, fallback_contract_num,
order_map, order_contract_map, tree_uuid_to_id, sit_pos_to_tree_id, limit=0
):
self.log("Lecture du fichier GeoJSON des interventions...")
with open(interventions_file, 'r', encoding='utf-8') as f:
data = json.load(f)
features = data.get('features', [])
total_feats = len(features)
self.log(f"Nombre total d'interventions dans GeoJSON : {total_feats}")
company = Company.objects.filter(name__icontains=company_name).first()
fallback_contract = Contract.objects.filter(contract_number=fallback_contract_num).first()
source_fulcrum, _ = SourceCategory.objects.get_or_create(name_fr="Fulcrum", defaults={"name_nl": "Fulcrum"})
# Symptômes
symptom_elagage = Symptom.objects.filter(name_fr__icontains="élaguer").first()
symptom_abattage = Symptom.objects.filter(name_fr__icontains="Arbre endommagé ou tombé").first()
symptom_bois_mort = Symptom.objects.filter(name_fr__icontains="Branche(s) au sol").first()
symptom_autre = Symptom.objects.filter(name_fr__icontains="Arbre - Autre").first()
content_type_tree = ContentType.objects.get_for_model(NatureTree)
self.log("Indexation des interventions existantes en base...")
existing_itvs = {
itv.source_ref: itv
for itv in Intervention.objects.exclude(source_ref__isnull=True).exclude(source_ref="")
}
self.log(f"Interventions existantes indexées : {len(existing_itvs)}")
itvs_to_update = []
itvs_to_create = []
itv_tree_links = []
processed = 0
for feat in features:
if limit and processed >= limit:
break
processed += 1
props = feat.get('properties') or {}
geom = feat.get('geometry') or {}
coords = geom.get('coordinates')
if not coords or len(coords) < 2:
continue
lon, lat = float(coords[0]), float(coords[1])
rec_id = props.get('_record_id')
title = props.get('_title') or "Intervention arboricole Fulcrum"
desc = props.get('rec_interv_com') or title
sit_nom = props.get('sit_nom') or ''
comm_commune = props.get('comm_commune') or ''
sit_id = props.get('sit_id') or ''
pos_id = str(props.get('pos_id') or '')
dia_comma = (props.get('dia_comma') or '').strip()
address = f"{sit_nom} ({comm_commune})" if sit_nom and comm_commune else (sit_nom or comm_commune)
# Identification du bon de commande et du contrat
order_id = order_map.get(dia_comma)
contract_id = order_contract_map.get(dia_comma) or (fallback_contract.id if fallback_contract else None)
# Priorité
rec_delai = props.get('rec_delai')
if rec_delai in ('Urg-0', 'Urg-1'):
priority = '1'
elif rec_delai == 'Urg-2':
priority = '2'
elif rec_delai == 'Pla-1':
priority = '3'
else:
priority = '4'
# Type d'intervention
rec_interv = props.get('rec_interv') or ''
if rec_interv in ('AR', 'AA'):
maintain_type = 'corrective'
action_type = 'replace'
symptom = symptom_abattage or symptom_autre
elif rec_interv in ('EB', 'BO'):
maintain_type = 'corrective'
action_type = 'care'
symptom = symptom_bois_mort or symptom_elagage or symptom_autre
elif rec_interv in ('FO', 'EN', 'ED', 'TA', 'BM'):
maintain_type = 'preventive'
action_type = 'care'
symptom = symptom_elagage or symptom_autre
elif rec_interv in ('ET', 'HM'):
maintain_type = 'ameliorative'
action_type = 'repair'
symptom = symptom_autre
else:
maintain_type = 'corrective'
action_type = 'care'
symptom = symptom_autre
# Statut
in_fin_yn = props.get('in_fin_yn')
if in_fin_yn == 'yes':
status = 'finished'
status_order = 25
else:
status = 'to_be_planned'
status_order = 5
# Date limite
deadline_str = props.get('in_deadline')
expected_end_time = None
if deadline_str:
try:
expected_end_time = datetime.fromisoformat(deadline_str.replace('Z', '+00:00'))
except Exception:
pass
# Géométrie MultiPolygon SRID 3812
pt_3812 = Point(lon, lat, srid=4326).transform(3812, clone=True)
buffered_geom = MultiPolygon(pt_3812.buffer(1))
# Recherche de l'arbre associé
tree_uuid_raw = props.get('mater_arbres_lk')
tree_uuid = tree_uuid_raw[0] if isinstance(tree_uuid_raw, list) and tree_uuid_raw else tree_uuid_raw
matched_tree_id = tree_uuid_to_id.get(tree_uuid)
if not matched_tree_id and sit_id and pos_id:
matched_tree_id = sit_pos_to_tree_id.get(f"{sit_id}.{pos_id}")
existing_itv = existing_itvs.get(rec_id)
if existing_itv:
existing_itv.title = title
existing_itv.description = desc
existing_itv.address = address
existing_itv.location_code = sit_id
existing_itv.priority = priority
existing_itv.maintain_type = maintain_type
existing_itv.type = action_type
existing_itv.status = status
existing_itv.status_order = status_order
existing_itv.expected_end_time = expected_end_time
existing_itv.order_id = order_id
existing_itv.contract_id = contract_id
existing_itv.source_detail = dia_comma
existing_itv.lon = lon
existing_itv.lat = lat
existing_itv.geom = buffered_geom
itvs_to_update.append(existing_itv)
if matched_tree_id:
itv_tree_links.append((existing_itv.id, matched_tree_id))
else:
new_itv = Intervention(
title=title,
description=desc,
address=address,
location_code=sit_id,
priority=priority,
maintain_type=maintain_type,
type=action_type,
status=status,
status_order=status_order,
expected_end_time=expected_end_time,
thematic=thematic_nature,
asset_category=cat_arbres,
symptom=symptom,
order_id=order_id,
contract_id=contract_id,
assigned_provider=company,
source_category=source_fulcrum,
source_ref=rec_id,
source_detail=dia_comma,
lon=lon,
lat=lat,
geom=buffered_geom
)
itvs_to_create.append((new_itv, matched_tree_id))
self.log(f"Interventions à mettre à jour : {len(itvs_to_update)}")
self.log(f"Interventions à créer : {len(itvs_to_create)}")
# Mettre à jour en lots
BATCH_SIZE = 2000
for i in range(0, len(itvs_to_update), BATCH_SIZE):
chunk = itvs_to_update[i:i + BATCH_SIZE]
Intervention.objects.bulk_update(
chunk,
fields=[
'title', 'description', 'address', 'location_code', 'priority',
'maintain_type', 'type', 'status', 'status_order', 'expected_end_time',
'order_id', 'contract_id', 'source_detail', 'lon', 'lat', 'geom'
],
batch_size=BATCH_SIZE
)
self.log(f" Mise à jour interventions : {min(i + BATCH_SIZE, len(itvs_to_update))} / {len(itvs_to_update)}")
# Création des nouvelles interventions
if itvs_to_create:
self.log("Attribution des codes d'intervention séquentiels...")
with transaction.atomic():
seq, _ = InterventionSequence.objects.select_for_update().get_or_create(id=1)
start_num = seq.current_number + 1
seq.current_number += len(itvs_to_create)
seq.save(update_fields=['current_number'])
for idx, (itv, _) in enumerate(itvs_to_create):
code_num = start_num + idx
itv.code = f"I{str(code_num).zfill(6)}"
created_objs = []
for i in range(0, len(itvs_to_create), BATCH_SIZE):
chunk = itvs_to_create[i:i + BATCH_SIZE]
objs = [c[0] for c in chunk]
batch_created = Intervention.objects.bulk_create(objs, batch_size=BATCH_SIZE)
created_objs.extend(batch_created)
for (itv, tree_id), obj in zip(chunk, batch_created):
if tree_id:
itv_tree_links.append((obj.id, tree_id))
self.log(f" Création interventions : {min(i + BATCH_SIZE, len(itvs_to_create))} / {len(itvs_to_create)}")
# Lier les interventions aux arbres via InterventionAsset
self.log(f"Création des liens Intervention <-> Arbre ({len(itv_tree_links)} liens)...")
existing_links = set(
InterventionAsset.objects.filter(
content_type=content_type_tree,
intervention_id__in=[l[0] for l in itv_tree_links[:50000]]
).values_list('intervention_id', 'object_id')
)
assets_to_create = []
for itv_id, tree_id in itv_tree_links:
if (itv_id, tree_id) not in existing_links:
assets_to_create.append(
InterventionAsset(
intervention_id=itv_id,
content_type=content_type_tree,
object_id=tree_id
)
)
existing_links.add((itv_id, tree_id))
self.log(f"Nouveaux liens InterventionAsset à insérer : {len(assets_to_create)}")
for i in range(0, len(assets_to_create), BATCH_SIZE):
chunk = assets_to_create[i:i + BATCH_SIZE]
InterventionAsset.objects.bulk_create(chunk, batch_size=BATCH_SIZE)
self.log(f" Insertion liens : {min(i + BATCH_SIZE, len(assets_to_create))} / {len(assets_to_create)}")
self.log(f"Importation des interventions terminée. Total en base : {Intervention.objects.count()}")

184
loko/assets/tests_import_fulcrum.py Normal file
View file

@ -0,0 +1,184 @@
import json
import tempfile
import os
from datetime import date
from django.test import TestCase
from django.core.management import call_command
from django.contrib.gis.geos import Point
from common.models import Thematic
from assets.models import NatureTree, AssetCategory
from contracts.models import Contract, ContractOrder, Company
from interventions.models import Intervention, InterventionAsset
class ImportFulcrumTreesCommandTest(TestCase):
def setUp(self):
self.thematic = Thematic.objects.create(name_fr="Nature", name_nl="Natuur")
self.category = AssetCategory.objects.create(id=60, name_fr="Arbres", name_nl="Bomen")
self.company = Company.objects.create(name="Krinkels")
self.contract = Contract.objects.create(
company=self.company,
contract_number="E19.051",
description="Marché E19.051",
start_date=date(2020, 1, 1),
end_date=date(2030, 12, 31)
)
self.contract.thematics.add(self.thematic)
def test_import_fulcrum_trees_and_interventions(self):
# 1. Préparer gestion.csv temporaire
gestion_content = (
"com_id,com_typ,com_nom,_status,com_dat,com_ech\n"
"BC-TEST-001,Marché E19.051 - Diagnostic,Élagage test,commande,2024-05-10,2024-06-15\n"
)
with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False) as f_ges:
f_ges.write(gestion_content)
ges_path = f_ges.name
# 2. Préparer arbes.geojson temporaire
mock_tree_uuid = "11111111-2222-3333-4444-555555555555"
tree_geojson = {
"type": "FeatureCollection",
"features": [
{
"type": "Feature",
"geometry": {
"type": "Point",
"coordinates": [4.3517, 50.8503]
},
"properties": {
"_record_id": mock_tree_uuid,
"id_sit_pos": "099.01.001",
"id_visuel": "V-9901",
"sit_id": "099.01",
"pos_id": "001",
"sit_nom": "Boulevard du Test",
"comm_commune": "Bruxelles",
"arb_ess": "Platanus x hispanica",
"arb_vern": "Platane commun",
"arb_tax": "Platanus hispanica",
"arb_haut": 18.5,
"arb_circ": 140.0,
"arb_diam_tmp": 8.0,
"arb_st": "V",
"arb_vital": "Bo",
"arb_ont": "Ad",
"ges_conduite": "LI",
"cf_sit": 4.5,
"arb_anpla": "1995"
}
}
]
}
with tempfile.NamedTemporaryFile(mode='w', suffix='.geojson', delete=False) as f_arb:
json.dump(tree_geojson, f_arb)
arb_path = f_arb.name
# 3. Préparer interventions.geojson temporaire
mock_itv_uuid = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
itv_geojson = {
"type": "FeatureCollection",
"features": [
{
"type": "Feature",
"geometry": {
"type": "Point",
"coordinates": [4.3517, 50.8503]
},
"properties": {
"_record_id": mock_itv_uuid,
"_title": "Élagage de formation",
"rec_interv_com": "Taille branches basses",
"sit_nom": "Boulevard du Test",
"comm_commune": "Bruxelles",
"sit_id": "099.01",
"pos_id": "001",
"dia_comma": "BC-TEST-001",
"rec_delai": "Pla-1",
"rec_interv": "FO",
"in_fin_yn": "no",
"mater_arbres_lk": [mock_tree_uuid]
}
}
]
}
with tempfile.NamedTemporaryFile(mode='w', suffix='.geojson', delete=False) as f_itv:
json.dump(itv_geojson, f_itv)
itv_path = f_itv.name
try:
# Exécuter la commande
call_command(
'import_fulcrum_trees',
arbes_file=arb_path,
gestion_file=ges_path,
interventions_file=itv_path,
managing_authority="Test Authority",
management_type="regional",
code_prefix="TEST-ARB-"
)
# Vérifications gestion / bon de commande
order = ContractOrder.objects.filter(order_code="BC-TEST-001").first()
self.assertIsNotNone(order)
self.assertEqual(order.contract.contract_number, "E19.051")
self.assertEqual(order.order_status, "sent")
# Vérifications arbre
tree = NatureTree.objects.filter(external_reference=mock_tree_uuid).first()
self.assertIsNotNone(tree)
self.assertEqual(tree.code, "TEST-ARB-099.01.001")
self.assertEqual(tree.managing_authority, "Test Authority")
self.assertEqual(tree.management_type, "regional")
self.assertEqual(tree.circumference, 140.0)
self.assertEqual(tree.height, 18.5)
self.assertIsNotNone(tree.amenity_value)
self.assertGreater(tree.amenity_value, 0)
self.assertIsNotNone(tree.carbon_stock)
self.assertGreater(tree.carbon_stock, 0)
self.assertIsNotNone(tree.cooling_indicator)
# Vérifications intervention
itv = Intervention.objects.filter(source_ref=mock_itv_uuid).first()
self.assertIsNotNone(itv)
self.assertEqual(itv.order_id, order.id)
self.assertEqual(itv.priority, "3") # Pla-1 -> 3
self.assertEqual(itv.status, "to_be_planned")
# Vérification liaison InterventionAsset
links = InterventionAsset.objects.filter(intervention=itv)
self.assertEqual(links.count(), 1)
self.assertEqual(links.first().object_id, tree.id)
# Test mise à jour sans duplication
tree_geojson['features'][0]['properties']['arb_circ'] = 145.0
with open(arb_path, 'w') as f_arb_update:
json.dump(tree_geojson, f_arb_update)
call_command(
'import_fulcrum_trees',
arbes_file=arb_path,
gestion_file=ges_path,
interventions_file=itv_path,
skip_gestion=True,
skip_interventions=True
)
# Le nombre d'arbres ne doit pas avoir augmenté
self.assertEqual(NatureTree.objects.filter(external_reference=mock_tree_uuid).count(), 1)
tree.refresh_from_db()
self.assertEqual(tree.circumference, 145.0)
finally:
if os.path.exists(ges_path): os.remove(ges_path)
if os.path.exists(arb_path): os.remove(arb_path)
if os.path.exists(itv_path): os.remove(itv_path)
def test_import_fulcrum_regional_trees_alias(self):
from io import StringIO
out = StringIO()
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())