372 lines
13 KiB
Python
372 lines
13 KiB
Python
# tracker.py
|
||
# -------------------------------------------------------------
|
||
# Streamlit — Gestion de la table MySQL Sondes.tracker (avec address_hyphen)
|
||
# -------------------------------------------------------------
|
||
# Schéma attendu :
|
||
# id (PK), address (ROM {0x..}), address_hyphen (28-..-..-..-..-..-..-..),
|
||
# lieu, repere, mise_en_service (DATE), res_bits, date (timestamp)
|
||
# Authentification intégrée via .env (AUTH_USERS JSON)
|
||
# -------------------------------------------------------------
|
||
import re
|
||
import time
|
||
import json
|
||
import hmac
|
||
from typing import Optional
|
||
from datetime import date
|
||
|
||
import pandas as pd
|
||
import streamlit as st
|
||
import mysql.connector as mysql
|
||
from contextlib import contextmanager
|
||
from dotenv import load_dotenv
|
||
import os
|
||
from pathlib import Path
|
||
|
||
# 1) .env (avec recherche automatique depuis le fichier courant vers la racine)
|
||
try:
|
||
from dotenv import load_dotenv, find_dotenv
|
||
load_dotenv(find_dotenv(usecwd=True), override=False)
|
||
except Exception:
|
||
pass # si python-dotenv n'est pas installé, on passe (Streamlit peut fournir st.secrets)
|
||
|
||
# 2) Streamlit secrets (optionnel si tu l’utilises)
|
||
try:
|
||
import streamlit as st
|
||
st_secrets_mysql = st.secrets.get("mysql", {})
|
||
except Exception:
|
||
st_secrets_mysql = {}
|
||
|
||
def get_db_cfg():
|
||
# Priorité à st.secrets si dispo, sinon variables d’environnement (.env)
|
||
cfg = {
|
||
"host": st_secrets_mysql.get("host") or os.getenv("DB_HOST"),
|
||
"user": st_secrets_mysql.get("user") or os.getenv("DB_USER"),
|
||
"password": st_secrets_mysql.get("password") or os.getenv("DB_PASSWORD"),
|
||
"database": st_secrets_mysql.get("database") or os.getenv("DB_NAME") or os.getenv("DB_DATABASE"),
|
||
"port": int(st_secrets_mysql.get("port") or os.getenv("DB_PORT") or 3306),
|
||
"auth_plugin": os.getenv("DB_AUTH_PLUGIN") or None, # optionnel
|
||
}
|
||
# Nettoyage: retire les clés None
|
||
return {k: v for k, v in cfg.items() if v not in (None, "")}
|
||
# ==========================
|
||
# Configuration / Constantes
|
||
# ==========================
|
||
load_dotenv() # lit .env si présent
|
||
|
||
TABLE_NAME = "tracker"
|
||
COL_ID = "id"
|
||
COL_ADDRESS = "address" # format ROM : {0x28,0xFF,...}
|
||
COL_ADDR_HYPHEN = "address_hyphen" # format hyphen : 28-xx-xx-xx-xx-xx-xx-xx
|
||
COL_LIEU = "lieu"
|
||
COL_REPERE = "repere"
|
||
COL_MES = "mise_en_service" # DATE
|
||
COL_RESBITS = "res_bits"
|
||
COL_DATE = "date"
|
||
|
||
# Configuration BDD (standardisée sur les variables d'env MYSQL_*)
|
||
DB_CFG = dict(
|
||
host=os.getenv("DB_HOST"),
|
||
user=os.getenv("DB_USER"),
|
||
password=os.getenv("DB_PASS"),
|
||
database=os.getenv("DB_NAME"),
|
||
port=int(os.getenv("MYSQL_PORT", "3306")),
|
||
)
|
||
|
||
# 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}}$")
|
||
# Adresse hyphen : 8 octets hexa séparés par des tirets
|
||
HYPHEN_REGEX = re.compile(r"^[0-9A-Fa-f]{2}(?:-[0-9A-Fa-f]{2}){7}$")
|
||
|
||
# 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},
|
||
}
|
||
|
||
# ==================
|
||
# Authentification via .env (AUTH_USERS)
|
||
# ==================
|
||
AUTH_USERS_RAW = os.getenv("AUTH_USERS", "[]")
|
||
|
||
def _load_users() -> dict:
|
||
try:
|
||
data = json.loads(AUTH_USERS_RAW)
|
||
return {str(d.get("user", "")).strip(): str(d.get("pass", "")) for d in data if d.get("user") and d.get("pass")}
|
||
except Exception:
|
||
return {}
|
||
|
||
USERS = _load_users()
|
||
|
||
def _constant_time_equals(a: str, b: str) -> bool:
|
||
return hmac.compare_digest(a.encode("utf-8"), b.encode("utf-8"))
|
||
|
||
def verify_credentials(username: str, password: str) -> bool:
|
||
if not username or not password:
|
||
return False
|
||
expected = USERS.get(username.strip())
|
||
if expected is None:
|
||
return False
|
||
return _constant_time_equals(password, expected)
|
||
|
||
def require_login() -> Optional[str]:
|
||
if st.session_state.get("auth_ok") and st.session_state.get("auth_user"):
|
||
return st.session_state.get("auth_user")
|
||
|
||
st.markdown("<h2 style='text-align:center;'>🔒 Tracker</h2>", unsafe_allow_html=True)
|
||
_, col2, _ = st.columns([1, 2, 1])
|
||
with col2:
|
||
with st.form("login_form", clear_on_submit=False):
|
||
username = st.text_input("Utilisateur")
|
||
password = st.text_input("Mot de passe", type="password")
|
||
ok = st.form_submit_button("Se connecter")
|
||
if ok:
|
||
if verify_credentials(username, password):
|
||
st.session_state["auth_ok"] = True
|
||
st.session_state["auth_user"] = username.strip()
|
||
st.success("Connexion réussie.")
|
||
time.sleep(0.3)
|
||
st.rerun()
|
||
else:
|
||
st.error("Identifiants invalides.")
|
||
st.stop()
|
||
|
||
# ==================
|
||
# Accès Base de Données
|
||
# ==================
|
||
@contextmanager
|
||
def get_conn():
|
||
cfg = get_db_cfg()
|
||
# Astuce debug si besoin :
|
||
# print({k: ('***' if k=='password' else v) for k,v in cfg.items()})
|
||
conn = mysql.connect(**cfg)
|
||
try:
|
||
yield conn
|
||
finally:
|
||
conn.close()
|
||
|
||
# -----------------
|
||
# Utilitaires
|
||
# -----------------
|
||
def rom_help() -> str:
|
||
return (
|
||
"Format ROM attendu : `{0x28,0xFF,0xAA,0xBB,0xCC,0xDD,0xEE,0x12}` "
|
||
"(8 octets en hex). Le premier octet est souvent 0x28 pour DS18B20."
|
||
)
|
||
|
||
def is_valid_rom(address: str) -> bool:
|
||
return bool(ROM_REGEX.match(str(address).strip()))
|
||
|
||
def is_valid_hyphen(address_h: str) -> bool:
|
||
return bool(HYPHEN_REGEX.match(str(address_h).strip()))
|
||
|
||
def rom_to_hyphen(rom: str) -> str:
|
||
hexes = re.findall(r"0x([0-9A-Fa-f]{2})", str(rom))
|
||
if len(hexes) != 8:
|
||
return ""
|
||
return "-".join(h.lower() for h in hexes)
|
||
|
||
def hyphen_to_rom(h: str) -> str:
|
||
parts = str(h).strip().split("-")
|
||
if len(parts) != 8 or not all(re.fullmatch(r"[0-9A-Fa-f]{2}", p) for p in parts):
|
||
return ""
|
||
return "{" + ",".join(f"0x{p.upper()}" for p in parts) + "}"
|
||
|
||
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)"
|
||
|
||
# -----------------
|
||
# Fonctions SQL
|
||
# -----------------
|
||
def fetch_trackers(where_lieu: str | None = None) -> pd.DataFrame:
|
||
query = (
|
||
f"SELECT {COL_ID}, {COL_ADDRESS}, {COL_ADDR_HYPHEN}, {COL_LIEU}, "
|
||
f"{COL_REPERE}, {COL_MES}, {COL_RESBITS}, {COL_DATE} "
|
||
f"FROM {TABLE_NAME}"
|
||
)
|
||
params = []
|
||
if where_lieu:
|
||
query += f" WHERE {COL_LIEU} = %s"
|
||
params.append(where_lieu)
|
||
query += f" ORDER BY {COL_LIEU}, {COL_REPERE}, {COL_ADDR_HYPHEN}, {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,
|
||
repere: str | None = None, mise_en_service: date | None = None,
|
||
address_hyphen: str | None = None) -> int:
|
||
addr_rom = (address or "").strip() if address else ""
|
||
addr_hyp = (address_hyphen or "").strip() if address_hyphen else ""
|
||
if addr_rom and not addr_hyp:
|
||
addr_hyp = rom_to_hyphen(addr_rom)
|
||
if addr_hyp and not addr_rom:
|
||
addr_rom = hyphen_to_rom(addr_hyp)
|
||
if not is_valid_rom(addr_rom) or not is_valid_hyphen(addr_hyp):
|
||
raise ValueError("Adresse invalide (ROM ou hyphen).")
|
||
sql = f"""
|
||
INSERT INTO {TABLE_NAME}
|
||
({COL_ADDRESS}, {COL_ADDR_HYPHEN}, {COL_LIEU}, {COL_REPERE}, {COL_MES}, {COL_RESBITS})
|
||
VALUES (%s, %s, %s, %s, %s, %s)
|
||
"""
|
||
with get_conn() as conn:
|
||
cur = conn.cursor()
|
||
cur.execute(sql, (
|
||
addr_rom,
|
||
addr_hyp.lower(),
|
||
lieu,
|
||
(repere.strip() if repere and str(repere).strip() else None),
|
||
mise_en_service,
|
||
res_bits,
|
||
))
|
||
conn.commit()
|
||
return cur.lastrowid
|
||
|
||
def update_tracker(row_id: int, address: str, lieu: str, res_bits: int,
|
||
repere: str | None, mise_en_service: date | None,
|
||
address_hyphen: str | None = None) -> None:
|
||
addr_rom = (address or "").strip()
|
||
addr_hyp = (address_hyphen or "").strip() if address_hyphen else ""
|
||
if addr_rom and not addr_hyp:
|
||
addr_hyp = rom_to_hyphen(addr_rom)
|
||
if addr_hyp and not addr_rom:
|
||
addr_rom = hyphen_to_rom(addr_hyp)
|
||
if not is_valid_rom(addr_rom) or not is_valid_hyphen(addr_hyp):
|
||
raise ValueError("Adresse invalide (ROM ou hyphen).")
|
||
sql = f"""
|
||
UPDATE {TABLE_NAME}
|
||
SET {COL_ADDRESS}=%s, {COL_ADDR_HYPHEN}=%s, {COL_LIEU}=%s, {COL_REPERE}=%s, {COL_MES}=%s, {COL_RESBITS}=%s
|
||
WHERE {COL_ID}=%s
|
||
"""
|
||
with get_conn() as conn:
|
||
cur = conn.cursor()
|
||
cur.execute(sql, (
|
||
addr_rom,
|
||
addr_hyp.lower(),
|
||
lieu,
|
||
(repere.strip() if repere and str(repere).strip() else None),
|
||
mise_en_service,
|
||
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()
|
||
|
||
# ==================
|
||
# Application Streamlit
|
||
# ==================
|
||
st.set_page_config(page_title="Tracker", page_icon="🌡️", layout="wide")
|
||
user = require_login()
|
||
|
||
st.title("🌡️ Gestion du parc sondes (stock ou installées)")
|
||
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")
|
||
|
||
st.sidebar.header("Filtres & Actions")
|
||
st.sidebar.caption(f"Connecté en tant que **{user}**")
|
||
|
||
_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)
|
||
|
||
# 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())
|
||
preview_h = rom_to_hyphen(new_address) if new_address else ""
|
||
st.text_input("Adresse hyphen (auto)", value=preview_h, disabled=True)
|
||
new_lieu = st.text_input("Lieu d'installation")
|
||
new_repere = st.text_input("Repère (optionnel)")
|
||
new_mes = st.date_input("Mise en service (optionnel)", value=None, format="YYYY-MM-DD")
|
||
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.")
|
||
elif not new_lieu.strip():
|
||
st.warning("Lieu requis.")
|
||
else:
|
||
rid = insert_tracker(
|
||
new_address.strip(),
|
||
new_lieu.strip(),
|
||
int(new_res),
|
||
new_repere,
|
||
new_mes if isinstance(new_mes, date) else None,
|
||
address_hyphen=rom_to_hyphen(new_address.strip()),
|
||
)
|
||
st.success(f"Sonde ajoutée (id={rid}).")
|
||
time.sleep(0.6)
|
||
st.rerun()
|
||
|
||
st.sidebar.divider()
|
||
if st.sidebar.button("Se déconnecter"):
|
||
for _k in list(st.session_state.keys()):
|
||
st.session_state.pop(_k, None)
|
||
st.success("Déconnecté.")
|
||
time.sleep(0.3)
|
||
st.rerun()
|
||
|
||
# Vue principale
|
||
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:
|
||
df["resolution"] = df[COL_RESBITS].apply(res_label)
|
||
st.subheader("Enregistrements")
|
||
edited = st.data_editor(
|
||
df[[COL_ID, COL_ADDRESS, COL_ADDR_HYPHEN, COL_LIEU, COL_REPERE, COL_MES, COL_RESBITS, "resolution", COL_DATE]],
|
||
hide_index=True,
|
||
use_container_width=True,
|
||
num_rows="dynamic",
|
||
)
|
||
removed_ids = set(df[COL_ID]) - set(edited[COL_ID])
|
||
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_ADDR_HYPHEN] != orig[COL_ADDR_HYPHEN]) or
|
||
(row[COL_LIEU] != orig[COL_LIEU]) or
|
||
(row.get(COL_REPERE) or "") != (orig.get(COL_REPERE) or "") or
|
||
(str(row.get(COL_MES) or "")[:10] != str(orig.get(COL_MES) or "")[:10]) or
|
||
(int(row[COL_RESBITS]) != int(orig[COL_RESBITS]))
|
||
)
|
||
if changed:
|
||
mes_val = row.get(COL_MES)
|
||
if pd.isna(mes_val):
|
||
mes_val = None
|
||
elif hasattr(mes_val, "date"):
|
||
mes_val = mes_val.date()
|
||
to_update.append((int(row[COL_ID]), str(row[COL_ADDRESS]), str(row[COL_ADDR_HYPHEN] or ""),
|
||
str(row[COL_LIEU]), int(row[COL_RESBITS]),
|
||
str(row.get(COL_REPERE) or None), mes_val))
|
||
col1, col2 = st.columns([1,1])
|
||
if st.button("Enregistrer les modifications"):
|
||
for rid, addr_rom, addr_hyp, lieu, rbits, repere, mes in to_update:
|
||
update_tracker(rid, addr_rom, lieu, rbits, repere, mes, address_hyphen=addr_hyp)
|
||
for rid in removed_ids:
|
||
delete_tracker(rid)
|
||
st.success("Modifications enregistrées ✔️")
|
||
time.sleep(0.6)
|
||
st.rerun()
|
||
if st.button("Annuler"):
|
||
st.rerun()
|
||
|
||
st.divider()
|
||
st.caption("Astuce : collez une adresse ROM {0x..,...} → la version hyphen est générée automatiquement.")
|