275 lines
9.4 KiB
Python
275 lines
9.4 KiB
Python
# 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("🌡️ Tracker - Gestion du parc de sondes")
|
||
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()
|
||
|
||
# Bouton de déconnexion (EXIT)
|
||
st.sidebar.divider()
|
||
st.sidebar.subheader("Sécurité")
|
||
if st.sidebar.button("EXIT / Déconnexion", type="primary"):
|
||
# Efface l'état de session Streamlit
|
||
for _k in list(st.session_state.keys()):
|
||
try:
|
||
del st.session_state[_k]
|
||
except Exception:
|
||
pass
|
||
# Redirige vers /logout (géré par Nginx pour renvoyer 401 et re‑demander l'auth Basic)
|
||
st.markdown('<meta http-equiv="refresh" content="0; url=/logout">', unsafe_allow_html=True)
|
||
st.stop()
|
||
|
||
# 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."
|
||
)
|