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

feat: automate asset parent data synchronization for geometry and location fields with new management command and model hooks

Browse source
This commit is contained in:
kdeterme 2026-08-17 09:32:30 +02:00
parent 016ce202b4
commit 1e7e8ae697
16 changed files with 1032 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

379
loko/assets/management/commands/sync_asset_parent_data.py Normal file
View file

@ -0,0 +1,379 @@
import logging
from django.core.management.base import BaseCommand
from django.db import transaction
from assets.models.trafficlights import (
TrafficLightLantern,
TrafficLightAccessory,
TrafficLightDetector,
TrafficLightControllerHardware,
TrafficLightNetworkHardware,
TrafficLightVoltageHardware,
TrafficLightElectricalCabinetContent,
)
from assets.models.sign import SignPanel
from assets.models.publiclighting import PublicLightingLightFixture
from assets.models.its import ITSGeoAsset
from assets.models.structures import StructureGeoAsset
from assets.models.buildings import BuildingAsset
from assets.models.outdoor import OutdoorAsset
logger = logging.getLogger(__name__)
class Command(BaseCommand):
help = "Synchronizes coordinates (geom, lon, lat, geojson) and locations from parent supports to child assets."
def add_arguments(self, parser):
parser.add_argument(
'--batch-size',
type=int,
default=1000,
help="Batch size for processing records."
)
def handle(self, *args, **options):
batch_size = options['batch_size']
self.stdout.write(self.style.NOTICE("Starting parent-child asset data synchronization..."))
# 1. Traffic Light Lanterns
self.stdout.write("1. Syncing Traffic Light Lanterns...")
lanterns = TrafficLightLantern.objects.select_related('pole').filter(pole__geom__isnull=False)
count = 0
to_update = []
for lantern in lanterns.iterator(chunk_size=batch_size):
if lantern.pole and lantern.pole.geom:
if lantern.geom != lantern.pole.geom or lantern.lon != lantern.pole.lon or lantern.lat != lantern.pole.lat:
lantern.geom = lantern.pole.geom
lantern.lon = lantern.pole.lon
lantern.lat = lantern.pole.lat
lantern.geojson = lantern.pole.geojson
to_update.append(lantern)
count += 1
if len(to_update) >= batch_size:
TrafficLightLantern.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson'])
to_update = []
if to_update:
TrafficLightLantern.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson'])
self.stdout.write(self.style.SUCCESS(f" ✓ Updated {count} Traffic Light Lanterns"))
# 2. Traffic Light Accessories
self.stdout.write("2. Syncing Traffic Light Accessories...")
accessories = TrafficLightAccessory.objects.select_related('pole', 'intersection').filter(pole__geom__isnull=False)
count = 0
to_update = []
for acc in accessories.iterator(chunk_size=batch_size):
changed = False
if acc.pole and acc.pole.geom:
if acc.geom != acc.pole.geom or acc.lon != acc.pole.lon or acc.lat != acc.pole.lat:
acc.geom = acc.pole.geom
acc.lon = acc.pole.lon
acc.lat = acc.pole.lat
acc.geojson = acc.pole.geojson
changed = True
if acc.pole and acc.pole.intersection and not acc.intersection_id:
acc.intersection = acc.pole.intersection
changed = True
if changed:
to_update.append(acc)
count += 1
if len(to_update) >= batch_size:
TrafficLightAccessory.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson', 'intersection'])
to_update = []
if to_update:
TrafficLightAccessory.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson', 'intersection'])
self.stdout.write(self.style.SUCCESS(f" ✓ Updated {count} Traffic Light Accessories"))
# 3. Traffic Light Detectors
self.stdout.write("3. Syncing Traffic Light Detectors...")
detectors = TrafficLightDetector.objects.select_related('pole', 'controller')
count = 0
to_update = []
for det in detectors.iterator(chunk_size=batch_size):
changed = False
support = det.pole or det.controller
if support and getattr(support, 'geom', None):
if det.geom != support.geom or det.lon != support.lon or det.lat != support.lat:
det.geom = support.geom
det.lon = support.lon
det.lat = support.lat
det.geojson = support.geojson
changed = True
if det.pole and det.pole.intersection and not det.intersection_id:
det.intersection = det.pole.intersection
changed = True
elif det.controller and det.controller.intersection and not det.intersection_id:
det.intersection = det.controller.intersection
changed = True
if changed:
to_update.append(det)
count += 1
if len(to_update) >= batch_size:
TrafficLightDetector.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson', 'intersection'])
to_update = []
if to_update:
TrafficLightDetector.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson', 'intersection'])
self.stdout.write(self.style.SUCCESS(f" ✓ Updated {count} Traffic Light Detectors"))
# 4. Traffic Light Controller Hardware
self.stdout.write("4. Syncing Traffic Light Controller Hardware...")
hws = TrafficLightControllerHardware.objects.select_related('controller').filter(controller__geom__isnull=False)
count = 0
to_update = []
for hw in hws.iterator(chunk_size=batch_size):
if hw.controller and hw.controller.geom:
if hw.geom != hw.controller.geom or hw.lon != hw.controller.lon or hw.lat != hw.controller.lat:
hw.geom = hw.controller.geom
hw.lon = hw.controller.lon
hw.lat = hw.controller.lat
hw.geojson = hw.controller.geojson
to_update.append(hw)
count += 1
if len(to_update) >= batch_size:
TrafficLightControllerHardware.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson'])
to_update = []
if to_update:
TrafficLightControllerHardware.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson'])
self.stdout.write(self.style.SUCCESS(f" ✓ Updated {count} Traffic Light Controller Hardwares"))
# 5. Traffic Light Network Hardware
self.stdout.write("5. Syncing Traffic Light Network Hardware...")
nets = TrafficLightNetworkHardware.objects.select_related('controller').filter(controller__geom__isnull=False)
count = 0
to_update = []
for net in nets.iterator(chunk_size=batch_size):
changed = False
if net.controller and net.controller.geom:
if net.geom != net.controller.geom or net.lon != net.controller.lon or net.lat != net.controller.lat:
net.geom = net.controller.geom
net.lon = net.controller.lon
net.lat = net.controller.lat
net.geojson = net.controller.geojson
changed = True
if net.controller and net.controller.intersection and not net.intersection_id:
net.intersection = net.controller.intersection
changed = True
if changed:
to_update.append(net)
count += 1
if len(to_update) >= batch_size:
TrafficLightNetworkHardware.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson', 'intersection'])
to_update = []
if to_update:
TrafficLightNetworkHardware.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson', 'intersection'])
self.stdout.write(self.style.SUCCESS(f" ✓ Updated {count} Traffic Light Network Hardwares"))
# 6. Traffic Light Voltage Hardware
self.stdout.write("6. Syncing Traffic Light Voltage Hardware...")
volts = TrafficLightVoltageHardware.objects.select_related('controller').filter(controller__geom__isnull=False)
count = 0
to_update = []
for volt in volts.iterator(chunk_size=batch_size):
changed = False
if volt.controller and volt.controller.geom:
if volt.geom != volt.controller.geom or volt.lon != volt.controller.lon or volt.lat != volt.controller.lat:
volt.geom = volt.controller.geom
volt.lon = volt.controller.lon
volt.lat = volt.controller.lat
volt.geojson = volt.controller.geojson
changed = True
if volt.controller and volt.controller.intersection and not volt.intersection_id:
volt.intersection = volt.controller.intersection
changed = True
if changed:
to_update.append(volt)
count += 1
if len(to_update) >= batch_size:
TrafficLightVoltageHardware.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson', 'intersection'])
to_update = []
if to_update:
TrafficLightVoltageHardware.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson', 'intersection'])
self.stdout.write(self.style.SUCCESS(f" ✓ Updated {count} Traffic Light Voltage Hardwares"))
# 7. Traffic Light Electrical Cabinet Contents
self.stdout.write("7. Syncing Traffic Light Electrical Cabinet Contents...")
contents = TrafficLightElectricalCabinetContent.objects.select_related('cabinet').filter(cabinet__geom__isnull=False)
count = 0
to_update = []
for content in contents.iterator(chunk_size=batch_size):
if content.cabinet and content.cabinet.geom:
if content.geom != content.cabinet.geom or content.lon != content.cabinet.lon or content.lat != content.cabinet.lat:
content.geom = content.cabinet.geom
content.lon = content.cabinet.lon
content.lat = content.cabinet.lat
content.geojson = content.cabinet.geojson
to_update.append(content)
count += 1
if len(to_update) >= batch_size:
TrafficLightElectricalCabinetContent.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson'])
to_update = []
if to_update:
TrafficLightElectricalCabinetContent.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson'])
self.stdout.write(self.style.SUCCESS(f" ✓ Updated {count} Electrical Cabinet Contents"))
# 8. Road Sign Panels
self.stdout.write("8. Syncing Road Sign Panels...")
panels = SignPanel.objects.select_related('pole', 'trafficlight_pole', 'publiclighting_pole')
count = 0
to_update = []
for panel in panels.iterator(chunk_size=batch_size):
changed = False
support = panel.pole or panel.trafficlight_pole or panel.publiclighting_pole
if support and getattr(support, 'geom', None):
if panel.geom != support.geom or panel.lon != support.lon or panel.lat != support.lat:
panel.geom = support.geom
panel.lon = support.lon
panel.lat = support.lat
panel.geojson = support.geojson
changed = True
if panel.pole and panel.pole.street and not panel.street_id:
panel.street = panel.pole.street
changed = True
if changed:
to_update.append(panel)
count += 1
if len(to_update) >= batch_size:
SignPanel.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson', 'street'])
to_update = []
if to_update:
SignPanel.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson', 'street'])
self.stdout.write(self.style.SUCCESS(f" ✓ Updated {count} Road Sign Panels"))
# 9. Public Lighting Light Fixtures (Luminaires)
self.stdout.write("9. Syncing Public Lighting Light Fixtures...")
fixtures = PublicLightingLightFixture.objects.select_related('pole', 'traffic_light_pole')
count = 0
to_update = []
for fix in fixtures.iterator(chunk_size=batch_size):
changed = False
support = fix.pole or fix.traffic_light_pole
if support and getattr(support, 'geom', None):
if fix.geom != support.geom or fix.lon != support.lon or fix.lat != support.lat:
fix.geom = support.geom
fix.lon = support.lon
fix.lat = support.lat
fix.geojson = support.geojson
changed = True
if fix.pole and fix.pole.street and not fix.street_id:
fix.street = fix.pole.street
changed = True
if changed:
to_update.append(fix)
count += 1
if len(to_update) >= batch_size:
PublicLightingLightFixture.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson', 'street'])
to_update = []
if to_update:
PublicLightingLightFixture.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson', 'street'])
self.stdout.write(self.style.SUCCESS(f" ✓ Updated {count} Public Lighting Light Fixtures"))
# 10. ITS Assets
self.stdout.write("10. Syncing ITS Assets...")
its_assets = ITSGeoAsset.objects.select_related('trafficlights_pole', 'lighting_pole')
count = 0
to_update = []
for its in its_assets.iterator(chunk_size=batch_size):
support = its.trafficlights_pole or its.lighting_pole
if support and getattr(support, 'geom', None):
if its.geom != support.geom or its.lon != support.lon or its.lat != support.lat:
its.geom = support.geom
its.lon = support.lon
its.lat = support.lat
its.geojson = support.geojson
to_update.append(its)
count += 1
if len(to_update) >= batch_size:
ITSGeoAsset.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson'])
to_update = []
if to_update:
ITSGeoAsset.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson'])
self.stdout.write(self.style.SUCCESS(f" ✓ Updated {count} ITS Assets"))
# 11. Structure Assets
self.stdout.write("11. Syncing Structure Assets...")
struct_assets = StructureGeoAsset.objects.select_related('parent_asset')
count = 0
to_update = []
for sa in struct_assets.iterator(chunk_size=batch_size):
changed = False
if sa.parent_asset and getattr(sa.parent_asset, 'geom', None) and not sa.geom:
sa.geom = sa.parent_asset.geom
sa.lon = sa.parent_asset.lon
sa.lat = sa.parent_asset.lat
sa.geojson = sa.parent_asset.geojson
changed = True
if sa.parent_asset and sa.parent_asset.location and not sa.location_id:
sa.location = sa.parent_asset.location
changed = True
if sa.parent_asset and sa.parent_asset.structure and not sa.structure_id:
sa.structure = sa.parent_asset.structure
changed = True
if changed:
to_update.append(sa)
count += 1
if len(to_update) >= batch_size:
StructureGeoAsset.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson', 'location', 'structure'])
to_update = []
if to_update:
StructureGeoAsset.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson', 'location', 'structure'])
self.stdout.write(self.style.SUCCESS(f" ✓ Updated {count} Structure Assets"))
# 12. Building Assets
self.stdout.write("12. Syncing Building Assets...")
bld_assets = BuildingAsset.objects.select_related('parent_asset')
count = 0
to_update = []
for ba in bld_assets.iterator(chunk_size=batch_size):
changed = False
if ba.parent_asset and getattr(ba.parent_asset, 'geom', None) and not ba.geom:
ba.geom = ba.parent_asset.geom
ba.lon = ba.parent_asset.lon
ba.lat = ba.parent_asset.lat
ba.geojson = ba.parent_asset.geojson
changed = True
if ba.parent_asset and ba.parent_asset.location and not ba.location_id:
ba.location = ba.parent_asset.location
changed = True
if ba.parent_asset and ba.parent_asset.building and not ba.building_id:
ba.building = ba.parent_asset.building
changed = True
if changed:
to_update.append(ba)
count += 1
if len(to_update) >= batch_size:
BuildingAsset.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson', 'location', 'building'])
to_update = []
if to_update:
BuildingAsset.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson', 'location', 'building'])
self.stdout.write(self.style.SUCCESS(f" ✓ Updated {count} Building Assets"))
# 13. Outdoor Assets
self.stdout.write("13. Syncing Outdoor Assets...")
out_assets = OutdoorAsset.objects.select_related('parent_asset')
count = 0
to_update = []
for oa in out_assets.iterator(chunk_size=batch_size):
changed = False
if oa.parent_asset and getattr(oa.parent_asset, 'geom', None) and not oa.geom:
oa.geom = oa.parent_asset.geom
oa.lon = oa.parent_asset.lon
oa.lat = oa.parent_asset.lat
oa.geojson = oa.parent_asset.geojson
changed = True
if oa.parent_asset and oa.parent_asset.location and not oa.location_id:
oa.location = oa.parent_asset.location
changed = True
if oa.parent_asset and oa.parent_asset.site and not oa.site_id:
oa.site = oa.parent_asset.site
changed = True
if changed:
to_update.append(oa)
count += 1
if len(to_update) >= batch_size:
OutdoorAsset.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson', 'location', 'site'])
to_update = []
if to_update:
OutdoorAsset.objects.bulk_update(to_update, ['geom', 'lon', 'lat', 'geojson', 'location', 'site'])
self.stdout.write(self.style.SUCCESS(f" ✓ Updated {count} Outdoor Assets"))
self.stdout.write(self.style.SUCCESS("All asset parent data synchronized successfully!"))

