Merge pull request #698 from Emantor/topic/tasmota

mqtt: support for tasmota power ports

Author: jluebbe

dev-requirements.txt

@@ -15,3 +15,4 @@ psutil==5.6.6
 -r docker-requirements.txt
 -r pyvisa-requirements.txt
 -r vxi11-requirements.txt
+-r mqtt-requirements.txt

doc/configuration.rst

@@ -253,6 +253,32 @@ The example describes port 1 on the hub with the ID_PATH
 Used by:
   - `SiSPMPowerDriver`_
 
+TasmotaPowerPort
+++++++++++++++++
+A :any:`TasmotaPowerPort` resource describes a switchable Tasmora power outlet
+accessed over MQTT.
+
+.. code-block:: yaml
+
+   TasmotaPowerPort:
+     host: this.is.an.example.host.com
+     status_topic: stat/tasmota_575A2B/POWER
+     power_topic: cmnd/tasmota_575A2B/POWER
+     avail_topic: tele/tasmota_575A2B/LWT
+
+The example uses a mosquitto server at "this.is.an.example.host.com" and has the
+topics setup for a tasmota power port that has the ID 575A2B.
+
+- host (str): hostname of the MQTT server
+- status_topic (str): topic that signals the current status as "ON" of "OFF"
+- power_topic (str): topic that allows switchting the status between "ON" and
+  "OFF"
+- avail_topic (str): topic that signals the availability of the Tasmota power
+  outlet
+
+Used by:
+  - `TasmotaPowerDriver`_
+
 Digital Outputs
 ~~~~~~~~~~~~~~~
 
@@ -1499,7 +1525,7 @@ Arguments:
   - delay (float): optional delay in seconds between off and on
 
 SiSPMPowerDriver
-~~~~~~~~~~~~~~~~~~~
+~~~~~~~~~~~~~~~~
 A SiSPMPowerDriver controls a `SiSPMPowerPort`, allowing control of the target
 power state without user interaction.
 
@@ -1517,6 +1543,25 @@ Implements:
 Arguments:
   - delay (float): optional delay in seconds between off and on
 
+TasmotaPowerDriver
+~~~~~~~~~~~~~~~~~~
+A TasmotaPowerDriver contols a `TasmotaPowerPort`, allowing the outlet to be
+switched on and off.
+
+Binds to:
+  - `TasmotaPowerPort`_
+
+Implements:
+  - :any:`PowerProtocol`
+
+.. code-block:: yaml
+
+   TasmotaPowerDriver:
+     delay: 5.0
+
+Arguments:
+  - delay (float): optional delay in seconds between off and on
+
 GpioDigitalOutputDriver
 ~~~~~~~~~~~~~~~~~~~~~~~
 The GpioDigitalOutputDriver writes a digital signal to a GPIO line.

labgrid/driver/__init__.py

@@ -36,3 +36,4 @@
 from .usbaudiodriver import USBAudioInputDriver
 from .networkinterfacedriver import NetworkInterfaceDriver
 from .provider import TFTPProviderDriver
+from .mqtt import TasmotaPowerDriver

labgrid/driver/mqtt.py

