Tibber API - Real Time Subscription mit Python

edit: Hier ein WIKI Eintag mit einer Komplettlösung über VenusOS.

Moin zusammen.

Ich frage seit einiger Zeit die Tibber API nach Daten wie Strompreis etc mit einem Python Script ab. Klappt super, siehe ersten Teil meinem Codes.

Nun, da ich neben meinem Discovergy-Power-Up auch den Pulse laufen habe, möchte ich die Real Time Daten nicht nur über die Discovergy-API, sondern vorsichtshalber auch über die Tibber API abfragen. Das folgende Programmierbeispiel kommt mit den Strompreisdaten etc. prima klar, allerdings bekomme ich für den aktuellen Stromverbrauch keinen gültigen Wert (NONE).

Hat das von Euch schon jemand hinbekommen und kann mir sagen, was ich hier falsch mache?

Den Graph habe ich hierher:

https://developer.tibber.com/explorer

So wie ich das verstanden habe, loggt man sich dort mit dem Tibber Account ein, wählt dann rechts real time subscription und erhält den Graph, mit der entsprechenden Home-ID. Da kann ich mich aber auch irren. Allerings: auf den Play Button gedrückt erhalte ich auch die Realtime Daten. Ich wüsste aber nicht, woher man sonst seine Home-ID bekommen könnte, falls diese nicht stimmt.

Hin und wieder bekomm eich über den Explorer auch mal diese Fehlermeldung: "Your subscription data will appear here after server publication"

Das Ganze mag damit zusammenhängen, dass ich sowohl das Discovergy- als auch das Pulse-Power-Up im Home eingebunden habe. Ich habe schon bei Tibber nachgefragt, wie man die beiden Power-Ups getrennt voneinander auslesen kann. Noch hat's da niemand gelesen.

Falls Ihr den Code testen wollt, dann müsst Ihr natürlich noch Token und ggf. die Home-ID anpassen.

Ich hab's auch mal hiermit verucht:

https://github.com/Danielhiversen/pyTibber

Allerdings hagelt es da Fehlermeldungen dass es einem schwindelt. Vermutlich mal wieder ein Python-Versionsproblem. Ich habe 3.9.13 auf dem PC laufen.