34
loko/assets/migrations/0020_trafficlightlantern_geojson_trafficlightlantern_geom_and_more.py Normal file
View file

@ -0,0 +1,34 @@
# Generated by Django 6.0.7 on 2026-08-17 07:23
import django.contrib.gis.db.models.fields
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('assets', '0019_naturerwiasset_location'),
]
operations = [
migrations.AddField(
model_name='trafficlightlantern',
name='geojson',
field=models.TextField(blank=True, null=True),
),
migrations.AddField(
model_name='trafficlightlantern',
name='geom',
field=django.contrib.gis.db.models.fields.PointField(blank=True, null=True, srid=3812),
),
migrations.AddField(
model_name='trafficlightlantern',
name='lat',
field=models.FloatField(blank=True, null=True),
),
migrations.AddField(
model_name='trafficlightlantern',
name='lon',
field=models.FloatField(blank=True, null=True),
),
]

20
loko/assets/migrations/0021_sync_asset_parent_data.py Normal file
View file

@ -0,0 +1,20 @@
from django.db import migrations
from django.core.management import call_command
def sync_parent_data(apps, schema_editor):
try:
call_command('sync_asset_parent_data')
except Exception as e:
print(f"Warning during sync_asset_parent_data migration: {e}")
class Migration(migrations.Migration):
dependencies = [
('assets', '0020_trafficlightlantern_geojson_trafficlightlantern_geom_and_more'),
]
operations = [
migrations.RunPython(sync_parent_data, migrations.RunPython.noop),
]

