251 lines
6.9 KiB
Python
251 lines
6.9 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):
|
||
"""
|
||
Charge automatiquement toute la configuration TP (det_type = 1)
|
||
pour un con_id, sans coder les groupes/celfs en dur.
|
||
|
||
Structure produite (identique à ton ancienne fonction) :
|
||
{
|
||
tp_id: {
|
||
"tp_name": "...",
|
||
"config": {
|
||
<det_groupe>: {
|
||
<det_clef>: [
|
||
{
|
||
det_id,
|
||
value1, value2, value3
|
||
}
|
||
]
|
||
}
|
||
}
|
||
}
|
||
}
|
||
"""
|
||
|
||
# 1) Récupérer tous les TP liés au con_id
|
||
cur.execute("""
|
||
SELECT tp_id, tp_name
|
||
FROM reader_config_timingpoint
|
||
WHERE con_id = %s
|
||
""", (con_id,))
|
||
tps = cur.fetchall()
|
||
|
||
tp_config = {}
|
||
|
||
for tp in tps:
|
||
tp_id = tp["tp_id"]
|
||
tp_config[tp_id] = {
|
||
"tp_name": tp["tp_name"],
|
||
"config": {}
|
||
}
|
||
|
||
if not tps:
|
||
log_info(f"[CONFIG] Aucun TP trouvé pour con_id={con_id}")
|
||
return {}
|
||
|
||
# 2) Récupérer toutes les configs det_type = 1 liées aux TP
|
||
cur.execute("""
|
||
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_ref1_id, d.det_groupe, d.det_clef, d.det_id
|
||
""", (con_id,))
|
||
|
||
rows = cur.fetchall()
|
||
|
||
# 3) Construction automatique
|
||
for row in rows:
|
||
tp_id = row["tp_id"]
|
||
|
||
group = row["det_groupe"] or "default"
|
||
clef = row["det_clef"]
|
||
|
||
# Créer les niveaux si nécessaires
|
||
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] = []
|
||
|
||
# Ajouter l’item
|
||
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"],
|
||
})
|
||
|
||
return tp_config
|
||
|
||
|
||
def load_tp_config_for_conbad(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
|