send command setting curtain motor's position when receiving msg on topic `homeassistant/cover/switchbot-curtain/+/position/set-percent`

https://github.com/fphammerle/switchbot-mqtt/issues/59
https://github.com/fphammerle/switchbot-mqtt/pull/64

.pylintrc

@@ -8,7 +8,7 @@ disable=missing-function-docstring,
 # avoid marking param list of implementation as duplicate of abstract declaration
 # https://github.com/PyCQA/pylint/issues/3239
 # https://github.com/PyCQA/pylint/issues/214
-min-similarity-lines=6
+min-similarity-lines=7
 
 [DESIGN]
 

CHANGELOG.md

@@ -6,6 +6,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 ### Added
+- MQTT messages on topic `homeassistant/cover/switchbot-curtain/MAC_ADDRESS/position/set-percent`
+  trigger command to set curtain motors' position (payload: decimal integer in range `[0, 100]`)
 - support `PySwitchbot` `v0.11.0` and `v0.12.0`
 
 ### Removed

README.md

@@ -45,12 +45,19 @@ The report may be requested manually by sending a MQTT message to the topic
 
 ### Curtain Motor
 
-Send `OPEN`, `CLOSE`, or `STOP` to topic `homeassistant/cover/switchbot-curtain/aa:bb:cc:dd:ee:ff/set`.
+Send `OPEN`, `CLOSE`, or `STOP` to topic `homeassistant/cover/switchbot-curtain/aa:bb:cc:dd:ee:ff/set`:
 
 ```sh
 $ mosquitto_pub -h MQTT_BROKER -t homeassistant/cover/switchbot-curtain/aa:bb:cc:dd:ee:ff/set -m CLOSE
 ```
 
+Or a position in percent (0 fully closed, 100 fully opened) to topic
+`homeassistant/cover/switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set-percent`:
+
+```sh
+$ mosquitto_pub -h MQTT_BROKER -t homeassistant/cover/switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set-percent -m 42
+```
+
 The command-line option `--fetch-device-info` enables position reports on topic
 `homeassistant/cover/switchbot-curtain/MAC_ADDRESS/position` after `STOP` commands
 and battery level reports on topic `homeassistant/cover/switchbot-curtain/MAC_ADDRESS/battery-percentage`
@@ -128,6 +135,7 @@ cover:
 - platform: mqtt
   name: switchbot_curtains
   command_topic: homeassistant/cover/switchbot-curtain/11:22:33:44:55:66/set
