100 lines
3.8 KiB
Python
100 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Cariflex Simulator - Posts data to FlexMeasures via session cookies."""
|
|
import json, requests, random, math, time, warnings, re
|
|
from datetime import datetime, timedelta, timezone
|
|
import pytz
|
|
|
|
warnings.filterwarnings("ignore")
|
|
|
|
with open("/tmp/fm_creds.json") as f:
|
|
_creds = json.load(f)
|
|
|
|
FM_HOST = "https://cariflex.digitribe.fr"
|
|
FM_EMAIL = _creds["email"]
|
|
FM_PASSWORD=_creds["password"]
|
|
MARTINIQUE_TZ = pytz.timezone("America/Martinique")
|
|
|
|
SENSORS = {}
|
|
for i in range(1, 11):
|
|
SENSORS[40 + i] = {"type": "pv", "unit": "kW", "min": 0, "max": 5}
|
|
for i in range(1, 11):
|
|
SENSORS[50 + i] = {"type": "battery", "unit": "kWh", "min": 10, "max": 100}
|
|
for i in range(1, 11):
|
|
SENSORS[60 + i] = {"type": "ev_charger", "unit": "kW", "min": 0, "max": 22}
|
|
for i in range(1, 11):
|
|
SENSORS[70 + i] = {"type": "ev_v2g", "unit": "kWh", "min": 15, "max": 75}
|
|
|
|
def generate_value(cfg, hour):
|
|
t = cfg["type"]
|
|
if t == "pv":
|
|
factor = max(0, math.sin((hour - 6) * math.pi / 12)) if 6 <= hour <= 18 else 0
|
|
return round(max(0, cfg["max"] * factor + random.gauss(0, 0.3)), 2)
|
|
elif t == "battery":
|
|
return round(max(cfg["min"], min(cfg["max"], 50 + 30 * math.sin((hour - 6) * math.pi / 12) + random.gauss(0, 2))), 2)
|
|
elif t == "ev_charger":
|
|
factor = random.uniform(0.2, 1.0) if 8 <= hour <= 22 else random.uniform(0, 0.15)
|
|
return round(max(0, cfg["max"] * factor + random.gauss(0, 0.5)), 2)
|
|
elif t == "ev_v2g":
|
|
base = 60 if 0 <= hour <= 6 else 30 if 17 <= hour <= 21 else 45
|
|
return round(max(cfg["min"], min(cfg["max"], base + random.gauss(0, 3))), 2)
|
|
return 0
|
|
|
|
def login(session):
|
|
try:
|
|
r = session.get(f"{FM_HOST}/login", timeout=10)
|
|
if r.status_code != 200:
|
|
return False
|
|
match = re.search(r'<input[^>]*csrf_token[^>]*value="([^"]+)"', r.text)
|
|
if not match:
|
|
return False
|
|
r = session.post(f"{FM_HOST}/login", data={"email": FM_EMAIL, "password": FM_PASSWORD, "csrf_token": match.group(1), "remember": "y"}, allow_redirects=True, timeout=10)
|
|
return "dashboard" in r.url or "login" not in r.url
|
|
except Exception as e:
|
|
print(f" Login error: {e}", flush=True)
|
|
return False
|
|
|
|
def main():
|
|
print(f"Cariflex Simulator -> {FM_HOST}", flush=True)
|
|
print(f" Sensors: {len(SENSORS)}", flush=True)
|
|
print(f" Timezone: America/Martinique", flush=True)
|
|
|
|
session = requests.Session()
|
|
session.verify = False
|
|
|
|
if not login(session):
|
|
print(" ERROR: Initial login failed!", flush=True)
|
|
return
|
|
print(" Logged in successfully", flush=True)
|
|
|
|
iteration = 0
|
|
while True:
|
|
now_local = datetime.now(MARTINIQUE_TZ)
|
|
now_utc = datetime.now(timezone.utc)
|
|
hour = now_local.hour
|
|
success = 0
|
|
failed = 0
|
|
|
|
for sensor_id, cfg in SENSORS.items():
|
|
value = generate_value(cfg, hour)
|
|
start = (now_utc - timedelta(minutes=5)).isoformat()
|
|
try:
|
|
r = session.post(f"{FM_HOST}/api/v3_0/sensors/{sensor_id}/data",
|
|
json={"values": [value], "start": start, "duration": "PT5M", "unit": cfg["unit"], "type": "PostSensorDataRequest"}, timeout=30)
|
|
if r.status_code in [200, 201]:
|
|
success += 1
|
|
else:
|
|
failed += 1
|
|
if failed <= 3:
|
|
print(f" Sensor {sensor_id}: HTTP {r.status_code}", flush=True)
|
|
except Exception as e:
|
|
failed += 1
|
|
if failed <= 3:
|
|
print(f" Sensor {sensor_id}: {e}", flush=True)
|
|
|
|
iteration += 1
|
|
print(f" Iteration {iteration}: {success} OK, {failed} failed (hour={hour})", flush=True)
|
|
time.sleep(30)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|