24
loko/assets/models/buildings.py
View file

@ -211,6 +211,30 @@ class BuildingAsset(AbstractGeoAsset):
last_control_date = models.DateField(null=True, blank=True, verbose_name=_("Dernier contrôle réglementaire"))
next_control_date = models.DateField(null=True, blank=True, verbose_name=_("Prochain contrôle réglementaire"))
def clean(self):
super().clean()
if self.parent_asset and getattr(self.parent_asset, 'geom', None) and not self.geom:
self.geom = self.parent_asset.geom
self.lon = self.parent_asset.lon
self.lat = self.parent_asset.lat
self.geojson = self.parent_asset.geojson
if self.parent_asset and self.parent_asset.location and not self.location_id:
self.location = self.parent_asset.location
if self.parent_asset and self.parent_asset.building and not self.building_id:
self.building = self.parent_asset.building
def save(self, *args, **kwargs):
if self.parent_asset and getattr(self.parent_asset, 'geom', None) and not self.geom:
self.geom = self.parent_asset.geom
self.lon = self.parent_asset.lon
self.lat = self.parent_asset.lat
self.geojson = self.parent_asset.geojson
if self.parent_asset and self.parent_asset.location and not self.location_id:
self.location = self.parent_asset.location
if self.parent_asset and self.parent_asset.building and not self.building_id:
self.building = self.parent_asset.building
super().save(*args, **kwargs)
class Meta:
verbose_name = _("Asset / Équipement de bâtiment")
verbose_name_plural = _("Assets / Équipements de bâtiment")

44
loko/assets/models/core.py
View file

