142 lines
11 KiB
Markdown
142 lines
11 KiB
Markdown
# 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
|
|
import random
|
|
import datetime
|
|
|
|
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}")
|
|
|
|
# --- Alertes de test dynamiques (sonde choisie) ---
|
|
if st.session_state.get("authenticated"):
|
|
try:
|
|
conn = mysql.connector.connect(**db_config)
|
|
cursor = conn.cursor()
|
|
site_actuel = st.session_state.get("lieu_autorise")
|
|
if site_actuel:
|
|
cursor.execute(f"SELECT DISTINCT Sonde FROM `{site_actuel}` ORDER BY Sonde ASC")
|
|
sondes_dispo = [row[0] for row in cursor.fetchall()]
|
|
else:
|
|
sondes_dispo = []
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
if sondes_dispo:
|
|
st.sidebar.markdown("---")
|
|
st.sidebar.subheader("🧪 Test alerte manuelle")
|
|
sonde_test = st.sidebar.selectbox("Choisir une sonde :", sondes_dispo)
|
|
if st.sidebar.button("🔔 Forcer une alerte de test"):
|
|
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
try:
|
|
conn = mysql.connector.connect(**db_config)
|
|
cursor = conn.cursor()
|
|
table_alertes = f"Alertes_{site_actuel}"
|
|
cursor.execute(f"""
|
|
INSERT INTO {table_alertes} (Sonde, Debut_defaut, Etat)
|
|
VALUES (%s, %s, %s)
|
|
""", (sonde_test, now, "Test"))
|
|
conn.commit()
|
|
cursor.close()
|
|
conn.close()
|
|
st.success(f"Alerte de test créée pour {sonde_test} à {now}")
|
|
except Exception as e:
|
|
st.error(f"Erreur lors de la création de l'alerte : {e}")
|
|
except Exception as e:
|
|
st.sidebar.warning(f"Impossible de charger les sondes pour test : {e}")
|
|
|
|
|