datalogger/mqttinflux/main.py

49 lines
1.2 KiB
Python
Raw Normal View History

2021-02-19 11:06:59 +00:00
import paho.mqtt.client as mqtt
from influxdb import InfluxDBClient
import datetime
import logging
import json
import os
def on_connect(client, userdata, flags, rc):
for topic in topics:
client.subscribe(topic)
def on_message(client, userdata, msg):
current_time = datetime.datetime.utcnow().isoformat()
data = json.loads(msg.payload)
print(data)
json_body = [
{
"measurement": msg.topic,
"tags": {},
"time": current_time,
"fields": data
}
]
logging.info(json_body)
influx_client.write_points(json_body)
if 'TOPICS' not in os.environ:
print("At least one topic is needed.")
exit(1)
topics = os.environ['TOPICS'].split(' ')
print(topics)
logging.basicConfig(level=logging.INFO)
influx_client = InfluxDBClient(os.environ['INFLUX_HOST'], int(os.environ['INFLUX_PORT']), database=os.environ["INFLUX_DB"])
influx_client.create_database(os.environ["INFLUX_DB"])
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.connect(os.environ['MQTT_HOST'], int(os.environ["MQTT_PORT"]), 60)
print("Starting")
client.loop_forever()