From 4b9958dd25bf67ec2a9fb6bbb9f1e2c7e9ee9321 Mon Sep 17 00:00:00 2001 From: Dennis Eriksen Date: Tue, 31 Jan 2023 07:49:27 +0100 Subject: handle HAN as well --- mqtt2pgsql.py | 73 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ mqtt2pgsql.service | 19 ++++++++++++++ mqtt_temps.py | 67 ------------------------------------------------- mqtt_temps.service | 19 -------------- 4 files changed, 92 insertions(+), 86 deletions(-) create mode 100644 mqtt2pgsql.py create mode 100644 mqtt2pgsql.service delete mode 100644 mqtt_temps.py delete mode 100644 mqtt_temps.service diff --git a/mqtt2pgsql.py b/mqtt2pgsql.py new file mode 100644 index 0000000..004216e --- /dev/null +++ b/mqtt2pgsql.py @@ -0,0 +1,73 @@ +#!/usr/bin/env python3 + +import os +import sys +import json +import psycopg2 +import paho.mqtt.client as mqtt + +mqtt_server = os.environ['el_mqtt_server'] +mqtt_port = int(os.environ['el_mqtt_port']) +keepalive = int(os.environ['el_mqtt_keepalive']) +mqtt_topic = os.environ['el_mqtt_topic'] + +pg_db = os.environ['el_pg_db'] +pg_host = os.environ['el_pg_host'] +pg_user = os.environ['el_pg_user'] +pg_pass = os.environ['el_pg_pass'] +pg_table = "mqtt_temps" + + +# The callback for when the client receives a CONNACK response from the server. +def on_connect(client, userdata, flags, rc): + print("Connected with result code "+str(rc)) + + # Subscribing in on_connect() means that if we lose the connection and + # reconnect then subscriptions will be renewed. + client.subscribe(mqtt_topic) + +# The callback for when a PUBLISH message is received from the server. +def on_message(client, userdata, msg): + name = msg.topic.split('/')[1] + data = json.loads(msg.payload) + if name.startswith('tmp') and 'temperature' in data and 'humidity' in data: + sql = "INSERT INTO mqtt_temps (name, temperature, humidity, battery, linkquality, voltage) VALUES(%s,%s,%s,%s,%s,%s)" + values = (name, data['temperature'], data['humidity'], data['battery'], data['linkquality'], data['voltage']) + todb(sql,values) + + elif name == 'HAN' and 'current' in data: + sql = "INSERT INTO mqtt_han (name, current, power, voltage, linkquality) VALUES(%s,%s,%s,%s,%s)" + values = (name, data['current'], data['power'], data['voltage'], data['linkquality']) + todb(sql,values) + + +# Write values to database +def todb(sql, values): + print(values) + conn = psycopg2.connect(database=pg_db, host=pg_host, user=pg_user, password=pg_pass) + cur = conn.cursor() + try: + cur.execute(sql, values) + conn.commit() + except (Exception, psycopg2.Error) as error: + conn.rollback() + print("Error while connecting to PostgreSQL", error) + raise error + finally: + cur.close() + conn.close() + + + +# mqtt +client = mqtt.Client() +client.on_connect = on_connect +client.on_message = on_message + +client.connect(mqtt_server, mqtt_port, keepalive) + +# Blocking call that processes network traffic, dispatches callbacks and +# handles reconnecting. +# Other loop*() functions are available that give a threaded interface and a +# manual interface. +client.loop_forever() diff --git a/mqtt2pgsql.service b/mqtt2pgsql.service new file mode 100644 index 0000000..d400b10 --- /dev/null +++ b/mqtt2pgsql.service @@ -0,0 +1,19 @@ +[Unit] +Description=Simple service to start python-listener +StartLimitIntervalSec=100 +StartLimitBurst=5 + +[Service] +Type=simple +EnvironmentFile=%h/energyscripts/.env +WorkingDirectory=%h/energyscripts +ExecStart=%h/energyscripts/venv/bin/python3 -u mqtt2pgsql.py +Restart=on-failure +TimeoutStopSec=70 +RestartSec=30 +SyslogIdentifier=mqtt2pgsql +#StandardOutput=journal +#StandardError=journal + +[Install] +WantedBy=default.target diff --git a/mqtt_temps.py b/mqtt_temps.py deleted file mode 100644 index bda7b0f..0000000 --- a/mqtt_temps.py +++ /dev/null @@ -1,67 +0,0 @@ -#!/usr/bin/env python3 - -import os -import sys -import json -import psycopg2 -import paho.mqtt.client as mqtt - -mqtt_server = os.environ['el_mqtt_server'] -mqtt_port = int(os.environ['el_mqtt_port']) -keepalive = int(os.environ['el_mqtt_keepalive']) -mqtt_topic = os.environ['el_mqtt_topic'] - -pg_db = os.environ['el_pg_db'] -pg_host = os.environ['el_pg_host'] -pg_user = os.environ['el_pg_user'] -pg_pass = os.environ['el_pg_pass'] -pg_table = "mqtt_temps" - - -# The callback for when the client receives a CONNACK response from the server. -def on_connect(client, userdata, flags, rc): - print("Connected with result code "+str(rc)) - - # Subscribing in on_connect() means that if we lose the connection and - # reconnect then subscriptions will be renewed. - client.subscribe(mqtt_topic) - -# The callback for when a PUBLISH message is received from the server. -def on_message(client, userdata, msg): - if msg.topic.startswith('zigbee2mqtt/tmp'): - #print("TMP!! - " + msg.topic.split('/')[1]) - data = json.loads(msg.payload) - if 'temperature' in data and 'humidity' in data: - values = (msg.topic.split('/')[1], data['temperature'], data['humidity'], data['battery'], data['linkquality'], data['voltage']) - print(values) - todb(values) - -# Write values to database -def todb(values): - conn = psycopg2.connect(database=pg_db, host=pg_host, user=pg_user, password=pg_pass) - cur = conn.cursor() - try: - cur.execute("INSERT INTO " + pg_table + " (name, temperature, humidity, battery, linkquality, voltage) VALUES(%s,%s,%s,%s,%s,%s)", values) - conn.commit() - except (Exception, psycopg2.Error) as error: - conn.rollback() - print("Error while connecting to PostgreSQL", error) - raise error - finally: - cur.close() - conn.close() - - - -# mqtt -client = mqtt.Client() -client.on_connect = on_connect -client.on_message = on_message - -client.connect(mqtt_server, mqtt_port, keepalive) - -# Blocking call that processes network traffic, dispatches callbacks and -# handles reconnecting. -# Other loop*() functions are available that give a threaded interface and a -# manual interface. -client.loop_forever() diff --git a/mqtt_temps.service b/mqtt_temps.service deleted file mode 100644 index d6a859e..0000000 --- a/mqtt_temps.service +++ /dev/null @@ -1,19 +0,0 @@ -[Unit] -Description=Simple service to start python-listener -StartLimitIntervalSec=100 -StartLimitBurst=5 - -[Service] -Type=simple -EnvironmentFile=%h/energyscripts/.env -WorkingDirectory=%h/energyscripts -ExecStart=%h/energyscripts/venv/bin/python3 -u mqtt_temps.py -Restart=on-failure -TimeoutStopSec=70 -RestartSec=30 -SyslogIdentifier=mqtt_temps -#StandardOutput=journal -#StandardError=journal - -[Install] -WantedBy=default.target -- cgit v1.2.3