From f381921875cb33663a5b27fb901bf2550b4134d1 Mon Sep 17 00:00:00 2001 From: Michel Date: Fri, 9 May 2025 10:07:24 +0200 Subject: [PATCH] Ajout ou correction Domo91.py sur develop --- .gitignore | 1 + domo91.py | 91 ++++++++++++++++++++++++++++++++++++++---------------- 2 files changed, 66 insertions(+), 26 deletions(-) diff --git a/.gitignore b/.gitignore index d193444..b5c8fd9 100644 --- a/.gitignore +++ b/.gitignore @@ -20,6 +20,7 @@ __pycache__/ *.log *.tmp *.bak +Logs/monitor.csv # 👉 Archives et fichiers générés *.zip diff --git a/domo91.py b/domo91.py index 4c7a1fc..a20c7f2 100644 --- a/domo91.py +++ b/domo91.py @@ -2,14 +2,13 @@ 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 import random -import datetime from dotenv import load_dotenv +from datetime import datetime, date, time # Charger les variables d'environnement load_dotenv() @@ -32,37 +31,54 @@ def get_connection(): return mysql.connector.connect(**db_config) # --- 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}") +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) - cursor.execute(f"SELECT Sonde, Date, Temperature FROM `{site}` WHERE DATE(Date) = %s ORDER BY Sonde, Date", - (date_str,)) - rows = cursor.fetchall() + # 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}" - cursor.execute(f"SELECT Sonde, Debut_defaut, Status FROM {table_alertes} WHERE DATE(Debut_defaut) = %s", - (date_str,)) - alertes = cursor.fetchall() + pdf_cursor.execute(f"SELECT Sonde, Debut_defaut, Status FROM {table_alertes} WHERE DATE(Debut_defaut) = %s", (date_str,)) + alertes = pdf_cursor.fetchall() - cursor.close() + 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"Période : {getattr(self, 'periode', '')}", ln=1, align="C") self.ln(5) def site_info(self, site_name): @@ -73,13 +89,37 @@ def generer_pdf(site, date_str): 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 heure, temp in mesures: - self.cell(0, 6, f"{heure} - {temp} °C", ln=1) - self.ln(2) + 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) @@ -88,7 +128,9 @@ def generer_pdf(site, date_str): 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) @@ -96,7 +138,7 @@ def generer_pdf(site, date_str): file_name = f"rapport_{site}_{date_str}.pdf" output_dir = "PDF" - os.makedirs(output_dir, exist_ok=True) # 🔧 Crée le dossier si absent + os.makedirs(output_dir, exist_ok=True) output_path = os.path.join(output_dir, file_name) pdf.output(output_path) @@ -139,7 +181,7 @@ if not st.session_state.get("authenticated"): st.success(f"Connecté comme {result['role']} ({result['Lieu']})") # Enregistrement de la connexion dans Connexion_Log try: - now = datetime.datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") cursor.execute(""" INSERT INTO Connexion_Log (Utilisateur, Lieu, Date_Connexion) VALUES (%s, %s, %s) @@ -173,7 +215,8 @@ if site_pdf and date_pdf: st.sidebar.markdown("---") st.sidebar.subheader("📄 Rapport PDF") if st.sidebar.button("📥 Télécharger l’état du jour (PDF)", key="pdf_btn"): - generer_pdf(site_pdf, date_pdf.strftime("%Y-%m-%d")) + periode = st.session_state.get("selected_periode", "Toute la journée") + generer_pdf(site_pdf, date_pdf.strftime("%Y-%m-%d"), periode) # --- Forcer une alerte de test dynamique (réservé aux superviseurs) if st.session_state.get("authenticated") and st.session_state.get("role") == "superviseur": @@ -210,7 +253,7 @@ if st.session_state.get("authenticated") and st.session_state.get("role") == "su sonde_test = st.sidebar.selectbox("Choisir une sonde :", sondes_dispo) if st.sidebar.button("🔔 Forcer une alerte de test"): - now = datetime.datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") try: conn = mysql.connector.connect(**db_config) cursor = conn.cursor() @@ -261,7 +304,7 @@ def maj_entretien(sonde_id, statut, utilisateur): action = "Mise en entretien" if statut else "Sortie d'entretien" cursor.execute( "INSERT INTO Connexion_Log (Utilisateur, Lieu, Date_Connexion, Action) VALUES (%s, %s, %s, %s)", - (utilisateur, lieu, datetime.datetime.now(), action) + (utilisateur, lieu, datetime.now(), action) ) conn.commit() conn.close() @@ -349,7 +392,7 @@ if st.session_state["authenticated"]: 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)": @@ -614,9 +657,5 @@ if st.session_state["authenticated"]: except Exception as e: st.error(f"Erreur chargement des connexions : {e}") - - except Exception as err: - st.error(f"Erreur MySQL : {err}") - - except Exception as err: - st.error(f"Erreur MySQL : {err}") \ No newline at end of file + except Exception as er: + st.error(f"Erreur MySQL : {er}")