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

🆕 Ajout de fonction 'Test sondes'.py

Browse Source
This commit is contained in:
Michel
2025-04-19 10:13:04 +02:00
parent dd08735902
commit c460925f9f
10 changed files with 76 additions and 768 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
Chat-Id.py
View File

@@ -1,36 +0,0 @@
import requests
import json
# Remplace par ton token de bot Telegram
token = "8128378340:AAF2sO3gaH1XpMNya_pEslzerqokoCiFRGs"
url = f"https://api.telegram.org/bot{token}/getUpdates"
try:
response = requests.get(url)
data = response.json()
print("🔍 Récupération des chats Telegram récents...\n")
if "result" in data:
chats = set()
for update in data["result"]:
message = update.get("message") or update.get("edited_message")
if not message:
continue
chat = message["chat"]
chat_id = chat["id"]
chat_title = chat.get("title") or chat.get("username") or chat.get("first_name") or "Inconnu"
if chat_id not in chats:
chats.add(chat_id)
print(f"➡️ Chat : {chat_title} | ID : {chat_id}")
if not chats:
print("⚠️ Aucun chat trouvé. Envoie un message depuis un groupe ou une discussion avec le bot.")
else:
print("❌ Erreur : aucune donnée 'result' trouvée dans la réponse.")
except Exception as e:
print(f"Erreur lors de la récupération des chats : {e}")

30
Chaufferie.py
View File

@@ -1,30 +0,0 @@
import paho.mqtt.client as mqtt
import mysql.connector
# Configuration de la connexion MySQL
mydb = mysql.connector.connect(
host="54.36.188.119",
user="michel",
password="#SO2&1nf%mZ@jfh",
database="Sondes"
)
# Fonction de callback quand un message est reçu
def on_message(_client, _userdata, msg):
print(f"Message reçu sur {msg.topic}: {msg.payload.decode()}")
cursor = mydb.cursor()
frigo_name = msg.topic.split('/')[-1] # Prend la dernière partie après le "/"
sql = "INSERT INTO Chaufferie (Sonde, Temperature) VALUES (%s, %s)"
val = (frigo_name, msg.payload.decode())
cursor.execute(sql, val)
mydb.commit()
# Configuration du client MQTT
client = mqtt.Client()
client.username_pw_set("Bwps", "scJ5ACj2keRfI^")
client.on_message = on_message
client.connect("54.36.188.119", 1883, 60)
client.subscribe("Module_01/#") # S'abonner à tous les topics commençant par Saclay
client.loop_forever() # Rester connecté en continu pour écouter les messages

18
Purge_Alertes_saclay.py
View File

@@ -1,18 +0,0 @@
# Purge de la table Sondes.Alertes_Saclay.
import mysql.connector
config = {
"host": "54.36.188.119",
"user": "michel",
"password": "#SO2&1nf%mZ@jfh",
"database": "Sondes"
}
conn = mysql.connector.connect(**config)
cursor = conn.cursor()
cursor.execute("DELETE FROM Alertes_Saclay WHERE Debut_defaut < NOW() - INTERVAL 7 DAY")
conn.commit()
cursor.close()
conn.close()
print("✅ Alertes de plus de 7 jours supprimées.")

107
Societe_Generale.py
View File