@@ -0,0 +1,97 @@
+#!/usr/bin/env python3
+
+import threading
+import time
+
+import attr
+
+from .common import Driver
+from ..factory import target_factory
+from ..protocol import PowerProtocol
+from ..step import step
+from ..util import Timeout
+
+
+class MQTTError(Exception):
+    pass
+
+@target_factory.reg_driver
+@attr.s(eq=False)
+class TasmotaPowerDriver(Driver, PowerProtocol):
+    bindings = {
+            "power": {"TasmotaPowerPort"}
+    }
+    delay = attr.ib(default=2.0, validator=attr.validators.instance_of(float))
+    _client = attr.ib(default=None)
+    _status = attr.ib(default=None)
+
+    def __attrs_post_init__(self):
+        super().__attrs_post_init__()
+        import paho.mqtt.client as mqtt
+        self._client = mqtt.Client()
+
+    def on_activate(self):
+        self._client.on_message = self._on_message
+        self._client.on_connect = self._on_connect
+        self._client.connect(self.power.host)
+        self._client.loop_start()
+
+    def on_deactivate(self):
+        self._client.loop_stop()
+
+    def _on_message(self, client, userdata, msg):
+        if msg.payload == b'ON':
+            status = True
+        elif msg.payload == b'OFF':
+            status = False
+        self._status = status
+
+    def _on_connect(self, client, userdata, flags, rc):
+        client.subscribe(self.power.status_topic)
+
+    def _publish(self, topic, payload):
+        msg = self._client.publish(topic, payload=payload)
+        timeout = Timeout(3.0)
+        while not msg.is_published:
+            time.sleep(0.1)
+            if timeout.expired:
+                raise MQTTError("publish timed out")
+        return msg
+
+    @Driver.check_active
+    @step()
+    def on(self):
+        self._publish(self.power.power_topic, "ON")
+        timeout = Timeout(3.0)
+        while self._status is False:
+            time.sleep(0.1)
+            if timeout.expired:
+                raise MQTTError("Port did not change status within 3 seconds")
+
+    @Driver.check_active
+    @step()
+    def off(self):
+        self._publish(self.power.power_topic, "OFF")
+        timeout = Timeout(3.0)
+        while self._status is True:
+            time.sleep(0.1)
+            if timeout.expired:
+                raise MQTTError("Port did not change status within 3 seconds")
+
+    @Driver.check_active
+    @step()
+    def cycle(self):
+        self.off()
+        time.sleep(self.delay)
+        self.on()
+
+    @Driver.check_active
+    @step()
+    def get(self):
+        self._client.publish(self.power.power_topic)
+        timeout = Timeout(3.0)
+        while self._status is None:
+            time.sleep(0.1)
+            if timeout.expired:
+                raise MQTTError("Could not get initial status")
+        return self._status

labgrid/remote/client.py

@@ -721,8 +721,10 @@ def power(self):
         target = self._get_target(place)
         from ..driver.powerdriver import (NetworkPowerDriver, PDUDaemonDriver,
                                           USBPowerDriver, SiSPMPowerDriver)
+        from ..driver.mqtt import TasmotaPowerDriver
         from ..resource.power import NetworkPowerPort, PDUDaemonPort
-        from ..resource.remote import NetworkUSBPowerPort, NetworkSiSPMPowerPort
+        from ..resource.remote import (NetworkUSBPowerPort, NetworkSiSPMPowerPort)
+        from ..resource.mqtt import TasmotaPowerPort
 
         drv = None
         try:
@@ -753,6 +755,12 @@ def power(self):
                     except NoDriverFoundError:
                         drv = PDUDaemonDriver(target, name=None)
                         break
+                elif isinstance(resource, TasmotaPowerPort):
+                    try:
+                        drv = target.get_driver(TasmotaPowerDriver)
+                    except NoDriverFoundError:
+                        drv = TasmotaPowerDriver(target, name=None)
+                        break
         if not drv:
             raise UserError("target has no compatible resource available")
         if delay is not None:

labgrid/resource/__init__.py

@@ -19,3 +19,4 @@
 from .lxaiobus import LXAIOBusPIO
 from .pyvisa import PyVISADevice
 from .provider import TFTPProvider
+from .mqtt import TasmotaPowerPort

labgrid/resource/mqtt.py

