Merge pull request #484 from joerggollnick/main

new service `mqtt_filter` and add mqttwarn.openrc script

Author: jpmens

HANDBOOK.md

@@ -326,6 +326,7 @@ _mqttwarn_ supports a number of services (listed alphabetically below):
 * mastodon (see [tootpaste](#tootpaste))
 * [mattermost](#mattermost)
 * [mqtt](#mqtt)
+* [mqtt_filter](#mqtt_filter)
 * [mqttpub](#mqttpub)
 * [mysql](#mysql)
 * [mysql_dynamic](#mysql_dynamic)
@@ -1423,6 +1424,74 @@ This shows the currently full configuration possible. Global values from the
 authentication (`auth`) or (`tls`) you may omit those sections. (The `defaults`
 section must exist.)
 
+### `mqtt_filter`
+
+The `mqtt_filter` target executes the specified program and its arguments. It is similar
+to `pipe` but it doesn't open a pipe to the program. It provides stdout as response
+to a configured queue.
+Example use cases are e.g. IoT buttons which publish a message when they are pushed
+and they execute an external program. It is also a clone of [mqtt-launcher](https://github.com/jpmens/mqtt-launcher).
+With no response configured it acts like `execute` with multiple arguments.
+
+To pass the published data (json args array) to the command, use `{args[0]}` and `{args[1]}` which then gets replaced. Message looks like `'{ "args" : ["' + temp + '","' + room + '"] }'` for `fr
+itzctl`.
+
+outgoing_topic is constructed by parts of incoming topic or as full_incoming topic.
+
+```ini
+[config:mqtt_filter]
+targets = {
+   # full_topic, topic[0], topic[1], args[0], .....
+   # touch file /tmp/executed
+   'touch'    : [ None,0,0,'touch', '/tmp/executed' ],
+   # uses firtzctl to set temperature of a room
+   'fritzctl' : [ None,0,0,'/usr/bin/fritzctl','--loglevel=ERROR','temperature', "{args[0]}", "{args[1]}" ]
+   # performs a dirvish backup and writes stdout as a new messages to response topic
+   'backup'   : ["response/{topic[1]}/{topic[2]}",0,0,'/usr/bin/sudo','/usr/sbin/dirvish','--vault', "{args[0]}" ],
+   }
+```
+
+Use case for fritzctl is to change the requested temperature for a connected thermostat.
+Topic is constructed as /home/{room}/temperature/{action}.
+
+```
+def TemperatureConvert( data=None, srv=None):
+
+    # optional debug logger
+    if srv is not None:
+        srv.logging.debug('data={data}, srv={srv}'.format(**locals()))
+
+    topic = str( data.get('topic','') )
+
+    # init
+    room = ''
+    action = 'status'
+
+    # /home/{room}/temperature/{action}
+    parts = topic.split('/')
+
+    for idx, part in enumerate( parts ):
+         if idx == 1:
+             room = part
+
+         if idx == 3:
+             action = part
+
+    temp = str( data.get('payload','sav') )
+    if temp == '':
+        temp = 'sav'
+
+    if action == 'set':
+        cmd = '{ "args" : ["' + temp + '","' + room + '"] }'
+
+    return cmd
+```
+
+Use case for backup is to run a dirvish backup triggered by a simple mqtt message.
+
+Note, that for each message targeted to the `mqtt_filter` service, a new process is
+spawned (fork/exec), so it is quite "expensive".
+
 ### `mqttpub`
 
 This service publishes a message to the broker _mqttwarn_ is connected to. (To

etc/mqttwarn.openrc

@@ -0,0 +1,14 @@
+#!/sbin/openrc-run
+
+command="/usr/bin/python /opt/mqttwarn/mqttwarn.py"
+command_args="${MQTTWARN_OPTIONS}"
+command_background=yes
+pidfile=/run/mqttwarn.pid
+directory=/opt/mqttwarn
+
+name="mqttwarn"
+description="mqttwarn pluggable mqtt notification service"
+
+depend() {
+  need net
+}

mqttwarn/services/mqtt_filter.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+__author__    = 'Joerg Gollnick <github+mqttwarn()wurzelbenutzer.de>'
+__copyright__ = 'Copyright 2016 Tobias Brunner / 2021 Joerg Gollnick'
+__license__   = """Eclipse Public License - v 1.0 (http://www.eclipse.org/legal/epl-v10.html)"""
+
+import subprocess
+import json
+from pipes import quote
+from six import string_types
+
+def plugin(srv, item):
+
+    srv.logging.debug("*** MODULE=%s: service=%s, target=%s", __file__, item.service, item.target)
+
+    # same as for ssh
+    json_message=json.loads(item.message)
+
+    args = None
+    if json_message is not None:
+        args = json_message["args"]
+
+        if type(args) is list and len(args) == 1:
+            args=args[0]
+
+        if type(args) is list:
+            args=tuple([ quote(v) for v  in args ]) #escape the shell args
+        elif type(args) is str or type(args) is unicode:
+            args=(quote(args),)
+
+    # parse topic
+    topic=list(map( lambda x: quote(x), item.topic.split('/') ))
+
+    outgoing_topic = None
+    # replace palceholders args[0], args[1] ..., full_topic, topic[0],
+    if item.addrs[0] is not None:
+        outgoing_topic = item.addrs[0].format(args=args,full_topic=quote(item.topic),topic=topic)
+    qos            = item.addrs[1]
+    retain         = item.addrs[2]
+    addrs          = item.addrs[3:]
+
+    cmd = None
+    if addrs is not None:
+        cmd = [i.format(args=args, full_topic=quote(item.topic),topic=topic) for i in addrs]
+
+    srv.logging.debug("*** MODULE=%s: service=%s, command=%s outgoing_topic=%s", __file__, item.service, str( cmd ),outgoing_topic)
+
+    try:
+        res = subprocess.check_output(cmd, stdin=None, stderr=subprocess.STDOUT, shell=False, universal_newlines=True, cwd='/tmp')
+    except Exception as e:
+        srv.logging.warning("Cannot execute %s because %s" % (cmd, e))
+        return False
+
+    if outgoing_topic is not None:
+        outgoing_payload = res.rstrip('\n')
+        if isinstance(outgoing_payload, string_types):
+            outgoing_payload = bytearray(outgoing_payload, encoding='utf-8')
+        try:
+            srv.mqttc.publish(outgoing_topic, outgoing_payload, qos=qos, retain=retain)
+        except Exception as e:
+            srv.logging.warning("Cannot PUBlish response %s: %s" % (outgoing_topic, e))
+            return False
+
+    return True