139 lines
4.7 KiB
Python
139 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Cariflex Simulator - Posts data to FlexMeasures via session cookies."""
|
|
import requests
|
|
import random
|
|
import math
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
FM_HOST = "https://cariflex.digitribe.fr"
|
|
FM_EMAIL = "admin@digitribe.fr"
|
|
with open("/tmp/fm_pass.txt") as f:
|
|
FM_PASSWORD=f.read().strip()
|
|
|
|
SENSORS = {}
|
|
for i in range(1, 11):
|
|
SENSORS[40 + i] = {"name": f"pv_{i:02d}_power", "type": "pv", "unit": "kW", "min": 0, "max": 5}
|
|
for i in range(1, 11):
|
|
SENSORS[50 + i] = {"name": f"bat_{i:02d}_power", "type": "battery", "unit": "kWh", "min": 10, "max": 100}
|
|
for i in range(1, 11):
|
|
SENSORS[60 + i] = {"name": f"chg_{i:02d}_power", "type": "ev_charger", "unit": "kW", "min": 0, "max": 22}
|
|
for i in range(1, 11):
|
|
SENSORS[70 + i] = {"name": f"ev_{i:02d}_power", "type": "ev_v2g", "unit": "kWh", "min": 15, "max": 75}
|
|
|
|
def generate_value(cfg, hour):
|
|
t = cfg["type"]
|
|
if t == "pv":
|
|
if 6 <= hour <= 18:
|
|
factor = max(0, math.sin((hour - 6) * math.pi / 12))
|
|
else:
|
|
factor = 0
|
|
return round(max(0, cfg["max"] * factor + random.gauss(0, 0.3)), 2)
|
|
elif t == "battery":
|
|
base = 50 + 30 * math.sin((hour - 6) * math.pi / 12)
|
|
return round(max(cfg["min"], min(cfg["max"], base + random.gauss(0, 2))), 2)
|
|
elif t == "ev_charger":
|
|
if 8 <= hour <= 22:
|
|
factor = random.uniform(0.2, 1.0)
|
|
else:
|
|
factor = random.uniform(0, 0.15)
|
|
return round(max(0, cfg["max"] * factor + random.gauss(0, 0.5)), 2)
|
|
elif t == "ev_v2g":
|
|
if 0 <= hour <= 6:
|
|
base = 60
|
|
elif 17 <= hour <= 21:
|
|
base = 30
|
|
else:
|
|
base = 45
|
|
return round(max(cfg["min"], min(cfg["max"], base + random.gauss(0, 3))), 2)
|
|
return 0
|
|
|
|
def main():
|
|
print(f"Cariflex Simulator -> {FM_HOST}", flush=True)
|
|
print(f" Sensors: {len(SENSORS)}", flush=True)
|
|
|
|
# Create session with CSRF token handling
|
|
session = requests.Session()
|
|
session.verify = False # Self-signed cert
|
|
|
|
# Get login page to extract CSRF token
|
|
login_url = f"{FM_HOST}/login"
|
|
r = session.get(login_url)
|
|
print(f" Login page: {r.status_code}", flush=True)
|
|
|
|
# Extract CSRF token
|
|
from html.parser import HTMLParser
|
|
class CSRFParser(HTMLParser):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.token = None
|
|
def handle_starttag(self, tag, attrs):
|
|
if tag == "input":
|
|
attrs = dict(attrs)
|
|
if attrs.get("name") == "csrf_token":
|
|
self.token = attrs.get("value")
|
|
|
|
parser = CSRFParser()
|
|
parser.feed(r.text)
|
|
csrf_token = parser.token
|
|
print(f" CSRF token: {csrf_token[:20] if csrf_token else 'NOT FOUND'}...", flush=True)
|
|
|
|
# Login
|
|
r = session.post(login_url, data={
|
|
"email": FM_EMAIL,
|
|
"password": FM_PASSWORD,
|
|
"csrf_token": csrf_token,
|
|
"remember": "y",
|
|
}, allow_redirects=True)
|
|
print(f" Login: {r.status_code} (url: {r.url})", flush=True)
|
|
|
|
# Check if logged in
|
|
if "dashboard" in r.url or "login" not in r.url:
|
|
print(f" Logged in successfully!", flush=True)
|
|
else:
|
|
print(f" WARNING: May not be logged in (url: {r.url})", flush=True)
|
|
|
|
# Post sensor data
|
|
iteration = 0
|
|
while True:
|
|
now = datetime.now(timezone.utc)
|
|
hour = now.hour
|
|
|
|
success = 0
|
|
failed = 0
|
|
|
|
for sensor_id, cfg in SENSORS.items():
|
|
value = generate_value(cfg, hour)
|
|
start = (now - 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} - {r.text[:100]}", 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)
|
|
|
|
import time
|
|
time.sleep(30)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|