478 lines
20 KiB
Python
478 lines
20 KiB
Python
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 any column
|
|
cable_sections = []
|
|
max_row = sheet.max_row
|
|
max_col = sheet.max_column
|
|
|
|
for r in range(1, max_row + 1):
|
|
for c in range(1, max_col + 1):
|
|
val = sheet.cell(row=r, column=c).value
|
|
if val and isinstance(val, str) and ("câble" in val.lower() or "cable" in val.lower()):
|
|
cable_sections.append((r, val.strip()))
|
|
break
|
|
|
|
# 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 dynamically
|
|
m_double = re.search(r'(\d+)\s*x\s*(\d+)\s*x\s*[\d,.]+', cable_model_name, re.IGNORECASE)
|
|
m_multi = re.search(r'(\d+)\s*x[^\d]+(\d+)\s*x\s*[\d,.]+', cable_model_name, re.IGNORECASE)
|
|
m_single = re.search(r'(\d+)\s*x\s*[\d,.]+', cable_model_name, re.IGNORECASE)
|
|
if m_double:
|
|
strands = int(m_double.group(1)) * int(m_double.group(2))
|
|
elif m_multi:
|
|
strands = int(m_multi.group(1)) * int(m_multi.group(2))
|
|
elif m_single:
|
|
strands = int(m_single.group(1))
|
|
else:
|
|
strands = 61
|
|
|
|
# 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(" ", "")
|
|
|
|
# Check if RADAR power supply (Alimentation radar, not a detector)
|
|
if normalized_upper == "RADAR" or normalized_upper.startswith("RADAR") or (bool(re.search(r'\bradar\b', richting, re.IGNORECASE)) and not bool(re.match(r'^(?:RA|RB|RC|RD|RM|RP)\d*$', normalized_upper, re.IGNORECASE))):
|
|
continue
|
|
|
|
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", "RA", "RB", "RC", "RD", "VD", "VC", "VT")):
|
|
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
|
|
}
|