163 lines
4.6 KiB
Python
163 lines
4.6 KiB
Python
|
|
from typing import Dict, Any, List
|
|
from engine.logger import log_debug, log_info
|
|
|
|
|
|
def load_tp_config_for_con(cur, con_id: int) -> Dict[int, Dict[str, Dict[str, List[Dict[str, Any]]]]]:
|
|
"""
|
|
Charge TOUTE la configuration TP (det_type = 1) pour un con_id.
|
|
|
|
Structure retournée :
|
|
{
|
|
tp_id: {
|
|
"tp_name": "...",
|
|
"config": {
|
|
"config_tp": {
|
|
"clef_controleur": [
|
|
{ det_id, value1, value2, value3 }
|
|
],
|
|
"clef_temp_read": [
|
|
...
|
|
]
|
|
},
|
|
"config_tp_interval": {
|
|
"clef_temp_minimum_read_entre": [
|
|
{ det_id, value1, value2, value3 }
|
|
]
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
# 1) Lire la liste des TP pour ce con_id
|
|
sql_tp = """
|
|
SELECT tp_id, tp_name
|
|
FROM reader_config_timingpoint
|
|
WHERE con_id = %s
|
|
"""
|
|
cur.execute(sql_tp, (con_id,))
|
|
tps = cur.fetchall()
|
|
|
|
if not tps:
|
|
log_info(f"[CONFIG] Aucun TP trouvé pour con_id={con_id}")
|
|
return {}
|
|
|
|
# Préparer structure finale
|
|
tp_config: Dict[int, Dict] = {}
|
|
|
|
for tp in tps:
|
|
tp_id = tp["tp_id"]
|
|
tp_config[tp_id] = {
|
|
"tp_name": tp["tp_name"],
|
|
"config": {}
|
|
}
|
|
|
|
# 2) Lire toutes les configs det_type = 1 pour ce con_id
|
|
sql_conf = """
|
|
SELECT
|
|
d.det_id,
|
|
d.det_ref1_id AS tp_id,
|
|
d.det_groupe,
|
|
d.det_clef,
|
|
d.det_value1,
|
|
d.det_value2,
|
|
d.det_value3
|
|
FROM reader_config_detail d
|
|
JOIN reader_config_timingpoint tp ON tp.tp_id = d.det_ref1_id
|
|
WHERE tp.con_id = %s
|
|
AND d.det_type = 1
|
|
ORDER BY d.det_groupe, d.det_clef, d.det_id
|
|
"""
|
|
cur.execute(sql_conf, (con_id,))
|
|
rows = cur.fetchall()
|
|
|
|
# 3) Assembler
|
|
for row in rows:
|
|
tp_id = row["tp_id"]
|
|
group = row["det_groupe"] or "default"
|
|
clef = row["det_clef"]
|
|
|
|
if group not in tp_config[tp_id]["config"]:
|
|
tp_config[tp_id]["config"][group] = {}
|
|
|
|
if clef not in tp_config[tp_id]["config"][group]:
|
|
tp_config[tp_id]["config"][group][clef] = []
|
|
|
|
tp_config[tp_id]["config"][group][clef].append({
|
|
"det_id": row["det_id"],
|
|
"value1": row["det_value1"],
|
|
"value2": row["det_value2"],
|
|
"value3": row["det_value3"],
|
|
})
|
|
|
|
# 4) Affichage debug clair
|
|
log_info(f"[CONFIG] Chargé config TP pour con_id={con_id}")
|
|
for tp_id, data in tp_config.items():
|
|
log_info(f" TP {tp_id} ({data['tp_name']})")
|
|
for group, clefs in data["config"].items():
|
|
log_debug(f" [{group}]")
|
|
for clef, items in clefs.items():
|
|
for it in items:
|
|
log_debug(
|
|
f" {clef}: det_id={it['det_id']} "
|
|
f"v1={it['value1']} v2={it['value2']} v3={it['value3']}"
|
|
)
|
|
|
|
return tp_config
|
|
|
|
|
|
def resolve_tp_for_lecture(tp_config, con_id: int, chr_location: str):
|
|
"""
|
|
Retourne l'entrée TP correspondant à la location lue.
|
|
|
|
On cherche dans les configs TP la clef :
|
|
config_tp → clef_controleur → value1 = chr_location
|
|
|
|
On choisit automatiquement :
|
|
- la priorité la plus basse (1 est meilleur que 2)
|
|
- si ties : on prend la première
|
|
|
|
Retourne un dict complet :
|
|
{
|
|
"tp_id": ...,
|
|
"tp_name": "...",
|
|
"priority": ...,
|
|
"controller_name": "...",
|
|
"det_id": ...
|
|
}
|
|
"""
|
|
|
|
best = None
|
|
|
|
for tp_id, data in tp_config.items():
|
|
|
|
# Toute config pour ce timing point
|
|
groups = data.get("config", {})
|
|
|
|
# Groupe config_tp (celui des contrôleurs)
|
|
cfg_tp = groups.get("config_tp", {})
|
|
|
|
# Liste des clefs clef_controleur
|
|
controllers = cfg_tp.get("clef_controleur", [])
|
|
|
|
for ctrl in controllers:
|
|
if ctrl["value1"] != chr_location:
|
|
continue
|
|
|
|
prio = int(ctrl["value2"]) if ctrl["value2"] else 999
|
|
|
|
item = {
|
|
"tp_id": tp_id,
|
|
"tp_name": data["tp_name"],
|
|
"priority": prio,
|
|
"controller_name": ctrl["value3"],
|
|
"det_id": ctrl["det_id"],
|
|
}
|
|
|
|
# Choisir le meilleur contrôleur (prio la plus petite)
|
|
if best is None or prio < best["priority"]:
|
|
best = item
|
|
|
|
return best
|