+  set_position_topic: homeassistant/cover/switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set-percent
   state_topic: homeassistant/cover/switchbot-curtain/11:22:33:44:55:66/state
 ```
 

setup.py

@@ -72,7 +72,7 @@ setuptools.setup(
         "Topic :: Home Automation",
     ],
     entry_points={"console_scripts": ["switchbot-mqtt = switchbot_mqtt._cli:_main"]},
-    # >=3.6 variable type hints, f-strings & * to force keyword-only arguments
+    # >=3.6 variable type hints, f-strings, typing.Collection & * to force keyword-only arguments
     # >=3.7 postponed evaluation of type annotations (PEP563) & dataclass
     python_requires=">=3.7",
     install_requires=[
@@ -81,6 +81,7 @@ setuptools.setup(
         # https://github.com/IanHarvey/bluepy/tree/v/1.3.0#release-notes
         "bluepy>=1.3.0,<2",
         # >=0.10.0 for SwitchbotCurtain.{update,get_position}
+        # >=0.9.0 for SwitchbotCurtain.set_position
         "PySwitchbot>=0.10.0,<0.13",
         "paho-mqtt<2",
     ],

switchbot_mqtt/_actors/__init__.py

@@ -23,7 +23,7 @@ import bluepy.btle
 import paho.mqtt.client
 import switchbot
 
-from switchbot_mqtt._actors._base import _MQTTControlledActor
+from switchbot_mqtt._actors._base import _MQTTCallbackUserdata, _MQTTControlledActor
 from switchbot_mqtt._utils import (
     _join_mqtt_topic_levels,
     _MQTTTopicLevel,
@@ -119,9 +119,13 @@ class _ButtonAutomator(_MQTTControlledActor):
 
 
 class _CurtainMotor(_MQTTControlledActor):
-    # https://www.home-assistant.io/integrations/cover.mqtt/
 
+    # https://www.home-assistant.io/integrations/cover.mqtt/
     MQTT_COMMAND_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + ["set"]
+    _MQTT_SET_POSITION_TOPIC_LEVELS = tuple(_CURTAIN_TOPIC_LEVELS_PREFIX) + (
+        "position",
+        "set-percent",
+    )
     _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + [
         "request-device-info"
     ]
@@ -221,3 +225,51 @@ class _CurtainMotor(_MQTTControlledActor):
             self._update_and_report_device_info(
                 mqtt_client=mqtt_client, report_position=report_position
             )
+
+    @classmethod
+    def _mqtt_set_position_callback(
+        cls,
+        mqtt_client: paho.mqtt.client.Client,
+        userdata: _MQTTCallbackUserdata,
+        message: paho.mqtt.client.MQTTMessage,
+    ) -> None:
+        # pylint: disable=unused-argument; callback
+        # https://github.com/eclipse/paho.mqtt.python/blob/v1.6.1/src/paho/mqtt/client.py#L3556
+        _LOGGER.debug("received topic=%s payload=%r", message.topic, message.payload)
+        if message.retain:
+            _LOGGER.info("ignoring retained message on topic %s", message.topic)
+            return
+        actor = cls._init_from_topic(
+            userdata=userdata,
+            topic=message.topic,
+            expected_topic_levels=cls._MQTT_SET_POSITION_TOPIC_LEVELS,
+        )
+        if not actor:
+            return  # warning in _init_from_topic
+        position_percent = int(message.payload.decode(), 10)
+        if position_percent < 0 or position_percent > 100:
+            _LOGGER.warning("invalid position %u%%, ignoring message", position_percent)
+            return
+        # pylint: disable=protected-access; own instance
+        if actor._get_device().set_position(position_percent):
+            _LOGGER.info(
+                "set position of switchbot curtain %s to %u%%",
+                actor._mac_address,
+                position_percent,
+            )
+        else:
+            _LOGGER.error(
+                "failed to set position of switchbot curtain %s", actor._mac_address
+            )
+
+    @classmethod
+    def _get_mqtt_message_callbacks(
+        cls,
+        *,
+        enable_device_info_update_topic: bool,
+    ) -> typing.Dict[typing.Tuple[_MQTTTopicLevel, ...], typing.Callable]:
+        callbacks = super()._get_mqtt_message_callbacks(
+            enable_device_info_update_topic=enable_device_info_update_topic
+        )
+        callbacks[cls._MQTT_SET_POSITION_TOPIC_LEVELS] = cls._mqtt_set_position_callback
+        return callbacks

switchbot_mqtt/_actors/_base.py

@@ -137,7 +137,7 @@ class _MQTTControlledActor(abc.ABC):
         cls,
         userdata: _MQTTCallbackUserdata,
         topic: str,
-        expected_topic_levels: typing.List[_MQTTTopicLevel],
+        expected_topic_levels: typing.Collection[_MQTTTopicLevel],
     ) -> typing.Optional[_MQTTControlledActor]:
         try:
             mac_address = _parse_mqtt_topic(
@@ -211,6 +211,23 @@ class _MQTTControlledActor(abc.ABC):
                 update_device_info=userdata.fetch_device_info,
             )
 
+    @classmethod
+    def _get_mqtt_message_callbacks(
+        cls,
+        *,
+        enable_device_info_update_topic: bool,
+    ) -> typing.Dict[typing.Tuple[_MQTTTopicLevel, ...], typing.Callable]:
+        # returning dict because `paho.mqtt.client.Client.message_callback_add` overwrites
+        # callbacks with same topic pattern
+        # https://github.com/eclipse/paho.mqtt.python/blob/v1.6.1/src/paho/mqtt/client.py#L2304
+        # https://github.com/eclipse/paho.mqtt.python/blob/v1.6.1/src/paho/mqtt/matcher.py#L19
+        callbacks = {tuple(cls.MQTT_COMMAND_TOPIC_LEVELS): cls._mqtt_command_callback}
+        if enable_device_info_update_topic:
+            callbacks[
+                tuple(cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS)
+            ] = cls._mqtt_update_device_info_callback
+        return callbacks
+
     @classmethod
     def mqtt_subscribe(
         cls,
@@ -218,15 +235,9 @@ class _MQTTControlledActor(abc.ABC):
         *,
         enable_device_info_update_topic: bool,
     ) -> None:
-        topics = [(cls.MQTT_COMMAND_TOPIC_LEVELS, cls._mqtt_command_callback)]
-        if enable_device_info_update_topic:
-            topics.append(
-                (
-                    cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS,
-                    cls._mqtt_update_device_info_callback,
-                )
-            )
-        for topic_levels, callback in topics:
+        for topic_levels, callback in cls._get_mqtt_message_callbacks(
+            enable_device_info_update_topic=enable_device_info_update_topic
+        ).items():
             topic = _join_mqtt_topic_levels(topic_levels, mac_address="+")
             _LOGGER.info("subscribing to MQTT topic %r", topic)
             mqtt_client.subscribe(topic)

switchbot_mqtt/_utils.py

@@ -37,7 +37,7 @@ _MQTTTopicLevel = typing.Union[str, _MQTTTopicPlaceholder]
 
 
 def _join_mqtt_topic_levels(
-    topic_levels: typing.List[_MQTTTopicLevel], mac_address: str
+    topic_levels: typing.Iterable[_MQTTTopicLevel], mac_address: str
 ) -> str:
     return "/".join(
         mac_address if l == _MQTTTopicPlaceholder.MAC_ADDRESS else typing.cast(str, l)
@@ -46,7 +46,7 @@ def _join_mqtt_topic_levels(
 
 
 def _parse_mqtt_topic(
-    topic: str, expected_levels: typing.List[_MQTTTopicLevel]
+    topic: str, expected_levels: typing.Collection[_MQTTTopicLevel]
 ) -> typing.Dict[_MQTTTopicPlaceholder, str]:
     attrs: typing.Dict[_MQTTTopicPlaceholder, str] = {}
     topic_split = topic.split("/")

tests/test_mqtt.py

@@ -77,10 +77,11 @@ def test__run(
     with caplog.at_level(logging.DEBUG):
         mqtt_client_mock().on_connect(mqtt_client_mock(), userdata, {}, 0)
     subscribe_mock = mqtt_client_mock().subscribe
-    assert subscribe_mock.call_count == (4 if fetch_device_info else 2)
+    assert subscribe_mock.call_count == (5 if fetch_device_info else 3)
     for topic in [
         "homeassistant/switch/switchbot/+/set",
         "homeassistant/cover/switchbot-curtain/+/set",
+        "homeassistant/cover/switchbot-curtain/+/position/set-percent",
     ]:
         assert unittest.mock.call(topic) in subscribe_mock.call_args_list
     for topic in [
@@ -90,6 +91,14 @@ def test__run(
         assert (
             unittest.mock.call(topic) in subscribe_mock.call_args_list
         ) == fetch_device_info
+    callbacks = {
+        c[1]["sub"]: c[1]["callback"]
+        for c in mqtt_client_mock().message_callback_add.call_args_list
+    }
+    assert (  # pylint: disable=comparison-with-callable; intended
+        callbacks["homeassistant/cover/switchbot-curtain/+/position/set-percent"]
+        == _CurtainMotor._mqtt_set_position_callback
+    )
     mqtt_client_mock().loop_forever.assert_called_once_with()
     assert caplog.record_tuples[:2] == [
         (
@@ -103,7 +112,7 @@ def test__run(
             f"connected to MQTT broker {mqtt_host}:{mqtt_port}",
         ),
     ]
-    assert len(caplog.record_tuples) == (6 if fetch_device_info else 4)
+    assert len(caplog.record_tuples) == (7 if fetch_device_info else 5)
     assert (
         "switchbot_mqtt._actors._base",
         logging.INFO,

tests/test_switchbot_curtain_motor_position.py

@@ -0,0 +1,243 @@
+# switchbot-mqtt - MQTT client controlling SwitchBot button & curtain automators,
+# compatible with home-assistant.io's MQTT Switch & Cover platform
+#
+# Copyright (C) 2022 Fabian Peter Hammerle <fabian@hammerle.me>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+import logging
+import unittest.mock
+
+import _pytest.logging
+import pytest
+from paho.mqtt.client import MQTTMessage
+
+from switchbot_mqtt._actors import _CurtainMotor
+from switchbot_mqtt._actors._base import _MQTTCallbackUserdata
+
+# pylint: disable=protected-access
+
+
+@pytest.mark.parametrize(
+    ("topic", "payload", "expected_mac_address", "expected_position_percent"),
+    [
+        (
+            b"homeassistant/cover/switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set-percent",
+            b"42",
+            "aa:bb:cc:dd:ee:ff",
+            42,
+        ),
+        (
+            b"homeassistant/cover/switchbot-curtain/11:22:33:44:55:66/position/set-percent",
+            b"0",
+            "11:22:33:44:55:66",
+            0,
+        ),
+        (
+            b"homeassistant/cover/switchbot-curtain/11:22:33:44:55:66/position/set-percent",
+            b"100",
+            "11:22:33:44:55:66",
+            100,
+        ),
+    ],
+)
+@pytest.mark.parametrize("retry_count", (3, 42))
+def test__mqtt_set_position_callback(
+    caplog: _pytest.logging.LogCaptureFixture,
+    topic: bytes,
+    payload: bytes,
+    expected_mac_address: str,
+    retry_count: int,
+    expected_position_percent: int,
+) -> None:
+    callback_userdata = _MQTTCallbackUserdata(
+        retry_count=retry_count,
+        device_passwords={},
+        fetch_device_info=False,
+    )
+    message = MQTTMessage(topic=topic)
+    message.payload = payload
+    with unittest.mock.patch(
+        "switchbot.SwitchbotCurtain"
+    ) as device_init_mock, caplog.at_level(logging.DEBUG):
+        _CurtainMotor._mqtt_set_position_callback(
+            mqtt_client="client dummy", userdata=callback_userdata, message=message
+        )
+    device_init_mock.assert_called_once_with(
+        mac=expected_mac_address,
+        password=None,
+        retry_count=retry_count,
+        reverse_mode=True,
+    )
+    device_init_mock().set_position.assert_called_once_with(expected_position_percent)
+    assert caplog.record_tuples == [
+        (
+            "switchbot_mqtt._actors",
+            logging.DEBUG,
+            f"received topic=homeassistant/cover/switchbot-curtain/{expected_mac_address}"
+            f"/position/set-percent payload=b'{expected_position_percent}'",
+        ),
+        (
+            "switchbot_mqtt._actors",
+            logging.INFO,
+            f"set position of switchbot curtain {expected_mac_address}"
+            f" to {expected_position_percent}%",
+        ),
+    ]
+
+
+def test__mqtt_set_position_callback_ignore_retained(
+    caplog: _pytest.logging.LogCaptureFixture,
+) -> None:
+    callback_userdata = _MQTTCallbackUserdata(
+        retry_count=3,
+        device_passwords={},
+        fetch_device_info=False,
+    )
+    message = MQTTMessage(
+        topic=b"homeassistant/cover/switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set-percent"
+    )
+    message.payload = b"42"
+    message.retain = True
+    with unittest.mock.patch(
+        "switchbot.SwitchbotCurtain"
+    ) as device_init_mock, caplog.at_level(logging.INFO):
+        _CurtainMotor._mqtt_set_position_callback(
+            mqtt_client="client dummy", userdata=callback_userdata, message=message
+        )
+    device_init_mock.assert_not_called()
+    assert caplog.record_tuples == [
+        (
+            "switchbot_mqtt._actors",
+            logging.INFO,
+            "ignoring retained message on topic"
+            " homeassistant/cover/switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set-percent",
+        ),
+    ]
+
+
+def test__mqtt_set_position_callback_unexpected_topic(
+    caplog: _pytest.logging.LogCaptureFixture,
+) -> None:
+    callback_userdata = _MQTTCallbackUserdata(
+        retry_count=3,
+        device_passwords={},
+        fetch_device_info=False,
+    )
+    message = MQTTMessage(topic=b"switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set")
+    message.payload = b"42"
+    with unittest.mock.patch(
+        "switchbot.SwitchbotCurtain"
+    ) as device_init_mock, caplog.at_level(logging.INFO):
+        _CurtainMotor._mqtt_set_position_callback(
+            mqtt_client="client dummy", userdata=callback_userdata, message=message
+        )
+    device_init_mock.assert_not_called()
+    assert caplog.record_tuples == [
+        (
+            "switchbot_mqtt._actors._base",
+            logging.WARN,
+            "unexpected topic switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set",
+        ),
+    ]
+
+
+def test__mqtt_set_position_callback_invalid_mac_address(
+    caplog: _pytest.logging.LogCaptureFixture,
+) -> None:
+    callback_userdata = _MQTTCallbackUserdata(
+        retry_count=3,
+        device_passwords={},
+        fetch_device_info=False,
+    )
+    message = MQTTMessage(
+        topic=b"homeassistant/cover/switchbot-curtain/aa:bb:cc:dd:ee/position/set-percent"
+    )
+    message.payload = b"42"
+    with unittest.mock.patch(
+        "switchbot.SwitchbotCurtain"
+    ) as device_init_mock, caplog.at_level(logging.INFO):
+        _CurtainMotor._mqtt_set_position_callback(
+            mqtt_client="client dummy", userdata=callback_userdata, message=message
+        )
+    device_init_mock.assert_not_called()
+    assert caplog.record_tuples == [
+        (
+            "switchbot_mqtt._actors._base",
+            logging.WARN,
+            "invalid mac address aa:bb:cc:dd:ee",
+        ),
+    ]
+
+
+@pytest.mark.parametrize("payload", [b"-1", b"123"])
+def test__mqtt_set_position_callback_invalid_position(
+    caplog: _pytest.logging.LogCaptureFixture,
+    payload: bytes,
+) -> None:
+    callback_userdata = _MQTTCallbackUserdata(
+        retry_count=3,
+        device_passwords={},
+        fetch_device_info=False,
+    )
+    message = MQTTMessage(
+        topic=b"homeassistant/cover/switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set-percent"
+    )
+    message.payload = payload
+    with unittest.mock.patch(
+        "switchbot.SwitchbotCurtain"
+    ) as device_init_mock, caplog.at_level(logging.INFO):
+        _CurtainMotor._mqtt_set_position_callback(
+            mqtt_client="client dummy", userdata=callback_userdata, message=message
+        )
+    device_init_mock.assert_called_once()
+    device_init_mock().set_position.assert_not_called()
+    assert caplog.record_tuples == [
+        (
+            "switchbot_mqtt._actors",
+            logging.WARN,
+            f"invalid position {payload.decode()}%, ignoring message",
+        ),
+    ]
+
+
+def test__mqtt_set_position_callback_command_failed(
+    caplog: _pytest.logging.LogCaptureFixture,
+) -> None:
+    callback_userdata = _MQTTCallbackUserdata(
+        retry_count=3,
+        device_passwords={},
+        fetch_device_info=False,
+    )
+    message = MQTTMessage(
+        topic=b"homeassistant/cover/switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set-percent"
+    )
+    message.payload = b"21"
+    with unittest.mock.patch(
+        "switchbot.SwitchbotCurtain"
+    ) as device_init_mock, caplog.at_level(logging.INFO):
+        device_init_mock().set_position.return_value = False
+        device_init_mock.reset_mock()
+        _CurtainMotor._mqtt_set_position_callback(
+            mqtt_client="client dummy", userdata=callback_userdata, message=message
+        )
+    device_init_mock.assert_called_once()
+    device_init_mock().set_position.assert_called_with(21)
+    assert caplog.record_tuples == [
+        (
+            "switchbot_mqtt._actors",
+            logging.ERROR,
+            "failed to set position of switchbot curtain aa:bb:cc:dd:ee:ff",
+        ),
+    ]