diff --git a/engine/engine_components/__pycache__/detail_writer.cpython-313.pyc b/engine/engine_components/__pycache__/detail_writer.cpython-313.pyc index 3af2d2c..62e6e0b 100644 Binary files a/engine/engine_components/__pycache__/detail_writer.cpython-313.pyc and b/engine/engine_components/__pycache__/detail_writer.cpython-313.pyc differ diff --git a/engine/engine_components/__pycache__/process_course.cpython-313.pyc b/engine/engine_components/__pycache__/process_course.cpython-313.pyc index b34f119..b8adb22 100644 Binary files a/engine/engine_components/__pycache__/process_course.cpython-313.pyc and b/engine/engine_components/__pycache__/process_course.cpython-313.pyc differ diff --git a/engine/engine_components/__pycache__/tp_config.cpython-313.pyc b/engine/engine_components/__pycache__/tp_config.cpython-313.pyc index 5578785..47fb95d 100644 Binary files a/engine/engine_components/__pycache__/tp_config.cpython-313.pyc and b/engine/engine_components/__pycache__/tp_config.cpython-313.pyc differ diff --git a/engine/engine_components/process_course.py b/engine/engine_components/process_course.py index 9a7851e..cc1591d 100644 --- a/engine/engine_components/process_course.py +++ b/engine/engine_components/process_course.py @@ -9,30 +9,44 @@ from datetime import datetime, timezone from zoneinfo import ZoneInfo from decimal import Decimal from engine.utils import format_temps_decimal + LOCAL_TZ = ZoneInfo("America/Toronto") + +# ------------------------------------------------- +# Mémoire locale pour la dernière lecture valide +# ------------------------------------------------- _last_valid_read = {} # (tp_id, dossard) -> datetime locale +# ------------------------------------------------- +# Conversion d'un timestamp Decimal en datetime UTC +# ------------------------------------------------- def ts_to_utc(ts): ts = Decimal(str(ts)) seconds = int(ts) micro = int((ts - Decimal(seconds)) * Decimal(1_000_000)) - return datetime.fromtimestamp(seconds, tz=timezone.utc).replace(microsecond=micro) - +# ------------------------------------------------- +# Normalisation du temps de course (start/end) +# ------------------------------------------------- def _normalize_course_time(dt): if dt is None: return None, None dt_utc = dt.replace(tzinfo=timezone.utc) - dt_local = dt_utc.astimezone(LOCAL_TZ) + dt_local = dt_utc return dt_local, dt_utc - +# ------------------------------------------------- +# Moteur principal : traitement d'une course +# ------------------------------------------------- def process_course(cur, cnx, con_id: int): + # ------------------------------------------------- + # Lecture de la configuration TP + configuration course + # ------------------------------------------------- tp_config = load_tp_config_for_con(cur, con_id) course_cfg = load_course_config(cur, con_id) @@ -45,13 +59,18 @@ def process_course(cur, cnx, con_id: int): engine_table = f"resultv2_{con_id}_lecture" classement_table = f"resultv2_{con_id}_classement" - # last_processed + # ------------------------------------------------- + # Lire dernier ID traité (mécanisme incremental) + # ------------------------------------------------- last_processed = get_last_processed_read_id(cur, engine_table) if last_processed is None: last_processed = 0 log_info(f"[STATE] con_id={con_id} last_processed={last_processed}") + # ------------------------------------------------- + # Récupère les lectures non traitées + # ------------------------------------------------- lectures = get_new_lectures(cur, last_processed) log_info(f"[NEW LECTURES] con_id={con_id} count={len(lectures)}") @@ -60,53 +79,81 @@ def process_course(cur, cnx, con_id: int): max_id = last_processed + # ------------------------------------------------- + # Boucle sur chaque lecture brute + # ------------------------------------------------- for lec in lectures: rcourse_id = int(lec["rcourse_id"]) chr_loc = lec["chrinfo_location"] - timestamp = Decimal(lec["chrinfo_time"]) + timestamp = Decimal(lec["chrinfo_time"]) + max_id = max(max_id, rcourse_id) - dt_utc = ts_to_utc(timestamp) + dt_utc = ts_to_utc(timestamp) dt_local = dt_utc.astimezone(LOCAL_TZ) log_debug(f"[READ] id={rcourse_id} utc={dt_utc} loc={chr_loc}") - # 1) start/end + # ------------------------------------------------- + # Validation start/end + # ------------------------------------------------- if start_utc and dt_utc < start_utc: - log_debug(f"[REJECT START] id={rcourse_id} dt_utc={dt_utc} < start_utc={start_utc}") - continue + log_debug(f"[REJECT START] id={rcourse_id} dt_utc={dt_utc} < start_utc={start_utc}") + continue if end_utc and dt_utc > end_utc: - log_debug(f"[REJECT END] id={rcourse_id} dt_utc={dt_utc} > end_utc={end_utc}") - continue + log_debug(f"[REJECT END] id={rcourse_id} dt_utc={dt_utc} > end_utc={end_utc}") + continue status = 1 reason_code = None reason_ref_id = None - # 2) coureur (BIB doit exister dans le classement) + # ------------------------------------------------- + # Vérifier que le BIB existe dans la table classement + # ------------------------------------------------- tag = lec["chrinfo_tag"] coureur = get_classement_by_tag(cur, classement_table, tag) if not coureur: - # IMPORTANT : on ignore complètement continue dossard = coureur["cla_info_dossard"] - # 3) TP obligatoire + # ------------------------------------------------- + # Résoudre le timing point selon la config + # ------------------------------------------------- tp_info = resolve_tp_for_lecture(tp_config, con_id, chr_loc) if not tp_info: continue - tp_id = tp_info["tp_id"] - tp_name = tp_info["tp_name"] - prio = tp_info["priority"] + tp_id = tp_info["tp_id"] + tp_name = tp_info["tp_name"] + prio = tp_info["priority"] controller = tp_info["controller_name"] - det_id = tp_info["det_id"] + det_id = tp_info["det_id"] - # 4) interval rule + detail_table = f"resultv2_{con_id}_detail" + + # ------------------------------------------------- + # Règle de priorité (peut rejeter ou forcer tour) + # ------------------------------------------------- + status, reason_code, reason_ref_id, forced_tour = resolve_priority( + cur, + cnx, + detail_table, + tp_id, + dossard, + timestamp, + prio, + det_id, + tp_config + ) + + # ------------------------------------------------- + # Règle d'intervalle minimum entre lectures + # ------------------------------------------------- interval_cfg = ( tp_config[tp_id]["config"] .get("config_tp_interval", {}) @@ -124,19 +171,21 @@ def process_course(cur, cnx, con_id: int): if last_dt and interval_sec > 0: delta = (dt_local - last_dt).total_seconds() + if delta < interval_sec: status = 0 reason_code = "SKIP_INTERVAL" reason_ref_id = det_id log_debug(f"[SKIP INTERVAL] id={rcourse_id} {delta:.1f}s < {interval_sec}s") - # 5) INSERT - - detail_table = f"resultv2_{con_id}_detail" - - # 5) tours + tempstour + # ------------------------------------------------- + # Calcul du tour + temps de tour + # ------------------------------------------------- tour, prev_ts = compute_tour(cur, detail_table, tp_id, dossard) + if forced_tour is not None: + tour = forced_tour + if prev_ts is None: tempstour = Decimal("0") else: @@ -144,12 +193,15 @@ def process_course(cur, cnx, con_id: int): tempstour_str = format_temps_decimal(tempstour) + # ------------------------------------------------- + # Écriture dans la table detail + # ------------------------------------------------- insert_detail_record( cur, cnx, con_id, dossard, lec, tp_id, tp_name, chr_loc, prio, controller, - dt_local, + dt_utc, status=status, reason_code=reason_code, reason_ref_id=reason_ref_id, @@ -158,23 +210,24 @@ def process_course(cur, cnx, con_id: int): tempstour_str=tempstour_str ) - - - - - # Mise à jour du dernier read VALIDE + # ------------------------------------------------- + # Mise à jour mémoire locale dernière lecture valide + # ------------------------------------------------- if status == 1: _last_valid_read[(tp_id, dossard)] = dt_local log_info(f"[OK] id={rcourse_id} tag={tag} dossard={dossard} tp={tp_name} prio={prio}") + # ------------------------------------------------- + # Mise à jour du pointeur de traitement + # ------------------------------------------------- update_last_processed_read_id(cur, cnx, engine_table, max_id) + +# ------------------------------------------------- +# Recherche du dernier tour pour un coureur à un TP +# ------------------------------------------------- def compute_tour(cur, detail_table, tp_id, dossard): - """ - Retourne (tour, prev_timestamp_decimal) - - tour = dernier_tour + 1 - - prev_timestamp_decimal = timestamp_read du tour précédent (Decimal) ou None si premier tour - """ + cur.execute(f""" SELECT tour, timestamp_read FROM `{detail_table}` @@ -187,7 +240,6 @@ def compute_tour(cur, detail_table, tp_id, dossard): row = cur.fetchone() - # Première lecture → tour 1, pas de temps précédent if not row or row["tour"] is None: return 1, None @@ -195,3 +247,96 @@ def compute_tour(cur, detail_table, tp_id, dossard): prev_ts = Decimal(str(row["timestamp_read"])) return last_tour + 1, prev_ts + + +# ------------------------------------------------- +# Règle de priorité : remplace, rejette, ou force tour +# ------------------------------------------------- +def resolve_priority( + cur, + cnx, + detail_table, + tp_id, + dossard, + new_timestamp, + new_priority, + det_id, + tp_config +): + + # ------------------------------------------------- + # Lecture de la fenêtre backup pour priorité + # ------------------------------------------------- + cfg = ( + tp_config[tp_id]["config"] + .get("config_tp", {}) + .get("clef_temp_bbk", []) + ) + + backup_window = 0 + if cfg: + try: + backup_window = int(cfg[0]["value1"]) + except: + backup_window = 0 + + log_debug(f"[resolve_priority] backup_window={backup_window}") + + # ------------------------------------------------- + # Si backup_window = 0 → toujours accepter + # ------------------------------------------------- + if backup_window <= 0: + return 1, None, None, None + + # ------------------------------------------------- + # Recherche d'une lecture existante dans la fenêtre + # ------------------------------------------------- + cur.execute(f""" + SELECT detail_id, controller_priority, timestamp_read, tour + FROM `{detail_table}` + WHERE tp_id = %s + AND cla_info_dossard = %s + AND status = 1 + AND ABS(timestamp_read - %s) <= %s + LIMIT 1 + """, (tp_id, dossard, str(new_timestamp), backup_window)) + + row = cur.fetchone() + + # ------------------------------------------------- + # Aucun doublon → accepter + # ------------------------------------------------- + if not row: + return 1, None, None, None + + old_id = row["detail_id"] + old_priority = row["controller_priority"] + old_tour = row["tour"] + + # ------------------------------------------------- + # Comparaison des priorités + # ------------------------------------------------- + + # Nouvelle lecture moins prioritaire → rejet + if new_priority > old_priority: + return 0, "LOWER_PRIORITY", det_id, None + + # Nouvelle lecture plus prioritaire → remplacer + if new_priority < old_priority: + + cur.execute(f""" + UPDATE `{detail_table}` + SET + status = 0, + reason_code = 'REPLACED_BY_HIGHER_PRIORITY', + reason_ref_id = %s + WHERE detail_id = %s + """, (det_id, old_id)) + + cnx.commit() + + forced_tour = old_tour + return 1, None, None, forced_tour + + # Même priorité → rejet + return 0, "EQUAL_PRIORITY", det_id, None diff --git a/engine/engine_components/tp_config.py b/engine/engine_components/tp_config.py index 924c33d..697d38d 100644 --- a/engine/engine_components/tp_config.py +++ b/engine/engine_components/tp_config.py @@ -2,8 +2,96 @@ 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. -def load_tp_config_for_con(cur, con_id: int) -> Dict[int, Dict[str, Dict[str, List[Dict[str, Any]]]]]: + Structure produite (identique à ton ancienne fonction) : + { + tp_id: { + "tp_name": "...", + "config": { + : { + : [ + { + 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.