Add MQTT dedup republish filter example (closes #2990)

examples/mqtt_filter.py

@@ -0,0 +1,190 @@
+#!/usr/bin/env python3
+""" MQTT republishing filter to rename and deduplicate rtl_433 data
+
+Example program for receiving, filtering and republishing sensor data
+from rtl_433 using MQTT network messages.
+
+An MQTT broker e.g. 'mosquitto' must be running and rtl_433 must publish
+to that broker, e.g. using -F mqtt.
+"""
+
+import json
+import logging
+import sys
+import time
+import socket
+import re
+
+import paho.mqtt.client as mqtt
+from paho.mqtt.subscribeoptions import SubscribeOptions
+
+HOSTNAME = socket.getfqdn()
+
+MQTT_SERVER = "127.0.0.1" # set your MQTT broker address here
+
+MQTT_TOPIC_PREFIX = "rtl_433/+" # default to use all rtl_433 senders
+# MQTT_TOPIC_PREFIX = "rtl_433/" + HOSTNAME # alternative: use just the local host topics
+# MQTT_TOPIC_PREFIX = "rtl_433/MYSERVER" # alternativ: use a named rtl_433 sender
+
+MQTT_TOPIC_DEVICES = MQTT_TOPIC_PREFIX + "/devices" # default "devices" topic tree base
+MQTT_TOPIC_EVENTS = MQTT_TOPIC_PREFIX + "/events" # default "events" topic
+
+# set source and target topics as well as deduplication in seconds, 0 disables deduplication
+DEVICE_MAPPINGS = [
+ { "source": "Bresser-3CH/1/130", "target": "Livingroom", "dedup": 1.5},
+ { "source": "Bresser-3CH/2/175", "target": "Bedroom", "dedup": 1.5},
+ { "source": "LaCrosse-TX141THBv2/2/122", "target": "Garage", "dedup": 0},
+]
+
+EVENT_MAPPINGS = [
+ { "source": { "model": "Bresser-3CH", "id": 130 }, "target": { "nickname": "Livingroom" }, "dedup": 1.5},
+ { "source": { "model": "Bresser-3CH", "id": 175 }, "target": { "nickname": "Bedroom" }, "dedup": 1.5},
+ { "source": { "model": "LaCrosse-TX141THBv2", "id": 122 }, "target": { "nickname": "Garage" }, "dedup": 0},
+]
+
+EVENT_REPUBLISHED_KEY = "republished" # a JSON key only present in republished messages
+# EVENT_REPUBLISHED_KEY = "nickname" # can also be key added with "target"
+
+if hasattr(mqtt, 'CallbackAPIVersion'): # paho >= 2.0.0
+ mqtt_client = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION1, client_id="RTL_433_Filter")
+else:
+ mqtt_client = mqtt.Client(client_id="RTL_433_Filter")
+
+DEVICES_RE = MQTT_TOPIC_DEVICES.replace("+", ".+")
+EVENTS_RE = MQTT_TOPIC_EVENTS.replace("+", ".+")
+
+dedup_cache = {}
+
+
+def publish_dedup(topic, payload, timeout):
+ """ Deduplicate and publish a message. """
+ global dedup_cache
+ global mqtt_client
+
+ if timeout <= 0:
+ logging.debug("republishing " + topic + " : " + str(payload))
+ mqtt_client.publish(topic, payload)
+ return
+
+ key = topic + str(payload)
+ now = time.time()
+ if key in dedup_cache and dedup_cache[key] > now:
+ logging.info("dedup " + topic + " : " + str(payload))
+ else:
+ logging.debug("republishing " + topic + " : " + str(payload))
+ mqtt_client.publish(topic, payload)
+
+ dedup_cache[key] = now + timeout
+ for k in list(dedup_cache.keys()):
+ if dedup_cache[k] < now:
+ del dedup_cache[k]
+
+
+def filter_devices(topic, payload):
+ """ Deduplicate and republish device messages. """
+ global mqtt_client
+
+ if not re.match(DEVICES_RE, topic):
+ return
+
+ # Loop through all device mappings
+ for map in DEVICE_MAPPINGS:
+ path = "(" + DEVICES_RE + "/)" + map["source"] + "(.+)"
+ m = re.match(path, topic)
+ # On match republish same payload to different topic
+ if m:
+ target = m.group(1) + map["target"] + m.group(2)
+ logging.debug("republishing " + map["source"] + " : " + target)
+ publish_dedup(target, payload, map["dedup"])
+
+
+def is_sub_dict(needle, haystack):
+ """ Test if all key-value pairs of needle match haystack. """
+ for k, v in needle.items():
+ if k not in haystack or haystack[k] != v:
+ return False
+ return True
+
+
+def filter_events(topic, payload):
+ """ Deduplicate and republish event messages. """
+ global mqtt_client
+
+ if not re.match(EVENTS_RE, topic):
+ return
+
+ try:
+ data = json.loads(payload.decode())
+
+ # MQTT v5 noLocal might not work
+ # Ignore republished events, there should be a better way?
+ if EVENT_REPUBLISHED_KEY in data:
+ return
+
+ # Loop through all event mappings
+ for map in EVENT_MAPPINGS:
+ # Ensure all JSON data keys match
+ if not is_sub_dict(map["source"], data):
+ continue
+ # On match republish modified payload to same topic
+
+ # Tag as repbulished event
+ data[EVENT_REPUBLISHED_KEY] = 1
+ # Add all "target" keys, e.g. a nickname
+ for k, v in map["target"].items():
+ data[k] = v
+ logging.debug("republishing to " + topic)
+ publish_dedup(topic, json.dumps(data), map["dedup"])
+
+ except json.decoder.JSONDecodeError:
+ logging.warning("JSON decode error: " + payload.decode())
+
+
+def on_connect(client, userdata, flags, rc):
+ """ Callback for when the client receives a CONNACK response from the server. """
+ logging.info("MQTT Connection: " + mqtt.connack_string(rc))
+ if rc != 0:
+ logging.error("Could not connect. RC: " + str(rc))
+ exit()
+ # Subscribing in on_connect() means that if we lose the connection and reconnect then subscriptions will be renewed.
+ options = SubscribeOptions(qos=1, noLocal=True)
+ logging.info("Subscribing to " + MQTT_TOPIC_EVENTS)
+ client.subscribe(MQTT_TOPIC_EVENTS, options=options)
+ logging.info("Subscribing to " + MQTT_TOPIC_DEVICES + "/#")
+ client.subscribe(MQTT_TOPIC_DEVICES + "/#", options=options)
+
+
+def on_disconnect(client, userdata, rc):
+ if rc != 0:
+ logging.error("Unexpected disconnection. RC: " + str(rc))
+
+
+def on_message(client, userdata, msg):
+ """ Callback for when a PUBLISH message is received from the server. """
+ logging.debug("Received: " + msg.topic + "\t" + msg.payload.decode())
+ filter_devices(msg.topic, msg.payload)
+ filter_events(msg.topic, msg.payload)
+
+
+# Setup MQTT client
+mqtt_client.on_connect = on_connect
+mqtt_client.on_disconnect = on_disconnect
+mqtt_client.on_message = on_message
+mqtt_client.connect(MQTT_SERVER)
+mqtt_client.loop_start()
+
+
+def main():
+ """MQTT republishing filter"""
+ logging.basicConfig(format='[%(asctime)s] %(levelname)s:%(name)s:%(message)s',datefmt='%Y-%m-%dT%H:%M:%S%z')
+ logging.getLogger().setLevel(logging.INFO)
+
+ try:
+ while True:
+ time.sleep(1)
+ except KeyboardInterrupt:
+ pass
+
+
+if __name__ == "__main__":
+ main()