feature: restructured Datapoints

DataPoints.py

@@ -0,0 +1,24 @@
+from influxdb_client import Point
+class Pulse:
+    label = str(__name__).lower()
+    def __init__(self,data):
+        measurement = data['liveMeasurement']
+        timestamp = measurement['timestamp']
+        power = measurement['power']
+        cost = measurement.get('accumulatedCost',None)
+        lastMeterConsumption = measurement['lastMeterConsumption']
+        tags = {self.label: ""}
+        fields = {
+            "power": float(power),
+            "lastMeterConsumption": float(lastMeterConsumption)
+        }
+        if cost is not None:
+            fields.update({'accumulatedCost':cost})
+        self.datapoint = {"fields": fields,
+                     "tags": tags,
+                     "measurement": "pulse",
+                     "time": timestamp
+                     }
+    def get_datapoint(self):
+        return Point(measurement_name=str(self.label)).from_dict(self.datapoint)
+

pulse.py

@@ -1,86 +1,59 @@
-#!/usr/bin/python3
 # -*- coding: utf-8 -*-
 import os
-import time
+import random
 
-from dateutil.parser import parse
-
-
-print("tibber_v001")
 import tibber.const
 import asyncio
 import aiohttp
 import tibber
-# settings from EnvionmentValue
+from influxdb_client import InfluxDBClient
+from influxdb_client.client.write_api import SYNCHRONOUS
+from DataPoints import Pulse
+
 TOKEN=os.getenv('TOKEN', '')
 TIBBERTOKEN=os.getenv('TIBBERTOKEN', '')
 URL = os.getenv('URL',"" )
 BUCKET = os.getenv('BUCKET',"tibber" )
 ORG = os.getenv('ORG',"Default" )
 
-from influxdb_client import InfluxDBClient, Point
+__version__ = "v0.0.2"
-from influxdb_client.client.write_api import SYNCHRONOUS
+print(f"tibber_{__version__}")
-
-bucket = "Default"
 
 client = InfluxDBClient(url=URL, token=TOKEN, org=ORG)
 
 write_api = client.write_api(write_options=SYNCHRONOUS)
 query_api = client.query_api()
 
-def _callback(pkg):
+def _incoming(pkg):
     try:
         data = pkg.get("data")
         if data is None:
-            return
+            exit(1)
-        lm=data.get("liveMeasurement")
+        p = Pulse(data).get_datapoint()
-        print(lm)
+        write_api.write(record=p, bucket=BUCKET)
-        write_to_db(data)
+        print(p)
-        return lm
+        return True
-    except Exception as e:
+    except:
-        raise e
+        exit(1)
-
-
-def write_to_db(data,bucket = BUCKET):
-
-    measurement = data['liveMeasurement']
-    timestamp = measurement['timestamp']
-    power = measurement['power']
-    lastMeterConsumption = measurement['lastMeterConsumption']
-
-    tags = {"adress":""}
-    fields = {
-            "power": float(power),
-            "lastMeterConsumption": float(lastMeterConsumption)
-        }
-    datapoint = {"fields": fields,
-                 "tags": tags,
-                 "measurement": "pulse",
-                 time: timestamp
-
-                 }
-    p = Point(measurement_name="pulse").from_dict(datapoint)
-    write_api.write(record=p, bucket=bucket)
-
-
-
-
 
 async def run():
     async with aiohttp.ClientSession() as session:
+        if session.closed:
+            print("session closed")
+            exit(1)
         tibber_connection = tibber.Tibber(TIBBERTOKEN, websession=session, user_agent="python")
         await tibber_connection.update_info()
     home = tibber_connection.get_homes()[0]
     try:
-        await home.rt_subscribe(_callback)
+         await home.rt_subscribe(_incoming)
-    except:
-        raise Exception("Session gone")
 
-    while True:
-      await asyncio.sleep(4)
-
-loop = asyncio.get_event_loop()
-try:
-    loop.run_until_complete(run())
     except:
         exit(1)
+
+    while 1:
+        await asyncio.sleep(5)
+
+
+
+loop = asyncio.get_event_loop()
+loop.run_until_complete(run())