feat: handle common wires and intelligent lantern matching in cross plan parser

loko/assets/tests.py

@@ -2046,6 +2046,85 @@ class TrafficLightCrossPlanImportTest(TestCase):
             if os.path.exists(tmp_path):
                 os.remove(tmp_path)
 
+    def test_parse_cross_plan_hp_ls_common_wire_not_an_equipment(self):
+        import openpyxl
+        import tempfile
+        from assets.utils.cross_plan_parser import parse_cross_plan
+
+        # Workbook with pedestrian pole B01, vehicular-only pole A01,
+        # and a common wire HP/LS (42V) spanning across both poles.
+        wb = openpyxl.Workbook()
+        ws = wb.active
+        ws.title = "kruisjesplan"
+        ws.cell(row=1, column=1, value="Sleutel:")
+        ws.cell(row=1, column=2, value="SWB01")
+        ws.cell(row=5, column=1, value="Câble 01 SVAVB")
+        ws.cell(row=6, column=3, value="richting")
+        ws.cell(row=6, column=4, value="kringen")
+        ws.cell(row=7, column=5, value="B01")
+        ws.cell(row=7, column=6, value="A01")
+
+        # Row 8: Vehicle phase A on A01
+        ws.cell(row=8, column=2, value="1")
+        ws.cell(row=8, column=3, value="A")
+        ws.cell(row=8, column=4, value="G")
+        ws.cell(row=8, column=6, value="X")
+
+        # Row 9: Pedestrian phase a on B01
+        ws.cell(row=9, column=2, value="2")
+        ws.cell(row=9, column=3, value="a")
+        ws.cell(row=9, column=4, value="gr")
+        ws.cell(row=9, column=5, value="X")
+
+        # Row 10: Common wire HP/LS (42V) connected to both B01 and A01
+        ws.cell(row=10, column=2, value="55")
+        ws.cell(row=10, column=3, value="HP/LS")
+        ws.cell(row=10, column=4, value="42V")
+        ws.cell(row=10, column=5, value="X")
+        ws.cell(row=10, column=6, value="X")
+
+        # Row 11: Common wire RADAR (42V)
+        ws.cell(row=11, column=2, value="56")
+        ws.cell(row=11, column=3, value="RADAR")
+        ws.cell(row=11, column=4, value="42V")
+        ws.cell(row=11, column=5, value="X")
+
+        # Row 12: Common wire Com (42V)
+        ws.cell(row=12, column=2, value="58")
+        ws.cell(row=12, column=3, value="Com")
+        ws.cell(row=12, column=4, value="42V")
+        ws.cell(row=12, column=5, value="X")
+        ws.cell(row=12, column=6, value="X")
+
+        with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as tmp:
+            tmp_path = tmp.name
+        wb.save(tmp_path)
+
+        try:
+            deduced = parse_cross_plan(tmp_path)
+            lanterns = deduced["lanterns"]
+            detectors = deduced["detectors"]
+
+            # Exactly 2 lanterns: 1 vehicle for A on A01, 1 pedestrian for a on B01
+            self.assertEqual(len(lanterns), 2)
+            
+            lan_a = next(l for l in lanterns if l["phase"] == "A")
+            self.assertEqual(lan_a["code"], "SWB01_A01_LAN01_A")
+            self.assertEqual(lan_a["model_name"], "3V200")
+
+            lan_ped = next(l for l in lanterns if l["phase"] == "a")
+            self.assertEqual(lan_ped["code"], "SWB01_B01_LAN01_a")
+            # B01 has HP/LS common wire, so pedestrian lantern gets piéton+HP
+            self.assertEqual(lan_ped["model_name"], "2V200 piéton+HP")
+
+            # Crucial assertion: NO equipment (lantern or detector) created for HP/LS, RADAR, or Com
+            self.assertFalse(any("HP" in l["code"] or "LS" in l["code"] for l in lanterns))
+            self.assertEqual(len(detectors), 0)
+        finally:
+            import os
+            if os.path.exists(tmp_path):
+                os.remove(tmp_path)
+
     def test_parse_cross_plan_double_cross_lanterns(self):
         import openpyxl
         import tempfile
@@ -2492,6 +2571,150 @@ class TrafficLightCrossPlanImportTest(TestCase):
             if os.path.exists(tmp_path):
                 os.remove(tmp_path)
 