@@ -1,107 +0,0 @@
# Insertion de données da l'app SG.
import streamlit as st
import mysql.connector
from dotenv import load_dotenv
import os
from datetime import datetime
load_dotenv()
st.set_page_config(page_title="Insertion Mysql", layout="centered")
st.title("🗓️ Insertion dans une base MySQL")
# Charger les identifiants
DB_HOST = os.getenv("DB_HOST")
DB_USER = os.getenv("DB_USER")
DB_PASSWORD = os.getenv("DB_PASSWORD")
# Connexion sans DB initiale pour lister les bases
@st.cache_data
def get_databases():
conn = mysql.connector.connect(
host=DB_HOST,
user=DB_USER,
password=DB_PASSWORD
)
cursor = conn.cursor()
cursor.execute("SHOW DATABASES")
bases = [db[0] for db in cursor.fetchall()]
conn.close()
return bases
# Connexion avec DB sélectionnée
def connect_to_db(db_name):
return mysql.connector.connect(
host=DB_HOST,
user=DB_USER,
password=DB_PASSWORD,
database=db_name
)
# --- Interface principale ---
base = st.selectbox("Choisir une base de données", get_databases())
if base:
conn = connect_to_db(base)
cursor = conn.cursor()
cursor.execute("SHOW TABLES")
tables = [table[0] for table in cursor.fetchall()]
table = st.selectbox("Choisir une table", tables)
if table:
cursor.execute(f"DESCRIBE {table}")
colonnes = cursor.fetchall()
# Détection clé primaire
cursor.execute(f"SHOW INDEX FROM {table} WHERE Key_name = 'PRIMARY'")
pk = cursor.fetchone()
primary_key = pk[4] if pk else None
champ_date_candidates = [col[0] for col in colonnes if "date" in col[1].lower()]
champ_date = st.selectbox("Champ de date", champ_date_candidates)
nb_mois = st.number_input("Nombre d'insertions mensuelles", 1, 36, 6)
date_depart = st.date_input("Date de départ des insertions", value=datetime.today().date())
st.subheader("Champs à insérer (hors clé primaire et champ de date)")
champs_choisis = []
champs_editables = [col for col in colonnes if col[0] != primary_key and col[0] != champ_date]
for col in champs_editables:
if st.checkbox(f"{col[0]} ({col[1]})", value=True):
champs_choisis.append(col[0])
with st.form("formulaire_insertion"):
st.markdown("**Valeurs fixes pour les champs cochés**")
valeurs_fixes = {}
for nom in champs_choisis:
val = st.text_input(nom)
valeurs_fixes[nom] = val if val != "" else None
submit = st.form_submit_button("Insérer")
try:
cursor = conn.cursor()
insert_count = 0
for i in range(nb_mois):
year = date_depart.year + (date_depart.month + i - 1) // 12
month = (date_depart.month + i - 1) % 12 + 1
day = min(date_depart.day, 28)
date_cible = datetime(year, month, day).date()
champs = [champ_date] + list(valeurs_fixes.keys())
valeurs = [date_cible] + list(valeurs_fixes.values())
sql = f"INSERT INTO {table} ({', '.join(champs)}) VALUES ({', '.join(['%s'] * len(valeurs))})"
cursor.execute(sql, valeurs)
insert_count += 1
conn.commit()
st.success(f"{insert_count} lignes insérées avec succès dans `{table}`.")
except Exception as e:
st.error(f"❌ Erreur lors de l’insertion : {e}")
conn.commit()
st.success(f"{insert_count} lignes insérées avec succès dans `{table}`.")
st.success(f"{insert_count} lignes insérées avec succès.")

324
Telegram_sondes.py
View File

