Add skills, docs, and tool references
- OpenOCPP skill for EV charger firmware - Energy markets documentation (EPEX SPOT, EEX GO, PPA) - R&D tools: HAMLET, OPLEM, OpenSTEF, OpenDSM - EV tools: CitrineOS, OpenOCPP, EVerest - Standards: Flex Ready, S2, OpenADR - Architecture and deployment docs
This commit is contained in:
77
scripts/influxdb_to_fm.py
Normal file
77
scripts/influxdb_to_fm.py
Normal file
@@ -0,0 +1,77 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Cariflex - Post InfluxDB sensor data to FlexMeasures API.
|
||||
Reads from InfluxDB and posts to FlexMeasures via REST API.
|
||||
"""
|
||||
import requests
|
||||
import json
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
# Config
|
||||
FM_HOST = "https://flexmeasures.digitribe.fr"
|
||||
FM_EMAIL = "admin@digitribe.fr"
|
||||
FM_PASSWORD = "Digitribe972"
|
||||
|
||||
INFLUXDB_HOST = "http://localhost:8086"
|
||||
INFLUXDB_TOKEN = "my-super-token"
|
||||
INFLUXDB_ORG = "digitribe"
|
||||
INFLUXDB_BUCKET = "smartcity"
|
||||
|
||||
def get_fm_token():
|
||||
"""Get FlexMeasures auth token."""
|
||||
r = requests.post(f"{FM_HOST}/api/v3_0/requestAuthToken",
|
||||
json={"email": FM_EMAIL, "password": FM_PASSWORD}, verify=False)
|
||||
if r.status_code == 200:
|
||||
return r.json().get("token")
|
||||
# Try alternative auth
|
||||
r = requests.post(f"{FM_HOST}/api/v3_0/login",
|
||||
json={"email": FM_EMAIL, "password": FM_PASSWORD}, verify=False)
|
||||
if r.status_code == 200:
|
||||
return r.json().get("token")
|
||||
print(f"Auth failed: {r.status_code} - {r.text[:200]}")
|
||||
return None
|
||||
|
||||
def get_influxdb_data(sensor_name, hours=1):
|
||||
"""Read sensor data from InfluxDB."""
|
||||
query = f'''
|
||||
from(bucket: "{INFLUXDB_BUCKET}")
|
||||
|> range(start: -{hours}h)
|
||||
|> filter(fn: (r) => r["_measurement"] == "mqtt_consumer")
|
||||
|> filter(fn: (r) => r["topic"] =~ /{sensor_name}/)
|
||||
|> limit(n: 10)
|
||||
'''
|
||||
r = requests.post(f"{INFLUXDB_HOST}/api/v2/query?org={INFLUXDB_ORG}",
|
||||
headers={"Authorization": f"Token {INFLUXDB_TOKEN}", "Content-Type": "application/vnd.flux"},
|
||||
data=query, verify=False)
|
||||
return r.text
|
||||
|
||||
def post_sensor_data(token, sensor_id, values, unit="kW"):
|
||||
"""Post sensor data to FlexMeasures."""
|
||||
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
|
||||
|
||||
# Format data for FlexMeasures
|
||||
data = {
|
||||
"sensor": sensor_id,
|
||||
"values": values,
|
||||
"unit": unit,
|
||||
"horizon": "PT0H"
|
||||
}
|
||||
|
||||
r = requests.post(f"{FM_HOST}/api/v3_0/sensors/data",
|
||||
headers=headers, json=data, verify=False)
|
||||
return r.status_code, r.text
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("Cariflex - InfluxDB → FlexMeasures bridge")
|
||||
print("=" * 50)
|
||||
|
||||
# Test auth
|
||||
token = get_fm_token()
|
||||
if token:
|
||||
print(f"✅ Auth OK: {token[:20]}...")
|
||||
else:
|
||||
print("❌ Auth failed")
|
||||
|
||||
# Test InfluxDB read
|
||||
data = get_influxdb_data("pv_41")
|
||||
print(f"\n📊 InfluxDB data sample:\n{data[:500]}")
|
||||
Reference in New Issue
Block a user