# -*- 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, time from fpdf import FPDF import bcrypt import traceback import random # Charger les variables d'environnement load_dotenv() st.set_page_config(page_title="Domo91 - Surveillance", layout="wide") st.title("📊 Domo91 - Surveillance des sondes") st.write("Bienvenue sur l’application de supervision.") # Initialisation session state avec valeurs sĂ»res for key, default in { "authenticated": False, "role": None, "site_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) # Configuration MySQL db_config = { "host": os.getenv("DB_HOST"), "user": os.getenv("DB_USER"), "password": os.getenv("DB_PASS"), "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.get("authenticated", False): 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) # On interroge la bonne table cursor.execute(""" SELECT NomUtilisateur, role, MotDePasseHash, Site, DateExpiration FROM Acces.Utilisateurs WHERE NomUtilisateur = %s LIMIT 1 """, (login,)) result = cursor.fetchone() if not result: st.sidebar.error("Identifiants invalides") elif result["DateExpiration"] and result["DateExpiration"] < date.today(): st.sidebar.error("⛔ AccĂšs expirĂ©.") elif not verifier_password(password, result["MotDePasseHash"]): st.sidebar.error("Identifiants invalides") else: # Authentification rĂ©ussie st.session_state.update({ "authenticated": True, "role": result["role"], "site_autorise": result["Site"] }) now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") cursor.execute(""" INSERT INTO Sondes.Connexion_Log (Utilisateur, Lieu, Date_Connexion) VALUES (%s, %s, %s) """, (result["NomUtilisateur"], result["Site"], now_str)) conn.commit() st.rerun() cursor.close() conn.close() except Exception as e: st.sidebar.error(f"Erreur connexion : {e}") else: st.sidebar.success(f"ConnectĂ© ({st.session_state.get('role')})") if st.sidebar.button("🔓 DĂ©connexion"): for key in ["authenticated", "role", "site_autorise"]: st.session_state[key] = False if key == "authenticated" else None st.rerun() # --- Fonction de gĂ©nĂ©ration PDF --- def generer_pdf(site, date_str, periode): st.info(f"GĂ©nĂ©ration du rapport PDF pour {site} Ă  la date {date_str} ({periode})") try: conn = mysql.connector.connect(**db_config) pdf_cursor = conn.cursor(dictionary=True) # RequĂȘte principale pdf_cursor.execute(f"SELECT Sonde, Date, Temperature FROM `{site}` WHERE DATE(Date) = %s ORDER BY Sonde, Date", (date_str,)) rows = pdf_cursor.fetchall() df = pd.DataFrame(rows) df["Heure"] = pd.to_datetime(df["Date"]).dt.strftime("%H:%M") df["Heure_obj"] = pd.to_datetime(df["Date"]).dt.time # --- Filtrage par pĂ©riode --- plages = { "Toute la journĂ©e": (time(0, 0), time(23, 59)), "Matin (6h-12h)": (time(6, 0), time(12, 0)), "AprĂšs-midi (12h-18h)": (time(12, 0), time(18, 0)), "Nuit (18h-6h)": (time(18, 0), time(6, 0)), } heure_debut, heure_fin = plages.get(periode, (time(0, 0), time(23, 59))) if heure_debut < heure_fin: df = df[(df["Heure_obj"] >= heure_debut) & (df["Heure_obj"] <= heure_fin)] else: df = df[(df["Heure_obj"] >= heure_debut) | (df["Heure_obj"] <= heure_fin)] # --- Structuration des relevĂ©s --- releves = {} for sonde in df["Sonde"].unique(): df_sonde = df[df["Sonde"] == sonde] releves[sonde] = list(zip(df_sonde["Heure"], df_sonde["Temperature"])) # --- RequĂȘte alertes --- table_alertes = f"Alertes_{site}" pdf_cursor.execute(f"SELECT Sonde, Debut_defaut, Etat FROM {table_alertes} WHERE DATE(Debut_defaut) = %s", (date_str,)) alertes = pdf_cursor.fetchall() pdf_cursor.close() conn.close() # --- Classe PDF --- class RapportPDF(FPDF): def header(self): self.set_font("Arial", "B", 14) self.cell(0, 10, "Rapport de surveillance des sondes", ln=1, align="C") self.set_font("Arial", "", 12) self.cell(0, 10, f"Date : {date_str}", ln=1, align="C") self.cell(0, 10, f"Periode : {getattr(self, 'periode', '')}", ln=1, align="C") self.ln(5) def site_info(self, site_name): self.set_font("Arial", "B", 12) self.cell(0, 10, f"Site : {site_name}", ln=1) self.ln(2) def releves_section(self, data): self.set_font("Arial", "B", 12) self.cell(0, 10, "RelevĂ©s de tempĂ©rature", ln=1) for sonde, mesures in data.items(): self.set_font("Arial", "B", 11) self.cell(0, 8, f"Sonde : {sonde}", ln=1) col1 = mesures[::2] col2 = mesures[1::2] self.set_font("Arial", "B", 10) self.cell(40, 6, "Heure", border=1) self.cell(30, 6, "Temp (°C)", border=1) self.cell(20, 6, "", border=0) self.cell(40, 6, "Heure", border=1) self.cell(30, 6, "Temp (°C)", border=1) self.ln() self.set_font("Arial", "", 10) for i in range(max(len(col1), len(col2))): if i < len(col1): h1, t1 = col1[i] self.cell(40, 6, h1, border=1) self.cell(30, 6, f"{t1:.2f}", border=1) else: self.cell(70, 6, "", border=0) self.cell(20, 6, "", border=0) if i < len(col2): h2, t2 = col2[i] self.cell(40, 6, h2, border=1) self.cell(30, 6, f"{t2:.2f}", border=1) self.ln() self.ln(4) def alertes_section(self, data): self.set_font("Arial", "B", 12) self.cell(0, 10, "Alertes enregistrĂ©es", ln=1) self.set_font("Arial", "", 10) for a in data: self.cell(0, 6, f"{a['Sonde']} - {a['Debut_defaut']} - {a['Status']}", ln=1) # --- GĂ©nĂ©ration du PDF --- pdf = RapportPDF() pdf.periode = periode pdf.add_page() pdf.site_info(site) pdf.releves_section(releves) pdf.alertes_section(alertes) file_name = f"rapport_{site}_{date_str}.pdf" output_dir = "PDF" os.makedirs(output_dir, exist_ok=True) output_path = os.path.join(output_dir, file_name) pdf.output(output_path) with open(output_path, "rb") as f: st.download_button( label="đŸ“„ TĂ©lĂ©charger le rapport PDF", data=f, file_name=file_name, mime="application/pdf" ) except Exception as err1: st.error(f"Erreur lors de la gĂ©nĂ©ration du PDF : {err1}") # --- AFFICHAGE GLOBAL DES ALERTES NON ACQUITTÉES --- if st.session_state.get("authenticated"): try: role = st.session_state.get("role") # si superviseur → utilise selected_site (par dĂ©faut Saclay) # sinon → utilise site_autorise (imposĂ© Ă  la connexion) site_selectionne = ( st.session_state.get("site_autorise") if role != "superviseur" else st.session_state.get("selected_site", "Saclay") ) if not site_selectionne: # pas de site → on n’essaie pas de lire la table st.info("Connectez-vous et choisissez un site pour afficher les alertes.") else: conn = mysql.connector.connect(**db_config) cursor = conn.cursor(dictionary=True) # Optionnel: whitelist pour Ă©viter toute injection sur le nom de table SITES_AUTORISES = {"Saclay", "Meudon", "Roissy"} if site_selectionne not in SITES_AUTORISES: raise ValueError(f"Site invalide: {site_selectionne}") table_alertes = f"Alertes_{site_selectionne}" cursor.execute( f"SELECT Sonde, Debut_defaut, Etat " f"FROM `{table_alertes}` " f"WHERE Etat != 'AcquittĂ©' " f"ORDER BY Debut_defaut DESC" ) alertes = cursor.fetchall() if alertes: df_alertes = pd.DataFrame(alertes) st.subheader("🚹 Alertes non acquittĂ©es") st.dataframe(df_alertes, use_container_width=True) else: st.success("✅ Aucune alerte en cours.") cursor.close() conn.close() except Exception as e: st.error(f"Erreur lors de la rĂ©cupĂ©ration des alertes : {e}") else: st.info("Connectez-vous pour voir les alertes en cours.") # --- Navigation --- if st.session_state["authenticated"]: onglets = ["Accueil", "Entretien"] if st.session_state["role"] != "superviseur" else ["Accueil", "Statistiques", "Entretien", "Traffic"] 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("site_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 # Bouton de gĂ©nĂ©ration du PDF if st.button("đŸ§Ÿ GĂ©nĂ©rer le PDF du jour"): generer_pdf( site_actuel, date_selectionnee.strftime("%Y-%m-%d"), st.session_state.get("selected_periode", "Toute la journĂ©e") ) 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 Sondes.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") site = ( st.session_state["site_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()) try: conn = mysql.connector.connect(**db_config) cursor = conn.cursor(dictionary=True) site = ( st.session_state["site_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}") # Tableau consignes chambres froides if st.session_state["role"] == "superviseur": with st.expander("đŸ› ïž Gestion des chambres froides (administrateur)", expanded=True): if st.button("🔄 Actualiser la liste"): st.session_state["refresh_admin"] = random.randint(0, 9999) try: conn_admin = mysql.connector.connect(**db_config) cursor_admin = conn_admin.cursor(dictionary=True) cursor_admin.execute("SELECT * FROM Sondes.Chambres_froides WHERE Lieu = %s", (site,)) chambres = cursor_admin.fetchall() if not chambres: st.warning("Aucune chambre froide pour ce site.") else: for chambre in chambres: col1, col2, col3 = st.columns([3, 1, 2]) with col1: st.markdown(f"**{chambre['Sonde']}**") with col2: etat = st.checkbox("ON", value=(chambre["Etat"] == "ON"), key=f"etat_{chambre['Id']}_{st.session_state.get('refresh_admin', 0)}") new_etat = "ON" if etat else "OFF" with col3: temp_max = chambre["Temp_Max"] moins, temp_display, plus = st.columns([1, 2, 1]) with moins: if st.button("â–Œ", key=f"moins_{chambre['Id']}"): temp_max -= 1 with temp_display: st.markdown(f"
{temp_max}°C
", unsafe_allow_html=True) with plus: if st.button("â–Č", key=f"plus_{chambre['Id']}"): temp_max += 1 if new_etat != chambre["Etat"] or temp_max != chambre["Temp_Max"]: cursor_admin.execute( "UPDATE Sondes.Chambres_froides SET Etat = %s, Temp_Max = %s WHERE Id = %s", (new_etat, temp_max, chambre["Id"]) ) conn_admin.commit() st.success(f"{chambre['Sonde']} mise Ă  jour") cursor_admin.close() conn_admin.close() except Exception as e: st.error(f"Erreur SQL (admin) : {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 Sondes.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 Sondes.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 Sondes.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())