diff --git a/app/magasin_sondes.py b/app/magasin_sondes.py new file mode 100644 index 0000000..7874379 --- /dev/null +++ b/app/magasin_sondes.py @@ -0,0 +1,260 @@ +# tracker_app.py +# ------------------------------------------------------------- +# Streamlit — Gestion de la table MySQL Sondes.tracker +# ------------------------------------------------------------- +# Hypothèses colonnes existantes : id (PK), address, lieu, res_bits, date (timestamp) +# ------------------------------------------------------------- + +import os +import re +import time +import pandas as pd +import streamlit as st +import mysql.connector as mysql +from contextlib import contextmanager +from dotenv import load_dotenv + +# ========================== +# Configuration / Constantes +# ========================== +TABLE_DB = os.getenv("MYSQL_DB", "Sondes") +TABLE_NAME = "tracker" +COL_ID = "id" +COL_ADDRESS = "address" +COL_LIEU = "lieu" +COL_RESBITS = "res_bits" +COL_DATE = "date" + +# Regex d'une ROM code DS18B20 au format {0x28,0xFF,0xAA,0xBB,0xCC,0xDD,0xEE,0xCRC} +ROM_REGEX = re.compile(r"^\{(?:0x[0-9A-Fa-f]{2},){7}0x[0-9A-Fa-f]{2}\}$") + +# Mapping résolution DS18B20 (bits -> infos) +RES_MAP = { + 9: {"precision": 0.5, "tconv_ms": 94}, + 10: {"precision": 0.25, "tconv_ms": 188}, + 11: {"precision": 0.125, "tconv_ms": 375}, + 12: {"precision": 0.0625,"tconv_ms": 750}, +} + +# ================== +# Accès Base de Données +# ================== +load_dotenv() # lit .env si présent + +DB_CFG = dict( + host=os.getenv("DB_HOST"), + user=os.getenv("DB_USER"), + password=os.getenv("DB_PASSWORD"), + database=os.getenv("DB_NAME"), + port=int(os.getenv("MYSQL_PORT", "3306")), +) + +@contextmanager +def get_conn(): + conn = mysql.connect(**DB_CFG) + try: + yield conn + finally: + conn.close() + +# -------------- +# Fonctions SQL +# -------------- + +def fetch_trackers(where_lieu: str | None = None) -> pd.DataFrame: + query = f"SELECT {COL_ID}, {COL_ADDRESS}, {COL_LIEU}, {COL_RESBITS}, {COL_DATE} FROM {TABLE_NAME}" + params = [] + if where_lieu: + query += f" WHERE {COL_LIEU} = %s" + params.append(where_lieu) + query += f" ORDER BY {COL_LIEU}, {COL_ADDRESS}" + with get_conn() as conn: + df = pd.read_sql(query, conn, params=params) + return df + + +def insert_tracker(address: str, lieu: str, res_bits: int) -> int: + sql = f""" + INSERT INTO {TABLE_NAME} ({COL_ADDRESS}, {COL_LIEU}, {COL_RESBITS}) + VALUES (%s, %s, %s) + """ + with get_conn() as conn: + cur = conn.cursor() + cur.execute(sql, (address, lieu, res_bits)) + conn.commit() + return cur.lastrowid + + +def update_tracker(row_id: int, address: str, lieu: str, res_bits: int) -> None: + sql = f""" + UPDATE {TABLE_NAME} + SET {COL_ADDRESS}=%s, {COL_LIEU}=%s, {COL_RESBITS}=%s + WHERE {COL_ID}=%s + """ + with get_conn() as conn: + cur = conn.cursor() + cur.execute(sql, (address, lieu, res_bits, row_id)) + conn.commit() + + +def delete_tracker(row_id: int) -> None: + sql = f"DELETE FROM {TABLE_NAME} WHERE {COL_ID}=%s" + with get_conn() as conn: + cur = conn.cursor() + cur.execute(sql, (row_id,)) + conn.commit() + +# -------------- +# Utilitaires UI +# -------------- + +def is_valid_rom(address: str) -> bool: + return bool(ROM_REGEX.match(address.strip())) + + +def rom_help() -> str: + return ( + "Format attendu : `{0x28,0xFF,0xAA,0xBB,0xCC,0xDD,0xEE,0x12}` (8 octets en hex).\n" + "Le premier octet (famille) est souvent 0x28 pour DS18B20." + ) + + +def res_label(bits: int) -> str: + info = RES_MAP.get(bits) + if not info: + return f"{bits} bits (inconnu)" + return f"{bits} bits (±{info['precision']}°C, {info['tconv_ms']} ms)" + + +# ================== +# Application Streamlit +# ================== + +st.set_page_config(page_title="Gestion des sondes — tracker", page_icon="🌡️", layout="wide") + +st.title("🌡️ Gestion de Sondes.tracker") +with st.expander("Paramètres de connexion (lecture seule)"): + st.write({k: ("***" if k in {"password"} else v) for k, v in DB_CFG.items()}) + st.caption("Configurez ces valeurs via le fichier .env (MYSQL_HOST, MYSQL_DB, MYSQL_USER, MYSQL_PASSWORD, MYSQL_PORT)") + +# Barre latérale — Filtres et actions +st.sidebar.header("Filtres & Actions") + +# Récupération de la liste des lieux existants +_all = fetch_trackers() +lieux = sorted([x for x in _all[COL_LIEU].dropna().unique()]) if not _all.empty else [] +lieu_selected = st.sidebar.selectbox("Filtrer par lieu", options=["(Tous)"] + lieux, index=0) + +# Boutons d'export +col_exp1, col_exp2 = st.sidebar.columns(2) +with col_exp1: + if st.button("Exporter CSV"): + st.session_state["export_csv"] = True +with col_exp2: + if st.button("Recharger"): + st.cache_data.clear() + st.rerun() + +# Formulaire d'ajout +st.sidebar.subheader("Ajouter une sonde") +with st.sidebar.form("add_form", clear_on_submit=True): + new_address = st.text_input("Adresse ROM", placeholder="{0x28,0xFF,...}", help=rom_help()) + new_lieu = st.text_input("Lieu d'installation", placeholder="Ex: Chaufferie / Saclay") + new_res = st.selectbox("Résolution (bits)", options=[9,10,11,12]) + submitted = st.form_submit_button("Ajouter") + if submitted: + if not is_valid_rom(new_address): + st.warning("Adresse ROM invalide. Voir l'aide sur le format.") + elif not new_lieu.strip(): + st.warning("Lieu requis.") + else: + rid = insert_tracker(new_address.strip(), new_lieu.strip(), int(new_res)) + st.success(f"Sonde ajoutée (id={rid}).") + time.sleep(0.6) + st.rerun() + +# Vue principale (liste / édition) + +if lieu_selected != "(Tous)": + df = fetch_trackers(where_lieu=lieu_selected) +else: + df = _all.copy() + +if df.empty: + st.info("Aucune sonde enregistrée.") +else: + # Colonne lisible pour la résolution + df["resolution"] = df[COL_RESBITS].apply(res_label) + + st.subheader("Enregistrements") + st.caption("Double-cliquez pour éditer les cellules. Les colonnes *resolution* et *date* sont dérivées et non éditables.") + + edited = st.data_editor( + df[[COL_ID, COL_ADDRESS, COL_LIEU, COL_RESBITS, "resolution", COL_DATE]], + hide_index=True, + column_config={ + COL_ID: st.column_config.NumberColumn("ID", disabled=True), + COL_ADDRESS: st.column_config.TextColumn("Adresse (ROM)", help=rom_help()), + COL_LIEU: st.column_config.TextColumn("Lieu"), + COL_RESBITS: st.column_config.NumberColumn("Résolution (bits)", min_value=9, max_value=12, step=1), + "resolution": st.column_config.TextColumn("Détails", disabled=True), + COL_DATE: st.column_config.DatetimeColumn("Date insertion", disabled=True, format="YYYY-MM-DD HH:mm:ss"), + }, + use_container_width=True, + num_rows="dynamic", + ) + + # Détection des suppressions (si des lignes ont disparu) + removed_ids = set(df[COL_ID]) - set(edited[COL_ID]) + # Détection des modifications cellule par cellule + to_update = [] + for _, row in edited.iterrows(): + orig = df.loc[df[COL_ID] == row[COL_ID]].iloc[0] + changed = ( + (row[COL_ADDRESS] != orig[COL_ADDRESS]) or + (row[COL_LIEU] != orig[COL_LIEU]) or + (int(row[COL_RESBITS]) != int(orig[COL_RESBITS])) + ) + if changed: + to_update.append( + (int(row[COL_ID]), str(row[COL_ADDRESS]), str(row[COL_LIEU]), int(row[COL_RESBITS])) + ) + + # Validation des adresses modifiées + invalid_ids = [rid for (rid, addr, _, _) in to_update if not is_valid_rom(addr)] + if invalid_ids: + st.error(f"Adresses ROM invalides pour id: {sorted(invalid_ids)}. Aucune mise à jour effectuée.") + else: + col1, col2, col3 = st.columns([1,1,2]) + with col1: + if st.button("Enregistrer les modifications", disabled=(len(to_update)==0 and len(removed_ids)==0)): + for rid, addr, lieu, rbits in to_update: + update_tracker(rid, addr, lieu, rbits) + for rid in removed_ids: + delete_tracker(int(rid)) + st.success("Modifications enregistrées ✔️") + time.sleep(0.6) + st.rerun() + with col2: + if st.button("Annuler les modifs"): + st.rerun() + with col3: + st.caption(f"À enregistrer : {len(to_update)} mise(s) à jour, {len(removed_ids)} suppression(s)") + + # Export CSV si demandé + if st.session_state.get("export_csv"): + out = edited.drop(columns=["resolution"]).copy() + st.download_button( + label="Télécharger CSV", + data=out.to_csv(index=False).encode("utf-8"), + file_name="trackers_export.csv", + mime="text/csv", + ) + st.session_state["export_csv"] = False + +# Pied de page +st.divider() +st.caption( + "Astuce : vous pouvez coller directement une adresse ROM depuis vos logs au format {0x..,0x..,0x..,0x..,0x..,0x..,0x..,0x..}.\n" + "Si vos noms de colonnes diffèrent (ex. 'res.bits'), ajustez COL_RESBITS au début du fichier." +)