# -*- coding: utf-8 -*- import streamlit as st import mysql.connector import pandas as pd import matplotlib.pyplot as plt import matplotlib.dates as mdates import os from dotenv import load_dotenv from datetime import datetime, date import bcrypt import traceback # Charger les variables d'environnement load_dotenv() st.set_page_config(page_title="Domo91 - Surveillance", layout="wide") # Initialisation session state avec valeurs sûres for key, default in { "authenticated": False, "role": None, "lieu_autorise": None, "onglet_actif": "Accueil", "selected_date": date.today(), "selected_site": "Saclay", "selected_periode": "Toute la journée", }.items(): st.session_state.setdefault(key, default) st.title("📡 Supervision Températures") # Configuration MySQL db_config = { "host": os.getenv("DB_HOST"), "user": os.getenv("DB_USER"), "password": os.getenv("DB_PASSWORD"), "database": os.getenv("DB_NAME") } def get_connection(): return mysql.connector.connect(**db_config) def hash_password(plain_password): return bcrypt.hashpw(plain_password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') def verifier_password(input_password, hash_en_base): return bcrypt.checkpw(input_password.encode('utf-8'), hash_en_base.encode('utf-8')) # --- Connexion utilisateur --- if not st.session_state["authenticated"]: login = st.sidebar.text_input("Nom d'utilisateur") password = st.sidebar.text_input("Mot de passe", type="password") if st.sidebar.button("Se connecter"): try: conn = get_connection() cursor = conn.cursor(dictionary=True) cursor.execute("SELECT * FROM MotsDePasse WHERE utilisateur = %s", (login,)) result = cursor.fetchone() if result and verifier_password(password, result["mot_de_passe"]): if result["Expiration"] and result["Expiration"] < date.today(): st.sidebar.error("⛔ Accès expiré.") cursor.close() conn.close() st.stop() st.session_state.update({ "authenticated": True, "role": result["role"], "lieu_autorise": result["Lieu"] }) now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") cursor.execute("INSERT INTO Connexion_Log (Utilisateur, Lieu, Date_Connexion) VALUES (%s, %s, %s)", (login, result["Lieu"], now_str)) conn.commit() st.rerun() else: st.sidebar.error("Identifiants invalides") cursor.close() conn.close() except Exception as e: st.sidebar.error(f"Erreur connexion : {e}") else: st.sidebar.success(f"Connecté ({st.session_state['role']})") if st.sidebar.button("🔓 Déconnexion"): for key in ["authenticated", "role", "lieu_autorise"]: st.session_state[key] = False if key == "authenticated" else None st.rerun() # --- Navigation --- if st.session_state["authenticated"]: onglets = ["Accueil", "Entretien"] if st.session_state["role"] != "superviseur" else ["Accueil", "Statistiques", "Entretien", "Traffic", "Utilisateurs"] onglet_selectionne = st.sidebar.radio("📁 Navigation", onglets, index=onglets.index(st.session_state["onglet_actif"])) st.session_state["onglet_actif"] = onglet_selectionne site_actuel = st.session_state.get("lieu_autorise") if st.session_state["role"] != "superviseur" else st.session_state.get("selected_site", "Saclay") date_selectionnee = st.session_state.get("selected_date", date.today()) periode_selectionnee = st.session_state.get("selected_periode", "Toute la journée") # --- Onglet Accueil --- if onglet_selectionne == "Accueil": try: conn = get_connection() cursor = conn.cursor(dictionary=True) if st.session_state["role"] == "superviseur": site_actuel = st.selectbox("📍 Choisissez un site :", ["Saclay", "Meudon"], index=0) st.session_state["selected_site"] = site_actuel else: st.info(f"Site imposé : {site_actuel}") date_selectionnee = st.date_input("📅 Date du relevé", value=date_selectionnee) st.session_state["selected_date"] = date_selectionnee cursor.execute(f"SELECT * FROM `{site_actuel}` WHERE DATE(Date) = %s ORDER BY Sonde, Date DESC", (date_selectionnee.strftime("%Y-%m-%d"),)) rows = cursor.fetchall() if rows: df = pd.DataFrame(rows) df["Date"] = pd.to_datetime(df["Date"]) sondes = sorted(df["Sonde"].unique()) sonde_choisie = st.selectbox("🧪 Choisissez une sonde :", sondes) df_sonde = df[df["Sonde"] == sonde_choisie].copy() df_sonde["Heure"] = df_sonde["Date"].dt.hour tranche = st.radio("🕒 Tranche horaire :", ["Toute la journée", "Matin (6h-12h)", "Après-midi (12h-18h)", "Nuit (18h-6h)"]) st.session_state["selected_periode"] = tranche if tranche == "Matin (6h-12h)": df_sonde = df_sonde[(df_sonde["Heure"] >= 6) & (df_sonde["Heure"] < 12)] elif tranche == "Après-midi (12h-18h)": df_sonde = df_sonde[(df_sonde["Heure"] >= 12) & (df_sonde["Heure"] < 18)] elif tranche == "Nuit (18h-6h)": df_sonde = df_sonde[(df_sonde["Heure"] >= 18) | (df_sonde["Heure"] < 6)] seuil_temp = 10 cursor.execute("SELECT Temp_Max FROM Chambres_froides WHERE Lieu = %s AND Sonde = %s", (site_actuel, sonde_choisie)) seuil = cursor.fetchone() if seuil: seuil_temp = seuil["Temp_Max"] st.subheader("📊 Tableau des relevés") def surlignage_temp(val): try: if float(val) > seuil_temp: return "color: red; font-weight: bold" except: pass return "" styled_df = df_sonde.style.applymap(surlignage_temp, subset=["Temperature"]) st.dataframe(styled_df, use_container_width=True) st.subheader("📈 Évolution de la température") fig, ax = plt.subplots(figsize=(10, 4)) ax.plot(df_sonde["Date"], df_sonde["Temperature"], marker='o') ax.axhline(seuil_temp, color='red', linestyle='--', label=f"Seuil {seuil_temp}°C") ax.set_xlabel("Heure") ax.set_ylabel("Température (°C)") ax.set_title(f"{sonde_choisie} - {date_selectionnee.strftime('%d/%m/%Y')}") ax.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M')) ax.legend() st.pyplot(fig) cursor.close() conn.close() except Exception as e: st.error(f"Erreur : {e}") st.text(traceback.format_exc()) # --- Onglet Statistiques --- elif onglet_selectionne == "Statistiques": st.markdown("## 📈 Statistiques de température") try: conn = mysql.connector.connect(**db_config) cursor = conn.cursor(dictionary=True) site = ( st.session_state["lieu_autorise"] if st.session_state["role"] != "superviseur" else st.session_state.get("selected_site", "Saclay") ) date_val = st.session_state.get("selected_date", date.today()) cursor.execute( f"SELECT * FROM `{site}` WHERE DATE(Date) = %s ORDER BY Sonde, Date", (date_val.strftime("%Y-%m-%d"),) ) rows = cursor.fetchall() df = pd.DataFrame(rows) if df.empty: st.info("Aucune donnée pour cette date.") else: df["Date"] = pd.to_datetime(df["Date"]) sondes = sorted(df["Sonde"].unique()) sonde = st.selectbox("Choisir une sonde :", sondes, key="selectbox_stats") df_sonde = df[df["Sonde"] == sonde] st.subheader("Évolution journalière") fig, ax = plt.subplots(figsize=(10, 4)) ax.plot(df_sonde["Date"], df_sonde["Temperature"], marker='o') ax.set_title(f"{sonde} - {date_val.strftime('%d/%m/%Y')}") ax.set_xlabel("Heure") ax.set_ylabel("Température (°C)") ax.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M')) st.pyplot(fig) cursor.close() conn.close() except Exception as e: st.error(f"Erreur chargement statistiques : {e}") # --- Onglet Entretien --- elif onglet_selectionne == "Entretien": st.header("🧰 Gestion Entretien") try: conn = get_connection() cursor = conn.cursor(dictionary=True) cursor.execute("SELECT Id, Sonde, En_entretien FROM Chambres_froides WHERE Lieu = %s", (site_actuel,)) sondes = cursor.fetchall() for sonde in sondes: checked = st.checkbox(f"{sonde['Sonde']}", value=sonde['En_entretien']) if checked != sonde['En_entretien']: cursor.execute("UPDATE Chambres_froides SET En_entretien = %s WHERE Id = %s", (checked, sonde['Id'])) conn.commit() st.success(f"{sonde['Sonde']} {'mise' if checked else 'retirée'} en entretien.") cursor.close() conn.close() except Exception as e: st.error(f"Erreur : {e}") st.text(traceback.format_exc()) # --- Onglet Traffic --- elif onglet_selectionne == "Traffic": st.header("🚦 Connexions récentes") try: conn = get_connection() cursor = conn.cursor(dictionary=True) cursor.execute("SELECT Utilisateur, Lieu, Date_Connexion FROM Connexion_Log ORDER BY Date_Connexion DESC LIMIT 100") logs = cursor.fetchall() df_logs = pd.DataFrame(logs) st.dataframe(df_logs) cursor.close() conn.close() except Exception as e: st.error(f"Erreur : {e}") st.text(traceback.format_exc()) # --- Onglet Utilisateurs --- elif onglet_selectionne == "Utilisateurs": st.header("👥 Gestion des utilisateurs") with st.form("ajouter_utilisateur"): new_user = st.text_input("Nom d'utilisateur") new_pass = st.text_input("Mot de passe", type="password") new_role = st.selectbox("Rôle", ["utilisateur", "superviseur"]) new_lieu = st.selectbox("Lieu", ["Saclay", "Meudon", "Roissy"]) if st.form_submit_button("Ajouter"): try: conn = get_connection() cursor = conn.cursor() hash_mdp = hash_password(new_pass) cursor.execute("INSERT INTO MotsDePasse (utilisateur, mot_de_passe, role, Lieu) VALUES (%s, %s, %s, %s)", (new_user, hash_mdp, new_role, new_lieu)) conn.commit() cursor.close() conn.close() st.success("Utilisateur ajouté.") except Exception as e: st.error(f"Erreur : {e}") st.text(traceback.format_exc())