@@ -1,324 +0,0 @@
# Gestion des chambres froides & alertes Telegram
import requests
import mysql.connector
from datetime import datetime, timedelta
import time
import sys
import os
import schedule
def connect_db():
return mysql.connector.connect(
host="54.36.188.119",
user="michel",
password="#SO2&1nf%mZ@jfh",
database="Sondes"
)
def envoi_etat_quotidien(cursor, site):
token = "8128378340:AAF2sO3gaH1XpMNya_pEslzerqokoCiFRGs"
chat_id = get_chat_id(cursor, site)
etat_sondes(cursor, site, chat_id, token)
print(f"[INFO] État des sondes envoyé pour le site {site}.")
def get_active_sondes(cursor, site):
query = "SELECT Sonde, Temp_Max FROM Chambres_froides WHERE Lieu = %s AND Etat = 'On';"
cursor.execute(query, (site,))
result = cursor.fetchall()
return {row[0]: float(row[1]) for row in result}
def check_temperature_limits(cursor, table_historique, sonde, limite, duree=30):
temps_limite = datetime.now() - timedelta(minutes=duree)
query = f"""
SELECT Temperature
FROM {table_historique}
WHERE Sonde = %s AND Date >= %s
ORDER BY Date DESC LIMIT 6;
"""
cursor.execute(query, (sonde, temps_limite))
result = cursor.fetchall()
if len(result) == 6 and all(float(temp[0]) > limite for temp in result):
return True, float(result[0][0])
return False, None
def simulate_alert(db, cursor, site, sonde, chat_id, token):
table_alertes = f"Alertes_{site}"
print(f"[TEST] tentative d'insertion de sonde : {sonde} dans table {table_alertes}")
try:
query = f"""
INSERT INTO {table_alertes} (Sonde, Debut_defaut, Status)
VALUES (%s, NOW(), 'test')
ON DUPLICATE KEY UPDATE Debut_defaut = NOW(), Status = 'test';
"""
cursor.execute(query, (sonde,))
db.commit()
print(f"[TEST] insertion ou mise à jour réussie pour {sonde}.")
except Exception as e:
print(f"[ERREUR] lors de l'insertion test : {e}")
message = f"⚠️ TEST ALERTE : Simulation d'une alerte pour la sonde {sonde}."
url = f"https://api.telegram.org/bot{token}/sendMessage"
try:
requests.get(url, params={'chat_id': chat_id, 'text': message})
except Exception as e:
print(f"[ERREUR] lors de l'envoi Telegram : {e}")
def create_alert(db, cursor, table_alertes, sonde, temperature):
query = f"""
INSERT INTO {table_alertes} (Sonde, Debut_defaut, Status)
VALUES (%s, NOW(), 'en cours');
"""
cursor.execute(query, (sonde,))
send_telegram_message(sonde, temperature)
print(f"Alerte créée pour la sonde {sonde}")
db.commit()
def resolve_alert(cursor, table_alertes, sonde):
query = f"""
UPDATE {table_alertes}
SET Status = 'résolu'
WHERE Sonde = %s AND Status = 'en cours';
"""
cursor.execute(query, (sonde,))
print(f"Alerte résolue pour la sonde {sonde}")
def acquitter_alerte(cursor, site, sonde, chat_id, token):
table_alertes = f"Alertes_{site}"
query = f"""
UPDATE {table_alertes}
SET Status = 'acquittée'
WHERE Sonde = %s AND Status IN ('en cours', 'test');
"""
cursor.execute(query, (sonde,))
lignes_modifiees = cursor.rowcount
if lignes_modifiees > 0:
message = f"✅ Alerte acquittée pour la sonde {sonde} par commande Telegram."
else:
message = f"ℹ️ Aucune alerte active à acquitter pour la sonde {sonde}."
url = f"https://api.telegram.org/bot{token}/sendMessage"
try:
requests.get(url, params={'chat_id': chat_id, 'text': message})
except Exception as e:
print(f"[ERREUR] envoi message Telegram : {e}")
def send_telegram_message(sonde, temperature):
token = "5714323406:AAGSj9jrfBHbfxubz3ooabPEizI8aBOLnvE"
chat_id = "-1002442631825"
message = f"⚠️ Alerte température : La sonde {sonde} dépasse la limite avec une température de {temperature}°C."
url = f"https://api.telegram.org/bot{token}/sendMessage"
params = {'chat_id': chat_id, 'text': message}
try:
response = requests.get(url, params=params)
if response.status_code == 200:
print(f"Message envoyé via Telegram pour la sonde {sonde}.")
else:
print(f"Erreur lors de l'envoi du message Telegram : {response.status_code}")
except Exception as e:
print(f"Erreur lors de l'envoi du message Telegram : {e}")
def get_chat_id(cursor, site):
cursor.execute("SELECT Chat_ID FROM Sites WHERE Nom = %s;", (site,))
result = cursor.fetchone()
if result:
return result[0]
else:
print(f"Aucun Chat_ID trouvé pour le site {site}")
return None
def passer_en_maintenance(cursor, site, sonde, chat_id, token):
try:
query = """
UPDATE Chambres_froides
SET Etat = 'Off'
WHERE Lieu = %s AND Sonde = %s;
"""
cursor.execute(query, (site, sonde))
lignes_modifiees = cursor.rowcount
if lignes_modifiees > 0:
message = f"🛠️ La sonde {sonde} a été passée en mode maintenance (OFF)."
else:
message = f"⚠️ Sonde {sonde} introuvable pour le site {site}."
url = f"https://api.telegram.org/bot{token}/sendMessage"
requests.get(url, params={'chat_id': chat_id, 'text': message})
except Exception as e:
print(f"[ERREUR] lors du passage en maintenance : {e}")
def reactiver_sonde(cursor, site, sonde, chat_id, token):
try:
query = """
UPDATE Chambres_froides
SET Etat = 'On'
WHERE Lieu = %s AND Sonde = %s;
"""
cursor.execute(query, (site, sonde))
lignes_modifiees = cursor.rowcount
if lignes_modifiees > 0:
message = f"✅ La sonde {sonde} a été réactivée (ON)."
else:
message = f"⚠️ Sonde {sonde} introuvable pour le site {site}."
url = f"https://api.telegram.org/bot{token}/sendMessage"
requests.get(url, params={'chat_id': chat_id, 'text': message})
except Exception as e:
print(f"[ERREUR] lors de la réactivation : {e}")
def etat_sondes(cursor, site, chat_id, token):
query = "SELECT Sonde, Etat FROM Chambres_froides WHERE Lieu = %s ORDER BY Sonde;"
cursor.execute(query, (site,))
result = cursor.fetchall()
message = f"📊 État des sondes - {site} :\n"
for sonde, etat in result:
symbole = "🟢" if etat.upper() == "ON" else "🔴"
message += f"{symbole} {sonde} ({etat.upper()})\n"
url = f"https://api.telegram.org/bot{token}/sendMessage"
requests.get(url, params={'chat_id': chat_id, 'text': message})
def monitor_temperatures_simple(site, db, cursor):
table_historique = site
table_alertes = f"Alertes_{site}"
sondes = get_active_sondes(cursor, site)
print(f"[MONITORING] Sondes actives pour {site} : {sondes}")
for sonde, limite in sondes.items():
alert_needed, temperature = check_temperature_limits(cursor, table_historique, sonde, limite)
if alert_needed:
create_alert(db, cursor, table_alertes, sonde, temperature)
else:
resolve_alert(cursor, table_alertes, sonde)
db.commit()
print("[MONITORING] Vérification des sondes terminée.")
def monitor_temperatures(site):
db = connect_db()
cursor = db.cursor()
table_historique = site # Ex: Saclay
table_alertes = f"Alertes_{site}" # Ex: Alertes_Saclay
sondes = get_active_sondes(cursor, site)
print(f"Sondes actives pour le site {site} :", sondes)
for sonde, limite in sondes.items():
alert_needed, temperature = check_temperature_limits(cursor, table_historique, sonde, limite)
if alert_needed:
create_alert(db, cursor, table_alertes, sonde, temperature)
else:
resolve_alert(cursor, table_alertes, sonde)
db.commit()
listen_for_commands(db, cursor, site)
cursor.close()
db.close()
def listen_for_commands(db, cursor, site):
token = "5714323406:AAGSj9jrfBHbfxubz3ooabPEizI8aBOLnvE"
chat_id = get_chat_id(cursor, site)
offset_file = os.path.join(os.path.dirname(__file__), f"last_update_id_{site}.txt")
try:
with open(offset_file, 'r') as f:
last_update_id = int(f.read().strip())
except FileNotFoundError:
last_update_id = None
url = f"https://api.telegram.org/bot{token}/getUpdates"
if last_update_id is not None:
url += f"?offset={last_update_id + 1}"
try:
response = requests.get(url)
data = response.json()
if "result" in data:
for update in data["result"]:
update_id = update["update_id"]
if "message" in update:
message = update["message"]["text"]
print(f"[CMD] Message reçu : {message}")
if message.lower().startswith("/etat"):
etat_sondes(cursor, site, chat_id, token)
elif message.lower().startswith("/acquitter"):
parts = message.strip().split(" ", 1)
if len(parts) == 2 and parts[1].strip():
sonde = parts[1].strip()
acquitter_alerte(cursor, site, sonde, chat_id, token)
db.commit()
else:
erreur_msg = "❌ Utilisation incorrecte. Format attendu : /acquitter <nom_sonde>"
requests.get(f"https://api.telegram.org/bot{token}/sendMessage",
params={'chat_id': chat_id, 'text': erreur_msg})
elif message.lower().startswith("/maintenance"):
parts = message.strip().split(" ", 1)
if len(parts) == 2 and parts[1].strip():
sonde = parts[1].strip()
passer_en_maintenance(cursor, site, sonde, chat_id, token)
db.commit()
else:
erreur_msg = "❌ Utilisation incorrecte. Format attendu : /maintenance <nom_sonde>"
requests.get(f"https://api.telegram.org/bot{token}/sendMessage",
params={'chat_id': chat_id, 'text': erreur_msg})
elif message.lower().startswith("/reactiver"):
parts = message.strip().split(" ", 1)
if len(parts) == 2 and parts[1].strip():
sonde = parts[1].strip()
reactiver_sonde(cursor, site, sonde, chat_id, token)
db.commit()
else:
erreur_msg = "❌ Utilisation incorrecte. Format attendu : /reactiver <nom_sonde>"
requests.get(f"https://api.telegram.org/bot{token}/sendMessage",
params={'chat_id': chat_id, 'text': erreur_msg})
elif message.lower().startswith("/test"):
parts = message.strip().split(" ", 1)
if len(parts) == 2 and parts[1].strip():
sonde = parts[1].strip()
simulate_alert(db, cursor, site, sonde, chat_id, token)
else:
erreur_msg = "❌ Utilisation incorrecte. Format attendu : /test <nom_sonde>"
requests.get(f"https://api.telegram.org/bot{token}/sendMessage",
params={'chat_id': chat_id, 'text': erreur_msg})
with open(offset_file, 'w') as f:
f.write(str(update_id))
except Exception as e:
print(f"Erreur lors de la lecture des commandes Telegram : {e}")
def main():
site = sys.argv[1] if len(sys.argv) > 1 else "Saclay"
db = connect_db()
cursor = db.cursor()
# ✅ Planification : tous les jours à 07:00 → état des sondes
schedule.every().day.at("07:00").do(envoi_etat_quotidien, cursor=cursor, site=site)
# ✅ Planification : toutes les 2m30 → surveillance
schedule.every(2).minutes.do(monitor_temperatures_simple, site=site, db=db, cursor=cursor)
try:
while True:
schedule.run_pending()
listen_for_commands(db, cursor, site)
time.sleep(2) # petite pause pour ne pas surcharger Telegram
finally:
cursor.close()
db.close()
if __name__ == "__main__":
main()

