Inial version 1.0.0

README.md

@@ -1,2 +1,57 @@
-# tibberLive2mqtt
-Publishes tibber live consumption data from tibber api per MQTT.
+# tibber2mqtt
+Publishes Tibber live consumption data from Tibber API per MQTT.
+
+## Introduction
+
+If you are a customer of [Tibber](https://tibber.com/) and use a [Tibber Pulse](https://tibber.com/de/pulse), you can retrieve realtime consumption data via the Tibber API.
+
+This program publishes realtime consumption data from Tibber to one or more MQTT brokers of your choise.
+
+## Get tibber token and home id
+
+* Go to https://developer.tibber.com/settings/access-token , log in with your Tibber credentials and you will see the access token
+* After this, open https://developer.tibber.com/explorer and paste the following code into the left text box:
+```
+{
+  viewer {
+    homes {
+      id
+    }
+  }
+}
+```
+* Execute the query (play button above the text box) and you will see your home id in the left text box
+
+## **Prerequisites**
+
+- Python version 3.8 or newer with pip + venv
+
+This program should run in any OS, but I have no capacity to test this, so feedback is appreciated. My test machines run Ubuntu and Raspbian.
+
+## **Install**
+
+```
+git clone https://github.com/danielringch/tibber2mqtt.git
+python3 -m venv <path to virtual environment>
+source <path to virtual environment>/bin/activate
+python3 -m pip install -r requirements.txt
+```
+
+## **Configuration**
+
+The configuration is done via yaml file. The example file can be found in [config/sample.yaml](config/sample.yaml)
+
+To keep sensitive content out of config files, some parameters can also be passed using environment variables. See the example config file for further explanations.
+
+## **Usage**
+
+```
+source <path to virtual environment>/bin/activate
+python3 -B tibber2mqtt/tibber2mqtt.py --config /path/to/your/config/file.yaml
+```
+
+## **Get support**
+
+You have trouble getting started? Something does not work as expected? You have some suggestions or thoughts? Please let me know.
+
+Feel free to open an issue here on github or contact me on reddit: [3lr1ng0](https://www.reddit.com/user/3lr1ng0).

config/sample.yaml

@@ -0,0 +1,12 @@
+log:
+  path: "~/foo.log" # optional; alternatively, use environment variable T2M_LOG_PATH
+tibber:
+  token: "tibbertoken" # alternatively, use environment variable T2M_TIBBER_TOKEN
+  home: "homeid" # alternatively, use environment variable T2M_TIBBER_TOKEN
+mqtt:
+  server1: # a name of your choise, will be shown in the logs
+    host: "127.0.0.1:1883"
+    topic: "foo/abc"
+  server2:
+    host: "1.2.3.4:1883"
+    topic: "bar/def"

requirements.txt

@@ -0,0 +1,3 @@
+paho-mqtt
+pytibber
+pyyaml

tibber2mqtt/helpers.py

@@ -0,0 +1,27 @@
+import os
+from functools import reduce
+
+def get_argument(config: dict, keys: list, varname: str, optional=False):
+    try:
+        return os.environ[varname]
+    except:
+        try:
+            for key in keys:
+                dict = dict[key]
+            return str(dict)
+        except:
+            if optional:
+                return None
+            print(f'Error: Missing config entry or environment variable {varname}.')
+            exit()
+
+def get_recursive_key(dict, *keys, optional=False):
+    final_key = keys[-1]
+    for key in keys[:-1]:
+        dict = dict.get(key, {})
+    value = dict.get(final_key, None)
+    if optional or value:
+        return value
+    else:
+        print(f'Key {".".join(keys)} not found in configuration.')
+        exit()

tibber2mqtt/logger.py

@@ -0,0 +1,23 @@
+import datetime
+
+class Logger:
+    def __init__(self):
+        self.__handle = None
+
+    def __del__(self):
+        if self.__handle:
+            self.__handle.close()
+
+    def add_file(self, path):
+        if not path:
+            return
+        self.__handle = open(path, 'a')
+
+    def log(self, message):
+        message = f'[{datetime.datetime.now().strftime("%Y-%m-%dT%H:%M:%S")}] {message}\n'
+        print(message, end='')
+        if self.__handle:
+            self.__handle.write(message)
+            self.__handle.flush()
+
+logger = Logger()

tibber2mqtt/mqtt.py

@@ -0,0 +1,28 @@
+import struct
+import paho.mqtt.client as mqtt
+from helpers import get_recursive_key
+from logger import *
+
+class Mqtt():
+    def __init__(self, name: str, config: dict):
+        self.__name = name
+
+        self.__mqtt = mqtt.Client()
+        self.__mqtt.on_connect = self.__on_connect
+
+        ip, port = get_recursive_key(config, 'host').split(':')
+
+        self.__topic = get_recursive_key(config, 'topic')
+
+        self.__mqtt.connect(ip, int(port), 60)
+        self.__mqtt.loop_start()
+
+    def __del__(self):
+        self.__mqtt.loop_stop()
+
+    def send(self, value):
+        self.__mqtt.publish(self.__topic, struct.pack('!H', int(value)), qos=0, retain=False)
+        logger.log(f'[{self.__name}] Sent {value} to {self.__topic}')
+
+    def __on_connect(self, client, userdata, flags, rc):
+        logger.log(f'[{self.__name}] MQTT connected with code {rc}.')

tibber2mqtt/tibber2mqtt.py

@@ -0,0 +1,54 @@
+import argparse, asyncio, os, yaml
+from mqtt import Mqtt
+from helpers import get_argument
+from tibberlive import Tibberlive
+from watchdog import Watchdog
+from logger import Logger, logger
+
+__version__ = "0.1.0"
+
+
+async def main():
+    parser = argparse.ArgumentParser(description='Publishes tibber live consumption data from tibber api per MQTT.')
+    parser.add_argument('-c', '--config', type=str, required=True, help="Path to config file.")
+    args = parser.parse_args()
+
+    print(f'tibberLive2mqtt {__version__}')
+
+    try:
+        with open(args.config, "r") as stream:
+            config = yaml.safe_load(stream)
+    except Exception as e:
+        print(f'Failed to load config file {args.config}: {e}')
+        exit()
+
+    try:
+        log_file = get_argument(config, ('log', 'path'), 'T2M_LOG_PATH', optional=True)
+        logger.add_file(log_file)
+    except Exception as e:
+        print(f'Failed to open log file {log_file}: {e}')
+        exit()
+
+    mqtts = []
+    for mqttname, mqttconfig in config['mqtt'].items():
+        mqtts.append(Mqtt(mqttname, mqttconfig))
+
+    tibber = None
+    watchdog = Watchdog()
+
+    while True:
+        if tibber is None:
+            tibber = Tibberlive(config.get('tibber', {}), mqtts)
+            watchdog.subscription_success(await tibber.start())
+        if not watchdog.check(tibber.last_data):
+            tibber.stop()
+            tibber = None
+        await asyncio.sleep(2)
+
+if __name__ == "__main__":
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()

tibber2mqtt/tibberlive.py

@@ -0,0 +1,83 @@
+import aiohttp, datetime, tibber, threading, asyncio
+from enum import Enum
+
+from helpers import get_argument
+from mqtt import Mqtt
+from logger import *
+
+class Tibberlive:
+    def __init__(self, config: dict, mqtts: list):
+
+        home = get_argument(config, ('home'), "T2M_TIBBER_HOME")
+        self.__token = get_argument(config, ('token'), "T2M_TIBBER_TOKEN")
+
+        self.__available_request = {
+            'query': '{ viewer { home(id: "%s") { features { realTimeConsumptionEnabled } } } }' % home
+        }
+        self.__subscription_request = {
+            'query': 'subscription{ liveMeasurement( homeId:"%s" ) { timestamp power } }' % home
+        }
+
+        self.__home = None
+
+        self.__mqtts = mqtts
+
+        self.__last_timestamp = datetime.datetime.fromtimestamp(0)
+
+    def __del__(self):
+        self.stop()
+
+    async def start(self):
+        logger.log(f'Subscribing to tibber live data.')
+        async with aiohttp.ClientSession() as session:
+            try:
+                response_json = await self.__post(session, self.__available_request)
+            except Exception as e:
+                logger.log(f'Subscription to tibber live data failed: {e}')
+                return False
+            available = response_json['data']['viewer']['home']['features']['realTimeConsumptionEnabled']
+            if not available:
+                logger.log('No tibber live data available .')
+                return False
+
+            tibber_connection = tibber.Tibber(self.__token, websession=session, user_agent="HomeAssistant/2023.2")
+            await tibber_connection.update_info()
+
+            self.__home = tibber_connection.get_homes()[0]
+            try:
+                await self.__home.rt_subscribe(self.__power_callback)
+            except Exception as e:
+                logger.log(f'Subscription to tibber live data failed: {e}')
+                return False
+        return True
+
+    def stop(self):
+        if self.__home is not None:
+            try:
+                self.__home.rt_unsubscribe()
+            except:
+                pass
+
+    @property
+    def last_data(self):
+        return self.__last_timestamp
+
+    def __power_callback(self, data):
+        self.__last_timestamp = datetime.datetime.now()
+        power = data['data']['liveMeasurement']['power']
+        for mqtt in self.__mqtts:
+            mqtt.send(round(power + 0.5))
+
+    async def __post(self, session, query):
+        headers = {
+            'Authorization': f'Bearer {self.__token}',
+            'Content-Type': 'application/json'
+        }
+        async with session.post('https://api.tibber.com/v1-beta/gql', json=query, headers=headers) as response:
+            response_json = await response.json()
+            status = response.status
+        if not (status >= 200 and status <= 299):
+            raise Exception(status)
+        return response_json
+
+

tibber2mqtt/watchdog.py

@@ -0,0 +1,33 @@
+import datetime
+
+from logger import *
+
+class Watchdog:
+    def __init__(self):
+        self.__tolerance = datetime.timedelta(seconds=10)
+        self.__lost_at = None
+        self.__timeout = 5
+
+    def check(self, last_timestamp):
+        now = datetime.datetime.now()
+        if last_timestamp + self.__tolerance < now:
+            if self.__lost_at is None:
+                logger.log('Lost tibber live data.')
+                self.__lost_at = now
+            if self.__lost_at + datetime.timedelta(seconds=self.__timeout) < now:
+                return False
+        else:
+            self.__reset_timeout()
+            self.__lost_at = None
+        return True
+
+    def subscription_success(self, success):
+        self.__lost_at = datetime.datetime.now()
+        if success:
+            self.__reset_timeout()
+        else:
+            self.__timeout = min (self.__timeout * 2, 3600)
+            logger.log(f'Reconnect attempt in {self.__timeout} seconds.')
+
+    def __reset_timeout(self):
+        self.__timeout = 5