from flask import Flask, jsonify, render_template, request, redirect, url_for, session from flask_cors import CORS from config import Config from werkzeug.security import check_password_hash from functools import wraps from datetime import timedelta import psycopg2 app = Flask(__name__) app.config.from_object(Config) app.secret_key = app.config["SECRET_KEY"] app.config["SESSION_PERMANENT"] = True app.config["PERMANENT_SESSION_LIFETIME"] = timedelta(hours=2) app.config["SESSION_COOKIE_HTTPONLY"] = True app.config["SESSION_COOKIE_SAMESITE"] = "Lax" app.config["SESSION_COOKIE_SECURE"] = False CORS(app) def get_connection(): return psycopg2.connect( host=app.config["DB_HOST"], port=app.config["DB_PORT"], database=app.config["DB_NAME"], user=app.config["DB_USER"], password=app.config["DB_PASSWORD"] ) def safe_float(value): """Convertit une valeur en float de façon sûre. Accepte la virgule ou le point. Retourne None si impossible.""" if value is None: return None try: return float(str(value).replace(",", ".").strip()) except (TypeError, ValueError): return None def normalize_number(value): """Normalise une saisie utilisateur (virgule -> point).""" if value is None: return None return str(value).replace(",", ".").strip() def get_setting_value(cur, key, default_value): cur.execute("SELECT value FROM settings WHERE key = %s", (key,)) row = cur.fetchone() return row[0] if row else default_value def upsert_setting(cur, key, value): cur.execute(""" INSERT INTO settings (key, value) VALUES (%s, %s) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value """, (key, value)) def admin_required(view_func): @wraps(view_func) def wrapped_view(*args, **kwargs): if not session.get("admin_logged_in"): return redirect(url_for("admin_login")) return view_func(*args, **kwargs) return wrapped_view @app.before_request def make_session_permanent(): session.permanent = True @app.route("/") @admin_required def home(): return render_template("index.html") @app.route("/api/status") def status(): return jsonify({ "project": "EcoCharge", "api": "ok", "database": "postgresql_connected" }) @app.route("/api/data", methods=["POST"]) def receive_data(): data = request.get_json(silent=True) if not data: return jsonify({"error": "Aucune donnée JSON reçue"}), 400 # --- Noms JSON envoyés par l'ESP32 : INCHANGÉS --- temperature_ext = data.get("temperature_ext") humidity_ext = data.get("humidity_ext") system_status_msg = data.get("system_status") voltage_pv = data.get("voltage_pv") current_pv = data.get("current_pv") power_pv = data.get("power_pv", data.get("solar_power")) luminosity = data.get("luminosity") voltage_battery = data.get("voltage_battery", data.get("battery_voltage")) current_battery = data.get("current_battery") battery_temp = data.get("battery_temp", data.get("battery_temperature")) battery_level = data.get("battery_level") conn = get_connection() cur = conn.cursor() # --- Seuils visuels (cohérents avec les VRAIS capteurs) --- # Nouvelles clés pour ignorer d'anciennes valeurs incohérentes restées en base. max_temperature = safe_float(get_setting_value(cur, "max_temp", "60")) or 60.0 min_voltage = safe_float(get_setting_value(cur, "min_voltage", "0.2")) or 0.2 min_power = safe_float(get_setting_value(cur, "min_power", "0.1")) or 0.1 # --- Alerte basée UNIQUEMENT sur les mesures réelles --- # - température DHT11 (temperature_ext) # - tension mesurée INA219 (voltage_pv) # - puissance calculée (power_pv) # Plus aucune référence à une "batterie" (pas de BMS sur ce montage). temp_value = safe_float(temperature_ext) voltage_value = safe_float(voltage_pv) power_value = safe_float(power_pv) battery_alert = "none" # nom de colonne conservé pour ne rien casser en base if temp_value is not None and temp_value > max_temperature: battery_alert = "Température trop élevée" elif voltage_value is not None and voltage_value < min_voltage: battery_alert = "Tension mesurée trop faible" elif power_value is not None and power_value < min_power: battery_alert = "Puissance calculée insuffisante" cur.execute(""" INSERT INTO telemetry ( temperature_ext, humidity_ext, system_status_msg, voltage_pv, current_pv, power_pv, luminosity, voltage_battery, current_battery, battery_temp, battery_level, battery_alert ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) """, ( temperature_ext, humidity_ext, system_status_msg, voltage_pv, current_pv, power_pv, luminosity, voltage_battery, current_battery, battery_temp, battery_level, battery_alert )) conn.commit() cur.close() conn.close() return jsonify({ "message": "Données enregistrées avec succès", "battery_alert": battery_alert }), 201 @app.route("/api/latest", methods=["GET"]) def latest_data(): conn = get_connection() cur = conn.cursor() # data_age_seconds = ancienneté de la dernière mesure, calculée côté serveur. # Permet au dashboard de savoir si l'ESP32 envoie encore des données récentes. cur.execute(""" SELECT id, created_at, EXTRACT(EPOCH FROM (NOW() - created_at)) AS age_seconds, temperature_ext, humidity_ext, system_status_msg, voltage_pv, current_pv, power_pv, luminosity, voltage_battery, current_battery, battery_temp, battery_level, battery_alert FROM telemetry ORDER BY created_at DESC LIMIT 1 """) row = cur.fetchone() cur.close() conn.close() if not row: return jsonify({"message": "Aucune donnée disponible"}), 404 age = row[2] return jsonify({ "id": row[0], "created_at": str(row[1]), "data_age_seconds": float(age) if age is not None else None, "temperature_ext": row[3], "humidity_ext": row[4], "system_status": row[5], "voltage_pv": row[6], "current_pv": row[7], "power_pv": row[8], "luminosity": row[9], "voltage_battery": row[10], "current_battery": row[11], "battery_temp": row[12], "battery_level": row[13], "battery_alert": row[14] }) @app.route("/api/history", methods=["GET"]) def history_data(): limit = request.args.get("limit", default=15, type=int) if limit is None or limit <= 0: limit = 15 if limit > 50: limit = 50 conn = get_connection() cur = conn.cursor() # On renvoie maintenant current_pv (courant mesuré) pour le graphique électrique. cur.execute(""" SELECT id, created_at, temperature_ext, humidity_ext, power_pv, current_pv FROM telemetry ORDER BY created_at DESC LIMIT %s """, (limit,)) rows = cur.fetchall() cur.close() conn.close() history = [] for row in reversed(rows): history.append({ "id": row[0], "created_at": str(row[1]), "temperature_ext": row[2], "humidity_ext": row[3], "power_pv": row[4], "current_pv": row[5] }) return jsonify(history) @app.route("/admin/login", methods=["GET", "POST"]) def admin_login(): if session.get("admin_logged_in"): return redirect(url_for("home")) if request.method == "POST": username = request.form.get("username", "").strip() password = request.form.get("password", "") if not username or not password: return render_template("login.html", error="Veuillez remplir tous les champs") conn = get_connection() cur = conn.cursor() cur.execute( "SELECT id, username, password FROM users WHERE username = %s AND is_admin = TRUE", (username,) ) user = cur.fetchone() cur.close() conn.close() if user: user_id, db_username, password_hash = user if check_password_hash(password_hash, password): session.clear() session["admin_logged_in"] = True session["admin_id"] = user_id session["admin_username"] = db_username return redirect(url_for("home")) return render_template("login.html", error="Identifiants incorrects") return render_template("login.html") @app.route("/admin/dashboard", methods=["GET", "POST"]) @admin_required def admin_dashboard(): conn = get_connection() cur = conn.cursor() if request.method == "POST": # Normalisation : on accepte la virgule ET le point (ex : 0,3 -> 0.3) min_voltage = normalize_number(request.form.get("min_battery")) max_temp = normalize_number(request.form.get("max_temp")) min_power = normalize_number(request.form.get("min_power")) upsert_setting(cur, "min_voltage", min_voltage) upsert_setting(cur, "max_temp", max_temp) upsert_setting(cur, "min_power", min_power) conn.commit() cur.close() conn.close() # Post/Redirect/Get : évite le ré-envoi du formulaire si on actualise, # et affiche proprement le message de succès. return redirect(url_for("admin_dashboard", saved=1)) settings = { "min_battery": get_setting_value(cur, "min_voltage", "0.2"), "max_temp": get_setting_value(cur, "max_temp", "60"), "min_power": get_setting_value(cur, "min_power", "0.1") } cur.close() conn.close() saved = request.args.get("saved") == "1" return render_template("admin_dashboard.html", settings=settings, saved=saved) @app.route("/admin/logout", methods=["GET", "POST"]) @admin_required def admin_logout(): session.clear() return redirect(url_for("admin_login")) if __name__ == "__main__": app.run(host="0.0.0.0", port=5000, debug=True)