Add influxdb connector
This commit is contained in:
parent
b8e964078a
commit
e3e57c3b09
2 changed files with 35 additions and 0 deletions
|
@ -8,3 +8,6 @@ services:
|
||||||
clustering:
|
clustering:
|
||||||
build: .
|
build: .
|
||||||
command: python3 cluster.py
|
command: python3 cluster.py
|
||||||
|
influx:
|
||||||
|
build: .
|
||||||
|
command: python3 mqtt2influx.py
|
||||||
|
|
32
mqtt2influx.py
Normal file
32
mqtt2influx.py
Normal file
|
@ -0,0 +1,32 @@
|
||||||
|
import paho.mqtt.subscribe as sub
|
||||||
|
import json
|
||||||
|
from influxdb import InfluxDBClient
|
||||||
|
import datetime
|
||||||
|
|
||||||
|
|
||||||
|
DB = ('habctrl', 8086)
|
||||||
|
BROKER = ('homeproxy', 1883)
|
||||||
|
TOPIC_PREFIX = 'scalefix'
|
||||||
|
TOPIC_WEIGHTS = TOPIC_PREFIX+'/weights'
|
||||||
|
|
||||||
|
def handle_msg(client, userdata, message):
|
||||||
|
data = json.loads(message.payload.decode('utf-8'))
|
||||||
|
print(f'got data for {data["user"]}: {data["weight"]}')
|
||||||
|
client = InfluxDBClient(DB[0], DB[1], 'scalefix', 'weighting', 'scale')
|
||||||
|
json_body = [
|
||||||
|
{
|
||||||
|
"measurement": "weights",
|
||||||
|
"tags": {
|
||||||
|
"user": data['user'],
|
||||||
|
},
|
||||||
|
"time": datetime.datetime.fromtimestamp(data['timestamp']).isoformat(),
|
||||||
|
"fields": {
|
||||||
|
"value": data['weight']
|
||||||
|
}
|
||||||
|
}
|
||||||
|
]
|
||||||
|
client.write_points(json_body)
|
||||||
|
client.close()
|
||||||
|
|
||||||
|
|
||||||
|
sub.callback(handle_msg, [TOPIC_WEIGHTS], hostname=BROKER[0], port=BROKER[1])
|
Loading…
Reference in a new issue