fix(trafficlights): correct cyclopieton 'bF' detection and preserve STIB detectors by default

- Recognize combined 'bF' phases as cyclo-piétons with appropriate 2V200 cyclo-piéto model
- Improve lantern match scoring to pair LANxx_bF with existing LANxx lanterns on multi-lantern poles
- Normalize lantern model matching between Excel deduction and DB names
- Propose 'keep' by default for STIB tram detector loops ('dT') absent from the cross plan
- Add unit tests for cyclo-piéton deduction, lantern matching and STIB detector preservation
This commit is contained in:
kdeterme 2026-09-22 21:59:05 +02:00
parent c7c76f28dc
commit 61ef7fc3ea
4 changed files with 311 additions and 24 deletions

View file

@ -503,11 +503,11 @@
</thead>
<tbody>
{% for item in detectors %}
<tr class="{% if item.status == 'new' %}table-success-light{% elif item.status == 'modified' %}table-warning-light{% elif item.status == 'db_only' %}table-danger-light{% endif %}">
<tr class="{% if item.status == 'new' %}table-success-light{% elif item.status == 'modified' %}table-warning-light{% elif item.status == 'db_only' %}{% if item.action_default == 'keep' %}table-light{% else %}table-danger-light{% endif %}{% endif %}">
<td class="ps-4"><input type="checkbox" name="{% if item.status == 'db_only' %}selected_detector_db_{{ item.code }}{% else %}selected_detector_{{ item.code }}{% endif %}" class="form-check-input detector-checkbox" checked></td>
<td class="font-weight-medium"><code>{{ item.code }}</code><input type="hidden" name="db_id_detector_{{ item.code }}" value="{{ item.db_id|default:'' }}"></td>
<td>
<input type="text" name="name_detector_{{ item.code }}" class="form-control form-control-sm" value="{{ item.name }}">
<input type="text" name="name_detector_{{ item.code }}" class="form-control form-control-sm" value="{{ item.name }}">
</td>
<td><code>{{ item.pole_code|default:"—" }}</code></td>
<td>
@ -518,7 +518,11 @@
{% elif item.status == 'identical' %}
<span class="badge bg-light text-secondary border">{% translate "Identique" %}</span>
{% elif item.status == 'db_only' %}
<span class="badge bg-danger">{% translate "DB uniquement" %}</span>
{% if item.action_default == 'keep' %}
<span class="badge bg-secondary">{% translate "DB uniquement" %}</span>
{% else %}
<span class="badge bg-danger">{% translate "DB uniquement" %}</span>
{% endif %}
{% endif %}
{% for diff in item.details_diff %}
<div class="small text-muted mt-1" style="font-size: 0.75rem; line-height: 1.25;">
@ -531,8 +535,8 @@
<td>
{% if item.status == 'db_only' %}
<select name="action_detector_db_{{ item.code }}" class="form-select form-select-sm rounded-2">
<option value="keep">{% translate "Conserver en DB" %}</option>
<option value="archive" selected>{% translate "Archiver" %}</option>
<option value="keep" {% if item.action_default == 'keep' %}selected{% endif %}>{% translate "Conserver en DB" %}</option>
<option value="archive" {% if item.action_default != 'keep' %}selected{% endif %}>{% translate "Archiver" %}</option>
</select>
{% else %}
<select name="action_detector_{{ item.code }}" class="form-select form-select-sm rounded-2">

View file

