#!/usr/bin/env python3 """Cariflex - Script optimal de simulation avec page web intégrée.""" import json import requests import re import warnings import random import math from datetime import datetime, timezone, timedelta from flask import Flask, render_template_string, request, jsonify import pytz warnings.filterwarnings("ignore") FM_HOST = "https://cariflex.digitribe.fr" CREDS_FILE = "/tmp/fm_creds.json" MARTINIQUE_TZ = pytz.timezone("America/Martinique") # ======================================== # Web App # ======================================== app = Flask(__name__) DASHBOARD_HTML = """ Cariflex EMS - Simulation

🌴 Cariflex EMS - Simulation Temps Réel

Production PV (kW)

SOC Batteries (kWh)

Charge VE (kW)

Prix DSO (EUR/MWh)

Irradiance (W/m²)

Température (°C)

Flexibilité (kW)

Balance Réseau (kW)

📊 Données en temps réel

Chargement...

⚙️ Paramètres

""" # Global state sim_state = { "interval": 30, "pv_mult": 1.0, "ev_mult": 1.0, "use_real_weather": True, "running": False, "log": [], "last_data": {} } def fm_login(): session = requests.Session() session.verify = False r = session.get(f"{FM_HOST}/login") match = re.search(r']*csrf_token[^>]*value="([^"]+)"', r.text) csrf = match.group(1) r = session.post(f"{FM_HOST}/login", data={ "email": "admin@digitribe.fr", "password": "Digitribe972", "csrf_token": csrf, "remember": "y" }, allow_redirects=True) return session if "dashboard" in r.url else None def get_fm_data(session): """Get latest data from FM sensors.""" data = {} # Get sensor data via API sensors = { "pv": (41, 50), # PV sensors "bat": (51, 60), # Battery sensors "ev": (61, 70), # EV charger sensors "v2g": (71, 80), # V2G sensors "irradiance": 81, "temperature": 82, "consumption_price": 84, } for name, sensor_range in sensors.items(): if isinstance(sensor_range, tuple): # Multiple sensors - get aggregate total = 0 count = 0 for sid in range(sensor_range[0], sensor_range[1] + 1): r = session.get(f"{FM_HOST}/api/v3_0/sensors/{sid}/data?limit=1") if r.status_code == 200: try: vals = r.json() if vals and len(vals) > 0: total += vals[-1].get("event_value", 0) count += 1 except: pass data[name] = total if count > 0 else 0 else: # Single sensor r = session.get(f"{FM_HOST}/api/v3_0/sensors/{sensor_range}/data?limit=1") if r.status_code == 200: try: vals = r.json() if vals and len(vals) > 0: data[name] = vals[-1].get("event_value", 0) except: pass return data @app.route("/") def index(): return render_template_string(DASHBOARD_HTML) @app.route("/api/data") def api_data(): session = fm_login() if not session: return jsonify({"error": "Auth failed"}) data = get_fm_data(session) # Calculate derived values pv_power = data.get("pv", 0) bat_soc = data.get("bat", 0) ev_load = data.get("ev", 0) v2g_power = data.get("v2g", 0) price = data.get("consumption_price", 0) irradiance = data.get("irradiance", 0) temperature = data.get("temperature", 0) # Flexibility = total battery + V2G capacity flexibility = bat_soc + v2g_power # Balance = Production - Consumption balance = pv_power - ev_load result = { "pv_power": pv_power, "bat_soc": bat_soc, "ev_load": ev_load, "price": price, "irradiance": irradiance, "temperature": temperature, "flexibility": flexibility, "balance": balance, "log": "
".join(sim_state["log"][-20:]) if sim_state["log"] else "En attente de données..." } sim_state["last_data"] = result return jsonify(result) @app.route("/api/params", methods=["POST"]) def api_params(): params = request.json sim_state["interval"] = int(params.get("interval", 30)) sim_state["pv_mult"] = float(params.get("pv_mult", 1.0)) sim_state["ev_mult"] = float(params.get("ev_mult", 1.0)) sim_state["use_real_weather"] = bool(params.get("use_real_weather", True)) return jsonify({"status": "ok"}) if __name__ == "__main__": app.run(host="0.0.0.0", port=5001, debug=False)