# Application Gestion de sondes import streamlit as st import mysql.connector import pandas as pd from datetime import date import matplotlib.pyplot as plt import matplotlib.dates as mdates from fpdf import FPDF import os st.set_page_config(page_title="Domo91 - Surveillance", layout="wide") st.title("📡 Supervision Températures") # --- Configuration base de données --- db_config = { "host": "54.36.188.119", "user": "michel", "password": "#SO2&1nf%mZ@jfh", "database": "Sondes" } # --- Fonction de génération PDF --- def generer_pdf(site, date_str): st.info(f"Génération du rapport PDF pour {site} à la date {date_str}") try: conn = mysql.connector.connect(**db_config) cursor = conn.cursor(dictionary=True) cursor.execute(f"SELECT Sonde, Date, Temperature FROM `{site}` WHERE DATE(Date) = %s ORDER BY Sonde, Date", (date_str,)) rows = cursor.fetchall() df = pd.DataFrame(rows) df["Heure"] = pd.to_datetime(df["Date"]).dt.strftime("%H:%M") releves = {} for sonde in df["Sonde"].unique(): df_sonde = df[df["Sonde"] == sonde] releves[sonde] = list(zip(df_sonde["Heure"], df_sonde["Temperature"])) table_alertes = f"Alertes_{site}" cursor.execute(f"SELECT Sonde, Debut_defaut, Status FROM {table_alertes} WHERE DATE(Debut_defaut) = %s", (date_str,)) alertes = cursor.fetchall() cursor.close() conn.close() 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.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) self.set_font("Arial", "", 10) for heure, temp in mesures: self.cell(0, 6, f"{heure} - {temp} °C", ln=1) self.ln(2) 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) pdf = RapportPDF() 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) # 🔧 Crée le dossier si absent 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 e: st.error(f"Erreur lors de la génération du PDF : {e}") # --- Initialisation session --- if "authenticated" not in st.session_state: st.session_state["authenticated"] = False st.session_state["role"] = None st.session_state["lieu_autorise"] = None # --- Sidebar (connexion + bouton PDF) --- with st.sidebar: st.header("🔐 Connexion") if not st.session_state.get("authenticated"): login = st.text_input("Nom d'utilisateur") password = st.text_input("Mot de passe", type="password") if st.button("Se connecter"): try: conn = mysql.connector.connect(**db_config) cursor = conn.cursor(dictionary=True) cursor.execute("SELECT * FROM MotsDePasse WHERE utilisateur = %s", (login,)) result = cursor.fetchone() if result and result["mot_de_passe"] == password: st.session_state["authenticated"] = True st.session_state["role"] = result["role"] st.session_state["lieu_autorise"] = result["Lieu"] st.success(f"Connecté comme {result['role']} ({result['Lieu']})") else: st.error("Identifiants invalides") cursor.close() conn.close() except Exception as e: st.error(f"Erreur lors de la connexion à la base : {e}") else: st.success(f"Connecté ({st.session_state['role']})") if st.button("🔓 Déconnexion"): st.session_state["authenticated"] = False st.session_state["role"] = None st.session_state["lieu_autorise"] = None st.rerun() st.markdown("---") st.subheader("📄 Rapport PDF") if "selected_date" in st.session_state: if st.button("📥 Télécharger l’état du jour (PDF)"): site = st.session_state["lieu_autorise"] date_val = st.session_state["selected_date"].strftime("%Y-%m-%d") generer_pdf(site, date_val) else: st.info("Sélectionnez une date pour activer la génération PDF.") # --- CONTENU PRINCIPAL SI AUTHENTIFIÉ --- if st.session_state["authenticated"]: st.markdown("## Sélection du site et de la date") try: conn = mysql.connector.connect(**db_config) cursor = conn.cursor(dictionary=True) sites_possibles = ["Saclay", "Meudon"] if st.session_state["role"] == "superviseur": site_selectionne = st.selectbox("📍 Choisissez un site :", sites_possibles) else: site_selectionne = st.session_state["lieu_autorise"] st.info(f"Site imposé : {site_selectionne}") selected_date = st.date_input("📅 Date du relevé", value=date.today()) st.session_state["selected_date"] = selected_date cursor.execute( f"SELECT * FROM `{site_selectionne}` WHERE DATE(Date) = %s ORDER BY Sonde, Date", (selected_date.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] # Ajouter une colonne Heure pour faciliter les filtres df_sonde.loc[:, "Heure"] = df_sonde["Date"].dt.hour # Filtrage par tranche horaire tranche = st.radio("🕒 Tranche horaire :", ["Toute la journée", "Matin (6h-12h)", "Après-midi (12h-18h)", "Nuit (18h-6h)"]) 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)] df_sonde = df_sonde.copy() cursor.execute("SELECT Temp_Max FROM Chambres_froides WHERE Lieu = %s AND Sonde = %s", (site_selectionne, sonde_choisie)) seuil = cursor.fetchone() seuil_temp = seuil["Temp_Max"] if seuil else 10 st.subheader("📊 Tableau des relevés") df_filtré = df_sonde.copy() df_filtré = df_filtré.drop(columns="Id", errors="ignore") def surlignage_temp(val): try: if float(val) > seuil_temp: return "color: red; font-weight: bold" except: pass return "" # Appliquer le style uniquement à la colonne "Temperature" styled_df = df_filtré.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_filtré["Date"], df_filtré["Temperature"], marker='o', label="Température") 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} - {selected_date.strftime('%d/%m/%Y')}") ax.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M')) ax.legend() st.pyplot(fig) # --- Affichage automatique des alertes non acquittées --- try: conn = mysql.connector.connect(**db_config) cursor = conn.cursor(dictionary=True) table_alertes = f"Alertes_{site_selectionne}" cursor.execute( f"SELECT Sonde, Debut_defaut, Status FROM `{table_alertes}` WHERE Status != 'Acquitté' 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}") except Exception as e: st.error(f"Erreur MySQL : {e}")