@ -2008,7 +2008,7 @@ class TrafficLightCrossPlanImportTest(TestCase):
import tempfile
from assets.utils.cross_plan_parser import parse_cross_plan
# Workbook: Pedestrian phase aF with HP line
# Workbook: Cyclo-pedestrian phase bF with HP line
wb = openpyxl.Workbook()
ws = wb.active
ws.title = "kruisjesplan"
@ -2019,10 +2019,53 @@ class TrafficLightCrossPlanImportTest(TestCase):
ws.cell(row=6, column=4, value="kringen")
ws.cell(row=7, column=5, value="B01")
# Row 8: Pedestrian phase aF
# Row 8: Cyclo-pedestrian phase bF
ws.cell(row=8, column=1, value="com")
ws.cell(row=8, column=2, value="1")
ws.cell(row=8, column=3, value="aF")
ws.cell(row=8, column=3, value="bF")
ws.cell(row=8, column=5, value="X")
# Row 9: HP
ws.cell(row=9, column=1, value="com")
ws.cell(row=9, column=2, value="2")
ws.cell(row=9, column=3, value="HP")
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:
deduced = parse_cross_plan(tmp_path)
lanterns = deduced["lanterns"]
self.assertEqual(len(lanterns), 1)
self.assertEqual(lanterns[0]["model_name"], "2V200 cyclo-piéto + HP")
self.assertEqual(lanterns[0]["code"], "SWB01_B01_LAN01_bF")
finally:
import os
if os.path.exists(tmp_path):
os.remove(tmp_path)
def test_parse_cross_plan_pure_pedestrian_hp(self):
import openpyxl
import tempfile
from assets.utils.cross_plan_parser import parse_cross_plan
# Workbook: Pure pedestrian phase 'a' with HP line
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")
# Row 8: Pure pedestrian phase 'a'
ws.cell(row=8, column=1, value="com")
ws.cell(row=8, column=2, value="1")
ws.cell(row=8, column=3, value="a")
ws.cell(row=8, column=5, value="X")
# Row 9: HP
@ -2040,7 +2083,7 @@ class TrafficLightCrossPlanImportTest(TestCase):
lanterns = deduced["lanterns"]
self.assertEqual(len(lanterns), 1)
self.assertEqual(lanterns[0]["model_name"], "2V200 piéton+HP")
self.assertEqual(lanterns[0]["code"], "SWB01_B01_LAN01_aF")
self.assertEqual(lanterns[0]["code"], "SWB01_B01_LAN01_a")
finally:
import os
if os.path.exists(tmp_path):
@ -2618,6 +2661,115 @@ class TrafficLightCrossPlanImportTest(TestCase):
self.assertEqual(match_map['SEK02_B02_LAN01_B1'], 'SEK02_B02_LAN02')
self.assertEqual(match_map['SEK02_B02_LAN02_T2'], 'SEK02_B02_LAN01')
def test_match_pole_lanterns_cyclopieton_bf(self):
"""
Tests the user scenario on pole B01 with multiple lanterns:
DB has:
LAN01_A (3V200)
LAN02_A (3V300)
LAN03 (2V200 cyclo-piéto + HP)
Plan deduces:
LAN01_A (3V200)
LAN02_A (3V300)
LAN03_bF (2V200 cyclo-piéto + HP, phase bF)
Must match LAN03_bF -> DB LAN03 instead of archiving LAN03 and creating a new asset.
"""
from assets.utils.cross_plan_parser import match_pole_lanterns
class DummyModel:
def __init__(self, name_fr):
self.name_fr = name_fr
self.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)
pole_dls = [
{'code': 'SPW10_B01_LAN01_A', 'model_name': '3V200', 'name': '3V200 sur B01 - Phase A', 'phase': 'A', 'pole_code': 'SPW10_B01'},
{'code': 'SPW10_B01_LAN02_A', 'model_name': '3V300', 'name': '3V300 sur B01 - Phase A', 'phase': 'A', 'pole_code': 'SPW10_B01'},
{'code': 'SPW10_B01_LAN03_bF', 'model_name': '2V200 cyclo-piéto + HP', 'name': '2V200 cyclo-piéto + HP sur B01 - Phase bF', 'phase': 'bF', 'pole_code': 'SPW10_B01'},
]
el_v1 = DummyLantern('SPW10_B01_LAN01_A', '3V200', '3V200')
el_v2 = DummyLantern('SPW10_B01_LAN02_A', '3V300', '3V300')
el_cyclo = DummyLantern('SPW10_B01_LAN03', '2V200 cyclo-piéto + HP', '2V200 cyclo-piéto + HP')
pole_els = [el_v1, el_v2, el_cyclo]
matched_pairs, unmatched_dls, unmatched_els = match_pole_lanterns(pole_dls, pole_els)
self.assertEqual(len(matched_pairs), 3)
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['SPW10_B01_LAN03_bF'], 'SPW10_B01_LAN03')
def test_import_cross_plan_stib_detector_preserved_by_default(self):
"""
Tests that when a DB detector is named dT1 or dT2 (STIB tram loop),
the preview recognizes it as STIB, sets action_default='keep', and proposes
to preserve it by default in the HTML decision select box.
"""
import openpyxl
import tempfile
from django.core.files.uploadedfile import SimpleUploadedFile
from assets.models import TrafficLightDetector, TrafficLightDetectorModel
det_model = TrafficLightDetectorModel.objects.first()
stib_det = TrafficLightDetector.objects.create(
intersection=self.intersection,
code="SPW10_DET_1",
name="dT1",
name_fr="dT1",
model=det_model,
status="active"
)
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=8, column=3, value="A")
ws.cell(row=8, 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 = SimpleUploadedFile("test_cross_plan.xlsx", f.read(), content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")
url = reverse("assets:import_cross_plan", args=[self.intersection.id])
response = self.client.post(url, {"cross_plan_file_upload": uploaded})
self.assertEqual(response.status_code, 200)
detectors = response.context["detectors"]
stib_item = next((d for d in detectors if d["code"] == "SPW10_DET_1"), None)
self.assertIsNotNone(stib_item)
self.assertEqual(stib_item["status"], "db_only")
self.assertEqual(stib_item["action_default"], "keep")
self.assertTrue(stib_item["is_stib"])
self.assertIn("STIB", stib_item["details_diff"][0])
# Verify in HTML that 'Conserver en DB' is selected
html = response.content.decode("utf-8")
self.assertIn('name="action_detector_db_SPW10_DET_1"', html)
self.assertIn('value="keep" selected', html)
finally:
import os
if os.path.exists(tmp_path):
os.remove(tmp_path)
def test_import_cross_plan_multiple_lanterns_preview(self):
"""
End-to-end view test with multiple lanterns on the same pole.

View file

@ -26,6 +26,24 @@ def is_hp_or_ls(text):
return True
return False
def is_cyclopieton_phase(phase):
"""
Checks if a phase code corresponds to a combined pedestrian + cyclist signal (cyclo-piéton).
Standard notation: lowercase letter (pedestrian) + 'F' or 'f' (fiets/cyclist), e.g. 'bF', 'aF', 'cF', 'dF', 'bF1',
or vice versa ('Fa', 'Fb').
"""
if not phase:
return False
ph = str(phase).strip()
if re.match(r'^[a-z][Ff]\d*$', ph) or re.match(r'^[Ff][a-z]\d*$', ph):
return True
if re.match(r'^[a-z]\s*[\+/&]\s*[Ff]\d*$', ph) or re.match(r'^[Ff]\s*[\+/&]\s*[a-z]\d*$', ph):
return True
ph_lower = ph.lower()
if 'cyclo' in ph_lower and ('piet' in ph_lower or 'piéto' in ph_lower or 'ped' in ph_lower):
return True
return False
def parse_cross_plan(excel_path):
"""
Parses a crossroads cross plan (kruisjesplan) Excel file and returns
@ -375,9 +393,13 @@ def parse_cross_plan(excel_path):
# Now generate LANTERNS for each pole
deduced_lanterns = []
# helper to sort phases: vehicles first, then trams, then cycles, then pedestrians
# helper to sort phases: vehicles first, then trams, then cycles, then cyclo-piétons / pedestrians
def phase_sort_key(ph):
if ph and ph[0].islower(): # Pedestrian
if is_cyclopieton_phase(ph):
if ph and ph[0].islower():
return (3, ph)
return (2, ph)
elif ph and ph[0].islower(): # Pedestrian
return (3, ph)
elif ph.startswith("T"): # Tram
return (1, ph)
@ -404,7 +426,10 @@ def parse_cross_plan(excel_path):
if is_mast_arm and has_double_marker:
# Determine models for both lanterns
if phase and phase[0].islower(): # Pedestrian
if is_cyclopieton_phase(phase):
model_name_1 = "2V200 cyclo-piéto + HP" if has_hp else "2V200 cyclo-piéto"
model_name_2 = "2V200 cyclo-piéto + HP" if has_hp else "2V200 cyclo-piéto"
elif 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
@ -452,7 +477,12 @@ def parse_cross_plan(excel_path):
lantern_idx += 1
else:
# Deduce single model
if phase and phase[0].islower(): # Pedestrian
if is_cyclopieton_phase(phase):
if has_hp:
model_name = "2V200 cyclo-piéto + HP"
else:
model_name = "2V200 cyclo-piéto"
elif phase and phase[0].islower(): # Pedestrian
if has_hp:
model_name = "2V200 piéton+HP"
else:
@ -542,9 +572,23 @@ def classify_lantern(model_name="", phase="", code="", name=""):
)
)
# Check Cyclo-piéton:
is_cyclopieton = (
not is_tram and not is_bus and (
is_cyclopieton_phase(phase_clean)
or bool(re.search(r'[\b_]LAN\d+_[a-z][Ff]\d*', code_upper))
or "cyclo-piét" in combined_text
or "cyclo-piet" in combined_text
or "cyclopiet" in combined_text
or ("cyclo" in combined_text and any(w in combined_text for w in ["piét", "piet", "voetganger"]))
or (any(w in combined_text for w in ["cycliste", "velo", "vélo", "fiets"])
and any(w in combined_text for w in ["piéton", "pieton", "voetganger", "pedestrian"]))
)
)
# Check Cyclist:
is_cyclist = (
not is_tram and not is_bus and (
not is_tram and not is_bus and not is_cyclopieton 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))
@ -553,7 +597,7 @@ def classify_lantern(model_name="", phase="", code="", name=""):
# Check Pedestrian:
is_pedestrian = (
not is_tram and not is_bus and not is_cyclist and (
not is_tram and not is_bus and not is_cyclopieton 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")
@ -565,6 +609,8 @@ def classify_lantern(model_name="", phase="", code="", name=""):
category = "tram"
elif is_bus:
category = "bus"
elif is_cyclopieton:
category = "cyclopieton"
elif is_cyclist:
category = "cyclist"
elif is_pedestrian:
@ -634,7 +680,13 @@ def calculate_lantern_match_score(dl, el, is_single_on_pole=False):
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)
el_base_code = el_code.rsplit("_", 1)[0] if "_" in el_code else el_code
is_prefix_code_match = bool(
(dl_base_code and el_code and dl_base_code == el_code)
or (dl_code and el_base_code and dl_code == el_base_code)
or (dl_base_code and el_base_code and dl_base_code == el_base_code)
)
if is_single_on_pole and is_prefix_code_match:
return 1000
@ -664,10 +716,12 @@ def calculate_lantern_match_score(dl, el, is_single_on_pole=False):
# 2. Category matching
if dl_cat == el_cat:
if dl_cat in ("tram", "bus", "pedestrian", "cyclist"):
if dl_cat in ("tram", "bus", "pedestrian", "cyclist", "cyclopieton"):
score += 300
else:
score += 200
elif {dl_cat, el_cat} in ({"cyclopieton", "pedestrian"}, {"cyclopieton", "cyclist"}):
score += 50
else:
# Category mismatch penalty: e.g. tram vs car, pedestrian vs car
score -= 600
@ -707,9 +761,9 @@ def calculate_lantern_match_score(dl, el, is_single_on_pole=False):
else:
score -= 60
# 8. Prefix match bonus
# 8. Prefix match bonus (e.g. LAN03 vs LAN03_bF on the same pole)
if is_prefix_code_match:
score += 30
score += 250
# 9. LAN index tie-breaker (e.g. LAN01 vs LAN01)
m_dl = re.search(r'LAN(\d+)', dl_code, re.IGNORECASE)

View file

@ -26,6 +26,7 @@ from django.utils.translation import get_language, gettext as _
from django.utils.timezone import make_aware, get_current_timezone
import json
import re
from datetime import datetime
import unicodedata
from collections import defaultdict
@ -3744,6 +3745,77 @@ def update_staged_pole_position(request, intersection_id, pole_id):
})
def find_db_lantern_model(model_name):
"""
Finds the most suitable TrafficLightLanternModel from the database based on model_name.
Gracefully handles variants such as '2V200 cyclo-piéto + HP', '2V200 cyclo-piéton + HP',
'2V200 piéton+HP', tram, cyclist models, etc.
"""
if not model_name:
return TrafficLightLanternModel.objects.first()
m = TrafficLightLanternModel.objects.filter(name_fr__iexact=model_name).first()
if m:
return m
m_nl = TrafficLightLanternModel.objects.filter(name_nl__iexact=model_name).first()
if m_nl:
return m_nl
model_lower = str(model_name).lower()
if "cyclo" in model_lower:
qs = TrafficLightLanternModel.objects.filter(
Q(name_fr__icontains="cyclo") | Q(name_nl__icontains="cyclo") |
Q(name_fr__icontains="fietsers- en voetgangers") | Q(name_nl__icontains="fietsers- en voetgangers")
)
if "hp" in model_lower or "haut-parleur" in model_lower:
m = qs.filter(Q(name_fr__icontains="hp") | Q(name_fr__icontains="haut-parleur") | Q(name_nl__icontains="hp")).first()
else:
m = qs.exclude(Q(name_fr__icontains="hp") | Q(name_fr__icontains="haut-parleur") | Q(name_nl__icontains="hp")).first()
if m:
return m
if qs.exists():
return qs.first()
elif any(w in model_lower for w in ["piét", "piet", "voetganger"]):
qs = TrafficLightLanternModel.objects.filter(
Q(name_fr__icontains="piét") | Q(name_fr__icontains="piet") | Q(name_nl__icontains="voetganger")
).exclude(Q(name_fr__icontains="cyclo") | Q(name_nl__icontains="cyclo"))
if "hp" in model_lower or "haut-parleur" in model_lower:
m = qs.filter(Q(name_fr__icontains="hp") | Q(name_fr__icontains="haut-parleur") | Q(name_nl__icontains="hp")).first()
else:
m = qs.exclude(Q(name_fr__icontains="hp") | Q(name_fr__icontains="haut-parleur") | Q(name_nl__icontains="hp")).first()
if m:
return m
elif "tram" in model_lower:
if "300" in model_lower:
m = TrafficLightLanternModel.objects.filter(name_fr__icontains="tram").filter(name_fr__icontains="300").first()
if m:
return m
m = TrafficLightLanternModel.objects.filter(name_fr__icontains="tram").first()
if m:
return m
elif any(w in model_lower for w in ["cycliste", "cycl", "fiets"]):
m = TrafficLightLanternModel.objects.filter(
Q(name_fr__icontains="cycl") | Q(name_nl__icontains="fiets")
).exclude(Q(name_fr__icontains="piét") | Q(name_fr__icontains="piet") | Q(name_nl__icontains="voetganger")).first()
if m:
return m
return TrafficLightLanternModel.objects.first()
def is_stib_detector(detector_obj, code=""):
"""
Checks if a detector is a STIB/tram detection loop (e.g. dT1, dT2, DT1),
which is provided/managed by STIB and not present in cross plans.
"""
name = (getattr(detector_obj, "name", None) or getattr(detector_obj, "name_fr", None) or "").strip()
code_str = (code or getattr(detector_obj, "code", "") or "").strip()
if re.match(r'^dt[\s_\d]*$', name, re.IGNORECASE) or name.lower().startswith("dt"):
return True
if re.search(r'\bdt\d+\b', name, re.IGNORECASE) or "stib" in name.lower():
return True
if re.search(r'(?:^|_)dt\d*(?:_|$)', code_str, re.IGNORECASE) or "stib" in code_str.lower():
return True
return False
@check_thematic_access('trafficlights', edit_permission_required=True)
def import_cross_plan(request, intersection_id):
from django.shortcuts import render, get_object_or_404, redirect
@ -4413,7 +4485,7 @@ def import_cross_plan(request, intersection_id):
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()
db_model = find_db_lantern_model(model_name)
is_rename = (el.code != code)
is_model_diff = el.model_id != (db_model.id if db_model else None)
@ -4423,7 +4495,7 @@ def import_cross_plan(request, intersection_id):
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}")
details.append(f"Modèle : {el.model.name_fr if el.model else 'Aucun'} → {db_model.name_fr if db_model else 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']
@ -4434,7 +4506,7 @@ def import_cross_plan(request, intersection_id):
'code': code,
'name': display_name,
'status': status,
'deduced_model_name': model_name,
'deduced_model_name': db_model.name_fr if db_model else model_name,
'db_model_name': el.model.name_fr if el.model else '',
'db_id': el.id,
'pole_code': dl['pole_code'],
@ -4532,6 +4604,9 @@ def import_cross_plan(request, intersection_id):
'details_diff': []
})
for code, ed in existing_detectors.items():
is_stib = is_stib_detector(ed, code)
action_default = 'keep' if is_stib else 'archive'
diff_text = _("Détection STIB / Tram (fournie par la STIB, conservée par défaut)") if is_stib else _("Absent du fichier Excel (présent en DB)")
compared_detectors.append({
'code': code,
'name': ed.name or code,
@ -4540,7 +4615,9 @@ def import_cross_plan(request, intersection_id):
'db_model_name': ed.model.name_fr if ed.model else '',
'db_id': ed.id,
'pole_code': ed.pole.code if ed.pole else '',
'details_diff': [_("Absent du fichier Excel (présent en DB)")]
'details_diff': [diff_text],
'action_default': action_default,
'is_stib': is_stib,
})
# ── 5. OTHER ASSOCIATED ASSETS (NOT IN CROSS PLAN) ──
@ -4832,7 +4909,7 @@ def integrate_cross_plan(request, intersection_id):
return m
def resolve_lantern_model(name_fr):
m = TrafficLightLanternModel.objects.filter(name_fr__iexact=name_fr).first()
m = find_db_lantern_model(name_fr)
if not m:
m = TrafficLightLanternModel.objects.first()
if not m: