diff --git a/loko/assets/management/commands/sync_asset_parent_data.py b/loko/assets/management/commands/sync_asset_parent_data.py new file mode 100644 index 0000000..b86b872 --- /dev/null +++ b/loko/assets/management/commands/sync_asset_parent_data.py @@ -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!")) diff --git a/loko/assets/migrations/0020_trafficlightlantern_geojson_trafficlightlantern_geom_and_more.py b/loko/assets/migrations/0020_trafficlightlantern_geojson_trafficlightlantern_geom_and_more.py new file mode 100644 index 0000000..384decd --- /dev/null +++ b/loko/assets/migrations/0020_trafficlightlantern_geojson_trafficlightlantern_geom_and_more.py @@ -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), + ), + ] diff --git a/loko/assets/migrations/0021_sync_asset_parent_data.py b/loko/assets/migrations/0021_sync_asset_parent_data.py new file mode 100644 index 0000000..cd382a4 --- /dev/null +++ b/loko/assets/migrations/0021_sync_asset_parent_data.py @@ -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), + ] diff --git a/loko/assets/models/buildings.py b/loko/assets/models/buildings.py index 07ff187..08fd988 100644 --- a/loko/assets/models/buildings.py +++ b/loko/assets/models/buildings.py @@ -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") diff --git a/loko/assets/models/core.py b/loko/assets/models/core.py index b6e0a11..7714df2 100644 --- a/loko/assets/models/core.py +++ b/loko/assets/models/core.py @@ -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 diff --git a/loko/assets/models/its.py b/loko/assets/models/its.py index daa9c7b..23b298e 100644 --- a/loko/assets/models/its.py +++ b/loko/assets/models/its.py @@ -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"]), diff --git a/loko/assets/models/outdoor.py b/loko/assets/models/outdoor.py index 0b0bbd5..3d1441f 100644 --- a/loko/assets/models/outdoor.py +++ b/loko/assets/models/outdoor.py @@ -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") diff --git a/loko/assets/models/publiclighting.py b/loko/assets/models/publiclighting.py index 98fcddc..d06d4c4 100644 --- a/loko/assets/models/publiclighting.py +++ b/loko/assets/models/publiclighting.py @@ -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"]), diff --git a/loko/assets/models/sign.py b/loko/assets/models/sign.py index f47c86c..276d6ae 100644 --- a/loko/assets/models/sign.py +++ b/loko/assets/models/sign.py @@ -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"]), diff --git a/loko/assets/models/structures.py b/loko/assets/models/structures.py index 981abdb..32211c0 100644 --- a/loko/assets/models/structures.py +++ b/loko/assets/models/structures.py @@ -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"]), diff --git a/loko/assets/models/trafficlights.py b/loko/assets/models/trafficlights.py index 9ca2326..e083269 100644 --- a/loko/assets/models/trafficlights.py +++ b/loko/assets/models/trafficlights.py @@ -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") diff --git a/loko/assets/templates/assets/components/asset_location_link.html b/loko/assets/templates/assets/components/asset_location_link.html index 44c7e1e..6a1c179 100644 --- a/loko/assets/templates/assets/components/asset_location_link.html +++ b/loko/assets/templates/assets/components/asset_location_link.html @@ -1,6 +1,6 @@ {% load i18n %} -{% if location or can_edit_location %} +{% if location or parent_support or can_edit_location %}
@@ -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 %}
- {% if can_edit_location %} + {% if can_edit_location and not parent_support %} {% endif %}
+ {% if parent_support %} +
+ {% translate "Support / Équipement parent" %} : + {% if parent_support.get_absolute_url %} + + {{ parent_support.code }} + + {% else %} + {{ parent_support.code }} + {% endif %} + - {{ parent_support.get_name }} +
+ {% endif %} + {% if location %} -

+

+ {% if parent_support %}{% translate "Localisation" %} :{% endif %} {% if location.get_absolute_url %} @@ -30,8 +50,8 @@ {{ location.code }} {% endif %} - {{ location.get_name }} -

- {% else %} +
+ {% elif not parent_support %}

{% translate "Aucune localisation associée" %}

{% endif %}
diff --git a/loko/assets/templatetags/asset_tags.py b/loko/assets/templatetags/asset_tags.py index e748318..953fbec 100644 --- a/loko/assets/templatetags/asset_tags.py +++ b/loko/assets/templatetags/asset_tags.py @@ -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, } diff --git a/loko/assets/tests.py b/loko/assets/tests.py index c4ddac4..ab7d2a3 100644 --- a/loko/assets/tests.py +++ b/loko/assets/tests.py @@ -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) diff --git a/loko/assets/views/asset_editing.py b/loko/assets/views/asset_editing.py index 647abec..adafff4 100644 --- a/loko/assets/views/asset_editing.py +++ b/loko/assets/views/asset_editing.py @@ -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( diff --git a/loko/assets/views/trafficlights.py b/loko/assets/views/trafficlights.py index be3b9b9..29aea0c 100644 --- a/loko/assets/views/trafficlights.py +++ b/loko/assets/views/trafficlights.py @@ -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()