Ajout de l'app gestion des sondes
This commit is contained in:
260
app/magasin_sondes.py
Normal file
260
app/magasin_sondes.py
Normal file
@@ -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."
|
||||
)
|
||||
Reference in New Issue
Block a user