import openpyxl import re import os from django.conf import settings def parse_cross_plan(excel_path): """ Parses a crossroads cross plan (kruisjesplan) Excel file and returns a list of deduced assets structured for comparison/import. """ if not os.path.exists(excel_path): raise FileNotFoundError(f"Excel file not found at {excel_path}") wb = openpyxl.load_workbook(excel_path, data_only=True) target_sheet = None is_french = False for name in wb.sheetnames: normalized_name = re.sub(r'\s+', ' ', name).strip().lower() if normalized_name in ("kruisjesplan", "plan croix", "plan de croix"): target_sheet = name if normalized_name in ("plan croix", "plan de croix"): is_french = True break if not target_sheet: raise ValueError("La feuille 'kruisjesplan' ou 'Plan croix' est introuvable dans le fichier Excel.") sheet = wb[target_sheet] # 1. Deduce intersection code from the first row (e.g., sleutel: SWB01) intersection_code = None for r in range(1, 4): for c in range(1, 10): val = sheet.cell(row=r, column=c).value if val and any(k in str(val).lower() for k in ("sleutel", "clé", "cle", "key")): if any(k in str(val).lower() for k in ("clé", "cle")): is_french = True # Value is in the next cell or adjacent cells for offset in range(1, 4): next_val = sheet.cell(row=r, column=c + offset).value if next_val: intersection_code = str(next_val).strip() break if intersection_code: break if intersection_code: break if not intersection_code: # Fallback search in sheet dimensions for row in sheet.iter_rows(max_row=5, max_col=10, values_only=True): for cell in row: if cell and isinstance(cell, str): cleaned = cell.strip() if re.match(r'^[A-Z]{2,4}_?[A-Z]{0,3}\d{2,5}$|^[A-Z]_[A-Z]{1,3}\d{2,5}$', cleaned): intersection_code = cleaned break if intersection_code: break if not intersection_code: raise ValueError("Impossible de déterminer le code du carrefour (sleutel) dans le fichier Excel.") # 2. Scan sheet to identify cable sections # Cable sections start at rows containing 'Câble' or 'Cable' in the first columns cable_sections = [] max_row = sheet.max_row max_col = sheet.max_column for r in range(1, max_row + 1): val = sheet.cell(row=r, column=1).value if val and isinstance(val, str) and ("câble" in val.lower() or "cable" in val.lower()): cable_sections.append((r, val.strip())) # Parse each cable section deduced_poles = {} # pole_header -> {code, type} deduced_connections = [] # list of dicts for idx, (start_row, cable_label) in enumerate(cable_sections): end_row = cable_sections[idx+1][0] - 1 if idx + 1 < len(cable_sections) else max_row # We need to find the pole headers and the connection rows in this section. # The pole header row is a row below start_row that has non-empty values in columns 5+ # and has column 3 (Richting) / column 4 (Kringen) as labels. pole_row_num = None richting_col = 3 kringen_col = 4 for r in range(start_row + 1, end_row + 1): val_richting = sheet.cell(row=r, column=richting_col).value val_kringen = sheet.cell(row=r, column=kringen_col).value # If we find the row with "Richting" and "Kringen" if val_richting and isinstance(val_richting, str) and "richting" in val_richting.lower(): # The next row or the row after typically contains the pole list. # In SWB01, Row 4 is header, Row 5 is poles. pole_row_num = r + 1 break if not pole_row_num: # Fallback search: just search for any row in columns 5+ having pole codes for r in range(start_row + 1, start_row + 10): if r > end_row: break row_vals = [sheet.cell(row=r, column=c).value for c in range(5, min(max_col + 1, 30))] if any(v and isinstance(v, str) and re.match(r'^[AB]\d{2}$', v.strip()) for v in row_vals): pole_row_num = r break if not pole_row_num: continue # Read poles for this section section_poles = {} # col_index -> pole_name for col in range(5, max_col + 1): val_pole = sheet.cell(row=pole_row_num, column=col).value if val_pole: pole_name = str(val_pole).strip() if re.match(r'^[AB]\d{2}$', pole_name): section_poles[col] = pole_name # Save to global deduced poles deduced_poles[pole_name] = { "code": f"{intersection_code}_{pole_name}", "short_code": pole_name, "name": pole_name, "model": "Poteau gris avec bras horizontal de 7.5m" if pole_name.startswith("B") else "Poteau droit, gris 3.1m" } # Parse connections in this section last_richting = None is_detector_section = False if pole_row_num and pole_row_num > 1: header_val = sheet.cell(row=pole_row_num - 1, column=richting_col).value if header_val and isinstance(header_val, str) and any(k in header_val.lower() for k in ("detecteur", "détecteur")): is_detector_section = True for r in range(pole_row_num + 1, end_row + 1): val_richting = sheet.cell(row=r, column=richting_col).value val_kringen = sheet.cell(row=r, column=kringen_col).value # Check if this row is a "Detecteur" section header if val_richting and isinstance(val_richting, str) and any(k in val_richting.lower() for k in ("detecteur", "détecteur")): is_detector_section = True last_richting = None continue # If the row is empty or contains section markers, skip it if not val_kringen and not val_richting: # 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: last_richting = str(val_richting).strip() if not last_richting: continue kringen_val = str(val_kringen).strip() if val_kringen else "" # Check each pole column for connections (X or XX marker) for col, pole_name in section_poles.items(): marker = sheet.cell(row=r, column=col).value if marker: marker_str = str(marker).strip().upper() x_count = marker_str.count('X') if x_count > 0: deduced_connections.append({ "cable_label": cable_label, "pole": pole_name, "richting": last_richting, "kringen": kringen_val, "crosses": x_count, "is_detector": is_detector_section }) # 3. Post-process to deduce assets # A. CABLES deduced_cables = [] for idx, (start_row, cable_label) in enumerate(cable_sections): # e.g., "Câble 01 SVAVB 61x1,5mm²" m = re.search(r'(?:câble|cable)\s*(\d+)\s*(.*)', cable_label, re.IGNORECASE) cable_num = int(m.group(1)) if m else (idx + 1) cable_model_name = m.group(2).strip() if m else "SVAVB 61x1,5mm²" # Count number of strands strands = 61 if "10x2x0,8" in cable_model_name: strands = 20 elif "61x1,5" in cable_model_name: strands = 61 elif "14x1,5" in cable_model_name: strands = 14 # Find which poles are associated with this cable_label connected_poles = sorted(list(set([ f"{intersection_code}_{conn['pole']}" for conn in deduced_connections if conn["cable_label"] == cable_label ]))) deduced_cables.append({ "code": f"{intersection_code}_CAB{cable_num:02d}", "name": f"Câble {cable_num:02d}", "model_name": cable_model_name, "strands": strands, "connected_poles": connected_poles }) # B. POLES # Already populated in deduced_poles poles_list = sorted(deduced_poles.values(), key=lambda p: p["short_code"]) # C. LANTERNS & DETECTORS # Group connections by pole to identify lanes, pedestrian signals, etc. pole_phases = {} # pole_name -> set of phases (e.g. {'T1', 'a'}) pole_has_hp = {} # pole_name -> bool pole_detectors = {} # pole_name -> list of detectors for conn in deduced_connections: p = conn["pole"] richting = conn["richting"] kringen = conn["kringen"] # Check if HP is present on this pole if richting.lower() == "hp": pole_has_hp[p] = True continue # Detectors # Check if push button (e.g. Dka, Dkb, PBa1, BPf4...) if richting.lower().startswith(("dk", "pb", "bp")): if p not in pole_detectors: pole_detectors[p] = [] # Determine prefix if richting.lower().startswith("dk"): prefix = "DK" elif richting.lower().startswith("pb"): prefix = "PB" else: prefix = "BP" # Extract phase and potential index number phase_and_num = richting[2:] phase_char = "".join([c for c in phase_and_num if c.isalpha()]) det_code = f"{intersection_code}_{p}_{prefix}{phase_and_num}" if prefix == "BP": name_val = f"bouton poussoir NO (bleu) sur {p}" if phase_char: name_val += f" - Phase {phase_char}" elif prefix == "PB": name_val = f"bouton poussoir NF (jaune) sur {p}" if phase_char: name_val += f" - Phase {phase_char}" else: if is_french: name_val = f"Bouton Poussoir sur {p}" if phase_char: name_val += f" - Phase {phase_char}" else: name_val = f"Drukknop sur {p}" if phase_char: name_val += f" - Phase {phase_char}" if not any(d["code"] == det_code for d in pole_detectors[p]): pole_detectors[p].append({ "code": det_code, "name": name_val, "model_name": "Drukknop", "phases": [phase_char] if phase_char else [] }) continue # Other detectors (e.g. RA, VC, VT1, RC, RD, Rm...) is_detector = False det_code = None det_name = f"{richting} sur {p}" normalized_upper = richting.upper().replace(" ", "") if conn.get("is_detector", False): is_detector = True elif normalized_upper in ("RA", "RAB", "RB", "RC", "RD", "VD", "VC", "VT1", "VT2") or normalized_upper.startswith(("RM", "RP")): is_detector = True if is_detector: if normalized_upper.startswith("RM"): # Extract phase/direction (e.g., T2 from Rm T2,B1,B2) match = re.match(r'^Rm\s+([A-Z0-9]+)', richting, re.IGNORECASE) if match: phase = match.group(1) det_code = f"{intersection_code}_{p}_RM{phase.upper()}" else: det_code = f"{intersection_code}_{p}_{normalized_upper}" else: det_code = f"{intersection_code}_{p}_{normalized_upper}" if p not in pole_detectors: pole_detectors[p] = [] # Map model name if normalized_upper.startswith("RP"): model_name = "TMA-122 M" elif normalized_upper.startswith("R"): model_name = "TM60" elif normalized_upper.startswith("V"): model_name = "Traficam wide angle" else: model_name = "Traficam wide angle" if not any(d["code"] == det_code for d in pole_detectors[p]): pole_detectors[p].append({ "code": det_code, "name": det_name, "model_name": model_name, "phases": [] }) continue # 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. normalized = re.sub(r'\s+', ' ', richting).strip() if re.search(r'(?:alim|com|contact|radar|tension|comm)', normalized, re.IGNORECASE): continue if p not in pole_phases: pole_phases[p] = set() pole_phases[p].add(richting) # Now generate LANTERNS for each pole deduced_lanterns = [] # helper to sort phases: vehicles first, then trams, then cycles, then pedestrians def phase_sort_key(ph): if ph and ph[0].islower(): # Pedestrian return (3, ph) elif ph.startswith("T"): # Tram return (1, ph) elif ph.startswith("F"): # Cycle return (2, ph) else: # Vehicle return (0, ph) for p in sorted(pole_phases.keys()): phases = sorted(list(pole_phases[p]), key=phase_sort_key) lantern_idx = 1 for phase in phases: is_mast_arm = p.startswith("B") has_hp = pole_has_hp.get(p, False) # Find connections for this pole and phase phase_connections = [ conn for conn in deduced_connections if conn["pole"] == p and conn["richting"] == phase ] # A double lantern is indicated if any connection cell has 2 or more crosses (e.g. 'XX') has_double_marker = any(conn.get("crosses", 1) >= 2 for conn in phase_connections) if is_mast_arm and has_double_marker: # Determine models for both lanterns if phase and phase[0].islower(): # Pedestrian model_name_1 = "2V200 piéton+HP" if has_hp else "2V200" model_name_2 = "2V200 piéton+HP" if has_hp else "2V200" elif phase.startswith("T"): # Tram model_name_1 = "3V200 tram tout droite" model_name_2 = "3V300 tram tout droit" elif phase.startswith("F"): # Cycle model_name_1 = "3V200 cycliste" model_name_2 = "3V200 cycliste" else: # Vehicle model_name_1 = "3V200" model_name_2 = "3V300" # 1. Mast lantern (model_name_1) lantern_code_1 = f"{intersection_code}_{p}_LAN{lantern_idx:02d}_{phase}" display_model_1 = model_name_1 if "tram" in display_model_1.lower(): m = re.match(r'^(\d+V\d+)\s+tram', display_model_1, re.IGNORECASE) if m: display_model_1 = f"{m.group(1)} tram" lantern_name_1 = f"{display_model_1} sur {p} - Phase {phase}" deduced_lanterns.append({ "code": lantern_code_1, "name": lantern_name_1, "pole_code": f"{intersection_code}_{p}", "model_name": model_name_1, "phase": phase }) lantern_idx += 1 # 2. Arm lantern (model_name_2) lantern_code_2 = f"{intersection_code}_{p}_LAN{lantern_idx:02d}_{phase}" display_model_2 = model_name_2 if "tram" in display_model_2.lower(): m = re.match(r'^(\d+V\d+)\s+tram', display_model_2, re.IGNORECASE) if m: display_model_2 = f"{m.group(1)} tram" lantern_name_2 = f"{display_model_2} sur {p} - Phase {phase}" deduced_lanterns.append({ "code": lantern_code_2, "name": lantern_name_2, "pole_code": f"{intersection_code}_{p}", "model_name": model_name_2, "phase": phase }) lantern_idx += 1 else: # Deduce single model if phase and phase[0].islower(): # Pedestrian if has_hp: model_name = "2V200 piéton+HP" else: model_name = "2V200" elif phase.startswith("T"): # Tram if is_mast_arm: model_name = "3V300 tram tout droit" else: model_name = "3V200 tram tout droite" elif phase.startswith("F"): # Cycle model_name = "3V200 cycliste" else: # Vehicle if is_mast_arm: model_name = "3V300" else: model_name = "3V200" # Incorporate phase name in the code and name lantern_code = f"{intersection_code}_{p}_LAN{lantern_idx:02d}_{phase}" display_model = model_name if "tram" in display_model.lower(): m = re.match(r'^(\d+V\d+)\s+tram', display_model, re.IGNORECASE) if m: display_model = f"{m.group(1)} tram" lantern_name = f"{display_model} sur {p} - Phase {phase}" deduced_lanterns.append({ "code": lantern_code, "name": lantern_name, "pole_code": f"{intersection_code}_{p}", "model_name": model_name, "phase": phase }) lantern_idx += 1 # Compile all detectors into a single list deduced_detectors = [] for p, dets in pole_detectors.items(): for d in dets: deduced_detectors.append({ "code": d["code"], "name": d["name"], "pole_code": f"{intersection_code}_{p}", "model_name": d["model_name"], "phases": d["phases"] }) return { "intersection_code": intersection_code, "poles": poles_list, "cables": deduced_cables, "lanterns": deduced_lanterns, "detectors": deduced_detectors }