feat: changed tibber library to tibber.py

DataPoints.py

@@ -1,9 +1,10 @@
 from datetime import datetime
 from influxdb_client import Point
+import json
 class Pulse:
     label = str(__name__).lower()
     def __init__(self,data):
-        measurement = data['liveMeasurement']
+        measurement = data.cache
         timestamp = measurement['timestamp']
         power = measurement['power']
         cost = measurement.get('accumulatedCost',None)
@@ -26,13 +27,15 @@ class Pulse:
 class Price:
     label = str(__name__).lower()
     def __init__(self,data):
-        measurement = data['data']['viewer']['homes'][0]['currentSubscription']['priceInfo']['current']
+        measurement = data.cache
         current_time_utc = datetime.utcnow()
         timestamp = current_time_utc.isoformat()
-        total = measurement.get('total',0.34)
+        total = measurement["total"]
+        tax = measurement["tax"]
         tags = {self.label: ""}
         fields = {
             "total": float(total),
+            "tax": float(tax)
         }
         self.datapoint = {"fields": fields,
                      "tags": tags,

get_price.py

@@ -1,46 +1,35 @@
 import os
-import requests
+import sys
+import tibber
 from influxdb_client import InfluxDBClient
 from influxdb_client.client.write_api import SYNCHRONOUS
 from DataPoints import Price
+import logging
 
+TOKEN = os.getenv('TOKEN', '')
+TIBBERTOKEN = os.getenv('TIBBERTOKEN', '')
+URL = os.getenv('URL', "")
+BUCKET = os.getenv('BUCKET', "tibber")
+ORG = os.getenv('ORG', "Default")
 
-TOKEN=os.getenv('TOKEN', '')
-TIBBERTOKEN=os.getenv('TIBBERTOKEN', '')
-URL = os.getenv('URL',"" )
-BUCKET = os.getenv('BUCKET',"tibber" )
-ORG = os.getenv('ORG',"Default" )
-
-# URL of the Tibber API
-url = 'https://api.tibber.com/v1-beta/gql'
-
-# Request headers
-headers = {
-    'authority': 'api.tibber.com',
-    'accept': 'application/json',
-    'authorization': f'Bearer {TIBBERTOKEN}',
-    'content-type': 'application/json',
-    'user-agent': 'python 3.10',
-}
-
-# GraphQL query
-query = '{"query":"{ viewer { homes { currentSubscription { priceInfo { current { total } } } } } }"}'
-
+# Logging configuration
+logger = logging.getLogger("TibberInflux")
+formatter = logging.Formatter(
+    '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+ch = logging.StreamHandler(sys.stdout)
+ch.setFormatter(formatter)
+logger.addHandler(ch)
+logger.setLevel(logging.INFO)
 
+__version__ = "v0.3.0"
+logger.info(__version__)
 
 client = InfluxDBClient(url=URL, token=TOKEN, org=ORG)
-
 write_api = client.write_api(write_options=SYNCHRONOUS)
 query_api = client.query_api()
-# Make the HTTP POST request
-response = requests.post(url, headers=headers, data=query)
 
-# Check for successful response
-if response.status_code == 200:
-    # Print the response content (JSON data)
-    print(response.json())
-else:
-    print(f'Error: HTTP {response.status_code}')
-
-p = Price(response.json()).get_datapoint()
+account = tibber.Account(TIBBERTOKEN)
+home = account.homes[0]
+p = Price(home.current_subscription.price_info.current).get_datapoint()
 write_api.write(record=p, bucket=BUCKET)
+logger.info(vars(p))

pulse.py

@@ -1,10 +1,7 @@
-# -*- coding: utf-8 -*-
+import asyncio
 import os
 import sys
 
-import tibber.const
-import asyncio
-import aiohttp
 import tibber
 from influxdb_client import InfluxDBClient
 from influxdb_client.client.write_api import SYNCHRONOUS
@@ -16,71 +13,40 @@ TIBBERTOKEN = os.getenv('TIBBERTOKEN', '')
 URL = os.getenv('URL', "")
 BUCKET = os.getenv('BUCKET', "tibber")
 ORG = os.getenv('ORG', "Default")
-#logging
+
+# Logging configuration
 logger = logging.getLogger("TibberInflux")
 formatter = logging.Formatter(
     '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
 ch = logging.StreamHandler(sys.stdout)
-#fh = logging.FileHandler('log.log')
 ch.setFormatter(formatter)
-#fh.setFormatter(formatter)
 logger.addHandler(ch)
-#logger.addHandler(fh)
-
 logger.setLevel(logging.INFO)
 
-__version__ = "v0.2.3"
+__version__ = "v0.3.0"
 logger.info(__version__)
-client = InfluxDBClient(url=URL, token=TOKEN, org=ORG)
 
+client = InfluxDBClient(url=URL, token=TOKEN, org=ORG)
 write_api = client.write_api(write_options=SYNCHRONOUS)
 query_api = client.query_api()
 
-
-def _incoming(pkg):
+def _incoming(data):
     try:
-        data = pkg.get("data")
-        if data is None:
-            exit(1)
         p = Pulse(data).get_datapoint()
         write_api.write(record=p, bucket=BUCKET)
         logger.info(p)
-        return True
-    except:
-        exit(1)
-
-
-async def run():
-    try:
-        conn = aiohttp.TCPConnector(limit_per_host=3)
-        async with aiohttp.ClientSession(trust_env=True, connector=conn) as session:
-
-            logger.info("connecting to tibber...")
-            if session.closed:
-                logger.error("session closed")
-                exit(1)
-            try:
-                tibber_connection = tibber.Tibber(TIBBERTOKEN, user_agent="python", websession=session)
-
-            except Exception as e:
-                logger.info("error connecting to tibber...")
-
-                logger.info(e)
-                raise e
-            await tibber_connection.update_info()
-        home = tibber_connection.get_homes()[0]
-        await home.rt_subscribe(_incoming)
-
-        while True:
-            await asyncio.sleep(5)
     except Exception as e:
-        logger.exception("Error occurred in run():")
+        logger.exception("Error in _incoming():")
         raise e
 
-loop = asyncio.new_event_loop()
-asyncio.set_event_loop(loop)
-try:
-    loop.run_until_complete(asyncio.gather(run(), return_exceptions=False))
-except Exception as e:
-    raise e
+account = tibber.Account(TIBBERTOKEN)
+home = account.homes[0]
 
+@home.event("live_measurement")
+async def show_current_power(data):
+    _incoming(data)
+
+home.start_live_feed(user_agent="UserAgent/0.0.1")
+logger.info("Livefeed started")
+while home.live_feed.running:
+    event = home.event("live_measurement")

requirements.txt

@@ -1,6 +1,6 @@
 influxdb-client[ciso]
 requests
-pyTibber @ git+https://github.com/Danielhiversen/pyTibber@f64cb36
+tibber.py
 tzdata
 pytz
 certifi