1. #!/usr/bin/env /opt/rh/rh-python36/root/usr/bin/python3
  3. import paho.mqtt.client as mqtt
  4. import datetime
  5. import time
  6. from influxdb import InfluxDBClient
  8. def on_connect(client, userdata, flags, rc):
  9. print("Connected with result code "+str(rc))
  10. client.subscribe("home/#")
  13. def on_message(client, userdata, msg):
  14. print("Received a message on topic: " + msg.topic)
  15. # Use utc as timestamp
  16. receiveTime=datetime.datetime.utcnow()
  17. message=msg.payload.decode("utf-8")
  18. isfloatValue=False
  19. try:
  20. # Convert the string to a float so that it is stored as a number and not a string in the database
  21. val = float(message)
  22. isfloatValue=True
  23. except:
  24. print("Could not convert " + message + " to a float value")
  25. isfloatValue=False
  27. if isfloatValue:
  28. print(str(receiveTime) + ": " + msg.topic + " " + str(val))
  30. json_body = [
  31. {
  32. "measurement": msg.topic,
  33. "time": receiveTime,
  34. "fields": {
  35. "value": val
  36. }
  37. }
  38. ]
  40. dbclient.write_points(json_body)
  41. print("Finished writing to InfluxDB")
  43. # Set up a client for InfluxDB
  44. dbclient = InfluxDBClient('localhost', 8086, 'root', 'dusnikeair7', 'home')
  46. # Initialize the MQTT client that should connect to the Mosquitto broker
  47. client = mqtt.Client()
  48. client.username_pw_set("mqtt", "nikeair7")
  49. client.on_connect = on_connect
  50. client.on_message = on_message
  51. connOK=False
  52. while(connOK == False):
  53. try:
  54. client.connect("localhost", 1883, 60)
  55. connOK = True
  56. except:
  57. connOK = False
  58. time.sleep(2)
  60. # Blocking loop to the Mosquitto broker
  61. client.loop_forever()