from engine.engine_components.tp_config import load_tp_config_for_con, resolve_tp_for_lecture from engine.engine_components.course_config import load_course_config from engine.engine_components.engine_state import get_last_processed_read_id, update_last_processed_read_id from engine.engine_components.lectures import get_new_lectures from engine.engine_components.classement import get_classement_by_tag from engine.engine_components.detail_writer import insert_detail_record from engine.logger import log_info, log_debug 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 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) raw_start = course_cfg["start_time"] raw_end = course_cfg["end_time"] start_local, start_utc = _normalize_course_time(raw_start) end_local, end_utc = _normalize_course_time(raw_end) engine_table = f"resultv2_{con_id}_lecture" classement_table = f"resultv2_{con_id}_classement" # ------------------------------------------------- # 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)}") if not lectures: return 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"]) max_id = max(max_id, rcourse_id) 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}") # ------------------------------------------------- # 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 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 status = 1 reason_code = None reason_ref_id = None # ------------------------------------------------- # 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: continue dossard = coureur["cla_info_dossard"] # ------------------------------------------------- # 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"] controller = tp_info["controller_name"] det_id = tp_info["det_id"] 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", {}) .get("clef_temp_minimum_read_entre", []) ) interval_sec = 0 if interval_cfg: try: interval_sec = int(interval_cfg[0]["value1"]) except: interval_sec = 0 last_dt = _last_valid_read.get((tp_id, dossard)) 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") # ------------------------------------------------- # 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: tempstour = timestamp - prev_ts 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_utc, status=status, reason_code=reason_code, reason_ref_id=reason_ref_id, tour=tour, tempstour=tempstour, tempstour_str=tempstour_str ) # ------------------------------------------------- # 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): cur.execute(f""" SELECT tour, timestamp_read FROM `{detail_table}` WHERE tp_id = %s AND cla_info_dossard = %s AND status = 1 ORDER BY detail_id DESC LIMIT 1 """, (tp_id, dossard)) row = cur.fetchone() if not row or row["tour"] is None: return 1, None last_tour = row["tour"] 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