Michel/Gestion_sondes
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Ajout ou correction Domo91.py sur develop

Browse Source
This commit is contained in:
Michel
2025-05-09 10:07:24 +02:00
parent a33c8b9581
commit f381921875
2 changed files with 66 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -20,6 +20,7 @@ __pycache__/
*.log
*.tmp
*.bak
Logs/monitor.csv
# 👉 Archives et fichiers générés
*.zip

91
domo91.py
View File

@@ -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}")
except Exception as er:
st.error(f"Erreur MySQL : {er}")