diff --git a/scripts/__pycache__/cariflex_simulator.cpython-313.pyc b/scripts/__pycache__/cariflex_simulator.cpython-313.pyc new file mode 100644 index 0000000..d1f0fd9 Binary files /dev/null and b/scripts/__pycache__/cariflex_simulator.cpython-313.pyc differ diff --git a/scripts/cariflex_simulator.py b/scripts/cariflex_simulator.py index 745d0e0..d010e9e 100644 --- a/scripts/cariflex_simulator.py +++ b/scripts/cariflex_simulator.py @@ -1,128 +1,91 @@ #!/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 +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 = "admin@digitribe.fr" -with open("/tmp/fm_pass.txt") as f: - FM_PASSWORD=f.read().strip() +FM_EMAIL = _creds["email"] +FM_PASSWORD=_creds["password"] +MARTINIQUE_TZ = pytz.timezone("America/Martinique") SENSORS = {} for i in range(1, 11): - SENSORS[40 + i] = {"name": f"pv_{i:02d}_power", "type": "pv", "unit": "kW", "min": 0, "max": 5} + SENSORS[40 + i] = {"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} + SENSORS[50 + i] = {"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} + SENSORS[60 + i] = {"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} + SENSORS[70 + i] = {"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 + 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": - base = 50 + 30 * math.sin((hour - 6) * math.pi / 12) - return round(max(cfg["min"], min(cfg["max"], base + random.gauss(0, 2))), 2) + 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": - if 8 <= hour <= 22: - factor = random.uniform(0.2, 1.0) - else: - factor = random.uniform(0, 0.15) + 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": - if 0 <= hour <= 6: - base = 60 - elif 17 <= hour <= 21: - base = 30 - else: - base = 45 + 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']*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) - # Create session with CSRF token handling session = requests.Session() - session.verify = False # Self-signed cert + session.verify = False - # 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) + if not login(session): + print(" ERROR: Initial login failed!", flush=True) + return + print(" Logged in successfully", 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 - + 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 - timedelta(minutes=5)).isoformat() - + 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, - ) + 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) + print(f" Sensor {sensor_id}: HTTP {r.status_code}", flush=True) except Exception as e: failed += 1 if failed <= 3: @@ -130,8 +93,6 @@ def main(): iteration += 1 print(f" Iteration {iteration}: {success} OK, {failed} failed (hour={hour})", flush=True) - - import time time.sleep(30) if __name__ == "__main__": diff --git a/scripts/start_simulator.sh b/scripts/start_simulator.sh new file mode 100755 index 0000000..84a0428 --- /dev/null +++ b/scripts/start_simulator.sh @@ -0,0 +1,6 @@ +#!/bin/bash +# Cariflex Simulator Launcher +export FM_PASS="Digitribe972" +cd /home/eric/cariflex +nohup python3 -u scripts/cariflex_simulator.py > /tmp/simulator.log 2>&1 & +echo "Simulator started with PID $!"