feat: add duplicate sign panel suppression, inactive status filtering, and management command
This commit is contained in:
parent
82e20f648e
commit
025ddd09fc
5 changed files with 526 additions and 6 deletions
|
|
@ -49,6 +49,47 @@
|
|||
return geojsonData;
|
||||
}
|
||||
|
||||
// Deduplicate features by unique id per asset_type
|
||||
const seenFeatureIds = new Set();
|
||||
const uniqueFeatures = [];
|
||||
geojsonData.features.forEach(feature => {
|
||||
if (!feature) return;
|
||||
const fid = (feature.id != null) ? `${feature.properties?.asset_type || ''}_${feature.id}` : null;
|
||||
if (fid && seenFeatureIds.has(fid)) {
|
||||
return;
|
||||
}
|
||||
if (fid) {
|
||||
seenFeatureIds.add(fid);
|
||||
}
|
||||
uniqueFeatures.push(feature);
|
||||
});
|
||||
geojsonData.features = uniqueFeatures;
|
||||
|
||||
// Index attached panels by coordinates and type to suppress unattached duplicate icons
|
||||
const attachedKeys = new Set();
|
||||
geojsonData.features.forEach(feature => {
|
||||
const props = feature && feature.properties;
|
||||
if (props && props.asset_type === 'signpanel' && props.pole_id && feature.geometry && feature.geometry.coordinates) {
|
||||
const coords = feature.geometry.coordinates;
|
||||
const typeCode = (props.panel_type || '').trim().toUpperCase();
|
||||
attachedKeys.add(`${Number(coords[0]).toFixed(5)}_${Number(coords[1]).toFixed(5)}_${typeCode}`);
|
||||
}
|
||||
});
|
||||
|
||||
if (attachedKeys.size > 0) {
|
||||
geojsonData.features = geojsonData.features.filter(feature => {
|
||||
const props = feature && feature.properties;
|
||||
if (props && props.asset_type === 'signpanel' && !props.pole_id && feature.geometry && feature.geometry.coordinates) {
|
||||
const coords = feature.geometry.coordinates;
|
||||
const typeCode = (props.panel_type || '').trim().toUpperCase();
|
||||
if (attachedKeys.has(`${Number(coords[0]).toFixed(5)}_${Number(coords[1]).toFixed(5)}_${typeCode}`)) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
return true;
|
||||
});
|
||||
}
|
||||
|
||||
const panelsByPole = {};
|
||||
|
||||
geojsonData.features.forEach(feature => {
|
||||
|
|
|
|||
|
|
@ -628,8 +628,14 @@ def sign_assets_geojson(request):
|
|||
|
||||
return feature
|
||||
|
||||
seen_feature_keys = set()
|
||||
|
||||
def add_assets_to_features(queryset, asset_type, features, feature_count):
|
||||
for obj in queryset:
|
||||
key = (asset_type, obj.id)
|
||||
if key in seen_feature_keys:
|
||||
continue
|
||||
seen_feature_keys.add(key)
|
||||
if feature_count >= MAX_ASSET_FEATURES:
|
||||
return feature_count, True # early stop
|
||||
features.append(obj_to_feature(obj, asset_type=asset_type))
|
||||
|
|
@ -644,10 +650,17 @@ def sign_assets_geojson(request):
|
|||
)
|
||||
signpanel_qs = (
|
||||
SignPanel.objects
|
||||
.select_related('street', 'category', 'model', 'signpanel_type')
|
||||
.select_related('street', 'category', 'model', 'signpanel_type', 'pole')
|
||||
.all()
|
||||
)
|
||||
|
||||
if statuses:
|
||||
signpole_qs = signpole_qs.filter(status__in=statuses)
|
||||
signpanel_qs = signpanel_qs.filter(status__in=statuses)
|
||||
else:
|
||||
signpole_qs = signpole_qs.exclude(status__in=['removed', 'archived'])
|
||||
signpanel_qs = signpanel_qs.exclude(status__in=['removed', 'archived'])
|
||||
|
||||
# --------- Filtres métier existants
|
||||
if sign_location_ids:
|
||||
signpole_qs = signpole_qs.filter(street__id__in=sign_location_ids)
|
||||
|
|
@ -685,6 +698,9 @@ def sign_assets_geojson(request):
|
|||
signpole_qs = apply_search(signpole_qs, data)
|
||||
signpanel_qs = apply_search(signpanel_qs, data)
|
||||
|
||||
signpole_qs = signpole_qs.distinct()
|
||||
signpanel_qs = signpanel_qs.distinct()
|
||||
|
||||
# --------- Cap de sécurité
|
||||
signpole_qs = signpole_qs[0:MAX_ASSET_FEATURES+1]
|
||||
if len(signpole_qs) >= MAX_ASSET_FEATURES:
|
||||
|
|
@ -694,7 +710,26 @@ def sign_assets_geojson(request):
|
|||
if len(signpanel_qs) >= MAX_ASSET_FEATURES:
|
||||
return JsonResponse({"error": f"{_("Trop d'objets à afficher")} (> {MAX_ASSET_FEATURES})"}, status=413)
|
||||
|
||||
signpanel_list = list(signpanel_qs)
|
||||
raw_signpanel_list = list(signpanel_qs)
|
||||
|
||||
# Index attached panels by pole coordinates and signpanel_type to detect and suppress duplicate orphan panels
|
||||
attached_panel_keys = set()
|
||||
for p in raw_signpanel_list:
|
||||
if p.pole_id is not None and p.lon is not None and p.lat is not None and p.signpanel_type_id:
|
||||
coord_key = (round(float(p.lon), 5), round(float(p.lat), 5), p.signpanel_type_id)
|
||||
attached_panel_keys.add(coord_key)
|
||||
|
||||
signpanel_list = []
|
||||
seen_panel_ids = set()
|
||||
for p in raw_signpanel_list:
|
||||
if p.id in seen_panel_ids:
|
||||
continue
|
||||
if p.pole_id is None and p.lon is not None and p.lat is not None and p.signpanel_type_id:
|
||||
coord_key = (round(float(p.lon), 5), round(float(p.lat), 5), p.signpanel_type_id)
|
||||
if coord_key in attached_panel_keys:
|
||||
continue
|
||||
seen_panel_ids.add(p.id)
|
||||
signpanel_list.append(p)
|
||||
panel_render_meta = {}
|
||||
for panel in signpanel_list:
|
||||
panel_type_code = panel.signpanel_type.code.strip() if getattr(panel, "signpanel_type", None) and panel.signpanel_type.code else None
|
||||
|
|
|
|||
238
loko/sign/management/commands/check_duplicate_sign_panels.py
Normal file
238
loko/sign/management/commands/check_duplicate_sign_panels.py
Normal file
|
|
@ -0,0 +1,238 @@
|
|||
"""
|
||||
Management command to detect, report, and safely fix duplicate and orphan road sign panels.
|
||||
|
||||
Usage:
|
||||
# Inspection / Dry-run report:
|
||||
python manage.py check_duplicate_sign_panels [--org_id 1]
|
||||
|
||||
# Fix duplicate orphan panels by archiving them:
|
||||
python manage.py check_duplicate_sign_panels [--org_id 1] --fix
|
||||
|
||||
# Fix duplicate orphan panels by deleting them:
|
||||
python manage.py check_duplicate_sign_panels [--org_id 1] --fix --action delete
|
||||
|
||||
# Also automatically attach unattached non-duplicate panels to the nearest pole (within 1m):
|
||||
python manage.py check_duplicate_sign_panels [--org_id 1] --fix --attach-candidates
|
||||
"""
|
||||
|
||||
from django.core.management.base import BaseCommand
|
||||
from django.db import transaction
|
||||
from django.db.models import Count, Q
|
||||
|
||||
from sign.models import SignPanel, SignPole, Organization
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
help = "Détecte, affiche et corrige les panneaux de signalisation en double et orphelins."
|
||||
|
||||
def add_arguments(self, parser):
|
||||
parser.add_argument(
|
||||
"--org_id",
|
||||
type=int,
|
||||
default=None,
|
||||
help="ID de l'organisation à analyser (optionnel).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--fix",
|
||||
action="store_true",
|
||||
default=False,
|
||||
help="Applique les corrections en base de données. Sans cette option, la commande s'exécute en mode consultation (DRY-RUN).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--action",
|
||||
type=str,
|
||||
choices=["archive", "delete"],
|
||||
default="archive",
|
||||
help="Action pour les doublons orphelins : 'archive' (par défaut, conserve l'historique) ou 'delete'.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--attach-candidates",
|
||||
action="store_true",
|
||||
default=False,
|
||||
help="Rattache automatiquement les panneaux orphelins non-doublons au poteau situé à la même position (< 1m).",
|
||||
)
|
||||
|
||||
def handle(self, *args, **options):
|
||||
org_id = options.get("org_id")
|
||||
fix_mode = options.get("fix", False)
|
||||
action = options.get("action", "archive")
|
||||
attach_candidates = options.get("attach_candidates", False)
|
||||
|
||||
self.stdout.write("=" * 80)
|
||||
self.stdout.write(self.style.NOTICE("DIAGNOSTIC DES PANNEAUX DE SIGNALISATION (DOUBLONS ET ORPHELINS)"))
|
||||
self.stdout.write("=" * 80)
|
||||
|
||||
if not fix_mode:
|
||||
self.stdout.write(self.style.WARNING("Mode consultation (DRY-RUN) : aucune modification ne sera enregistrée.\nUtilisez --fix pour appliquer les corrections.\n"))
|
||||
else:
|
||||
self.stdout.write(self.style.SUCCESS(f"Mode CORRECTION actif (action doublons : {action.upper()}).\n"))
|
||||
|
||||
# Base querysets
|
||||
panels_qs = SignPanel.objects.select_related("pole", "signpanel_type", "street", "organization")
|
||||
poles_qs = SignPole.objects.select_related("street", "organization")
|
||||
|
||||
if org_id:
|
||||
panels_qs = panels_qs.filter(organization_id=org_id)
|
||||
poles_qs = poles_qs.filter(organization_id=org_id)
|
||||
self.stdout.write(f"Filtrage par organisation ID : {org_id}")
|
||||
|
||||
total_panels = panels_qs.count()
|
||||
total_poles = poles_qs.count()
|
||||
self.stdout.write(f"Total poteaux analysés : {total_poles}")
|
||||
self.stdout.write(f"Total panneaux analysés : {total_panels}\n")
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# 1. Doublons de panneaux attachés sur un même poteau
|
||||
# (même pole_id, même type de panneau, statut actif)
|
||||
# ---------------------------------------------------------------------
|
||||
self.stdout.write("-" * 80)
|
||||
self.stdout.write(self.style.NOTICE("1. RECHERCHE DES DOUBLONS ATTACHÉS AU MÊME POTEAU"))
|
||||
self.stdout.write("-" * 80)
|
||||
|
||||
attached_dupes_query = (
|
||||
panels_qs.filter(pole__isnull=False, signpanel_type__isnull=False, status="active")
|
||||
.values("pole_id", "signpanel_type_id")
|
||||
.annotate(count=Count("id"))
|
||||
.filter(count__gt=1)
|
||||
)
|
||||
|
||||
attached_dupe_groups = list(attached_dupes_query)
|
||||
self.stdout.write(f"Groupes de doublons sur un même poteau trouvés : {len(attached_dupe_groups)}")
|
||||
|
||||
attached_dupes_to_fix = []
|
||||
for group in attached_dupe_groups:
|
||||
p_id = group["pole_id"]
|
||||
type_id = group["signpanel_type_id"]
|
||||
matching_panels = list(
|
||||
panels_qs.filter(pole_id=p_id, signpanel_type_id=type_id, status="active").order_by("id")
|
||||
)
|
||||
# Keep the first (oldest), others are duplicates
|
||||
kept = matching_panels[0]
|
||||
dupes = matching_panels[1:]
|
||||
type_code = kept.signpanel_type.code if kept.signpanel_type else "?"
|
||||
self.stdout.write(
|
||||
f" • Poteau ID {p_id} (Code {kept.pole.code if kept.pole else '-'}) : "
|
||||
f"{len(matching_panels)} panneaux de type {type_code} (Conservé : ID {kept.id}, Doublon(s) : {[d.id for d in dupes]})"
|
||||
)
|
||||
attached_dupes_to_fix.extend(dupes)
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# 2. Panneaux orphelins (pole__isnull=True) à l'emplacement d'un poteau
|
||||
# ---------------------------------------------------------------------
|
||||
self.stdout.write("\n" + "-" * 80)
|
||||
self.stdout.write(self.style.NOTICE("2. RECHERCHE DES PANNEAUX ORPHELINS À L'EMPLACEMENT D'UN POTEAU"))
|
||||
self.stdout.write("-" * 80)
|
||||
|
||||
# Pre-load poles with coordinates
|
||||
poles_by_coord = {}
|
||||
for pole in poles_qs.iterator():
|
||||
if pole.lon is not None and pole.lat is not None:
|
||||
ck = (round(float(pole.lon), 4), round(float(pole.lat), 4))
|
||||
if ck not in poles_by_coord:
|
||||
poles_by_coord[ck] = []
|
||||
poles_by_coord[ck].append(pole)
|
||||
|
||||
# Pre-load attached panels per pole
|
||||
attached_types_by_pole = {}
|
||||
for p in panels_qs.filter(pole__isnull=False).iterator():
|
||||
if p.pole_id not in attached_types_by_pole:
|
||||
attached_types_by_pole[p.pole_id] = set()
|
||||
if p.signpanel_type_id:
|
||||
attached_types_by_pole[p.pole_id].add(p.signpanel_type_id)
|
||||
|
||||
orphan_panels = list(
|
||||
panels_qs.filter(pole__isnull=True, lon__isnull=False, lat__isnull=False)
|
||||
)
|
||||
self.stdout.write(f"Total panneaux orphelins avec coordonnées : {len(orphan_panels)}")
|
||||
|
||||
orphan_exact_duplicates = []
|
||||
orphan_attachable_candidates = []
|
||||
|
||||
for orphan in orphan_panels:
|
||||
ck = (round(float(orphan.lon), 4), round(float(orphan.lat), 4))
|
||||
nearby_poles = poles_by_coord.get(ck, [])
|
||||
if not nearby_poles:
|
||||
continue
|
||||
|
||||
target_pole = nearby_poles[0]
|
||||
target_types = attached_types_by_pole.get(target_pole.id, set())
|
||||
|
||||
type_code = orphan.signpanel_type.code if orphan.signpanel_type else "?"
|
||||
if orphan.signpanel_type_id and orphan.signpanel_type_id in target_types:
|
||||
# This orphan panel is an exact duplicate of a panel already attached to this pole!
|
||||
orphan_exact_duplicates.append((orphan, target_pole))
|
||||
self.stdout.write(
|
||||
f" • Doublon orphelin : Panneau ID {orphan.id} (Code {orphan.code}, Type {type_code}) "
|
||||
f"partage l'emplacement du Poteau ID {target_pole.id} (Code {target_pole.code}) qui possède DÉJÀ ce type !"
|
||||
)
|
||||
else:
|
||||
orphan_attachable_candidates.append((orphan, target_pole))
|
||||
|
||||
self.stdout.write(f"\nDoublons orphelins redondants détectés : {len(orphan_exact_duplicates)}")
|
||||
self.stdout.write(f"Panneaux orphelins pouvant être rattachés à un poteau : {len(orphan_attachable_candidates)}")
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# 3. Panneaux inactifs (removed / archived)
|
||||
# ---------------------------------------------------------------------
|
||||
self.stdout.write("\n" + "-" * 80)
|
||||
self.stdout.write(self.style.NOTICE("3. STATUTS INACTIFS"))
|
||||
self.stdout.write("-" * 80)
|
||||
inactive_count = panels_qs.filter(status__in=["removed", "archived"]).count()
|
||||
self.stdout.write(f"Panneaux retirés ou archivés en base : {inactive_count} (exclus de la carte par défaut)")
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# 4. Application des corrections si --fix est demandé
|
||||
# ---------------------------------------------------------------------
|
||||
if not fix_mode:
|
||||
self.stdout.write("\n" + "=" * 80)
|
||||
self.stdout.write(self.style.WARNING("FIN DU RAPPORT. Aucune modification effectuée (lancez avec --fix pour corriger)."))
|
||||
self.stdout.write("=" * 80)
|
||||
return
|
||||
|
||||
self.stdout.write("\n" + "=" * 80)
|
||||
self.stdout.write(self.style.NOTICE("APPLICATION DES CORRECTIONS EN BASE..."))
|
||||
self.stdout.write("=" * 80)
|
||||
|
||||
with transaction.atomic():
|
||||
fixed_attached = 0
|
||||
fixed_orphans = 0
|
||||
attached_count = 0
|
||||
|
||||
# Fix attached duplicates
|
||||
for dupe in attached_dupes_to_fix:
|
||||
if action == "delete":
|
||||
dupe.delete()
|
||||
else:
|
||||
dupe.status = "archived"
|
||||
dupe.note = (dupe.note or "") + " [Doublon archivé par check_duplicate_sign_panels]"
|
||||
dupe.save(update_fields=["status", "note"])
|
||||
fixed_attached += 1
|
||||
|
||||
# Fix orphan duplicates
|
||||
for orphan, target_pole in orphan_exact_duplicates:
|
||||
if action == "delete":
|
||||
orphan.delete()
|
||||
else:
|
||||
orphan.status = "archived"
|
||||
orphan.note = (orphan.note or "") + f" [Doublon orphelin du poteau {target_pole.code} archivé]"
|
||||
orphan.save(update_fields=["status", "note"])
|
||||
fixed_orphans += 1
|
||||
|
||||
# Attach candidates if requested
|
||||
if attach_candidates:
|
||||
for orphan, target_pole in orphan_attachable_candidates:
|
||||
orphan.pole = target_pole
|
||||
orphan.street = target_pole.street
|
||||
orphan.geom = target_pole.geom
|
||||
orphan.lon = target_pole.lon
|
||||
orphan.lat = target_pole.lat
|
||||
orphan.geojson = target_pole.geojson
|
||||
orphan.save()
|
||||
attached_count += 1
|
||||
|
||||
self.stdout.write(self.style.SUCCESS(
|
||||
f"\n✓ Succès :\n"
|
||||
f" - {fixed_attached} doublon(s) attaché(s) {action}s\n"
|
||||
f" - {fixed_orphans} doublon(s) orphelin(s) {action}s\n"
|
||||
f" - {attached_count} panneau(x) orphelin(s) rattaché(s) à leur poteau"
|
||||
))
|
||||
|
|
@ -1201,6 +1201,177 @@ class SupportRepresentationAndModelTests(TestCase):
|
|||
self.assertEqual(tl_support['properties']['support_model'], 'Feu Carrefour 1')
|
||||
self.assertTrue(tl_support['properties'].get('is_standalone_support'))
|
||||
|
||||
def test_sign_assets_geojson_excludes_inactive_statuses_by_default(self):
|
||||
"""Verify that sign_assets_geojson excludes removed/archived panels by default, but includes them when filtered."""
|
||||
from django.contrib.gis.geos import Point
|
||||
from sign.models import SignPanel
|
||||
import json
|
||||
|
||||
pt = Point(145000, 172000, srid=3812)
|
||||
active_panel = SignPanel.objects.create(
|
||||
code="PANEL-ACTIVE-01",
|
||||
street=self.adr_street,
|
||||
model=self.panel_model,
|
||||
signpanel_type=self.panel_type,
|
||||
organization=self.org,
|
||||
geom=pt,
|
||||
lon=4.32,
|
||||
lat=50.85,
|
||||
status="active",
|
||||
)
|
||||
archived_panel = SignPanel.objects.create(
|
||||
code="PANEL-ARCHIVED-01",
|
||||
street=self.adr_street,
|
||||
model=self.panel_model,
|
||||
signpanel_type=self.panel_type,
|
||||
organization=self.org,
|
||||
geom=pt,
|
||||
lon=4.32,
|
||||
lat=50.85,
|
||||
status="archived",
|
||||
)
|
||||
|
||||
self.client.force_login(self.user)
|
||||
|
||||
# Default query (no status specified): only active panel returned
|
||||
resp = self.client.post(
|
||||
reverse('sign:sign_assets_geojson'),
|
||||
data=json.dumps({'organization': [self.org.id]}),
|
||||
content_type='application/json'
|
||||
)
|
||||
self.assertEqual(resp.status_code, 200)
|
||||
features = resp.json().get('features', [])
|
||||
codes = [f['properties']['code'] for f in features if f['properties'].get('asset_type') == 'signpanel']
|
||||
self.assertIn("PANEL-ACTIVE-01", codes)
|
||||
self.assertNotIn("PANEL-ARCHIVED-01", codes)
|
||||
|
||||
# Explicit query for archived status: archived panel returned
|
||||
resp_archived = self.client.post(
|
||||
reverse('sign:sign_assets_geojson'),
|
||||
data=json.dumps({'organization': [self.org.id], 'asset_status': ['archived']}),
|
||||
content_type='application/json'
|
||||
)
|
||||
self.assertEqual(resp_archived.status_code, 200)
|
||||
features_archived = resp_archived.json().get('features', [])
|
||||
codes_archived = [f['properties']['code'] for f in features_archived if f['properties'].get('asset_type') == 'signpanel']
|
||||
self.assertIn("PANEL-ARCHIVED-01", codes_archived)
|
||||
self.assertNotIn("PANEL-ACTIVE-01", codes_archived)
|
||||
|
||||
def test_sign_assets_geojson_suppresses_orphan_duplicate_at_pole(self):
|
||||
"""Verify that sign_assets_geojson suppresses unlinked orphan panels that duplicate an attached panel on a pole."""
|
||||
from django.contrib.gis.geos import Point
|
||||
from sign.models import SignPole, SignPanel
|
||||
import json
|
||||
|
||||
pt = Point(145000, 172000, srid=3812)
|
||||
pole = SignPole.objects.create(
|
||||
code="POLE-DUP-01",
|
||||
street=self.adr_street,
|
||||
organization=self.org,
|
||||
model=self.pole_model,
|
||||
geom=pt,
|
||||
lon=4.35,
|
||||
lat=50.85,
|
||||
)
|
||||
attached_panel = SignPanel.objects.create(
|
||||
code="PANEL-ATT-01",
|
||||
street=self.adr_street,
|
||||
pole=pole,
|
||||
model=self.panel_model,
|
||||
signpanel_type=self.panel_type,
|
||||
organization=self.org,
|
||||
geom=pt,
|
||||
lon=4.35,
|
||||
lat=50.85,
|
||||
status="active",
|
||||
)
|
||||
# Duplicate orphan panel at exact same coordinates with same signpanel_type
|
||||
orphan_duplicate = SignPanel.objects.create(
|
||||
code="PANEL-ORPH-01",
|
||||
street=self.adr_street,
|
||||
pole=None,
|
||||
model=self.panel_model,
|
||||
signpanel_type=self.panel_type,
|
||||
organization=self.org,
|
||||
geom=pt,
|
||||
lon=4.35,
|
||||
lat=50.85,
|
||||
status="active",
|
||||
)
|
||||
|
||||
self.client.force_login(self.user)
|
||||
resp = self.client.post(
|
||||
reverse('sign:sign_assets_geojson'),
|
||||
data=json.dumps({'organization': [self.org.id]}),
|
||||
content_type='application/json'
|
||||
)
|
||||
self.assertEqual(resp.status_code, 200)
|
||||
features = resp.json().get('features', [])
|
||||
panel_features = [f for f in features if f['properties'].get('asset_type') == 'signpanel']
|
||||
panel_codes = [f['properties']['code'] for f in panel_features]
|
||||
|
||||
# The attached panel is kept, the redundant orphan is suppressed
|
||||
self.assertIn("PANEL-ATT-01", panel_codes)
|
||||
self.assertNotIn("PANEL-ORPH-01", panel_codes)
|
||||
|
||||
def test_check_duplicate_sign_panels_command(self):
|
||||
"""Verify that check_duplicate_sign_panels command detects and fixes duplicate orphan panels."""
|
||||
from django.contrib.gis.geos import Point
|
||||
from django.core.management import call_command
|
||||
from io import StringIO
|
||||
from sign.models import SignPole, SignPanel
|
||||
|
||||
pt = Point(145000, 172000, srid=3812)
|
||||
pole = SignPole.objects.create(
|
||||
code="POLE-CMD-01",
|
||||
street=self.adr_street,
|
||||
organization=self.org,
|
||||
model=self.pole_model,
|
||||
geom=pt,
|
||||
lon=4.36,
|
||||
lat=50.86,
|
||||
)
|
||||
SignPanel.objects.create(
|
||||
code="PANEL-CMD-ATT",
|
||||
street=self.adr_street,
|
||||
pole=pole,
|
||||
model=self.panel_model,
|
||||
signpanel_type=self.panel_type,
|
||||
organization=self.org,
|
||||
geom=pt,
|
||||
lon=4.36,
|
||||
lat=50.86,
|
||||
status="active",
|
||||
)
|
||||
orphan_panel = SignPanel.objects.create(
|
||||
code="PANEL-CMD-ORPH",
|
||||
street=self.adr_street,
|
||||
pole=None,
|
||||
model=self.panel_model,
|
||||
signpanel_type=self.panel_type,
|
||||
organization=self.org,
|
||||
geom=pt,
|
||||
lon=4.36,
|
||||
lat=50.86,
|
||||
status="active",
|
||||
)
|
||||
|
||||
# 1. Dry run
|
||||
out = StringIO()
|
||||
call_command("check_duplicate_sign_panels", org_id=self.org.id, stdout=out)
|
||||
output = out.getvalue()
|
||||
self.assertIn("Doublons orphelins redondants détectés : 1", output)
|
||||
orphan_panel.refresh_from_db()
|
||||
self.assertEqual(orphan_panel.status, "active")
|
||||
|
||||
# 2. Fix mode (archive)
|
||||
out_fix = StringIO()
|
||||
call_command("check_duplicate_sign_panels", org_id=self.org.id, fix=True, action="archive", stdout=out_fix)
|
||||
fix_output = out_fix.getvalue()
|
||||
self.assertIn("1 doublon(s) orphelin(s) archives", fix_output.replace("é", "e"))
|
||||
orphan_panel.refresh_from_db()
|
||||
self.assertEqual(orphan_panel.status, "archived")
|
||||
|
||||
|
||||
class SignPanelCategoryFilterTest(TestCase):
|
||||
"""Tests for panel type category grouping and filter controls."""
|
||||
|
|
@ -1270,6 +1441,3 @@ class SignPanelCategoryFilterTest(TestCase):
|
|||
self.assertIn('deselect-all-btn', content)
|
||||
# Verify data-group attribute on panel item
|
||||
self.assertIn('data-group="danger"', content)
|
||||
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -767,8 +767,14 @@ def sign_assets_geojson(request):
|
|||
})
|
||||
return feature
|
||||
|
||||
seen_feature_keys = set()
|
||||
|
||||
def add_assets_to_features(queryset, asset_type, features, feature_count):
|
||||
for obj in queryset:
|
||||
key = (asset_type, obj.id)
|
||||
if key in seen_feature_keys:
|
||||
continue
|
||||
seen_feature_keys.add(key)
|
||||
if feature_count >= MAX_ASSET_FEATURES:
|
||||
return feature_count, True
|
||||
features.append(obj_to_feature(obj, asset_type=asset_type))
|
||||
|
|
@ -781,6 +787,14 @@ def sign_assets_geojson(request):
|
|||
'trafficlight_pole', 'trafficlight_pole__model'
|
||||
).all()
|
||||
|
||||
if statuses:
|
||||
signpole_qs = signpole_qs.filter(status__in=statuses)
|
||||
signpanel_qs = signpanel_qs.filter(status__in=statuses)
|
||||
else:
|
||||
# By default, do not display removed or archived assets on the active map
|
||||
signpole_qs = signpole_qs.exclude(status__in=['removed', 'archived'])
|
||||
signpanel_qs = signpanel_qs.exclude(status__in=['removed', 'archived'])
|
||||
|
||||
if municipalities:
|
||||
signpole_qs = signpole_qs.filter(street__municipality_id__in=municipalities)
|
||||
signpanel_qs = signpanel_qs.filter(street__municipality_id__in=municipalities)
|
||||
|
|
@ -814,6 +828,9 @@ def sign_assets_geojson(request):
|
|||
signpole_qs = apply_search(signpole_qs, data, fields=['code'])
|
||||
signpanel_qs = apply_search(signpanel_qs, data, fields=['code'])
|
||||
|
||||
signpole_qs = signpole_qs.distinct()
|
||||
signpanel_qs = signpanel_qs.distinct()
|
||||
|
||||
signpole_qs = signpole_qs[0:MAX_ASSET_FEATURES+1]
|
||||
if len(signpole_qs) >= MAX_ASSET_FEATURES:
|
||||
return JsonResponse({"error": f"{_('Trop d\'objets à afficher')} (> {MAX_ASSET_FEATURES})"}, status=413)
|
||||
|
|
@ -822,7 +839,28 @@ def sign_assets_geojson(request):
|
|||
if len(signpanel_qs) >= MAX_ASSET_FEATURES:
|
||||
return JsonResponse({"error": f"{_('Trop d\'objets à afficher')} (> {MAX_ASSET_FEATURES})"}, status=413)
|
||||
|
||||
signpanel_list = list(signpanel_qs)
|
||||
raw_signpanel_list = list(signpanel_qs)
|
||||
|
||||
# Index attached panels by pole coordinates and signpanel_type to detect and suppress duplicate orphan panels
|
||||
attached_panel_keys = set()
|
||||
for p in raw_signpanel_list:
|
||||
if p.pole_id is not None and p.lon is not None and p.lat is not None and p.signpanel_type_id:
|
||||
coord_key = (round(float(p.lon), 5), round(float(p.lat), 5), p.signpanel_type_id)
|
||||
attached_panel_keys.add(coord_key)
|
||||
|
||||
# Filter out orphan panels that are exact duplicates of an attached panel on a pole at the same location
|
||||
signpanel_list = []
|
||||
seen_panel_ids = set()
|
||||
for p in raw_signpanel_list:
|
||||
if p.id in seen_panel_ids:
|
||||
continue
|
||||
if p.pole_id is None and p.lon is not None and p.lat is not None and p.signpanel_type_id:
|
||||
coord_key = (round(float(p.lon), 5), round(float(p.lat), 5), p.signpanel_type_id)
|
||||
if coord_key in attached_panel_keys:
|
||||
# Suppress orphan duplicate from map
|
||||
continue
|
||||
seen_panel_ids.add(p.id)
|
||||
signpanel_list.append(p)
|
||||
panel_render_meta = {}
|
||||
for panel in signpanel_list:
|
||||
panel_type_code = panel.signpanel_type.code.strip() if getattr(panel, "signpanel_type", None) and panel.signpanel_type.code else None
|
||||
|
|
|
|||
Loading…
Reference in a new issue