86 lines
3.2 KiB
Python
86 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Cariflex - Déclenche le forecasting et scheduling FlexMeasures."""
|
|
|
|
import subprocess
|
|
import json
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
FM_HOST = "https://cariflex.digitribe.fr"
|
|
|
|
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)} 2>&1"]
|
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
|
|
return result.stdout.strip(), result.stderr.strip(), result.returncode
|
|
|
|
# ========================================
|
|
# 1. Forecasting - Prévisions PV
|
|
# ========================================
|
|
print("=== Forecasting PV ===")
|
|
for sensor_id in range(41, 51):
|
|
stdout, stderr, rc = run_fm_cli([
|
|
"add", "forecasts",
|
|
"--sensor", str(sensor_id),
|
|
"--to-date", (datetime.now(timezone.utc) + timedelta(hours=24)).strftime("%Y-%m-%dT%H:%M:%S+00:00"),
|
|
])
|
|
if "Successfully" in stdout or "SAVED" in stdout:
|
|
print(f" Sensor {sensor_id}: OK")
|
|
else:
|
|
print(f" Sensor {sensor_id}: {stdout[:100]}")
|
|
|
|
# ========================================
|
|
# 2. Forecasting - Prévisions Charge VE
|
|
# ========================================
|
|
print("\n=== Forecasting Charge VE ===")
|
|
for sensor_id in range(61, 71):
|
|
stdout, stderr, rc = run_fm_cli([
|
|
"add", "forecasts",
|
|
"--sensor", str(sensor_id),
|
|
"--to-date", (datetime.now(timezone.utc) + timedelta(hours=24)).strftime("%Y-%m-%dT%H:%M:%S+00:00"),
|
|
])
|
|
if "Successfully" in stdout or "SAVED" in stdout:
|
|
print(f" Sensor {sensor_id}: OK")
|
|
else:
|
|
print(f" Sensor {sensor_id}: {stdout[:100]}")
|
|
|
|
# ========================================
|
|
# 3. Scheduling - Plans de charge Batteries
|
|
# ========================================
|
|
print("\n=== Scheduling Batteries ===")
|
|
for sensor_id in range(51, 61):
|
|
stdout, stderr, rc = run_fm_cli([
|
|
"add", "schedule",
|
|
"--sensor", str(sensor_id),
|
|
"--start", datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S+00:00"),
|
|
"--duration", "PT24H",
|
|
"--resolution", "PT15M",
|
|
"--soc-at-start", "0.5",
|
|
"--flex-model", '{"soc-min": "10 kWh", "soc-max": "100 kWh", "power-capacity": "50 kW"}',
|
|
])
|
|
if "Successfully" in stdout or "SAVED" in stdout:
|
|
print(f" Sensor {sensor_id}: OK")
|
|
else:
|
|
print(f" Sensor {sensor_id}: {stdout[:100]}")
|
|
|
|
# ========================================
|
|
# 4. Scheduling - Plans de charge V2G
|
|
# ========================================
|
|
print("\n=== Scheduling V2G ===")
|
|
for sensor_id in range(71, 81):
|
|
stdout, stderr, rc = run_fm_cli([
|
|
"add", "schedule",
|
|
"--sensor", str(sensor_id),
|
|
"--start", datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S+00:00"),
|
|
"--duration", "PT24H",
|
|
"--resolution", "PT15M",
|
|
"--soc-at-start", "0.5",
|
|
"--flex-model", '{"soc-min": "15 kWh", "soc-max": "75 kWh", "power-capacity": "11 kW"}',
|
|
])
|
|
if "Successfully" in stdout or "SAVED" in stdout:
|
|
print(f" Sensor {sensor_id}: OK")
|
|
else:
|
|
print(f" Sensor {sensor_id}: {stdout[:100]}")
|
|
|
|
print("\n=== Terminé ===")
|