24
Test_mail.py
View File

@@ -1,24 +0,0 @@
# Programme de test de la BAL alertes_saclay@domo91.fr.
import smtplib
from email.mime.text import MIMEText
def envoyer_mail(sujet, message, destinataires):
msg = MIMEText(message)
msg['Subject'] = sujet
msg['From'] = 'alertes_saclay@domo91.fr'
msg['To'] = ', '.join(destinataires)
try:
with smtplib.SMTP_SSL('smtp.mail.ovh.net', 465) as server:
server.login('alertes_saclay@domo91.fr', 'Kdpke674y23Feq^H')
server.sendmail(msg['From'], destinataires, msg.as_string())
print(f"📧 Mail envoyé à {destinataires}")
except Exception as e:
print(f"Erreur envoi mail : {e}")
# --- Test d'envoi ---
envoyer_mail(
sujet="🧪 Test d'envoi d'alerte",
message="Ceci est un test de l'envoi d'e-mail via alertes_saclay@domo91.fr.",
destinataires=["services@domo91.fr"]
)

106
Tracker.py
View File

@@ -1,106 +0,0 @@
# Programme de récupération des adresses de sondes DS18B20 et envoie dans table Sondes.Tracker.
import paho.mqtt.client as mqtt
import mysql.connector
import re # Import de la bibliothèque des expressions régulières
# Configuration de la base de données MySQL
db_config = {
'user': 'superviseur', # Remplacez par votre utilisateur MySQL
'password': 'Bto7Lm_z]m!BFH!*', # Remplacez par votre mot de passe MySQL
'host': '54.36.188.119', # Adresse de votre serveur MySQL (localhost si en local)
'database': 'Sondes' # Nom de la base de données MySQL
}
def convertir_rom_id_en_hexa(rom_id):
# On découpe la chaîne rom_id en paires de deux caractères
hex_parts = [f"0x{rom_id[i:i + 2]}" for i in range(0, len(rom_id), 2)]
return ",".join(hex_parts)
# Connexion à la base de données MySQL
def connect_db():
return mysql.connector.connect(**db_config)
# Fonction pour vérifier si la sonde est déjà présente dans la base
def sonde_deja_presente(rom_id):
conn = connect_db()
cursor = conn.cursor()
try:
# Vérifier si la sonde existe déjà dans la base
cursor.execute("SELECT COUNT(*) FROM Tracker WHERE rom_id = %s", (rom_id,))
result = cursor.fetchone()
return result[0] > 0 # Si le nombre est supérieur à 0, la sonde est déjà présente
except mysql.connector.Error as err:
print(f"Erreur MySQL lors de la vérification de la sonde : {err}")
return False
finally:
cursor.close()
conn.close()
# Fonction pour insérer les sondes dans la base de données
def inserer_sonde(rom_id):
if sonde_deja_presente(rom_id):
print(f"Sonde {rom_id} déjà présente dans la base de données. Aucune insertion effectuée.")
return
conn = connect_db()
cursor = conn.cursor()
try:
# Conversion en hexadécimal
rom_id_hexa = convertir_rom_id_en_hexa(rom_id)
# Insertion de rom_id et de sa version hexa dans la base
cursor.execute("INSERT INTO Tracker (rom_id, hexa) VALUES (%s, %s)", (rom_id, rom_id_hexa))
conn.commit()
print(f"Sonde {rom_id} insérée dans la base de données avec son format hexa {rom_id_hexa}.")
except mysql.connector.Error as err:
print(f"Erreur MySQL : {err}")
finally:
cursor.close()
conn.close()
# Fonction de callback pour gérer les messages MQTT
def on_message(_client, _userdata, message):
payload = message.payload.decode()
print(f"Message reçu : {payload}")
# Utilisation de re pour extraire les ROM IDs (Sonde [rom_id])
rom_ids = re.findall(r"[0-9a-fA-F]{16}", payload) # Capturer les IDs de 16 caractères
if rom_ids:
for rom_id in rom_ids:
inserer_sonde(rom_id)
else:
print("Aucun ROM ID trouvé dans le message.")
# Configuration du client MQTT
mqtt_broker = "46.105.92.116"
mqtt_port = 1883 # Vérifiez si votre broker utilise un autre port
mqtt_user = "Bwps"
mqtt_password = "scJ5ACj2keRfI^"
mqtt_topic = "Tracker/sondes"
client = mqtt.Client()
# Authentification MQTT
client.username_pw_set(mqtt_user, mqtt_password)
# Configuration du callback pour les messages MQTT
client.on_message = on_message
# Connexion au broker MQTT
client.connect(mqtt_broker, mqtt_port, 60)
# Souscription au topic Tracker/sondes
client.subscribe(mqtt_topic)
# Démarrer la boucle MQTT
client.loop_forever()

