MQTTClientHandler.py

import json
from datetime import datetime
import paho.mqtt.client as mqtt
from mqtt_influx_backend import InfluxDBWriter

class MQTTClientHandler:
    def __init__(self, broker_url: str, topic: str, influx_writer: InfluxDBWriter):
        self.broker_url = broker_url
        self.topic = topic
        self.influx_writer = influx_writer
        self.client = mqtt.Client()
        self.client.on_connect = self.on_connect
        self.client.on_message = self.on_message

    def on_connect(self, client, userdata, flags, rc):
        print("Connected with result code " + str(rc))
        client.subscribe(self.topic)

    def on_message(self, client, userdata, msg):

        msg = json.loads(msg.payload)
        print(msg)
        metadate = msg["metadata"]
        print(metadate["mac"])
        try:
            
            value = float(msg.get("co2"))

            self.influx_writer.write_point(
                measurement="sensor_data",
                tags={"time": metadate["mac"]},
                tags={"mac": metadate["mac"]},
                fields={"value": value},
                timestamp=metadate["timestamp"]
            )
            print("Wrote to InfluxDB:", value)
        except Exception as e:
            print("Error processing message:", e)

    def start(self):
        self.client.connect(self.broker_url)
        self.client.loop_forever()