Add support for custom shell commands as switches

README.md

@@ -67,6 +67,24 @@ Values in `<>` must be replaced.
 [List of GPIOs.](https://github.com/openlumi/xiaomi-gateway-openwrt#gpio)
 [List of device classes.](https://www.home-assistant.io/integrations/binary_sensor/#device-class)
 
+### Custom commands
+
+You can add an extra section with custom commands that are executed with 
+mqtt topics. Every command is exported as a switch entity in Home Assistant.
+If json is passed to the set topic, the command will be interpolated with 
+the values. Plain text is passed as {text} variable 
+```json
+{
+    <your configuration>,
+    "custom_commands": {
+      "tts": "echo \"Test TTS without MPD component for home assistant\" | python3 -c 'from urllib.parse import quote_plus;from sys import stdin;print(\"wget -O /tmp/tts.mp3 -U Mozilla \\\"http://translate.google.com/translate_tts?q=\"+quote_plus(stdin.read()[:100])+\"&ie=UTF-8&tl=en&total=1&idx=0&client=tw-ob&prev=input&ttsspeed=1\\\" && amixer set Master 200 && mpg123 /tmp/tts.mp3\")' | sh 2> /dev/null",
+      "tts_interpolate": "echo \"{text}\" | python3 -c 'from urllib.parse import quote_plus;from sys import stdin;print(\"wget -O /tmp/tts.mp3 -U Mozilla \\\"http://translate.google.com/translate_tts?q=\"+quote_plus(stdin.read()[:100])+\"&ie=UTF-8&tl=en&total=1&idx=0&client=tw-ob&prev=input&ttsspeed=1\\\" && amixer set Master {volume} && mpg123 /tmp/tts.mp3\")' | sh 2> /dev/null",
+      "restart_lumimqtt": "/etc/init.d/lumimqtt restart",
+      "reboot": "/sbin/reboot"
+    }
+}
+```
+
 ## OpenWrt installation
 
 ```sh 

lumimqtt/__main__.py

@@ -51,7 +51,7 @@ def main():
         'sensor_threshold': 50,  # 5% of illuminance sensor
         'sensor_debounce_period': 60,  # 1 minute
         'light_transition_period': 1.0,  # second
-        'light_notification_period': 60, # 1 minute
+        'light_notification_period': 60,  # 1 minute
         **config,
     }
 
@@ -75,7 +75,10 @@ def main():
         light_notification_period=float(config['light_notification_period']),
     )
 
-    for device in devices(config.get('binary_sensors', {})):
+    for device in devices(
+        binary_sensors=config.get('binary_sensors', {}),
+        custom_commands=config.get('custom_commands', {}),
+    ):
         server.register(device)
 
     try:

lumimqtt/commands.py

@@ -0,0 +1,64 @@
+"""
+LUMI light control
+"""
+import asyncio as aio
+import contextlib
+import logging
+import subprocess
+import sys
+from collections import defaultdict
+
+from .device import Device
+
+logger = logging.getLogger(__name__)
+
+
+class Command(Device):
+    """
+    Custom command control
+    """
+
+    def __init__(self, name, device_file, topic):
+        super().__init__(name, device_file, topic)
+        self.command = device_file
+
+    @property
+    def topic_set(self):
+        return f'{self.topic}/set'
+
+    @contextlib.contextmanager
+    def fix_watcher(self):
+        if sys.version_info < (3, 8, 0):
+            # https://github.com/aio-libs/aiohttp/pull/2075/files#diff-70599d14cae2351e35e46867bce26e325e84f3b84ce218718239c4bfeac4dcf5R445-R448
+            loop = aio.get_event_loop()
+            policy = aio.get_event_loop_policy()
+            watcher = policy.get_child_watcher()
+            watcher.attach_loop(loop)
+        yield
+
+    @staticmethod
+    def quote(s):
+        return s.replace('"', '\\"').replace("'", "\\'").replace('$', '')
+
+    async def run_command(self, value):
+        if isinstance(value, dict):
+            value = {
+                k: self.quote(v)
+                for k, v in value.items()
+            }
+            command = self.command.format_map(defaultdict(str, **value))
+        else:
+            command = self.command.format_map(defaultdict(str, text=value))
+
+        with self.fix_watcher():
+            proc = await aio.create_subprocess_shell(
+                command,
+                loop=aio.get_event_loop(),
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+            )
+        await proc.wait()
+
+    async def set(self, value):
+        logger.info(f'{self.name}: run command with params: {value}')
+        await self.run_command(value)

lumimqtt/lumimqtt.py

@@ -14,6 +14,7 @@
 
 from .__version__ import version
 from .button import Button
+from .commands import Command
 from .device import Device
 from .light import Light
 from .sensors import BinarySensor, Sensor
@@ -78,6 +79,7 @@ def __init__(
         self.sensors: ty.List[Sensor] = []
         self.lights: ty.List[Light] = []
         self.buttons: ty.List[Button] = []
+        self.custom_commands: ty.List[Command] = []
 
         self._debounce_sensors: ty.Dict[Sensor, DebounceSensor] = {}
 
@@ -120,6 +122,7 @@ def register(self, device: Device):
             Sensor: self.sensors,
             Button: self.buttons,
             Light: self.lights,
+            Command: self.custom_commands,
         }
         for typ, array in mapping.items():
             if isinstance(device, typ):
@@ -134,7 +137,8 @@ def _get_topic(self, subtopic):
     @property
     def subscribed_topics(self):
         # TODO: add SOUND/TTS topics ?
-        return (self._get_topic(light.topic_set) for light in self.lights)
+        return [self._get_topic(light.topic_set) for light in self.lights] + \
+            [self._get_topic(cmd.topic_set) for cmd in self.custom_commands]
 
     async def _handle_messages(self) -> None:
         async for message in self._client.delivered_messages(
@@ -147,28 +151,62 @@ async def _handle_messages(self) -> None:
                 for _light in self.lights:
                     if message.topic_name == self._get_topic(_light.topic_set):
                         light = _light
-                if not light:
-                    logger.error("Invalid topic for light")
+                if light:
+                    try:
+                        value = json.loads(message.payload)
+                    except ValueError as e:
+                        logger.exception(str(e))
+                        break
+
+                    try:
+                        await light.set(value, self._light_transition_period)
+                        await self._publish_light(light)
+                    except aio_mqtt.ConnectionClosedError as e:
+                        logger.error("Connection closed", exc_info=e)
+                        await self._client.wait_for_connect()
+                        continue
+
+                    except Exception as e:
+                        logger.error(
+                            "Unhandled exception during echo "
+                            "message publishing",
+                            exc_info=e)
                     break
 
-                try:
-                    value = json.loads(message.payload)
-                except ValueError as e:
-                    logger.exception(str(e))
+                command: ty.Optional[Command] = None
+                for _command in self.custom_commands:
+                    if message.topic_name == self._get_topic(
+                            _command.topic_set,
+                    ):
+                        command = _command
+                if command:
+                    try:
+                        value = json.loads(message.payload)
+                    except ValueError:
+                        value = message.payload.decode()
+                    try:
+                        await command.set(value)
+                        await self._client.publish(
+                            aio_mqtt.PublishableMessage(
+                                topic_name=self._get_topic(command.topic),
+                                payload='OFF',
+                                qos=aio_mqtt.QOSLevel.QOS_1,
+                            ),
+                        )
+                    except aio_mqtt.ConnectionClosedError as e:
+                        logger.error("Connection closed", exc_info=e)
+                        await self._client.wait_for_connect()
+                        continue
+
+                    except Exception as e:
+                        logger.error(
+                            "Unhandled exception during echo "
+                            "message publishing",
+                            exc_info=e,
+                        )
                     break
 
-                try:
-                    await light.set(value, self._light_transition_period)
-                    await self._publish_light(light)
-                except aio_mqtt.ConnectionClosedError as e:
-                    logger.error("Connection closed", exc_info=e)
-                    await self._client.wait_for_connect()
-                    continue
-
-                except Exception as e:
-                    logger.error(
-                        "Unhandled exception during echo message publishing",
-                        exc_info=e)
+                logger.error("Invalid topic for light")
                 break
 
     async def send_config(self):
@@ -270,6 +308,20 @@ def get_generic_vals(name):
                     retain=True,
                 ),
             )
+        for command in self.custom_commands:
+            await self._client.publish(
+                aio_mqtt.PublishableMessage(
+                    topic_name=f'homeassistant/switch/'
+                               f'{self.dev_id}_{command.name}/config',
+                    payload=json.dumps({
+                        **get_generic_vals(command.name),
+                        'state_topic': self._get_topic(command.topic),
+                        'command_topic': self._get_topic(command.topic_set),
+                    }),
+                    qos=aio_mqtt.QOSLevel.QOS_1,
+                    retain=True,
+                ),
+            )
 
     async def _periodic_publish(self, period=1):
         while True:
@@ -305,7 +357,8 @@ async def _periodic_publish(self, period=1):
                     now = datetime.now()
                     should_send = (
                         self._light_last_sent is None or
-                        (now - self._light_last_sent).seconds >= self._light_notification_period
+                        (now - self._light_last_sent).seconds >=
+                        self._light_notification_period
                     )
                     if should_send:
                         await self._publish_light(light)

lumimqtt/platform.py

@@ -3,15 +3,18 @@
 """
 import logging
 import os
+from typing import List
 
 from .button import Button
+from .commands import Command
+from .device import Device
 from .light import Light
 from .sensors import BinarySensor, IlluminanceSensor
 
 logger = logging.getLogger(__name__)
 
 
-def sensors(binary_sensors: dict):
+def sensors(binary_sensors: dict) -> List[Device]:
     sensors_ = list()
     for name, device_file in (
             ('illuminance', '/sys/bus/iio/devices/iio:device0/in_voltage5_raw'),
@@ -33,7 +36,7 @@ def sensors(binary_sensors: dict):
     return sensors_
 
 
-def buttons():
+def buttons() -> List[Device]:
     buttons_ = list()
     for name, device_file, scancodes in (
             ('btn0', '/dev/input/event0', ['BTN_0']),
@@ -45,7 +48,7 @@ def buttons():
     return buttons_
 
 
-def lights():
+def lights() -> List[Device]:
     led_r = '/sys/class/leds/red'
     led_g = '/sys/class/leds/green'
     led_b = '/sys/class/leds/blue'
@@ -73,5 +76,18 @@ def lights():
     return lights_
 
 
-def devices(binary_sensors: dict):
-    return sensors(binary_sensors) + buttons() + lights()
+def commands(params) -> List[Device]:
+    commands_ = []
+    for topic, command in params.items():
+        cmd_config = {
+            'name': topic,
+            'topic': topic,
+            'device_file': command,
+        }
+        commands_.append(Command(**cmd_config))
+    return commands_
+
+
+def devices(binary_sensors: dict, custom_commands: dict):
+    return sensors(binary_sensors) + buttons() + lights() + \
+        commands(custom_commands)

requirements.txt

@@ -1,3 +1,3 @@
 setuptools_scm
-aio-mqtt==0.2.0
-evdev==1.3.0
+aio-mqtt>=0.2.0
+evdev>=1.0.0