459 lines
19 KiB
Python
459 lines
19 KiB
Python
# 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")
|
||
if "authenticated" not in st.session_state:
|
||
st.session_state["authenticated"] = False
|
||
st.session_state["role"] = None
|
||
st.session_state["lieu_autorise"] = None
|
||
|
||
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):
|
||
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"])
|
||
|
||
# --- Appliquer les filtres de l'utilisateur ---
|
||
sonde_choisie = st.session_state.get("sonde_selectionnee")
|
||
tranche = st.session_state.get("tranche_horaire")
|
||
if sonde_choisie:
|
||
df = df[df["Sonde"] == sonde_choisie]
|
||
if tranche:
|
||
heure = df["Heure"].dt.hour
|
||
if tranche == "Matin (6h-12h)":
|
||
df = df[(heure >= 6) & (heure < 12)]
|
||
elif tranche == "Après-midi (12h-18h)":
|
||
df = df[(heure >= 12) & (heure < 18)]
|
||
elif tranche == "Nuit (18h-6h)":
|
||
df = df[(heure >= 18) | (heure < 6)]
|
||
|
||
df["Heure"] = df["Heure"].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)
|
||
col_width = self.w / 3.3
|
||
row_height = 6
|
||
for sonde, mesures in data.items():
|
||
if not mesures:
|
||
continue
|
||
self.set_font("Arial", "B", 11)
|
||
self.cell(0, 8, f"Sonde : {sonde}", ln=1, fill=True)
|
||
self.set_font("Arial", "", 10)
|
||
for i in range(0, len(mesures), 3):
|
||
ligne = mesures[i:i+3]
|
||
for heure, temp in ligne:
|
||
self.cell(col_width, row_height, f"{heure} : {temp:.1f}°C", border=1)
|
||
self.ln()
|
||
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)
|
||
output_path = os.path.join(output_dir, file_name)
|
||
pdf.output(output_path)
|
||
|
||
return output_path
|
||
|
||
except Exception as e:
|
||
st.error(f"Erreur lors de la génération du PDF : {e}")
|
||
return None
|
||
|
||
# Ajout du bouton dans la sidebar pour tous les utilisateurs connectés
|
||
if st.session_state.get("authenticated") and st.session_state.get("selected_date"):
|
||
site = (
|
||
st.session_state.get("lieu_autorise")
|
||
if st.session_state.get("role") != "superviseur"
|
||
else st.session_state.get("selected_site")
|
||
)
|
||
date_val = st.session_state["selected_date"].strftime("%Y-%m-%d")
|
||
st.session_state["site_selectionne"] = site
|
||
st.sidebar.markdown("---")
|
||
st.sidebar.subheader("📄 Impressions")
|
||
|
||
if st.sidebar.button("📥 Générer le PDF maintenant"):
|
||
pdf_path = generer_pdf(site, date_val)
|
||
if pdf_path:
|
||
st.session_state["pdf_path"] = pdf_path
|
||
|
||
if "pdf_path" in st.session_state:
|
||
with open(st.session_state["pdf_path"], "rb") as f:
|
||
st.sidebar.download_button(
|
||
label="📄 Télécharger le PDF",
|
||
data=f,
|
||
file_name=os.path.basename(st.session_state["pdf_path"]),
|
||
mime="application/pdf",
|
||
key="pdf_download_button"
|
||
)
|
||
|
||
# Ajout du bouton dans la sidebar pour tous les utilisateurs connectés
|
||
if st.session_state.get("authenticated") and st.session_state.get("selected_date"):
|
||
site = (
|
||
st.session_state.get("lieu_autorise")
|
||
if st.session_state.get("role") != "superviseur"
|
||
else st.session_state.get("selected_site")
|
||
)
|
||
date_val = st.session_state["selected_date"].strftime("%Y-%m-%d")
|
||
st.session_state["site_selectionne"] = site
|
||
st.sidebar.markdown("---")
|
||
st.sidebar.subheader("📄 Impressions")
|
||
generer_pdf(site, date_val)
|
||
# Ajout du bouton dans la sidebar pour tous les utilisateurs connectés
|
||
if st.session_state.get("authenticated") and st.session_state.get("selected_date"):
|
||
site = (
|
||
st.session_state.get("lieu_autorise")
|
||
if st.session_state.get("role") != "superviseur"
|
||
else st.session_state.get("selected_site")
|
||
)
|
||
date_val = st.session_state["selected_date"].strftime("%Y-%m-%d")
|
||
st.sidebar.markdown("---")
|
||
st.sidebar.subheader("📄 Impressions")
|
||
generer_pdf(site, date_val)
|
||
|
||
# --- Initialisation des variables de session ---
|
||
if "authenticated" not in st.session_state:
|
||
st.session_state["authenticated"] = False
|
||
if "role" not in st.session_state:
|
||
st.session_state["role"] = None
|
||
if "lieu_autorise" not in st.session_state:
|
||
st.session_state["lieu_autorise"] = None
|
||
|
||
# --- Sidebar (connexion + bouton PDF) ---
|
||
|
||
# 🔐 Connexion (toujours affichée)
|
||
st.sidebar.header("🔐 Connexion")
|
||
if not st.session_state.get("authenticated"):
|
||
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 = 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.sidebar.error("Identifiants invalides")
|
||
cursor.close()
|
||
conn.close()
|
||
except Exception as e:
|
||
st.sidebar.error(f"Erreur lors de la connexion à la base : {e}")
|
||
else:
|
||
st.sidebar.success(f"Connecté ({st.session_state['role']})")
|
||
if st.sidebar.button("🔓 Déconnexion"):
|
||
st.session_state["authenticated"] = False
|
||
st.session_state["role"] = None
|
||
st.session_state["lieu_autorise"] = None
|
||
st.rerun()
|
||
|
||
# 🔔 Forcer une alerte de test dynamique avec choix de la sonde
|
||
if st.session_state.get("authenticated"):
|
||
site_actuel = (
|
||
st.session_state.get("lieu_autorise")
|
||
if st.session_state.get("role") != "superviseur"
|
||
else st.session_state.get("selected_site")
|
||
)
|
||
if site_actuel:
|
||
try:
|
||
conn = mysql.connector.connect(**db_config)
|
||
cursor = conn.cursor()
|
||
cursor.execute(f"SELECT DISTINCT Sonde FROM `{site_actuel}` ORDER BY Sonde ASC")
|
||
sondes_dispo = [row[0] for row in cursor.fetchall()]
|
||
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, Status)
|
||
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"Erreur chargement des sondes : {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)
|
||
st.session_state["selected_site"] = site_selectionne
|
||
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)
|
||
st.session_state["sonde_selectionnee"] = sonde_choisie
|
||
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)"])
|
||
st.session_state["tranche_horaire"] = 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)":
|
||
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}")
|
||
if st.session_state["role"] == "superviseur":
|
||
with st.expander("+ Ajouter une nouvelle chambre froide", expanded=False):
|
||
with st.form("ajout_sonde"):
|
||
nouvelle_sonde = st.text_input("Nom de la sonde")
|
||
temp_max = st.number_input("Température maximale autorisée (°C)", value=4)
|
||
etat_on = st.checkbox("État actif (ON)", value=True)
|
||
|
||
submitted = st.form_submit_button("✅ Ajouter")
|
||
if submitted:
|
||
try:
|
||
conn_add = mysql.connector.connect(**db_config)
|
||
cursor_add = conn_add.cursor()
|
||
|
||
cursor_add.execute(
|
||
"INSERT INTO Chambres_froides (Lieu, Sonde, Temp_Max, Status) VALUES (%s, %s, %s, %s)",
|
||
(site_selectionne, nouvelle_sonde, temp_max, "ON" if etat_on else "OFF")
|
||
)
|
||
conn_add.commit()
|
||
cursor_add.close()
|
||
conn_add.close()
|
||
|
||
st.success(f"Sonde '{nouvelle_sonde}' ajoutée avec succès au site {site_selectionne} 🎉")
|
||
|
||
# Refresh immédiat
|
||
st.session_state["refresh_admin"] = random.randint(1000, 9999)
|
||
|
||
except Exception as e:
|
||
st.error(f"Erreur lors de l'ajout : {e}")
|
||
with st.expander("🛠️ Gestion des chambres froides (administrateur)", expanded=True):
|
||
|
||
# Bouton d'actualisation
|
||
if st.button("🔄 Actualiser la liste"):
|
||
st.session_state["refresh_admin"] = random.randint(0, 9999) # Force le rerun
|
||
|
||
try:
|
||
conn_admin = mysql.connector.connect(**db_config)
|
||
cursor_admin = conn_admin.cursor(dictionary=True)
|
||
|
||
cursor_admin.execute("SELECT * FROM Chambres_froides WHERE Lieu = %s", (site_selectionne,))
|
||
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, val, plus = st.columns([1, 2, 1])
|
||
with moins:
|
||
if st.button("-", key=f"moins_{chambre['Id']}"):
|
||
temp_max -= 1
|
||
with val:
|
||
st.markdown(f"<div style='text-align:center;font-size:20px'>{temp_max}°C</div>",
|
||
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 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}")
|
||
|
||
|