From 28d6664fbdb2d1b6d26501fa4a1ea07ca83a538e Mon Sep 17 00:00:00 2001 From: kdeterme Date: Mon, 21 Sep 2026 10:56:49 +0200 Subject: [PATCH] feat: handle common wires and intelligent lantern matching in cross plan parser --- loko/assets/tests.py | 223 +++++++++++++++++ loko/assets/utils/cross_plan_parser.py | 329 ++++++++++++++++++++++++- loko/assets/views/trafficlights.py | 114 ++++++--- 3 files changed, 625 insertions(+), 41 deletions(-) diff --git a/loko/assets/tests.py b/loko/assets/tests.py index 2f50dd8..0177e82 100644 --- a/loko/assets/tests.py +++ b/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): diff --git a/loko/assets/utils/cross_plan_parser.py b/loko/assets/utils/cross_plan_parser.py index 85b15a6..40ed41a 100644 --- a/loko/assets/utils/cross_plan_parser.py +++ b/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 - if richting.lower() == "hp": + # Check if HP / LS (haut-parleur / luidspreker / fil commun) is present on this pole + # 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 + diff --git a/loko/assets/views/trafficlights.py b/loko/assets/views/trafficlights.py index 276540e..246a567 100644 --- a/loko/assets/views/trafficlights.py +++ b/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')} - compared_lanterns = [] + all_db_lanterns = list(TrafficLightLantern.objects.filter(pole__intersection=intersection, status='active')) + + from collections import defaultdict + deduced_by_pole = defaultdict(list) + pole_order = [] for dl in deduced["lanterns"]: - code = dl["code"] - base_code = code.rsplit('_', 1)[0] if '_' in code else code - model_name = dl["model_name"] - db_model = TrafficLightLanternModel.objects.filter(name_fr__iexact=model_name).first() or TrafficLightLanternModel.objects.first() + p_code = dl["pole_code"] + if p_code not in deduced_by_pole: + pole_order.append(p_code) + deduced_by_pole[p_code].append(dl) - # Try exact match, then try prefix match (without the phase suffix) - match_key = None - if code in existing_lanterns: - match_key = code - elif base_code in existing_lanterns: - match_key = base_code + db_by_pole = defaultdict(list) + for el in all_db_lanterns: + p_code = el.pole.code if el.pole else "" + db_by_pole[p_code].append(el) + + 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({ - 'code': code, - 'name': el.name_fr or 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, - 'details_diff': [_("Absent du fichier Excel (présent en DB)")] - }) + + for el in unmatched_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_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}") - if db_id: - existing = TrafficLightLantern.objects.filter(pk=db_id, status='active').first() + existing = TrafficLightLantern.objects.filter(pole__intersection=intersection, code=code, status='active').first() selected_model_id = request.POST.get(f"model_lantern_{code}") db_model = None