+    def test_match_pole_lanterns_intelligence_user_scenario(self):
+        """
+        Tests the intelligent matching of lanterns when multiple lanterns are on the same pole.
+        In the user scenario:
+        - DB has:
+          LAN01: 3V200 tram tout droite
+          LAN02: 3V300 fl gauche
+        - Plan deduces:
+          LAN01_B1: 3V300 (vehicular phase B1)
+          LAN02_T2: 3V300 tram tout droit (tram phase T2)
+        The algorithm must NOT naively match LAN01_B1 to DB LAN01 and LAN02_T2 to DB LAN02.
+        Instead, it must match:
+          LAN01_B1 -> DB LAN02 (vehicular to vehicular)
+          LAN02_T2 -> DB LAN01 (tram to tram)
+        """
+        from assets.utils.cross_plan_parser import match_pole_lanterns
+
+        pole_dls = [
+            {'code': 'SEK02_B02_LAN01_B1', 'model_name': '3V300', 'name': '3V300 sur B02 - Phase B1', 'phase': 'B1', 'pole_code': 'SEK02_B02'},
+            {'code': 'SEK02_B02_LAN02_T2', 'model_name': '3V300 tram tout droit', 'name': '3V300 tram sur B02 - Phase T2', 'phase': 'T2', 'pole_code': 'SEK02_B02'}
+        ]
+
+        class DummyModel:
+            def __init__(self, name_fr, name_nl=''):
+                self.name_fr = name_fr
+                self.name_nl = name_nl
+
+        class DummyLantern:
+            def __init__(self, code, name_fr, model_name):
+                self.code = code
+                self.name_fr = name_fr
+                self.model = DummyModel(model_name)
+
+        el_tram = DummyLantern('SEK02_B02_LAN01', '3V200 tram tout droite', '3V200 tram tout droite')
+        el_car = DummyLantern('SEK02_B02_LAN02', '3V300 fl gauche', '3V300 fl gauche')
+        pole_els = [el_tram, el_car]
+
+        matched_pairs, unmatched_dls, unmatched_els = match_pole_lanterns(pole_dls, pole_els)
+
+        self.assertEqual(len(matched_pairs), 2)
+        self.assertEqual(len(unmatched_dls), 0)
+        self.assertEqual(len(unmatched_els), 0)
+
+        match_map = {dl['code']: el.code for dl, el in matched_pairs}
+        self.assertEqual(match_map['SEK02_B02_LAN01_B1'], 'SEK02_B02_LAN02')
+        self.assertEqual(match_map['SEK02_B02_LAN02_T2'], 'SEK02_B02_LAN01')
+
+    def test_import_cross_plan_multiple_lanterns_preview(self):
+        """
+        End-to-end view test with multiple lanterns on the same pole.
+        Verifies that import_cross_plan view correctly associates the lanterns in compared_lanterns
+        and populates the preview with the correct db_id and diffs.
+        """
+        import openpyxl
+        import tempfile
+        from django.core.files.uploadedfile import SimpleUploadedFile
+        from assets.models import TrafficLightPole, TrafficLightLantern, TrafficLightPoleModel, TrafficLightLanternModel, AssetCategory
+
+        pole_model = TrafficLightPoleModel.objects.first()
+        m_tram, _ = TrafficLightLanternModel.objects.get_or_create(code="3v200_tram", defaults={"name_fr": "3V200 tram tout droite", "voltage": 230})
+        m_turn, _ = TrafficLightLanternModel.objects.get_or_create(code="3v300_fl", defaults={"name_fr": "3V300 fl gauche", "voltage": 230})
+        TrafficLightLanternModel.objects.get_or_create(code="3v300_tram", defaults={"name_fr": "3V300 tram tout droit", "voltage": 230})
+        TrafficLightLanternModel.objects.get_or_create(code="3v300", defaults={"name_fr": "3V300", "voltage": 230})
+
+        category_pole = AssetCategory.objects.filter(code="TL_POLE").first()
+        category_lan = AssetCategory.objects.filter(code="TL_LANTERN").first() or AssetCategory.objects.filter(code="TL_LANTERNE").first()
+
+        pole_b02 = TrafficLightPole.objects.create(
+            intersection=self.intersection,
+            code="SWB01_B02",
+            model=pole_model,
+            category=category_pole,
+            status="active"
+        )
+        db_lan_tram = TrafficLightLantern.objects.create(
+            pole=pole_b02,
+            code="SWB01_B02_LAN01",
+            model=m_tram,
+            name_fr="3V200 tram tout droite",
+            category=category_lan,
+            status="active"
+        )
+        db_lan_car = TrafficLightLantern.objects.create(
+            pole=pole_b02,
+            code="SWB01_B02_LAN02",
+            model=m_turn,
+            name_fr="3V300 fl gauche",
+            category=category_lan,
+            status="active"
+        )
+
+        # Create Excel cross plan with pole B02 having phase B1 (vehicle) and T2 (tram)
+        wb = openpyxl.Workbook()
+        ws = wb.active
+        ws.title = "kruisjesplan"
+        ws.cell(row=1, column=1, value="Sleutel:")
+        ws.cell(row=1, column=2, value="SWB01")
+        ws.cell(row=5, column=1, value="Câble 01 SVAVB")
+        ws.cell(row=6, column=3, value="richting")
+        ws.cell(row=6, column=4, value="kringen")
+        ws.cell(row=7, column=5, value="B02")
+
+        # Row 8: Phase B1 (Vehicle)
+        ws.cell(row=8, column=2, value="1")
+        ws.cell(row=8, column=3, value="B1")
+        ws.cell(row=8, column=4, value="Y")
+        ws.cell(row=8, column=5, value="X")
+
+        # Row 9: Phase T2 (Tram)
+        ws.cell(row=9, column=2, value="2")
+        ws.cell(row=9, column=3, value="T2")
+        ws.cell(row=9, column=4, value="Y")
+        ws.cell(row=9, column=5, value="X")
+
+        with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as tmp:
+            tmp_path = tmp.name
+        wb.save(tmp_path)
+
+        try:
+            self.client.force_login(self.user)
+            with open(tmp_path, "rb") as f:
+                uploaded_file = SimpleUploadedFile("test_multi_lantern.xlsx", f.read(), content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")
+            url_import = reverse("assets:import_cross_plan", args=[self.intersection.id])
+            response = self.client.post(url_import, {"cross_plan_file_upload": uploaded_file})
+            self.assertEqual(response.status_code, 200)
+
+            lanterns = response.context["lanterns"]
+            lan_b1 = next(l for l in lanterns if "LAN01_B1" in l["code"])
+            lan_t2 = next(l for l in lanterns if "LAN02_T2" in l["code"])
+
+            # Verify that LAN01_B1 is paired with DB LAN02 (db_lan_car.id)
+            self.assertEqual(lan_b1["db_id"], db_lan_car.id)
+            self.assertEqual(lan_b1["status"], "modified")
+            self.assertTrue(any(f"Code : {db_lan_car.code} → {lan_b1['code']}" in d for d in lan_b1["details_diff"]))
+
+            # Verify that LAN02_T2 is paired with DB LAN01 (db_lan_tram.id)
+            self.assertEqual(lan_t2["db_id"], db_lan_tram.id)
+            self.assertEqual(lan_t2["status"], "modified")
+            self.assertTrue(any(f"Code : {db_lan_tram.code} → {lan_t2['code']}" in d for d in lan_t2["details_diff"]))
+        finally:
+            import os
+            if os.path.exists(tmp_path):
+                os.remove(tmp_path)
+
 
 class AssetDefaultPositionTest(TestCase):
     def test_assign_default_position_hierarchy(self):

loko/assets/utils/cross_plan_parser.py

@@ -1,8 +1,31 @@
 import openpyxl
 import re
 import os
+import itertools
+from collections import defaultdict
 from django.conf import settings
 
+def is_hp_or_ls(text):
+    """
+    Checks if the wire/direction/circuit text refers to loudspeakers (HP/LS).
+    In cross plans, HP (haut-parleur) / LS (luidspreker) is often a common wire (fil commun)
+    and must NOT be treated as an equipment (neither lantern nor detector).
+    """
+    if not text:
+        return False
+    t_str = str(text).strip()
+    t_lower = t_str.lower()
+    t_clean = re.sub(r'[\s/\\_\-\.:;]+', '', t_lower)
+    if t_clean in ('hp', 'ls', 'hpls', 'lshp', 'hpmute', 'lsmute', 'hautparleur', 'luidspreker'):
+        return True
+    if re.search(r'\b(?:hp\s*[/\\-]\s*ls|ls\s*[/\\-]\s*hp)\b', t_lower):
+        return True
+    if re.search(r'\b(?:hp|ls)\b', t_lower):
+        return True
+    if re.search(r'haut[\s-]*parleur|luidspreker', t_lower):
+        return True
+    return False
+
 def parse_cross_plan(excel_path):
     """
     Parses a crossroads cross plan (kruisjesplan) Excel file and returns
@@ -147,14 +170,21 @@ def parse_cross_plan(excel_path):
                 last_richting = None
                 continue
 
+            col2_val = sheet.cell(row=r, column=2).value if (richting_col and richting_col > 2) else None
+
             # If the row is empty or contains section markers, skip it
-            if not val_kringen and not val_richting:
+            if not val_kringen and not val_richting and not col2_val:
                 # check if the entire row is empty
                 if all(sheet.cell(row=r, column=c).value is None for c in range(1, max_col + 1)):
                     continue
 
-            if val_richting:
+            # Check if HP/LS is in richting, kringen, or col 2
+            if is_hp_or_ls(val_richting) or is_hp_or_ls(col2_val) or is_hp_or_ls(val_kringen):
+                last_richting = "HP/LS"
+            elif val_richting:
                 last_richting = str(val_richting).strip()
+            elif col2_val and isinstance(col2_val, str) and not str(col2_val).strip().isdigit():
+                last_richting = str(col2_val).strip()
             
             if not last_richting:
                 continue
@@ -229,8 +259,9 @@ def parse_cross_plan(excel_path):
         richting = conn["richting"]
         kringen = conn["kringen"]
         
-        # Check if HP is present on this pole
+        # Check if HP / LS (haut-parleur / luidspreker / fil commun) is present on this pole
-        if richting.lower() == "hp":
+        # It must NOT be treated as an equipment (neither lantern nor detector).
+        if is_hp_or_ls(richting) or is_hp_or_ls(kringen):
             pole_has_hp[p] = True
             continue
 
@@ -332,9 +363,9 @@ def parse_cross_plan(excel_path):
 
         # Standard traffic signals (vehicles, trams, cycles, pedestrians)
         # Richting codes are typically: A, B, C, T1, T2, F1, F2, a, b, c...
-        # Ignore wiring/cabling categories like ALIM, Radar, Com, contact, etc.
+        # Ignore wiring/cabling categories like ALIM, Radar, Com, contact, HP/LS, etc.
         normalized = re.sub(r'\s+', ' ', richting).strip()
-        if re.search(r'(?:alim|com|contact|radar|tension|comm)', normalized, re.IGNORECASE):
+        if is_hp_or_ls(richting) or is_hp_or_ls(kringen) or re.search(r'(?:alim|com|contact|radar|tension|comm|\bhp\b|\bls\b|hp/ls|ls/hp)', normalized, re.IGNORECASE):
             continue
             
         if p not in pole_phases:
@@ -476,3 +507,289 @@ def parse_cross_plan(excel_path):
         "lanterns": deduced_lanterns,
         "detectors": deduced_detectors
     }
+
+
+def classify_lantern(model_name="", phase="", code="", name=""):
+    """
+    Classifies a traffic light lantern based on its model name, phase, code, and description.
+    Returns:
+      category: 'tram', 'bus', 'pedestrian', 'cyclist', 'vehicle'
+      aspect_count: int (e.g. 1, 2, 3) or None
+      lens_size: int (e.g. 200, 300) or None
+      directions: set of strings in {'straight', 'left', 'right'}
+    """
+    model_lower = (model_name or "").lower().strip()
+    name_lower = (name or "").lower().strip()
+    code_upper = (code or "").upper().strip()
+    phase_clean = (phase or "").strip()
+
+    combined_text = f"{model_lower} {name_lower}"
+
+    # 1. Category
+    # Check Tram first:
+    is_tram = (
+        "tram" in combined_text
+        or phase_clean.upper().startswith("T")
+        or bool(re.search(r'\bT\d+\b', code_upper))
+    )
+
+    # Check Bus:
+    is_bus = (
+        not is_tram and (
+            "bus" in combined_text
+            or phase_clean.upper().startswith("BUS")
+            or "transport en commun" in combined_text
+        )
+    )
+
+    # Check Cyclist:
+    is_cyclist = (
+        not is_tram and not is_bus and (
+            any(w in combined_text for w in ["cycliste", "cycl", "velo", "vélo", "fiets"])
+            or phase_clean.upper().startswith("F")
+            or bool(re.search(r'\bF\d+\b', code_upper))
+        )
+    )
+
+    # Check Pedestrian:
+    is_pedestrian = (
+        not is_tram and not is_bus and not is_cyclist and (
+            any(w in combined_text for w in ["piéton", "pieton", "voetganger", "pedestrian"])
+            or (bool(phase_clean) and phase_clean[0].islower() and not phase_clean.startswith("fl"))
+            or phase_clean.upper().startswith("P")
+            or bool(re.search(r'\bP\d+\b', code_upper))
+        )
+    )
+
+    if is_tram:
+        category = "tram"
+    elif is_bus:
+        category = "bus"
+    elif is_cyclist:
+        category = "cyclist"
+    elif is_pedestrian:
+        category = "pedestrian"
+    else:
+        category = "vehicle"
+
+    # 2. Aspect count (e.g. 3V, 2V, 1V)
+    aspect_count = None
+    m_aspect = re.search(r'\b(\d+)V', f"{model_name} {name}", re.IGNORECASE)
+    if m_aspect:
+        aspect_count = int(m_aspect.group(1))
+
+    # 3. Lens size (e.g. 300, 200)
+    lens_size = None
+    m_size = re.search(r'\d+V(\d+)', f"{model_name} {name}", re.IGNORECASE)
+    if m_size:
+        lens_size = int(m_size.group(1))
+    else:
+        m_size_word = re.search(r'\b(200|300)\b', f"{model_name} {name}")
+        if m_size_word:
+            lens_size = int(m_size_word.group(1))
+
+    # 4. Directions: straight, left, right
+    directions = set()
+    if re.search(r'\b(tout\s+droit[es]?|rechtdoor|straight)\b', combined_text):
+        directions.add("straight")
+    if re.search(r'\b(fl\s*gauche|flèche\s*gauche|gauche|links|left)\b', combined_text):
+        directions.add("left")
+    if re.search(r'\b(fl\s*droite|flèche\s*droite|droite|rechts|right)\b', combined_text):
+        if not re.search(r'\btout\s+droite\b', combined_text):
+            directions.add("right")
+
+    return category, aspect_count, lens_size, directions
+
+
+def extract_phase_from_lantern(code="", name=""):
+    """
+    Extracts phase identifier (e.g. 'B1', 'T2', 'b', 'F1') from lantern code or name.
+    """
+    code_str = code or ""
+    name_str = name or ""
+
+    if "_" in code_str:
+        parts = code_str.rsplit("_", 1)
+        if len(parts) == 2 and not parts[1].upper().startswith("LAN"):
+            return parts[1]
+
+    m = re.search(r'Phase\s+([A-Za-z0-9]+)', name_str, re.IGNORECASE)
+    if m:
+        return m.group(1)
+
+    return ""
+
+
+def calculate_lantern_match_score(dl, el, is_single_on_pole=False):
+    """
+    Computes a match compatibility score between a deduced plan lantern (dl)
+    and an existing DB lantern (el).
+    Higher score indicates a more likely match. Negative score indicates a conflict.
+    """
+    dl_code = dl.get("code", "")
+    el_code = getattr(el, "code", "") or (el.get("code", "") if isinstance(el, dict) else "")
+
+    # 1. Exact full code match
+    if dl_code and el_code and dl_code == el_code:
+        return 10000
+
+    dl_base_code = dl_code.rsplit("_", 1)[0] if "_" in dl_code else dl_code
+    is_prefix_code_match = bool(dl_base_code and el_code and dl_base_code == el_code)
+
+    if is_single_on_pole and is_prefix_code_match:
+        return 1000
+
+    dl_model_name = dl.get("model_name", "")
+    dl_phase = dl.get("phase", "") or extract_phase_from_lantern(dl_code, dl.get("name", ""))
+    dl_name = dl.get("name", "")
+
+    el_name = getattr(el, "name_fr", "") or getattr(el, "name", "") or (el.get("name_fr", "") if isinstance(el, dict) else "")
+    el_model = getattr(el, "model", None)
+    if el_model:
+        el_model_name = getattr(el_model, "name_fr", "") or ""
+        el_model_name_nl = getattr(el_model, "name_nl", "") or ""
+    elif isinstance(el, dict):
+        el_model_name = el.get("model_name", "")
+        el_model_name_nl = el.get("model_name_nl", "")
+    else:
+        el_model_name = ""
+        el_model_name_nl = ""
+
+    el_phase = extract_phase_from_lantern(el_code, el_name)
+
+    dl_cat, dl_aspect, dl_size, dl_dirs = classify_lantern(dl_model_name, dl_phase, dl_code, dl_name)
+    el_cat, el_aspect, el_size, el_dirs = classify_lantern(el_model_name, el_phase, el_code, el_name)
+
+    score = 0
+
+    # 2. Category matching
+    if dl_cat == el_cat:
+        if dl_cat in ("tram", "bus", "pedestrian", "cyclist"):
+            score += 300
+        else:
+            score += 200
+    else:
+        # Category mismatch penalty: e.g. tram vs car, pedestrian vs car
+        score -= 600
+
+    # 3. Phase matching
+    if dl_phase and el_phase:
+        if dl_phase.upper() == el_phase.upper():
+            score += 500
+        else:
+            score -= 300
+
+    # 4. Model exact match
+    if dl_model_name and el_model_name:
+        if dl_model_name.lower().strip() == el_model_name.lower().strip():
+            score += 150
+        elif el_model_name_nl and dl_model_name.lower().strip() == el_model_name_nl.lower().strip():
+            score += 150
+
+    # 5. Aspect count matching (3V vs 3V, 2V vs 2V)
+    if dl_aspect and el_aspect:
+        if dl_aspect == el_aspect:
+            score += 50
+        else:
+            score -= 100
+
+    # 6. Lens size matching (300 vs 300, 200 vs 200)
+    if dl_size and el_size:
+        if dl_size == el_size:
+            score += 30
+
+    # 7. Directional matching (straight, left, right)
+    if dl_dirs and el_dirs:
+        if dl_dirs == el_dirs:
+            score += 80
+        elif dl_dirs.intersection(el_dirs):
+            score += 40
+        else:
+            score -= 60
+
+    # 8. Prefix match bonus
+    if is_prefix_code_match:
+        score += 30
+
+    # 9. LAN index tie-breaker (e.g. LAN01 vs LAN01)
+    m_dl = re.search(r'LAN(\d+)', dl_code, re.IGNORECASE)
+    m_el = re.search(r'LAN(\d+)', el_code, re.IGNORECASE)
+    if m_dl and m_el and m_dl.group(1) == m_el.group(1):
+        score += 15
+
+    # 10. Single lantern on pole baseline bonus
+    if is_single_on_pole:
+        score += 200
+
+    return score
+
+
+def match_pole_lanterns(pole_dls, pole_els):
+    """
+    Given a list of deduced lanterns (pole_dls) and DB lanterns (pole_els)
+    on the same pole, finds the optimal 1-to-1 matching based on scoring.
+    Returns:
+      matched_pairs: list of (dl, el) in the original order of pole_dls
+      unmatched_dls: list of dl (new lanterns)
+      unmatched_els: list of el (db_only lanterns)
+    """
+    if not pole_dls:
+        return [], [], list(pole_els)
+    if not pole_els:
+        return [], list(pole_dls), []
+
+    is_single_on_pole = (len(pole_dls) == 1 and len(pole_els) == 1)
+
+    scores = {}
+    for i, dl in enumerate(pole_dls):
+        for j, el in enumerate(pole_els):
+            scores[(i, j)] = calculate_lantern_match_score(dl, el, is_single_on_pole=is_single_on_pole)
+
+    m = len(pole_dls)
+    n = len(pole_els)
+
+    best_assignment = []
+    best_total_score = -float('inf')
+
+    if min(m, n) <= 8:
+        if m <= n:
+            for p in itertools.permutations(range(n), m):
+                total = sum(scores[(i, p[i])] for i in range(m))
+                if total > best_total_score:
+                    best_total_score = total
+                    best_assignment = [(i, p[i]) for i in range(m)]
+        else:
+            for p in itertools.permutations(range(m), n):
+                total = sum(scores[(p[j], j)] for j in range(n))
+                if total > best_total_score:
+                    best_total_score = total
+                    best_assignment = [(p[j], j) for j in range(n)]
+    else:
+        available_dls = set(range(m))
+        available_els = set(range(n))
+        sorted_pairs = sorted(scores.items(), key=lambda item: item[1], reverse=True)
+        best_assignment = []
+        for (i, j), score in sorted_pairs:
+            if i in available_dls and j in available_els:
+                best_assignment.append((i, j))
+                available_dls.remove(i)
+                available_els.remove(j)
+
+    matched_pairs = []
+    matched_dl_indices = set()
+    matched_el_indices = set()
+
+    for i, j in best_assignment:
+        if scores[(i, j)] >= 0:
+            matched_pairs.append((pole_dls[i], pole_els[j]))
+            matched_dl_indices.add(i)
+            matched_el_indices.add(j)
+
+    # Sort matched pairs to preserve the original order of pole_dls
+    matched_pairs.sort(key=lambda pair: pole_dls.index(pair[0]))
+
+    unmatched_dls = [dl for i, dl in enumerate(pole_dls) if i not in matched_dl_indices]
+    unmatched_els = [el for j, el in enumerate(pole_els) if j not in matched_el_indices]
+
+    return matched_pairs, unmatched_dls, unmatched_els
+

loko/assets/views/trafficlights.py

@@ -3752,7 +3752,7 @@ def import_cross_plan(request, intersection_id):
     from django.urls import reverse
     from django.conf import settings
     import json
-    from assets.utils.cross_plan_parser import parse_cross_plan
+    from assets.utils.cross_plan_parser import parse_cross_plan, match_pole_lanterns
     from assets.models.trafficlights import (
         TrafficLightIntersection, TrafficLightPole, TrafficLightLantern,
         TrafficLightDetector, TrafficLightCable, TrafficLightPoleModel,
@@ -4338,36 +4338,62 @@ def import_cross_plan(request, intersection_id):
             })
 
         # ── 3. LANTERNS COMPARISON ──
-        existing_lanterns = {l.code: l for l in TrafficLightLantern.objects.filter(pole__intersection=intersection, status='active')}
+        all_db_lanterns = list(TrafficLightLantern.objects.filter(pole__intersection=intersection, status='active'))
-        compared_lanterns = []
+
+        from collections import defaultdict
+        deduced_by_pole = defaultdict(list)
+        pole_order = []
         for dl in deduced["lanterns"]:
-            code = dl["code"]
+            p_code = dl["pole_code"]
-            base_code = code.rsplit('_', 1)[0] if '_' in code else code
+            if p_code not in deduced_by_pole:
-            model_name = dl["model_name"]
+                pole_order.append(p_code)
-            db_model = TrafficLightLanternModel.objects.filter(name_fr__iexact=model_name).first() or TrafficLightLanternModel.objects.first()
+            deduced_by_pole[p_code].append(dl)
 
-            # Try exact match, then try prefix match (without the phase suffix)
+        db_by_pole = defaultdict(list)
-            match_key = None
+        for el in all_db_lanterns:
-            if code in existing_lanterns:
+            p_code = el.pole.code if el.pole else ""
-                match_key = code
+            db_by_pole[p_code].append(el)
-            elif base_code in existing_lanterns:
+
-                match_key = base_code
+        def find_db_pole_key(target_pole_code, db_dict):
+            if target_pole_code in db_dict:
+                return target_pole_code
+            target_short = target_pole_code.split('_')[-1]
+            for k in db_dict:
+                if k.split('_')[-1] == target_short:
+                    return k
+            return None
+
+        compared_lanterns = []
+        for p_code in pole_order:
+            pole_dls = deduced_by_pole[p_code]
+            db_key = find_db_pole_key(p_code, db_by_pole)
+            pole_els = db_by_pole.pop(db_key, []) if db_key else []
+
+            matched_pairs, unmatched_dls, unmatched_els = match_pole_lanterns(pole_dls, pole_els)
+
+            for dl, el in matched_pairs:
+                code = dl["code"]
+                model_name = dl["model_name"]
+                db_model = TrafficLightLanternModel.objects.filter(name_fr__iexact=model_name).first() or TrafficLightLanternModel.objects.first()
 
-            if match_key:
-                el = existing_lanterns[match_key]
                 is_rename = (el.code != code)
-                is_model_diff = el.model_id != db_model.id
+                is_model_diff = el.model_id != (db_model.id if db_model else None)
                 status = 'modified' if (is_rename or is_model_diff) else 'identical'
-                
+
                 details = []
                 if is_rename:
                     details.append(f"Code : {el.code} → {code}")
                 if is_model_diff:
                     details.append(f"Modèle : {el.model.name_fr if el.model else 'Aucun'} → {model_name}")
 
+                if is_model_diff and el.name_fr and el.model and el.name_fr.strip().lower() == el.model.name_fr.strip().lower():
+                    display_name = dl['name']
+                else:
+                    display_name = el.name_fr or dl['name']
+
                 compared_lanterns.append({
                     'code': code,
-                    'name': el.name_fr or dl['name'],
+                    'name': display_name,
                     'status': status,
                     'deduced_model_name': model_name,
                     'db_model_name': el.model.name_fr if el.model else '',
@@ -4375,8 +4401,10 @@ def import_cross_plan(request, intersection_id):
                     'pole_code': dl['pole_code'],
                     'details_diff': details
                 })
-                existing_lanterns.pop(match_key)
+
-            else:
+            for dl in unmatched_dls:
+                code = dl["code"]
+                model_name = dl["model_name"]
                 compared_lanterns.append({
                     'code': code,
                     'name': dl['name'],
@@ -4387,17 +4415,32 @@ def import_cross_plan(request, intersection_id):
                     'pole_code': dl['pole_code'],
                     'details_diff': []
                 })
-        for code, el in existing_lanterns.items():
+
-            compared_lanterns.append({
+            for el in unmatched_els:
-                'code': code,
+                compared_lanterns.append({
-                'name': el.name_fr or code,
+                    'code': el.code,
-                'status': 'db_only',
+                    'name': el.name_fr or el.code,
-                'deduced_model_name': '',
+                    'status': 'db_only',
-                'db_model_name': el.model.name_fr if el.model else '',
+                    'deduced_model_name': '',
-                'db_id': el.id,
+                    'db_model_name': el.model.name_fr if el.model else '',
-                'pole_code': el.pole.code,
+                    'db_id': el.id,
-                'details_diff': [_("Absent du fichier Excel (présent en DB)")]
+                    'pole_code': el.pole.code if el.pole else p_code,
-            })
+                    'details_diff': [_("Absent du fichier Excel (présent en DB)")]
+                })
+
+        # Remaining DB lanterns on poles not present in the plan at all
+        for p_key, remaining_els in db_by_pole.items():
+            for el in remaining_els:
+                compared_lanterns.append({
+                    'code': el.code,
+                    'name': el.name_fr or el.code,
+                    'status': 'db_only',
+                    'deduced_model_name': '',
+                    'db_model_name': el.model.name_fr if el.model else '',
+                    'db_id': el.id,
+                    'pole_code': el.pole.code if el.pole else p_key,
+                    'details_diff': [_("Absent du fichier Excel (présent en DB)")]
+                })
 
         # ── 4. DETECTORS COMPARISON ──
         existing_detectors = {d.code: d for d in TrafficLightDetector.objects.filter(intersection=intersection, status='active')}
@@ -5073,11 +5116,12 @@ def integrate_cross_plan(request, intersection_id):
             for dl in deduced["lanterns"]:
                 code = dl["code"]
                 pole = active_poles.get(dl["pole_code"])
-                existing = TrafficLightLantern.objects.filter(pole__intersection=intersection, code=code, status='active').first()
+                db_id = request.POST.get(f"db_id_lantern_{code}")
+                existing = None
+                if db_id:
+                    existing = TrafficLightLantern.objects.filter(pk=db_id, status='active').first()
                 if not existing:
-                    db_id = request.POST.get(f"db_id_lantern_{code}")
+                    existing = TrafficLightLantern.objects.filter(pole__intersection=intersection, code=code, status='active').first()
-                    if db_id:
-                        existing = TrafficLightLantern.objects.filter(pk=db_id, status='active').first()
 
                 selected_model_id = request.POST.get(f"model_lantern_{code}")
                 db_model = None