@ -331,6 +331,50 @@ class AbstractAsset(models.Model):
return original
def get_parent_support(self):
"""
Returns the direct parent support object if this asset is mounted or attached to another asset.
e.g., pole, controller, cabinet, parent_asset, etc.
"""
for attr in (
'pole', 'trafficlight_pole', 'trafficlights_pole',
'publiclighting_pole', 'lighting_pole', 'controller',
'cabinet', 'parent_asset'
):
support = getattr(self, attr, None)
if support is not None:
return support
return None
def get_location(self):
"""
Resolves the associated location object (Street, Intersection, StructureLocation, PublicBuilding, etc.)
either directly from the asset or indirectly via its parent support.
"""
# 1. Direct location fields
for attr in ('location', 'street', 'intersection', 'building', 'site', 'structure'):
loc = getattr(self, attr, None)
if loc is not None:
if attr == 'structure' and hasattr(loc, 'main_location') and loc.main_location:
return loc.main_location
return loc
# 2. Indirectly via parent support
parent_support = self.get_parent_support()
if parent_support is not None:
if hasattr(parent_support, 'get_location'):
loc = parent_support.get_location()
if loc is not None:
return loc
for attr in ('location', 'street', 'intersection', 'building', 'site', 'structure'):
loc = getattr(parent_support, attr, None)
if loc is not None:
if attr == 'structure' and hasattr(loc, 'main_location') and loc.main_location:
return loc.main_location
return loc
return None
def get_name(self):
lang = get_language()
name = getattr(self, f"name_{lang}", self.name_fr) or "" # Fallback to name_fr

18
loko/assets/models/its.py
View file

