WeeWX MQTT driver

mqtt.py

"""
A custom weewx driver to support collecting data from a MQTT broker.
In combination with https://github.com/WouterGritter/433-mqtt-bridge, this driver can be used to
collect information from any 433, 868 or 915 MHz weather station using a Software Defined Radio dongle.
Using a MQTT broker as a go-between, the same raw data can be used in other places, for example in
Home Assistant or in Grafana.

This driver expects the units to be in a specific metric format. See conversion in the source-code
below (around line 70), and feel free to modify the multiplier values.

To install this driver, simply drop it in the weewx/drivers folder and update your configuration.
Typically, for a linux installation, this would be /usr/share/weewx/weewx/drivers/.

Make sure to install the paho-mqtt pip package as well.

Then, to configure the driver, use the following driver config in your configuration file (typically /etc/weewx/weewx.conf):
[Station]
    # ... your other configuration

    station_type = Mqtt

[Mqtt]
    mqtt_broker_host = localhost
    mqtt_broker_port = 1883

    temperature_topic = weatherstation/temperature
    humidity_topic = weatherstation/humidity
    gust_speed_topic = weatherstation/gustspeed
    wind_speed_topic = weatherstation/windspeed
    wind_direction_topic = weatherstation/winddirection
    uv_topic = weatherstation/uv
    light_topic = weatherstation/light
    rain_topic = weatherstation/rain
    rain_rate_topic = weatherstation/rain_rate
    pressure_topic = weatherstation/pressure

    hardware_name = My Weather Station Hardware

    driver = weewx.drivers.mqtt
"""

import time
from dataclasses import dataclass
from datetime import datetime
from typing import Optional

import paho.mqtt.client as mqtt

from typing_extensions import Any

import weewx.drivers

DRIVER_NAME = 'Mqtt'
DRIVER_VERSION = '1.0'


@dataclass
class PacketValueMap:
    name: str
    topic: str
    multiplier: float = 1.0


def loader(config_dict, engine):
    mqtt_config = config_dict['Mqtt']

    return Mqtt(
        mqtt_broker_host=config_dict['Mqtt']['mqtt_broker_host'],
        mqtt_broker_port=int(config_dict['Mqtt']['mqtt_broker_port']),
        value_maps=[
            PacketValueMap('outTemp', mqtt_config['temperature_topic']),  # Deg. Celsius
            PacketValueMap('outHumidity', mqtt_config['humidity_topic']),  # % RH
            PacketValueMap('windGust', mqtt_config['gust_speed_topic'], 3.6),  # m/s -> km/h
            PacketValueMap('windSpeed', mqtt_config['wind_speed_topic'], 3.6),  # m/s -> km/h
            PacketValueMap('windGustDir', mqtt_config['wind_direction_topic']),  # Degrees (0-360)
            PacketValueMap('windDir', mqtt_config['wind_direction_topic']),  # Degrees (0-360)
            PacketValueMap('UV', mqtt_config['uv_topic']),  # UV-Index
            PacketValueMap('radiation', mqtt_config['light_topic'], 0.01075),  # lux -> W/m^2
            PacketValueMap('rainTotal', mqtt_config['rain_topic']),  # mm
            PacketValueMap('rainRate', mqtt_config['rain_rate_topic']),  # mm/h
            PacketValueMap('pressure', mqtt_config['pressure_topic']),  # hPa
        ],
        hardware_name=config_dict['Mqtt']['hardware_name'],
    )


class Mqtt(weewx.drivers.AbstractDevice):

    def __init__(self, mqtt_broker_host: str, mqtt_broker_port: int, value_maps: list[PacketValueMap], hardware_name: str):
        self.__mqtt_broker_host = mqtt_broker_host
        self.__mqtt_broker_port = mqtt_broker_port
        self.__value_maps = value_maps
        self.__hardware_name = hardware_name

        self.__pending_values: Optional[dict[str, Any]] = None
        self.__last_value_received = datetime.now()

        self.__mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
        self.__mqttc.on_connect = self.__mqtt_on_connect
        self.__mqttc.on_message = self.__mqtt_on_message

        self.__mqtt_connect()

    def __mqtt_connect(self):
        self.__mqttc.connect(self.__mqtt_broker_host, self.__mqtt_broker_port, 60)
        self.__mqttc.loop_start()

    def __mqtt_on_connect(self, client, userdata, flags, reason_code, properties):
        topics = set([vm.topic for vm in self.__value_maps])
        for topic in topics:
            self.__mqttc.subscribe(topic)

    def __mqtt_on_message(self, client, userdata, msg):
        topic = msg.topic
        value = float(msg.payload.decode())

        for vm in self.__value_maps:
            if topic == vm.topic:
                self.__update_value(vm.name, value * vm.multiplier)

    def __update_value(self, name: str, value: float):
        if self.__pending_values is None:
            self.__pending_values = {}

        self.__pending_values[name] = value
        self.__last_value_received = datetime.now()

    def genLoopPackets(self):
        while True:
            time.sleep(0.25)

            if self.__pending_values is None:
                # No new values.
                continue

            now = datetime.now()
            if (now - self.__last_value_received).total_seconds() < 1.0:
                # Received a value less than 1 second ago. Expect more values to come in.
                continue

            packet = {
                'dateTime': int(time.time() + 0.5),
                'usUnits': weewx.METRIC
            }

            for key, value in self.__pending_values.items():
                packet[key] = value

            yield packet

    @property
    def hardware_name(self):
        return self.__hardware_name


def confeditor_loader():
    return MqttConfEditor()


class MqttConfEditor(weewx.drivers.AbstractConfEditor):

    @property
    def default_stanza(self):
        return """
[Mqtt]
    mqtt_broker_host = localhost
    mqtt_broker_port = 1883

    temperature_topic = weatherstation/temperature
    humidity_topic = weatherstation/humidity
    gust_speed_topic = weatherstation/gustspeed
    wind_speed_topic = weatherstation/windspeed
    wind_direction_topic = weatherstation/winddirection
    uv_topic = weatherstation/uv
    light_topic = weatherstation/light
    rain_topic = weatherstation/rain
    rain_rate_topic = weatherstation/rain_rate
    pressure_topic = weatherstation/pressure

    hardware_name = My Weather Station Hardware

    driver = weewx.drivers.mqtt
"""