@@ -0,0 +1,75 @@
+import logging
+import threading
+from time import monotonic
+
+import attr
+
+from .common import ManagedResource, ResourceManager
+from ..factory import target_factory
+
+@attr.s(eq=False)
+class MQTTManager(ResourceManager):
+    _available = attr.ib(default=attr.Factory(set), validator=attr.validators.instance_of(set))
+    _avail_lock = attr.ib(default=threading.Lock())
+    _clients = attr.ib(default=attr.Factory(dict), validator=attr.validators.instance_of(dict))
+    _topics = attr.ib(default=attr.Factory(list), validator=attr.validators.instance_of(list))
+    _topic_lock = attr.ib(default=threading.Lock())
+    _last = attr.ib(default=0.0, validator=attr.validators.instance_of(float))
+
+    def __attrs_post_init__(self):
+        super().__attrs_post_init__()
+        self.log = logging.getLogger('MQTTManager')
+
+    def _create_mqtt_connection(self, host):
+        import paho.mqtt.client as mqtt
+        client = mqtt.Client()
+        client.connect(host)
+        client.on_message = self._on_message
+        client.loop_start()
+        return client
+
+    def on_resource_added(self, resource):
+        host = resource.host
+        if host not in self._clients:
+            self._clients[host] = self._create_mqtt_connection(host)
+        self._clients[host].subscribe(resource.avail_topic)
+
+    def _on_message(self, client, userdata, msg):
+        payload = msg.payload.decode('utf-8')
+        topic = msg.topic
+        if payload == "Online":
+            with self._avail_lock:
+                self._available.add(topic)
+        if payload == "Offline":
+            with self._avail_lock:
+                self._available.discard(topic)
+
+    def poll(self):
+        if monotonic()-self._last < 2:
+            return  # ratelimit requests
+        self._last = monotonic()
+        with self._avail_lock:
+            for resource in self.resources:
+                resource.avail = resource.avail_topic in self._available
+
+
+@target_factory.reg_resource
+@attr.s(eq=False)
+class MQTTResource(ManagedResource):
+    manager_cls = MQTTManager
+
+    host = attr.ib(validator=attr.validators.instance_of(str))
+    avail_topic = attr.ib(validator=attr.validators.instance_of(str))
+
+    def __attrs_post_init__(self):
+        self.timeout = 30.0
+        super().__attrs_post_init__()
+
+
+@target_factory.reg_resource
+@attr.s(eq=False)
+class TasmotaPowerPort(MQTTResource):
+    power_topic = attr.ib(default=None,
+                         validator=attr.validators.instance_of(str))
+    status_topic = attr.ib(default=None,
+                         validator=attr.validators.instance_of(str))

labgrid/resource/remote.py

@@ -333,6 +333,14 @@ def __attrs_post_init__(self):
 @attr.s(eq=False)
 class NetworkUSBFlashableDevice(RemoteUSBResource):
     devnode = attr.ib(validator=attr.validators.optional(attr.validators.instance_of(str)))
+
+@attr.s(eq=False)
+class NetworkMQTTResource(ManagedResource):
+    manager_cls = RemotePlaceManager
+
+    host = attr.ib(validator=attr.validators.instance_of(str))
+    avail_topic = attr.ib(validator=attr.validators.instance_of(str))
+
     def __attrs_post_init__(self):
         self.timeout = 30.0
         super().__attrs_post_init__()

mqtt-requirements.txt

@@ -0,0 +1 @@
+paho-mqtt==1.5.1

tests/test_tasmota.py

@@ -0,0 +1,23 @@
+from labgrid.resource.mqtt import TasmotaPowerPort
+from labgrid.driver.mqtt import TasmotaPowerDriver
+
+import pytest
+
+pytest.importorskip("paho.mqtt.client")
+
+
+def test_tasmota_resource(target, mocker):
+    mocker.patch('paho.mqtt.client.Client.connect', return_value=None)
+    mocker.patch('paho.mqtt.client.Client.loop_start', return_value=None)
+    TasmotaPowerPort(target, name=None, host="localhost", avail_topic="test",
+                     power_topic="test", status_topic="test")
+
+
+def test_tasmota_driver(target, mocker):
+    mocker.patch('paho.mqtt.client.Client.connect', return_value=None)
+    mocker.patch('paho.mqtt.client.Client.loop_start', return_value=None)
+    res = TasmotaPowerPort(target, name=None, host="localhost", avail_topic="test",
+                           power_topic="test", status_topic="test")
+    res.manager._available.add("test")
+    driver = TasmotaPowerDriver(target, name=None)
+    target.activate(driver)