Vielen Dank vorab.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Sat Mar 25 14:25:03 2023
@author: nick
"""
import requests
import json
def mynin(liste,anzahl):
output=[]
if anzahl > len(liste):
output.append("n/a")
return(output)
else:
dummymax = max(liste) + 1
i = 1
while i <= anzahl:
index_min = min(range(len(liste)), key=liste.__getitem__)
output.append(index_min)
liste[index_min] = dummymax
i+=1
output = sorted(output)
return(output)
# Setzen Sie die Tibber-Token-ID als Umgebungsvariable oder geben Sie sie direkt ein.
# Hier wird davon ausgegangen, dass sie als Umgebungsvariable gesetzt wurde.
tibber_token = "Mein Tibber Token"
# Die URL für den Strompreis-Endpunkt von Tibber
tibber_url = "https://api.tibber.com/v1-beta/gql"
# Die Abfrage, um den Strompreis zu erhalten
tibber_query = """
{
viewer {
homes {
currentSubscription {
priceInfo {
current {
total
energy
level
tax
}
}
}
}
}
}
"""
PQ = """
subscription{
liveMeasurement(homeId: "Meine Home ID"){
timestamp
power
accumulatedConsumption
accumulatedCost
currency
minPower
averagePower
maxPower
}
}
"""
tibber_squery = """
{
viewer {
homes {
id
features {
realTimeConsumptionEnabled
}
}
}
}
"""
# Fügen Sie den Authentifizierungsheader hinzu
tibber_headers = {"Authorization": "Bearer {}".format(tibber_token)}
# Führen Sie die Anfrage aus und geben Sie die Ergebnisse aus
tibber_response = requests.post(tibber_url, json={"query": tibber_query}, headers=tibber_headers)
if tibber_response.status_code== 200:
data = json.loads(tibber_response.text)
price_data = data["data"]["viewer"]["homes"][0]["currentSubscription"]["priceInfo"]["current"]
sStrompreis = ("{:.2f}".format(price_data["total"] ))
print(data)
print(sStrompreis)
print("-------------------------------------------------------------")
tibber_headers = {"Authorization": "Bearer {}".format(tibber_token)}
# Führen Sie die Abfrage aus
tibber_response = requests.post(tibber_url, json={"query": tibber_squery}, headers=tibber_headers)
if tibber_response.status_code == 200:
data = tibber_response.json()
print(data)
else:
print(f"Fehler beim Abrufen der Daten. Statuscode: {tibber_response.status_code}")
print("++++++++++++++++++")
# Fügen Sie den Authentifizierungsheader hinzu
tibber_headers3 = {"Authorization": "Bearer {}".format(tibber_token)}
# Führen Sie die Abfrage für liveMeasurement aus
tibber_response3 = requests.post(tibber_url, json={"query": PQ}, headers=tibber_headers3)
if tibber_response3.status_code == 200:
data = json.loads(tibber_response3.text)
print(data)
print(tibber_response3)
live_measurement = data.get("data", {}).get("subscription", {}).get("liveMeasurement")
if live_measurement:
power = live_measurement.get("power")
spower = ("{:.2f}".format(power))
fpower = float(spower)
print(data)
print(spower)
print(fpower)
else:
print("Keine Daten für liveMeasurement verfügbar.")
else:
print(f"Fehler beim Abrufen des Stromverbrauchs. Statuscode: {tibber_response3.status_code}") 
1 „Gefällt mir“

Tja, so viele Leser und keine Antworten. Also, ich habe jetzt auch wirklich mal das Netz durchgesucht und keinen einzigen funktionierenden Code gefunden. Auch tibber.py oder mytibber oder wie sie auch heißen laufen allesamt nicht. Oder ich bin zu dumm, sie zu bedienen, das kann natürlich auch sein.

So treffunsicher wie die Antworten die vom Tibber-Chat kommen, scheint mir da eine etwas unausgegohrene künstliche Intelligenz dahinterzustecken. Aber immerhin ist sie nett, mit Smileys, und so. Aber geholfen hat's nisher noch nicht. Ich bohre mal weiter nach. Vielleicht ist ja inzwischen einem der Leser der Durchbruch gelungen.

Immerhin weiß ich jetzt, dass wenn sowohl das Discovergy- und das Pulse Power-Up installiert sind, zur Berechnung der Kosten immer die gerade funktionierenden Datenherangezogen werden. Ich nehme an, dass das dann bei der API auch so ist. Da habe ich aber jetzt auch noch einmal speizell nachgefragt.

Folgt also gerne meinem Monolog... ?

Ich benutzte das hier für meinen MPPSolar WR .. habs etwas anpassen müssen klappt aber

1 „Gefällt mir“

Vieleicht habe ich zu viel getextet. Kurzform: ich möchte mit Python diesen Graph auslesen

subscription{
liveMeasurement(homeId: "Meine Home ID"){
timestamp
power
accumulatedConsumption
accumulatedCost
currency
minPower
averagePower
maxPower
}
}

Wie hier beschrieben:

Tibber Developer. Home-ID und Tibber Token habe ich.

Mein Skript liefert:

{'data': {'liveMeasurement': None}}
<Response [200]>
Keine Daten für liveMeasurement verfügbar.

Trotzdem Danke. :wink:

1 „Gefällt mir“

So,

ich habe jetzt mal eine Antwort von Tibber auf mein Problemchen erhalten. Kurzgefasst heißt es in etwa: "Wir wissen zwar wie wir unsere Daten bereitstellen, aber nicht, wie man sie wieder abruft". ? Wie gesagt, etwas verkürzt.

Vielleicht findet sich ja doch noch jemand, der den oben angebildeten Python-Code soweit ändern kann, dass auch die RealTimeSubscription funktioniert. Ich hab's nach wie vor nicht hinbekommen... ?

Moin.

Mir hat dieses Projekt bzw. File geholfen, welches ich in mein Projekt übernommen und zum Laufen gebracht habe.
https://github.com/turbosnute/tibberpulse-influxdb/blob/master/pulse.py#L46

Hier wird zunächst einmal die WebSession URI ermittelt und dann ein Listener registriert, der kontinuierlich mit Infos versorgt wird.

Hoffe, das hilft weiter.

@luchbewohner

YES!! :grinning:

Das hilft schon mal ein gutes Stück weiter, vielen Dank! :+1: Ich habe den Code ein wenig umgemuddelt, sodass mir fetchdata den Power-Wert zurückgibt. Am Ende soll mir

    while True:
print("Sleep for 5 secs.")
time.sleep(5)
print("Run GQL query.")
myP = fetch_data(ws_uri, subscription_query, headers)
Power = str(myP)
print("Aktueller Bezug: "+ Power)

dann alle 5 Sekunden den Power Wert ausgeben.

Das funktioniert zwar auch ab und an, allerdings bekomme ich, wie man hier sieht, spätestens im zweiten Loop:

tibberpulse-influxdb
Sleep for 5 secs.
Run GQL query.
Exception ignored in: <async_generator object Client.subscribe_async at 0x754f7ad0>
RuntimeError: async generator ignored GeneratorExit
Aktueller Bezug: 261.3
Sleep for 5 secs.
Run GQL query.
Too many open connections. Sleeping 10 minutes...
If you continue to see this error you can fix it by recreating the tibber token

Tibber erlaubt ja nur (ich glaube) 60 Abfragen / Minute. Wenn ich 5 Sekunden warte, dann solten das eigentlichg nicht zu viele Abfragen sein. Oder starte ich mit jedem Mal so einen "Listener", der im Hintergrund die Tibber APi spamt?

Wie kann ich denn verhindern, dass es ständig zu zu vielen Verbindungen (oder Abfragen) kommt? Ich möchte ja einfach nur alle 5 Sekunden den Power-Wert auslesen.

Nochmals vielen Dank bisher und auch schon mal vorab. Vielleicht weißt Du ja was ich ändern muss...

PS: Hier auch noch mal meine geänderte fetch_data-Routine. Vielleicht habe ich da ja etwas falsch verstanden:

def fetch_data(url, subscription_query, headers):
transport = WebsocketsTransport(
url=url,
headers=headers,
keep_alive_timeout=120
)
ws_client = Client(
transport=transport,
fetch_schema_from_transport=True
)
subscription = gql(subscription_query)
#print(subscription)
try:
for result in ws_client.subscribe(subscription):
mypower =  result["liveMeasurement"]["power"]
return(mypower)
# console_handler(result)
# print(result)
except Exception as ex:
module = ex.__class__.__module__
#print(module + ex.__class__.__name__)
exargs = str(ex.args)
if exargs.find("Too many open connections") != -1:
print("Too many open connections. Sleeping 10 minutes...")
print("If you continue to see this error you can fix it by recreating the tibber token")
time.sleep(600)

Der einfache Weg wäre den Webserver der Bridge dauerhaft zu aktivieren und die Daten direkt lokal auszulesen.

christiankratzer/pulse2mqtt: Poll raw sml data from tibber pulse/bridge and publish on mqtt (github.com)

micw/homedatabroker: A middleware that reads, converts and publishes metrics from various sources to various targets (github.com)

1 „Gefällt mir“

@mdkeil

Moin! Ja, die Bridge direkt auszulesen wäre wohl die Königsdisziplin. Vielen Dank für den Link! Für jeden, der Zugriff auf seinen Puls hat, ist das Vermutlich das Beste.

Mir geht's leider so wie vielen: mein Zähler hängt im Keller und zwischen mir und ihm schirmen 3 jeweils 30cm dicke Betondecken jede Verbindung zu mir ab. Deshalb ist mein Nachbar so nett, meine Bridge-Daten im Erdgeschoss über seinen Router ins Netz zu schieben. Theoretisch könnte ich jetzt einen ESP32 daneben legen, den den Pulse ausliest und die Daten in die Cloud schiebt. Wäre jetzt aber ein Wenig zu aufwändig.

Deshalb wäre eine Lösung des oben angegebenen Problems, speziell für mich, die einfachere Lösung. Besser wäre aber eigentlich Dein Weg. Danke deshlab auch dafür! :blush: :+1:

Tja,

ich hab's noch mal zusammen mit meinem Kumpel ChatGPT probiert, aber wir beide kommen nicht darum herum, immer wieder bei

"Too many open connections. Sleeping 5 more seconds..."

zu enden. Alle Jubeljahre gibt's mal einen Wert, dann wieder das. A small step for a man like me, but still no giant leap for mankind. :expressionless:

Es gibt vielleicht eine elegantere (aber nicht ganz einfache) Lösung, sofern bei Dir lokal ein mqtt-Broker läuft, den du über das Inet erreichbar machen kannst.. Dann könntest Du die Bridge-Config ändern, dass diese sich nicht direkt mit dem tibber/amazon-mqtt-server verbindet, sondern mit deinem lokalen.. dein lokaler Broker wird zusätzlich als mqtt-bridge konfiguriert und baut ein zusätzliche Verbindung (mit den originalen credentials der bridge) zum tibber/amazon-mqtt-server auf, um Daten von bzw. an die Bridge weiterzuleiten.. Schon hast Du auf deinem lokalen Broker sämtliche Daten vom Pulse, die dann noch decodiert werden müssen :wink:

1 „Gefällt mir“

@monokristallin
Nein, die Daten kommen nicht von fetch_data() (der Name der Funktion führt in die Irre).
Das Konzept ist folgendes

  1. Initialisierung -> Get HomeID & WebSocket URI
  2. Registrierung der Subscriptions. Das passiert in fetch_data().
  3. Dann wird auf Daten gewartet (for result in ws_client.subscribe(subscription):slight_smile:
  4. Wenn Daten vorliegen, dann wird console_handler() aufgerufen - mit den ausgelesenen Daten. Das Intervall wird dabei von Tibber bestimmt (also wenn neue Daten da sind) und liegt bei mir bei ca 3 Sec.
    Das bekomme ich dann raus:
2023-12-20 19:52:30,831 - Rotating Multiplus Control - INFO - Tibber->console_handler()
---- Output, Date ----
[{'measurement': 'pulse', 'time': '2023-12-20T19:52:28.000+01:00', 'tags': {'address': 'Sackgasse 2b'}, 'fields': {'power': 501.0, 'consumption': 8.9216, 'cost': 0.0, 'voltagePhase1': 233.3, 'voltagePhase2': 234.7, 'voltagePhase3': 235.3, 'currentL1': 1.24, 'currentL2': 2.17, 'currentL3': 0.71, 'lastMeterConsumption': 2134.6849, 'hourmultiplier': 20, 'daysInMonth': 31}}]
{'liveMeasurement': {'timestamp': '2023-12-20T19:52:28.000+01:00', 'power': 501, 'accumulatedConsumption': 8.9216, 'accumulatedCost': 0, 'voltagePhase1': 233.3, 'voltagePhase2': 234.7, 'voltagePhase3': 235.3, 'currentL1': 1.24, 'currentL2': 2.17, 'currentL3': 0.71, 'lastMeterConsumption': 2134.6849}}
2023-12-20 19:52:32,275 - Rotating Multiplus Control - INFO - Tibber->fetch_data(): result({'liveMeasurement': {'timestamp': '2023-12-20T19:52:31.000+01:00', 'power': 501, 'accumulatedConsumption': 8.9219, 'accumulatedCost': 0, 'voltagePhase1': 233.2, 'voltagePhase2': 234.6, 'voltagePhase3': 235.3, 'currentL1': 1.25, 'currentL2': 2.16, 'currentL3': 0.71, 'lastMeterConsumption': 2134.6852}})

Die "Power" ist dabei.
Aber da du die Subscription durch die Loop jagst, sagt Timber spätestens beim 2. Mal Versuch zu subscriben: neeee...
Ich habe das Problem, dass ich noch nicht weiß, wie man die Subskription "ordentlich" verlässt d.h. wenn ich das Programm per Ctrl+C abbreche, dann muss ich ggf ein paar Minuten warten, bis ich wieder neu subscriben kann. Soll sicherlich Multi Connections und damit die Last auf dem Server verhindern.
@mdkeil

Die Idee mit dem direkten Abfragen / MQTT-Traffic verbiegen ist auch nice. :slight_smile:
Allerdings sehe ich das Risiko, das sich mit einem FW-Update dieses Interface bzw. die Vorgehensweise/Permissions (Zugänge zum Webserver via webserver_force_enable und Zugriff auf die Bridge-Console bekommen usw.) ändern könnte. Und das dann wieder zu debuggen, kann Zeit kosten.
Daher bevorzuge ich die "offizielle" Schnittstelle in der Hoffnung, in Problemsituationen schneller Lösungen finden zu können. Aber diese Meinung reflektiert nur mein Bauchgefühl ...

Okay,

das hilft schon mal weiter. Ich werde mir den code dahingehend noch mal genau anschauen.

Zu Deinem Problem, vielleicht hilf es, die fetch_data Routine so zu beenden:

        if exargs.find("Too many open connections") != -1:
print("Too many open connections. Sleeping 5 more seconds...")
print("If you continue to see this error you can fix it by recreating the tibber token")
time.sleep(5)
else:
print(f"Error: {ex}")
finally:
ws_client.transport.close() 

Da bin ich mal mit ChatGTP gelandet. Geholfen hat es mir natürlich nicht, weil ich den von Dir beschrieben und von mir schon irgendwie geahnten Fehler mache.

Ansonsten kann man noch websockets anstelle von websocket verwenden. Das wäre dann mit asyncio zu bewerkstelligen. Liegt aber noch etwas jenseits meines Verständnisses.

Vielen Dank nochmals! :slightly_smiling_face:

@mdkeil

Das wäre dann auch noch eine Möglichkeit.

Nebenbei läuft bei mir übrigens auch noch ein Victron VenusOS das sich prima auslesen lässt. Aber ein Plugin, mit dem ich die Power-Daten über die Tibber-API ins VenusOS bekomme ist mir so auch noch nicht über den Weg gelaufen. Falls da jemand einen Weg kennt, das wäre quasi noch einfacher. Wenn auch wieder noch ein Schritt dazwischenhängt...

@zx6r Hi, ich habe auch einen MPP Solar WR und Tibber im Einsatz. Die Tibberdaten kann ich ohne Probleme abrufen :slight_smile: Mich würde interessieren über welchen Weg / Programm Du den MPP Solar WR mit den optimalen Ladedaten versorgst. Beste Grüße und schon mal Danke für eine Antwort!

Da auf dem large-image von VenusOS auch ein Node Red läuft.. es gibt dort einen fertigen Flow, die sämtliche Tibber-Daten abholen können.. in Kombination mit:

mr-manuel/venus-os_dbus-mqtt-grid: This Venus OS driver gets the data from MQTT and displays it as grid meter. (github.com)

könntest Du dir aus den Tibber-Daten ein entsprechendes Grid-Meter erstellen und in VenusOS einbinden-

Hi.

Also, bei mir macht eigentlich alles ein zentrales, selbstgebasteltes Python-Script. Das ließt Daten vom Venus-OS und z.B. von Discovergy aus. Leteres ist in letzter Zeit etwas instabil, deshalb der Versuch, alternativ Tibberdaten auszulesen.

Dieses Pythonskript steurt dann auf den Raspberry Pi entsprechend dem aktuellen Verbrauch, dem Strompreis etc. den Soyosource WR über USB -> RS485 direkt an.

Tja, hätte ich mal das Large-Image installiert! :wink: Das wäre dann auch noch eine Möglichkeit. Das ließe sich ja noch schnell flashen. Allerdings müsste ich mich da noch in Node-Red einfuchsen. Eleganter fänd ich die Möglichkeit, mein zentrales Python-Script die Daten selbst abholen zu lassen. Das funktioniert ja auch schon fast...

Aber ja, falls alles andere nichts klappt, dann wäre das womöglich der nächste Schritt. Vielen dank also auch dafür! :+1:

@solarffm
Geht relativ einfach, wenn man Solpiplog am Laufen hat.

Das Skript https://github.com/christian1980nrw/Spotmarket-Switcher läuft Stündlich und schaltet zu den billigen Zeiten ja standardmäßig z.b. Steckdosen an oder aus .

Hier habe ich angesetzt und das Skript um die passende Zeile erweitert um per MQTT über Solpiplog den MPPSolar WR von SBU auf UTI zu schalten.

(mosquitto_pub -u emonpi -h localhost -t solpiplog/pip/status/set -m sbu)

Das war’s auch schon J

Klappt prima .. hab es aber trotzdem nicht mehr laufen .. weil ich meinen Akku nur mit max 30-40A laden möchte .. dann braucht es halt ein paar Stunden bis er einigermaßen voll ist!

Da reicht es dann wenn man im Winter immer zwischen 0 und 6 Uhr lädt.

Das kann man recht einfach in Solpiplog einstellen und bedarf keiner weiteren Handstände :wink:

@zx6r Wie oben beschrieben: Ich kann problemlos alle Daten von Tibber auslesen. Nur mit den Power-Daten, also der Realtime Subsciption, hapert es noch.