133 lines
4.3 KiB
Python
133 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Cariflex - Injecte les données météo réelles et prix DSO dans FM via le CLI."""
|
|
|
|
import subprocess
|
|
import json
|
|
import random
|
|
import math
|
|
from datetime import datetime, timezone, timedelta
|
|
import pytz
|
|
import requests
|
|
|
|
MARTINIQUE_TZ = pytz.timezone("America/Martinique")
|
|
|
|
# Sensor IDs (from previous creation)
|
|
SENSORS = {
|
|
"irradiance": 81,
|
|
"temperature": 82,
|
|
"wind_speed": 83,
|
|
"consumption_price": 84,
|
|
"production_price": 85,
|
|
}
|
|
|
|
def run_fm_cli(args):
|
|
"""Run FM CLI inside the container."""
|
|
cmd = ["docker", "exec", "flexmeasures-server", "bash", "-c",
|
|
f"cd /app && .venv/bin/flexmeasures {' '.join(args)}"]
|
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
|
|
return result.stdout.strip(), result.stderr.strip(), result.returncode
|
|
|
|
def post_sensor_data(sensor_id, values, unit, start, duration="PT1H"):
|
|
"""Post sensor data via FM API."""
|
|
import re
|
|
|
|
# Login first
|
|
session = requests.Session()
|
|
session.verify = False
|
|
r = session.get("https://cariflex.digitribe.fr/login")
|
|
match = re.search(r'<input[^>]*csrf_token[^>]*value="([^"]+)"', r.text)
|
|
if not match:
|
|
return False
|
|
csrf = match.group(1)
|
|
r = session.post("https://cariflex.digitribe.fr/login", data={
|
|
"email": "admin@digitribe.fr", "password": "Digitribe972",
|
|
"csrf_token": csrf, "remember": "y"
|
|
}, allow_redirects=True)
|
|
|
|
if "dashboard" not in r.url:
|
|
return False
|
|
|
|
# Post data - use proper ISO format with timezone
|
|
r = session.post(
|
|
f"https://cariflex.digitribe.fr/api/v3_0/sensors/{sensor_id}/data",
|
|
json={
|
|
"values": values,
|
|
"start": start,
|
|
"duration": duration,
|
|
"unit": unit
|
|
},
|
|
timeout=30
|
|
)
|
|
return r.status_code in [200, 201, 202]
|
|
|
|
# ========================================
|
|
# 1. Données météo réelles via API Open-Meteo
|
|
# ========================================
|
|
print("=== Récupération des données météo ===")
|
|
|
|
LAT, LON = 14.6091, -61.2155
|
|
today = datetime.now(MARTINIQUE_TZ).strftime("%Y-%m-%d")
|
|
|
|
resp = requests.get(
|
|
f"https://api.open-meteo.com/v1/forecast?"
|
|
f"latitude={LAT}&longitude={LON}"
|
|
f"&hourly=shortwave_radiation,temperature_2m,wind_speed_10m"
|
|
f"&timezone=America/Martinique"
|
|
f"&start_date={today}&end_date={today}",
|
|
timeout=15
|
|
)
|
|
|
|
if resp.status_code == 200:
|
|
wdata = resp.json()
|
|
times = wdata["hourly"]["time"]
|
|
|
|
for sensor_name, (key, unit) in {
|
|
"irradiance": ("shortwave_radiation", "W/m²"),
|
|
"temperature": ("temperature_2m", "°C"),
|
|
"wind_speed": ("wind_speed_10m", "m/s")
|
|
}.items():
|
|
sensor_id = SENSORS[sensor_name]
|
|
values = wdata["hourly"][key]
|
|
posted = 0
|
|
|
|
for i, t in enumerate(times):
|
|
# Parse datetime and format properly
|
|
dt = datetime.fromisoformat(t)
|
|
dt_utc = dt.astimezone(timezone.utc)
|
|
start_str = dt_utc.strftime("%Y-%m-%dT%H:%M:%S+00:00")
|
|
val = round(max(0, values[i]), 1)
|
|
|
|
if post_sensor_data(sensor_id, [val], unit, start_str):
|
|
posted += 1
|
|
|
|
print(f" '{sensor_name}': {posted}/{len(times)} posted")
|
|
else:
|
|
print(f" Erreur API: {resp.status_code}")
|
|
|
|
# ========================================
|
|
# 2. Prix DSO simulés (24h)
|
|
# ========================================
|
|
print("\n=== Injection des prix DSO ===")
|
|
|
|
now_utc = datetime.now(timezone.utc)
|
|
|
|
for sensor_name, sensor_id in [("consumption_price", SENSORS["consumption_price"]),
|
|
("production_price", SENSORS["production_price"])]:
|
|
posted = 0
|
|
for h in range(24):
|
|
dt = now_utc - timedelta(hours=23-h)
|
|
hour_of_day = dt.hour
|
|
|
|
if 6 <= hour_of_day <= 22:
|
|
price = round(random.uniform(80, 150), 2) if "consumption" in sensor_name else round(random.uniform(60, 120), 2)
|
|
else:
|
|
price = round(random.uniform(40, 80), 2) if "consumption" in sensor_name else round(random.uniform(30, 60), 2)
|
|
|
|
start_str = dt.strftime("%Y-%m-%dT%H:%M:%S+00:00")
|
|
if post_sensor_data(sensor_id, [price], "EUR/MWh", start_str):
|
|
posted += 1
|
|
|
|
print(f" '{sensor_name}': {posted}/24 posted")
|
|
|
|
print("\n=== Terminé ===")
|