Einleitung
Die Ecoflow Public API ermöglicht den programmatischen Zugriff auf Ecoflow-Geräte wie Powerstations und PowerOcean-Systeme. In diesem Beitrag zeige ich, wie man einen Python-Client für die Ecoflow Public API entwickelt, der die erforderliche HMAC-SHA256-Signaturlogik verwendet. Der Client eignet sich zum Abrufen von Gerätelisten, Quota-Daten (Gerätestatus) und MQTT-Zertifikaten.
Voraussetzungen: Developer-Zugang
Um die API nutzen zu können, benötigst du Developer-Zugang (Registrierung unter developer-eu.ecoflow.com). Nach der typischerweise 3-7 Tage dauernden Freischaltung kannst du im Portal deinen Access Key und Secret Key generieren.
Und: Nutzer eines PowerOcean Plus sind aktuell außen vor, selbst einfachste Dinge wie „gibt es gerade einen Stromausfall“ liefert das System nicht über Schnittstellen zurück. Das muss einem bewusst sein, dass die Integration in lokale Smarthome-Systeme wie Home-Assistant dadurch deutlich erschwert wird.
Die Authentifizierungsmechanik: HMAC-SHA256 & Flatten-Logik
Die API setzt auf HMAC-SHA256 zur Authentifizierung. Eine Besonderheit der Ecoflow-API ist die sogenannte „Flatten“-Logik: Verschachtelte Dictionaries (JSON-Daten) müssen vor der Signaturberechnung in eine flache, alphabetisch sortierte Darstellung überführt werden. Nur so stimmt die Signatur mit dem Backend überein.
ACHTUNG: Folgendes Script funktioniert bei mir, allerdings kann ich keine Gewähr übernehmen ob es überall der Fall ist. Der API-Zugang sowie die Möglichkeit Skripte zum auslesen der API zu verwenden ist etwas für erfahrene Anwender, wer nicht sicher ist sollte davon abstand nehmen.
def flatten_params(params, parent_key=''):
"""Flattent verschachtelte Dictionaries für die HMAC-Signatur."""
items = []
if isinstance(params, dict):
for k, v in params.items():
new_key = f"{parent_key}.{k}" if parent_key else k
if isinstance(v, (dict, list)):
items.extend(flatten_params(v, new_key).items())
else:
if isinstance(v, bool): v = str(v).lower()
elif v is None: v = ''
items.append((new_key, str(v)))
elif isinstance(params, list):
for i, v in enumerate(params):
new_key = f"{parent_key}[{i}]"
if isinstance(v, (dict, list)):
items.extend(flatten_params(v, new_key).items())
else:
if isinstance(v, bool): v = str(v).lower()
elif v is None: v = ''
items.append((new_key, str(v)))
return dict(items)
Die Client-Implementierung (Das Skript)
Hier ist die vollständige Implementierung der Client-Klasse, die die Signatur automatisch für jeden GET- und POST-Request erstellt:
import hmac
import hashlib
import time
import random
import json
import requests
# ==========================================
# 1. DEINE ZUGANGSDATEN HIER EINTRAGEN
# ==========================================
ACCESS_KEY = "DEIN_ACCESS_KEY"
SECRET_KEY = "DEIn_SECRET_KEY"
SERIAL_NUMBER = "SERIENNUMMER"
BASE_URL = "https://api-e.ecoflow.com"
# Sicherheits-Check: Unsichtbare Leerzeichen entfernen
ACCESS_KEY = ACCESS_KEY.strip()
SECRET_KEY = SECRET_KEY.strip()
SERIAL_NUMBER = SERIAL_NUMBER.strip()
# ==========================================
# 2. ECOFLOW API CLIENT (Angelehnt an hassio-ecoflow-cloud)
# ==========================================
class EcoflowPublicApiClient:
def __init__(self, access_key, secret_key, base_url):
self.access_key = access_key
self.secret_key = secret_key
self.base_url = base_url
def _flatten(self, d: dict, parent_key: str = '', sep: str = '.') -> dict:
"""
Exakte Flatten-Logik aus der hassio-ecoflow-cloud Integration.
"""
items = []
for k, v in d.items():
new_key = f"{parent_key}{sep}{k}" if parent_key else k
if isinstance(v, dict):
items.extend(self._flatten(v, new_key, sep=sep).items())
elif isinstance(v, list):
for i, elem in enumerate(v):
list_key = f"{new_key}[{i}]"
if isinstance(elem, dict):
items.extend(self._flatten(elem, list_key, sep=sep).items())
else:
items.append((list_key, str(elem)))
else:
if isinstance(v, bool):
items.append((new_key, str(v).lower()))
else:
items.append((new_key, str(v)))
return dict(items)
def _sign(self, params: dict):
"""
Generiert die Signatur exakt wie die Integration in Home Assistant.
"""
nonce = str(random.randint(100000, 999999))
timestamp = str(int(time.time() * 1000))
flat = self._flatten(params)
keys = sorted(flat.keys())
qs = "&".join([f"{k}={flat[k]}" for k in keys])
if qs:
qs += "&"
plain = f"{qs}accessKey={self.access_key}&nonce={nonce}×tamp={timestamp}"
sign = hmac.new(
self.secret_key.encode('utf-8'),
plain.encode('utf-8'),
hashlib.sha256
).hexdigest()
return {
"accessKey": self.access_key,
"nonce": nonce,
"timestamp": timestamp,
"sign": sign,
"Content-Type": "application/json"
}
def get(self, endpoint, query=None):
if query is None: query = {}
headers = self._sign(query)
url = f"{self.base_url}{endpoint}"
return requests.get(url, headers=headers, params=query)
def post(self, endpoint, payload=None):
if payload is None: payload = {}
headers = self._sign(payload)
url = f"{self.base_url}{endpoint}"
return requests.post(url, headers=headers, json=payload)
# ==========================================
# 3. TEST-ROUTINEN AUSFÜHREN
# ==========================================
if __name__ == "__main__":
if ACCESS_KEY == "DEIN_ACCESS_KEY" or not ACCESS_KEY:
print("Bitte trage zuerst deine Zugangsdaten im Skript ein!")
exit()
client = EcoflowPublicApiClient(ACCESS_KEY, SECRET_KEY, BASE_URL)
print("\n" + "="*50)
print("TEST 1: Geräteliste abrufen (GET)")
print("="*50)
res_list = client.get("/iot-open/sign/device/list")
print(f" Status: {res_list.status_code} | Antwort: {res_list.text}")
if res_list.json().get("code") == "0":
devices = res_list.json().get("data", [])
print("\n" + "="*50)
print("TEST 2: POST /quota für ALLE Geräte (Prüfung auf Fehler 1006)")
print("="*50)
for dev in devices:
dev_sn = dev.get("sn")
dev_name = dev.get("productName", "Unbekannt (Möglicherweise Teil des PowerOcean)")
print(f"\n--- Prüfe SN: {dev_sn} ({dev_name}) ---")
# Wir senden einen generischen POST-Request an jedes Gerät
payload = {
"sn": dev_sn,
"params": {
"quotas": ["sys"]
}
}
res_post = client.post("/iot-open/sign/device/quota", payload=payload)
out_post = res_post.text
if len(out_post) > 300:
print(f" Status: {res_post.status_code} | Antwort: {out_post[:300]} ... [Gekürzt]")
else:
print(f" Status: {res_post.status_code} | Antwort: {out_post}")
print("\n" + "="*50)
print("TEST 3: MQTT Zertifikat abrufen (Wichtig für Home Assistant)")
print("="*50)
# Endpunkt korrigiert auf /iot-open/sign/certification
res_cert = client.get("/iot-open/sign/certification")
out_cert = res_cert.text
if len(out_cert) > 300:
print(f" Status: {res_cert.status_code} | Antwort: {out_cert[:300]} ... [Gekürzt]")
else:
print(f" Status: {res_cert.status_code} | Antwort: {out_cert}")
Praktische Nutzung & Wichtige Hinweise
Mit diesem Skript kannst du nun ganz einfach Daten abrufen. Ein kurzes Beispiel für die Abfrage (die Seriennummern etc. wurden ersetzt), hier sieht man auch das aktuelle Problem, an dem die Integration meines PowerOcean Plus in Homeassistant scheitert, Fehler 1006:
TEST 1: Geräteliste abrufen (GET)
==================================================
Status: 200 | Antwort: {"code":"0","message":"Success","data":[{"sn":"AB12CDE3FG4H5678","online":1,"productName":"PowerPulse"},{"sn":"YZ98XWV7UT654321","online":0,"productName":"Smart Plug"},{"sn":"YZ98XWV7UT659876","online":1,"productName":"Smart Plug"},{"sn":"YZ98XWV7UT721122","online":1,"productName":"Smart Plug"},{"sn":"Q987AB2CDE4F1234","online":1},{"sn":"P5CDX5WA9949585","online":0,"productName":"RIVER"}],"eagleEyeTraceId":"","tid":""}
==================================================
TEST 2: POST /quota für ALLE Geräte (Prüfung auf Fehler 1006)
==================================================
--- Prüfe SN: AB12CDE3FG4H5678 (PowerPulse) ---
Status: 200 | Antwort: {"code":"1006","message":"current device is not allowed to get device info","eagleEyeTraceId":"","tid":""}
--- Prüfe SN: YZ98XWV7UT654321 (Smart Plug) ---
Status: 200 | Antwort: {"code":"0","message":"Success","data":{},"eagleEyeTraceId":"","tid":""}
--- Prüfe SN: YZ98XWV7UT659876 (Smart Plug) ---
Status: 200 | Antwort: {"code":"0","message":"Success","data":{},"eagleEyeTraceId":"","tid":""}
--- Prüfe SN: YZ98XWV7UT721122 (Smart Plug) ---
Status: 200 | Antwort: {"code":"0","message":"Success","data":{},"eagleEyeTraceId":"","tid":""}
--- Prüfe SN: Q987AB2CDE4F1234 (Unbekannt (Möglicherweise Teil des PowerOcean)) ---
Status: 200 | Antwort: {"code":"1006","message":"current device is not allowed to get device info","eagleEyeTraceId":"","tid":""}
--- Prüfe SN: P5CDX5WA9949585 (RIVER) ---
Status: 200 | Antwort: {"code":"1006","message":"current device is not allowed to get device info","eagleEyeTraceId":"","tid":""}
==================================================
TEST 3: MQTT Zertifikat abrufen (Wichtig für Home Assistant)
==================================================
Status: 200 | Antwort: {"code":"0","message":"Success","data":{"certificateAccount":"open-1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d","certificatePassword":"f1e2d3c4b5a697887766554433221100","url":"mqtt-e.ecoflow.com","port":"8883","protocol":"mqtts"},"eagleEyeTraceId":"","tid":""}
Fehler 1006 („current device is not allowed…“)
Wenn dieser Fehlercode beim Abrufen von Daten auftritt, handelt es sich um ein serverseitiges Whitelisting-Problem:
- Account-Whitelisting: Es dauert oft 3 bis 7 Tage, bis ein neuer Developer-Account von EcoFlow vollständig freigeschaltet ist. Diese Frist ist abgelaufen, der Account ist mehrere Monate alt.
- Geräte-Whitelisting: EcoFlow hat für bestimmte (oft neuere) Geräte wie PowerOcean-Systeme den API-Zugriff standardmäßig restriktiver eingestellt. Das muss laut meiner Recherche angefragt werden, hat bei mir allerdings nicht zum gewünschten Erfolg geführt.
Der Lösungsansatz: Ich habe eine E-Mail an den EcoFlow-Support, die Seriennummer (SN) meines Geräts genannt und explizit um ein Whitelisting für den Public API Developer-Zugang gebeten. Der Developer-Account muss auf jeden Fall mit der selben E-Mailadresse erstellt worden sein, wie für die Anmeldung bei den Geräten verwendet wird.
Die Rückmeldung von Ecoflow war jedoch ernüchternd:
Es liegen keine Informationen darüber vor ob dies überhaupt für den PO+ kommen wird.
Der Umstand dass viele Kunden eine lokale Schnittstelle zum auslesen wünschen ist jedoch bekannt und diesen haben wir schon an die Entwicklung weiter gegeben. Wie das dann zukünftig aussehen soll wissen wir aber leider auch nicht.
Aktueller Stand, zusammengefasst
Die „kleinen“ Ecoflow-Geräte lassen sich über den API-Zugang abfragen und darauf aufbauend Automatisierungen erstellen. Die „großen“ Systeme der PowerOcean-Reihe mit EMS, Ersatzstrom etc., wo es sinnvoll wäre direkt zu wissen wie es um das Stromnetz bestellt ist, bleiben aktuell außen vor. Vielleicht ändert sich daran noch etwas, aktuell habe ich nicht viele Hoffnungen.
Mein Plan ist es, über einen Lesekopf am Stromzähler zu ermitteln ob es gerade einen Ausfall gibt und auf Basis der Information große Verbraucher wie die Wärmepumpe auszuschalten, damit der Akku nicht zu schnell entleert wird.
Am Ende ist es bei der Wahl einer Solaranlage also wichtig, auch diese Details zu klären – auch wenn das System ansonsten reibungslos seit gut einem Jahr arbeitet, wäre das eine Information gewesen, die ich gerne am Anfang transparent gehabt hätte.