91
domo91.py
View File

@@ -11,6 +11,11 @@ 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 ---
@@ -100,36 +105,22 @@ def generer_pdf(site, date_str):
except Exception as e:
st.error(f"Erreur lors de la génération du PDF : {e}")
# --- Initialisation session ---
# --- 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) ---
with st.sidebar:
if st.sidebar.button("🔔 Forcer une alerte de test"):
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
sonde_test = "TEST_ALERTE"
try:
conn = mysql.connector.connect(**db_params)
cursor = conn.cursor()
table_alertes = f"Alertes_{site}"
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}")
st.header("🔐 Connexion")
# 🔐 Connexion (toujours affichée)
st.sidebar.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"):
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)
@@ -141,11 +132,58 @@ with st.sidebar:
st.session_state["lieu_autorise"] = result["Lieu"]
st.success(f"Connecté comme {result['role']} ({result['Lieu']})")
else:
st.error("Identifiants invalides")
st.sidebar.error("Identifiants invalides")
cursor.close()
conn.close()
except Exception as e:
st.error(f"Erreur lors de la connexion à la base : {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"):
@@ -173,6 +211,7 @@ if st.session_state["authenticated"]:
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}")
@@ -281,7 +320,7 @@ if st.session_state["role"] == "superviseur":
cursor_add = conn_add.cursor()
cursor_add.execute(
"INSERT INTO Chambres_froides (Lieu, Sonde, Temp_Max, Etat) VALUES (%s, %s, %s, %s)",
"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()

39
logs_analyse.py
View File

@@ -1,39 +0,0 @@
import streamlit as st
import pandas as pd
import matplotlib.pyplot as plt
# --- Chargement des données ---
try:
df = pd.read_csv("/home/debian/travail/logs_monitor.csv", sep=";", parse_dates=["Date"])
except FileNotFoundError:
st.error("📂 Le fichier de log n'a pas été trouvé.")
st.stop()
st.title("🧾 Analyse des scans Monitor.py")
# --- Filtres ---
df["Date_str"] = df["Date"].dt.date
dates_dispo = sorted(df["Date_str"].unique(), reverse=True)
date_selection = st.selectbox("📅 Sélectionnez une date :", dates_dispo)
lieux = sorted(df["Lieu"].unique())
lieu_selection = st.selectbox("📍 Site :", lieux)
df_filtré = df[(df["Date_str"] == date_selection) & (df["Lieu"] == lieu_selection)]
sondes = sorted(df_filtré["Sonde"].unique())
sonde_selection = st.selectbox("🧪 Sonde :", sondes)
df_sonde = df_filtré[df_filtré["Sonde"] == sonde_selection]
# --- Graphique température ---
fig, ax = plt.subplots(figsize=(10, 4))
ax.plot(df_sonde["Date"], df_sonde["Température"], marker='o')
ax.set_title(f"Évolution de la température - {sonde_selection}")
ax.set_xlabel("Heure")
ax.set_ylabel("Température (°C)")
st.pyplot(fig)
# --- Tableau complet du jour/sonde sélectionnés ---
st.markdown("### 📋 Détail des scans")
st.dataframe(df_sonde[["Date", "Sonde", "Température", "Seuil", "État"]], use_container_width=True)

47
supervisor_watchdog.py
View File

@@ -1,47 +0,0 @@
# Controle des services supervisor envoi en cas de disfonction et tous les jours à sept heures
# a faire tourner avec crontb -e en sudo
# */30 * * * * /home/debian/travail/myenv/bin/python3 /home/debian/travail/Scripts/supervisor_watchdog.py
import subprocess
import smtplib
from email.mime.text import MIMEText
from datetime import datetime
heure_actuelle = datetime.now().strftime("%H:%M")
etat_services = []
anomalies = []
try:
output = subprocess.check_output("/usr/bin/supervisorctl status", shell=True, text=True)
for line in output.splitlines():
parts = line.split()
if len(parts) >= 2:
nom, statut = parts[0], parts[1]
etat_services.append(f"{nom} ➤ {statut}")
if statut != "RUNNING":
anomalies.append(f"{nom} ➤ {statut}")
except Exception as e:
etat_services.append("❌ Impossible d'exécuter supervisorctl")
anomalies.append(f"Erreur : {e}")
# Déclenchement mail si anomalie ou à 07:00
envoyer_mail = bool(anomalies) or heure_actuelle == "07:00"
if envoyer_mail:
sujet = "⚠️ Alerte Supervisor" if anomalies else "✅ Rapport quotidien Supervisor"
intro = "🛑 Les services suivants ne sont pas en RUNNING :" if anomalies else "✅ Tous les services supervisés sont en RUNNING."
contenu = f"{intro}\n\n" + "\n".join(etat_services)
msg = MIMEText(contenu)
msg["Subject"] = sujet
msg["From"] = "alertes_saclay@domo91.fr"
msg["To"] = "services@domo91.fr"
try:
with smtplib.SMTP_SSL("smtp.mail.ovh.net", 465) as server:
server.login("alertes_saclay@domo91.fr", "Kdpke674y23Feq^H")
server.sendmail(msg["From"], [msg["To"]], msg.as_string())
print("📧 Mail envoyé.")
except Exception as e:
print(f"Erreur envoi mail : {e}")
else:
print("🕖 Aucun mail envoyé (tout est OK et ce n’est pas l’heure du rapport).")