@ -92,6 +92,24 @@ class ITSGeoAsset(AbstractGeoAsset):
trafficlights_pole = models.ForeignKey("assets.TrafficLightPole", on_delete=models.SET_NULL, null=True, blank=True, related_name="its_assets", verbose_name="Traffic Light Pole")
lighting_pole = models.ForeignKey("assets.PublicLightingPole", on_delete=models.SET_NULL, null=True, blank=True, related_name="its_assets", verbose_name="Lighting Pole")
def clean(self):
super().clean()
support = self.trafficlights_pole or self.lighting_pole
if support and getattr(support, 'geom', None):
self.geom = support.geom
self.lon = support.lon
self.lat = support.lat
self.geojson = support.geojson
def save(self, *args, **kwargs):
support = self.trafficlights_pole or self.lighting_pole
if support and getattr(support, 'geom', None):
self.geom = support.geom
self.lon = support.lon
self.lat = support.lat
self.geojson = support.geojson
super().save(*args, **kwargs)
class Meta:
indexes = [
models.Index(fields=["code"]),

24
loko/assets/models/outdoor.py
View file

@ -148,6 +148,30 @@ class OutdoorAsset(AbstractGeoAsset):
impact_surface_type = models.CharField(max_length=150, blank=True, null=True, verbose_name=_("Type de sol d'amortissement"))
critical_fall_height_m = models.FloatField(null=True, blank=True, verbose_name=_("Hauteur critique de chute (m)"))
def clean(self):
super().clean()
if self.parent_asset and getattr(self.parent_asset, 'geom', None) and not self.geom:
self.geom = self.parent_asset.geom
self.lon = self.parent_asset.lon
self.lat = self.parent_asset.lat
self.geojson = self.parent_asset.geojson
if self.parent_asset and self.parent_asset.location and not self.location_id:
self.location = self.parent_asset.location
if self.parent_asset and self.parent_asset.site and not self.site_id:
self.site = self.parent_asset.site
def save(self, *args, **kwargs):
if self.parent_asset and getattr(self.parent_asset, 'geom', None) and not self.geom:
self.geom = self.parent_asset.geom
self.lon = self.parent_asset.lon
self.lat = self.parent_asset.lat
self.geojson = self.parent_asset.geojson
if self.parent_asset and self.parent_asset.location and not self.location_id:
self.location = self.parent_asset.location
if self.parent_asset and self.parent_asset.site and not self.site_id:
self.site = self.parent_asset.site
super().save(*args, **kwargs)
class Meta:
verbose_name = _("Asset / Équipement plein air")
verbose_name_plural = _("Assets / Équipements plein air")

28
loko/assets/models/publiclighting.py
View file

@ -89,6 +89,12 @@ class PublicLightingPole(AbstractGeoAsset):
super().save(*args, **kwargs)
for radar in self.radars.all():
radar.save()
for lf in self.light_fixtures.all():
lf.save()
for sp in self.signpanels_attached.all():
sp.save()
for its in self.its_assets.all():
its.save()
def get_absolute_url(self):
return reverse('assets:publiclighting_assets_detail', kwargs={'asset_model': 'publiclightingpole', 'asset_id':self.id})
@ -105,6 +111,28 @@ class PublicLightingLightFixture(AbstractGeoAsset):
traffic_light_pole = models.ForeignKey(TrafficLightPole, related_name="light_fixtures", on_delete=models.SET_NULL, null=True, blank=True)
street = models.ForeignKey(PublicLightingStreet, related_name="light_fixtures", on_delete=models.PROTECT)
def clean(self):
super().clean()
support = self.pole or self.traffic_light_pole
if support and getattr(support, 'geom', None):
self.geom = support.geom
self.lon = support.lon
self.lat = support.lat
self.geojson = support.geojson
if self.pole and self.pole.street and not self.street_id:
self.street = self.pole.street
def save(self, *args, **kwargs):
support = self.pole or self.traffic_light_pole
if support and getattr(support, 'geom', None):
self.geom = support.geom
self.lon = support.lon
self.lat = support.lat
self.geojson = support.geojson
if self.pole and self.pole.street and not self.street_id:
self.street = self.pole.street
super().save(*args, **kwargs)
class Meta:
indexes = [
models.Index(fields=["code"]),

27
loko/assets/models/sign.py
View file

@ -105,6 +105,11 @@ class SignPole(AbstractGeoAsset):
models.Index(fields=["code"]),
]
def save(self, *args, **kwargs):
super().save(*args, **kwargs)
for panel in self.signpanels.all():
panel.save()
def get_absolute_url(self):
return reverse('assets:sign_assets_detail', kwargs={'asset_model': 'signpole', 'asset_id':self.id})
@ -165,6 +170,28 @@ class SignPanel(AbstractGeoAsset):
trafficlight_pole = models.ForeignKey("assets.TrafficLightPole", related_name="signpanels_attached", on_delete=models.SET_NULL, null=True, blank=True)
publiclighting_pole = models.ForeignKey("assets.PublicLightingPole", related_name="signpanels_attached", on_delete=models.SET_NULL, null=True, blank=True)
def clean(self):
super().clean()
support = self.pole or self.trafficlight_pole or self.publiclighting_pole
if support and getattr(support, 'geom', None):
self.geom = support.geom
self.lon = support.lon
self.lat = support.lat
self.geojson = support.geojson
if self.pole and self.pole.street and not self.street_id:
self.street = self.pole.street
def save(self, *args, **kwargs):
support = self.pole or self.trafficlight_pole or self.publiclighting_pole
if support and getattr(support, 'geom', None):
self.geom = support.geom
self.lon = support.lon
self.lat = support.lat
self.geojson = support.geojson
if self.pole and self.pole.street and not self.street_id:
self.street = self.pole.street
super().save(*args, **kwargs)
class Meta:
indexes = [
models.Index(fields=["code"]),

24
loko/assets/models/structures.py
View file

@ -169,6 +169,30 @@ class StructureGeoAsset(AbstractGeoAsset):
structure = models.ForeignKey(Structure, on_delete=models.SET_NULL, related_name="assets", null=True, blank=True)
parent_asset = models.ForeignKey('self', on_delete=models.SET_NULL, null=True, blank=True, related_name="child_assets")
def clean(self):
super().clean()
if self.parent_asset and getattr(self.parent_asset, 'geom', None) and not self.geom:
self.geom = self.parent_asset.geom
self.lon = self.parent_asset.lon
self.lat = self.parent_asset.lat
self.geojson = self.parent_asset.geojson
if self.parent_asset and self.parent_asset.location and not self.location_id:
self.location = self.parent_asset.location
if self.parent_asset and self.parent_asset.structure and not self.structure_id:
self.structure = self.parent_asset.structure
def save(self, *args, **kwargs):
if self.parent_asset and getattr(self.parent_asset, 'geom', None) and not self.geom:
self.geom = self.parent_asset.geom
self.lon = self.parent_asset.lon
self.lat = self.parent_asset.lat
self.geojson = self.parent_asset.geojson
if self.parent_asset and self.parent_asset.location and not self.location_id:
self.location = self.parent_asset.location
if self.parent_asset and self.parent_asset.structure and not self.structure_id:
self.structure = self.parent_asset.structure
super().save(*args, **kwargs)
class Meta:
indexes = [
models.Index(fields=["code"]),

168
loko/assets/models/trafficlights.py
View file

@ -297,6 +297,18 @@ class TrafficLightPole(TrafficLightAbstractGeoAsset):
super().save(*args, **kwargs)
for radar in self.radars.all():
radar.save()
for lantern in self.lanterns.all():
lantern.save()
for acc in self.accessories.all():
acc.save()
for det in self.detectors.all():
det.save()
for sp in self.signpanels_attached.all():
sp.save()
for lf in self.light_fixtures.all():
lf.save()
for its in self.its_assets.all():
its.save()
class Meta:
verbose_name = _("Traffic light pole")
@ -323,11 +335,31 @@ class TrafficLightLanternModel(AbstractAssetModel):
voltage = models.IntegerField(verbose_name=_("Tension (V)"))
# Traffic Light Lantern
class TrafficLightLantern(TrafficLightAbstractAsset):
class TrafficLightLantern(TrafficLightAbstractGeoAsset):
model = models.ForeignKey(TrafficLightLanternModel, on_delete=models.PROTECT)
pole = models.ForeignKey(TrafficLightPole, related_name="lanterns", on_delete=models.PROTECT)
orientation = models.FloatField(null=True, blank=True)
@property
def intersection(self):
return self.pole.intersection if self.pole else None
def clean(self):
super().clean()
if self.pole and getattr(self.pole, 'geom', None):
self.geom = self.pole.geom
self.lon = self.pole.lon
self.lat = self.pole.lat
self.geojson = self.pole.geojson
def save(self, *args, **kwargs):
if self.pole and getattr(self.pole, 'geom', None):
self.geom = self.pole.geom
self.lon = self.pole.lon
self.lat = self.pole.lat
self.geojson = self.pole.geojson
super().save(*args, **kwargs)
class Meta:
verbose_name = _("Lanterne")
verbose_name_plural = _("Lanternes")
@ -366,6 +398,17 @@ class TrafficLightController(TrafficLightAbstractGeoAsset):
def sorted_voltage_hardwares(self):
return self.voltage_hardwares.order_by('code')
def save(self, *args, **kwargs):
super().save(*args, **kwargs)
for hw in self.hardwares.all():
hw.save()
for det in self.detectors.all():
det.save()
for net in self.network_hardwares.all():
net.save()
for volt in self.voltage_hardwares.all():
volt.save()
class Meta:
verbose_name = _("Contrôleur")
verbose_name_plural = _("Contrôleurs")
@ -378,6 +421,22 @@ class TrafficLightControllerHardware(TrafficLightAbstractGeoAsset):
name = models.CharField(max_length=255, null=True, blank=True, verbose_name=_("Nom du matériel"))
description = models.TextField(null=True, blank=True, verbose_name=_("Description du matériel"))
def clean(self):
super().clean()
if self.controller and getattr(self.controller, 'geom', None):
self.geom = self.controller.geom
self.lon = self.controller.lon
self.lat = self.controller.lat
self.geojson = self.controller.geojson
def save(self, *args, **kwargs):
if self.controller and getattr(self.controller, 'geom', None):
self.geom = self.controller.geom
self.lon = self.controller.lon
self.lat = self.controller.lat
self.geojson = self.controller.geojson
super().save(*args, **kwargs)
class Meta:
verbose_name = _("Hardware contrôleur")
verbose_name_plural = _("Hardwares contrôleur")
@ -409,6 +468,32 @@ class TrafficLightDetector(TrafficLightAbstractGeoAsset):
phases = ArrayField(models.CharField(max_length=50), null=True, blank=True, verbose_name=_("Phases associées"))
mentionned_on_plan = models.BooleanField(default=False, verbose_name=_("Mentionné sur le plan"))
def clean(self):
super().clean()
support = self.pole or self.controller
if support and getattr(support, 'geom', None):
self.geom = support.geom
self.lon = support.lon
self.lat = support.lat
self.geojson = support.geojson
if self.pole and self.pole.intersection and not self.intersection_id:
self.intersection = self.pole.intersection
elif self.controller and self.controller.intersection and not self.intersection_id:
self.intersection = self.controller.intersection
def save(self, *args, **kwargs):
support = self.pole or self.controller
if support and getattr(support, 'geom', None):
self.geom = support.geom
self.lon = support.lon
self.lat = support.lat
self.geojson = support.geojson
if self.pole and self.pole.intersection and not self.intersection_id:
self.intersection = self.pole.intersection
elif self.controller and self.controller.intersection and not self.intersection_id:
self.intersection = self.controller.intersection
super().save(*args, **kwargs)
class Meta:
verbose_name = _("Détecteur")
verbose_name_plural = _("Détecteurs")
@ -422,6 +507,26 @@ class TrafficLightNetworkHardware(TrafficLightAbstractGeoAsset):
controller = models.ForeignKey(TrafficLightController, related_name="network_hardwares", on_delete=models.PROTECT, null=True, blank=True)
mentionned_on_plan = models.BooleanField(default=False, verbose_name=_("Mentionné sur le plan"))
def clean(self):
super().clean()
if self.controller and getattr(self.controller, 'geom', None):
self.geom = self.controller.geom
self.lon = self.controller.lon
self.lat = self.controller.lat
self.geojson = self.controller.geojson
if self.controller and self.controller.intersection and not self.intersection_id:
self.intersection = self.controller.intersection
def save(self, *args, **kwargs):
if self.controller and getattr(self.controller, 'geom', None):
self.geom = self.controller.geom
self.lon = self.controller.lon
self.lat = self.controller.lat
self.geojson = self.controller.geojson
if self.controller and self.controller.intersection and not self.intersection_id:
self.intersection = self.controller.intersection
super().save(*args, **kwargs)
def __str__(self):
return self.name or f"Matériel {self.id}"
@ -438,6 +543,26 @@ class TrafficLightVoltageHardware(TrafficLightAbstractGeoAsset):
controller = models.ForeignKey(TrafficLightController, related_name="voltage_hardwares", on_delete=models.PROTECT, null=True, blank=True)
mentionned_on_plan = models.BooleanField(default=False, verbose_name=_("Mentionné sur le plan"))
def clean(self):
super().clean()
if self.controller and getattr(self.controller, 'geom', None):
self.geom = self.controller.geom
self.lon = self.controller.lon
self.lat = self.controller.lat
self.geojson = self.controller.geojson
if self.controller and self.controller.intersection and not self.intersection_id:
self.intersection = self.controller.intersection
def save(self, *args, **kwargs):
if self.controller and getattr(self.controller, 'geom', None):
self.geom = self.controller.geom
self.lon = self.controller.lon
self.lat = self.controller.lat
self.geojson = self.controller.geojson
if self.controller and self.controller.intersection and not self.intersection_id:
self.intersection = self.controller.intersection
super().save(*args, **kwargs)
def __str__(self):
return self.name or f"Matériel {self.id}"
@ -494,6 +619,11 @@ class TrafficLightElectricalCabinet(TrafficLightAbstractGeoAsset):
visual_conformity = models.CharField(max_length=255, null=True, blank=True, verbose_name=_("Conformité visuelle"))
cleanliness = models.CharField(max_length=255, null=True, blank=True, verbose_name=_("Propreté"))
def save(self, *args, **kwargs):
super().save(*args, **kwargs)
for content in self.contents.all():
content.save()
class Meta:
verbose_name = _("Armoire")
verbose_name_plural = _("Armoires")
@ -509,6 +639,22 @@ class TrafficLightElectricalCabinetContent(TrafficLightAbstractGeoAsset):
number = models.CharField(max_length=100, null=True, blank=True, verbose_name=_("Numéro de référence"))
ean = models.CharField(max_length=100, null=True, blank=True, verbose_name=_("Code EAN"))
def clean(self):
super().clean()
if self.cabinet and getattr(self.cabinet, 'geom', None):
self.geom = self.cabinet.geom
self.lon = self.cabinet.lon
self.lat = self.cabinet.lat
self.geojson = self.cabinet.geojson
def save(self, *args, **kwargs):
if self.cabinet and getattr(self.cabinet, 'geom', None):
self.geom = self.cabinet.geom
self.lon = self.cabinet.lon
self.lat = self.cabinet.lat
self.geojson = self.cabinet.geojson
super().save(*args, **kwargs)
class Meta:
verbose_name = _("Contenu d'armoire")
verbose_name_plural = _("Contenus d'armoires")
@ -525,6 +671,26 @@ class TrafficLightAccessory(TrafficLightAbstractGeoAsset):
intersection = models.ForeignKey(TrafficLightIntersection, related_name="accessories", on_delete=models.PROTECT, null=True, blank=True)
orientation = models.FloatField(null=True, blank=True)
def clean(self):
super().clean()
if self.pole and getattr(self.pole, 'geom', None):
self.geom = self.pole.geom
self.lon = self.pole.lon
self.lat = self.pole.lat
self.geojson = self.pole.geojson
if self.pole and self.pole.intersection and not self.intersection_id:
self.intersection = self.pole.intersection
def save(self, *args, **kwargs):
if self.pole and getattr(self.pole, 'geom', None):
self.geom = self.pole.geom
self.lon = self.pole.lon
self.lat = self.pole.lat
self.geojson = self.pole.geojson
if self.pole and self.pole.intersection and not self.intersection_id:
self.intersection = self.pole.intersection
super().save(*args, **kwargs)
class Meta:
verbose_name = _("Accessoire")
verbose_name_plural = _("Accessoires")

30
loko/assets/templates/assets/components/asset_location_link.html
View file

@ -1,6 +1,6 @@
{% load i18n %}
{% if location or can_edit_location %}
{% if location or parent_support or can_edit_location %}
<div class="card mb-3">
<div class="card-header bg-light d-flex justify-content-between align-items-center">
<h5 class="mb-0">
@ -8,19 +8,39 @@
{% translate "Rue associée" %}
{% elif location_type == 'intersection' %}
{% translate "Carrefour associé" %}
{% elif location_type == 'building' %}
{% translate "Bâtiment associé" %}
{% elif location_type == 'site' %}
{% translate "Site associé" %}
{% else %}
{% translate "Localisation associée" %}
{% endif %}
</h5>
{% if can_edit_location %}
{% if can_edit_location and not parent_support %}
<button type="button" class="btn btn-sm btn-outline-primary d-none" id="changeLocationBtn" data-bs-toggle="modal" data-bs-target="#changeLocationModal" data-edit-mode-only="true">
<i class="bi bi-pencil me-1"></i>{% translate "Changer" %}
</button>
{% endif %}
</div>
<div class="card-body">
{% if parent_support %}
<div class="mb-2">
<small class="text-muted d-block">{% translate "Support / Équipement parent" %} :</small>
{% if parent_support.get_absolute_url %}
<a href="{{ parent_support.get_absolute_url }}" class="fw-bold text-decoration-none"
title="{% translate 'Voir le détail du support' %}">
<i class="bi bi-layers-fill me-1 text-primary"></i>{{ parent_support.code }}
</a>
{% else %}
<strong>{{ parent_support.code }}</strong>
{% endif %}
- {{ parent_support.get_name }}
</div>
{% endif %}
{% if location %}
<p class="mb-2">
<div>
{% if parent_support %}<small class="text-muted d-block">{% translate "Localisation" %} :</small>{% endif %}
{% if location.get_absolute_url %}
<a href="{{ location.get_absolute_url }}" class="fw-bold text-decoration-none"
title="{% translate 'Voir le détail de la localisation' %}">
@ -30,8 +50,8 @@
<strong>{{ location.code }}</strong>
{% endif %}
- {{ location.get_name }}
</p>
{% else %}
</div>
{% elif not parent_support %}
<p class="text-muted mb-0">{% translate "Aucune localisation associée" %}</p>
{% endif %}
</div>

71
loko/assets/templatetags/asset_tags.py
View file

@ -162,23 +162,46 @@ def get_asset_specific_fields(asset):
@register.inclusion_tag('assets/components/asset_location_link.html', takes_context=True)
def asset_location_link(context, asset):
"""
Display a link to the location associated with an asset.
Works with assets that have a 'street', 'location', or 'intersection' ForeignKey.
Display a link to the location and/or parent support associated with an asset.
Works with assets that have a direct location ('street', 'location', 'intersection', etc.)
or are attached to a parent support (pole, controller, cabinet, etc.).
"""
# Try different possible location field names
location = None
location_type = None
if hasattr(asset, 'street') and getattr(asset, 'street', None):
location = getattr(asset, 'street', None)
location_type = 'street'
elif hasattr(asset, 'location') and getattr(asset, 'location', None):
location = getattr(asset, 'location', None)
location_type = 'location'
elif hasattr(asset, 'intersection') and getattr(asset, 'intersection', None):
location = getattr(asset, 'intersection', None)
location_type = 'intersection'
if hasattr(asset, 'get_location'):
location = asset.get_location()
if not location:
if hasattr(asset, 'street') and getattr(asset, 'street', None):
location = getattr(asset, 'street', None)
elif hasattr(asset, 'location') and getattr(asset, 'location', None):
location = getattr(asset, 'location', None)
elif hasattr(asset, 'intersection') and getattr(asset, 'intersection', None):
location = getattr(asset, 'intersection', None)
elif hasattr(asset, 'building') and getattr(asset, 'building', None):
location = getattr(asset, 'building', None)
elif hasattr(asset, 'site') and getattr(asset, 'site', None):
location = getattr(asset, 'site', None)
if location:
loc_model = getattr(location._meta, 'model_name', '').lower()
if 'street' in loc_model:
location_type = 'street'
elif 'intersection' in loc_model:
location_type = 'intersection'
elif 'building' in loc_model:
location_type = 'building'
elif 'outdoor' in loc_model:
location_type = 'site'
else:
location_type = 'location'
# Parent support (pole, controller, cabinet, parent_asset...)
parent_support = None
if hasattr(asset, 'get_parent_support'):
parent_support = asset.get_parent_support()
# Determine if user can edit location (similar to model editing)
can_edit_location = False
user = context.get('request').user if context.get('request') else None
@ -198,12 +221,34 @@ def asset_location_link(context, asset):
thematic_code = 'trafficlights'
elif 'publiclighting' in model_name.lower():
thematic_code = 'publiclighting'
# Add more thematic mappings as needed
elif 'sign' in model_name.lower():
thematic_code = 'sign'
elif 'clean' in model_name.lower():
thematic_code = 'clean'
elif 'nature' in model_name.lower():
thematic_code = 'nature'
elif 'building' in model_name.lower():
thematic_code = 'buildings'
elif 'outdoor' in model_name.lower():
thematic_code = 'outdoor'
elif 'its' in model_name.lower():
thematic_code = 'its'
elif 'controlcenter' in model_name.lower():
thematic_code = 'controlcenters'
elif 'parking' in model_name.lower():
thematic_code = 'parking'
elif 'road' in model_name.lower():
thematic_code = 'roads'
elif 'artwork' in model_name.lower():
thematic_code = 'artworks'
elif 'water' in model_name.lower():
thematic_code = 'water'
return {
'asset': asset,
'location': location,
'location_type': location_type,
'parent_support': parent_support,
'can_edit_location': can_edit_location,
'thematic_code': thematic_code,
}

153
loko/assets/tests.py
View file

@ -2602,8 +2602,161 @@ class GeoSyncMixinTest(TestCase):
self.assertIn('4.36', pole.geojson)
class SyncGeoJsonCommandTest(TestCase):
def test_sync_all_geojson_command(self):
from io import StringIO
from django.core.management import call_command
from django.contrib.gis.geos import Point
from assets.models.publiclighting import PublicLightingPole, PublicLightingPoleModel, PublicLightingStreet
street = PublicLightingStreet.objects.create(code="ST-TEST", name_fr="Rue Test")
pole_model = PublicLightingPoleModel.objects.create(code="PM-TEST", name_fr="Model Test")
pt = Point(4.3517, 50.8467, srid=4326).transform(3812, clone=True)
pole = PublicLightingPole.objects.create(
code="TEST-SYNC-GEOJSON",
street=street,
model=pole_model,
geom=pt,
)
PublicLightingPole.objects.filter(id=pole.id).update(geojson=None)
pole.refresh_from_db()
self.assertIsNone(pole.geojson)
out = StringIO()
call_command("sync_all_geojson", "--dry-run", stdout=out)
self.assertIn("[DRY-RUN] Would update", out.getvalue())
pole.refresh_from_db()
self.assertIsNone(pole.geojson)
out = StringIO()
call_command("sync_all_geojson", stdout=out)
self.assertIn("Successfully updated", out.getvalue())
pole.refresh_from_db()
self.assertIsNotNone(pole.geojson)
self.assertIn("4.3517", pole.geojson)
class ParentChildAssetLocationAndGeoSyncTest(TestCase):
def setUp(self):
from django.contrib.gis.geos import Point
from assets.models.trafficlights import (
TrafficLightIntersection,
TrafficLightPole,
TrafficLightPoleModel,
TrafficLightLantern,
TrafficLightLanternModel,
)
from assets.models.sign import SignStreet, SignPole, SignPoleModel, SignPanel
from assets.models.publiclighting import (
PublicLightingStreet,
PublicLightingPole,
PublicLightingPoleModel,
PublicLightingLightFixture,
PublicLightingLightFixtureModel,
)
self.point_3812 = Point(649529.6, 669829.8, srid=3812)
# Traffic light setup
self.intersection = TrafficLightIntersection.objects.create(
code="TL-INT-01",
name_fr="Carrefour Test",
geom=self.point_3812
)
self.tl_pole_model = TrafficLightPoleModel.objects.create(code="TLPM-01", name_fr="Modèle Poteau TL")
self.tl_pole = TrafficLightPole.objects.create(
code="TL-POL-01",
intersection=self.intersection,
model=self.tl_pole_model,
geom=self.point_3812
)
self.lantern_model = TrafficLightLanternModel.objects.create(
code="LAN-MOD-01",
name_fr="Modèle Lanterne",
voltage=230
)
# Sign setup
self.sign_street = SignStreet.objects.create(code="STR-01", name_fr="Rue du Test")
self.sign_pole_model = SignPoleModel.objects.create(code="SPM-01", name_fr="Modèle Poteau Sign")
self.sign_pole = SignPole.objects.create(
code="SP-01",
street=self.sign_street,
model=self.sign_pole_model,
geom=self.point_3812
)
# Public lighting setup
self.pl_street = PublicLightingStreet.objects.create(code="PL-STR-01", name_fr="Rue Eclairage")
self.pl_pole_model = PublicLightingPoleModel.objects.create(code="PLPM-01", name_fr="Modèle Poteau PL")
self.pl_pole = PublicLightingPole.objects.create(
code="PL-POL-01",
street=self.pl_street,
model=self.pl_pole_model,
geom=self.point_3812
)
self.fixture_model = PublicLightingLightFixtureModel.objects.create(code="LFM-01", name_fr="Modèle Luminaire")
def test_lantern_inherits_coordinates_and_location_from_pole(self):
from assets.models.trafficlights import TrafficLightLantern
lantern = TrafficLightLantern.objects.create(
code="TL-LAN-01",
pole=self.tl_pole,
model=self.lantern_model
)
lantern.refresh_from_db()
self.assertIsNotNone(lantern.geom)
self.assertEqual(lantern.geom, self.tl_pole.geom)
self.assertIsNotNone(lantern.lon)
self.assertIsNotNone(lantern.lat)
self.assertIsNotNone(lantern.geojson)
self.assertEqual(lantern.intersection, self.intersection)
self.assertEqual(lantern.get_location(), self.intersection)
self.assertEqual(lantern.get_parent_support(), self.tl_pole)
def test_signpanel_attached_to_trafficlight_pole_inherits_coords(self):
from assets.models.sign import SignPanel
panel = SignPanel.objects.create(
code="PANEL-01",
trafficlight_pole=self.tl_pole
)
panel.refresh_from_db()
self.assertIsNotNone(panel.geom)
self.assertEqual(panel.geom, self.tl_pole.geom)
self.assertEqual(panel.get_parent_support(), self.tl_pole)
self.assertEqual(panel.get_location(), self.intersection)
def test_publiclighting_fixture_inherits_coords_and_street(self):
from assets.models.publiclighting import PublicLightingLightFixture
fixture = PublicLightingLightFixture.objects.create(
code="FIXTURE-01",
pole=self.pl_pole,
street=self.pl_street,
model=self.fixture_model
)
fixture.refresh_from_db()
self.assertIsNotNone(fixture.geom)
self.assertEqual(fixture.geom, self.pl_pole.geom)
self.assertEqual(fixture.get_parent_support(), self.pl_pole)
self.assertEqual(fixture.get_location(), self.pl_street)
def test_pole_save_cascades_to_children(self):
from django.contrib.gis.geos import Point
from assets.models.trafficlights import TrafficLightLantern
lantern = TrafficLightLantern.objects.create(
code="TL-LAN-02",
pole=self.tl_pole,
model=self.lantern_model
)
new_point = Point(650000.0, 670000.0, srid=3812)
self.tl_pole.geom = new_point
self.tl_pole.save()
lantern.refresh_from_db()
self.assertEqual(lantern.geom, new_point)

4
loko/assets/views/asset_editing.py
View file

@ -2437,7 +2437,9 @@ def create_intervention_for_asset(request, asset_model, asset_id):
content_type=content_type,
object_id=asset.pk
)
loc_obj = getattr(asset, 'street', None) or getattr(asset, 'location', None) or getattr(asset, 'intersection', None)
loc_obj = (
asset.get_location() if hasattr(asset, 'get_location') else None
) or getattr(asset, 'street', None) or getattr(asset, 'location', None) or getattr(asset, 'intersection', None)
if loc_obj:
loc_ct = ContentType.objects.get_for_model(loc_obj)
InterventionLocation.objects.get_or_create(

4
loko/assets/views/trafficlights.py
View file

@ -1289,6 +1289,10 @@ def trafficlights_assets_detail(request, asset_model, asset_id):
# Build GeoJSON of sibling assets at the same intersection (excluding current asset)
intersection_assets_geojson = None
asset_intersection = getattr(asset, 'intersection', None)
if not asset_intersection and hasattr(asset, 'get_location'):
loc = asset.get_location()
if loc and getattr(loc._meta, 'model_name', '') == 'trafficlightintersection':
asset_intersection = loc
if asset_intersection:
assets_features = []
active_assets = asset_intersection.get_active_assets()