implement _CurtainMotor._report_position()

rebased https://github.com/fphammerle/switchbot-mqtt/pull/31/commits/ab364bca38980aa8b568700cca4ceb6b54a1a7ab

Pipfile.lock

@@ -30,10 +30,10 @@
         },
         "pyswitchbot": {
             "hashes": [
-                "sha256:03fe4aa839a83e94ef455e562fe92427c4e16d41e055a59f7807f523752974b6",
-                "sha256:b24d3d2897d84cd51eb89e9b888043a71122ceebfb9a3f0c5e2f9bd98d412c81"
+                "sha256:5c2102b1c49f7e9b72d6213b9926a38b32897927569385f6e5830d0b1bb6762e",
+                "sha256:f2a80e123cf494a373dba17ba54995f18a024c774cc28f5b9f67a3bf0e08e7b5"
             ],
-            "version": "==0.9.1"
+            "version": "==0.10.0"
         },
         "sanitized-package": {
             "editable": true,

setup.py

@@ -72,8 +72,8 @@ setuptools.setup(
     ],
     entry_points={"console_scripts": ["switchbot-mqtt = switchbot_mqtt:_main"]},
     install_requires=[
-        # >=0.9.0 for SwitchbotCurtain
-        "PySwitchbot>=0.9.0,<0.10",
+        # >=0.10.0 for SwitchbotCurtain.get_position
+        "PySwitchbot>=0.10.0,<0.11",
         "paho-mqtt<2",
     ],
     setup_requires=["setuptools_scm"],

switchbot_mqtt/__init__.py

@@ -67,6 +67,7 @@ class _MQTTControlledActor(abc.ABC):
     def __init__(
         self, mac_address: str, retry_count: int, password: typing.Optional[str]
     ) -> None:
+        # alternative: pySwitchbot >=0.10.0 provides SwitchbotDevice.get_mac()
         self._mac_address = mac_address
 
     @abc.abstractmethod
@@ -228,16 +229,42 @@ class _CurtainMotor(_MQTTControlledActor):
         "state",
     ]
 
+    _MQTT_POSITION_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
+        "cover",
+        "switchbot-curtain",
+        _MQTTTopicPlaceholder.MAC_ADDRESS,
+        "position",
+    ]
+
     def __init__(
         self, mac_address: str, retry_count: int, password: typing.Optional[str]
     ) -> None:
+        # > The position of the curtain is saved in self._pos with 0 = open and 100 = closed.
+        # https://github.com/Danielhiversen/pySwitchbot/blob/0.10.0/switchbot/__init__.py#L150
         self._device = switchbot.SwitchbotCurtain(
-            mac=mac_address, password=password, retry_count=retry_count
+            mac=mac_address,
+            password=password,
+            retry_count=retry_count,
+            reverse_mode=True,
         )
         super().__init__(
             mac_address=mac_address, retry_count=retry_count, password=password
         )
 
+    def _report_position(self, mqtt_client: paho.mqtt.client.Client) -> None:
+        # > position_closed integer (Optional, default: 0)
+        # > position_open integer (Optional, default: 100)
+        # https://www.home-assistant.io/integrations/cover.mqtt/#position_closed
+        # SwitchbotCurtain.get_position() returns a cached value within [0, 100].
+        # SwitchbotCurtain.open() and .close() update the position optimistically,
+        # SwitchbotCurtain.update() fetches the real position via bluetooth.
+        # https://github.com/Danielhiversen/pySwitchbot/blob/0.10.0/switchbot/__init__.py#L202
+        self._mqtt_publish(
+            topic_levels=self._MQTT_POSITION_TOPIC_LEVELS,
+            payload=str(int(self._device.get_position())).encode(),
+            mqtt_client=mqtt_client,
+        )
+
     def execute_command(
         self, mqtt_message_payload: bytes, mqtt_client: paho.mqtt.client.Client
     ) -> None:

tests/test_switchbot_curtain_motor.py

@@ -28,6 +28,72 @@ import switchbot_mqtt
 # pylint: disable=too-many-arguments; these are tests, no API
 
 
+@pytest.mark.parametrize(
+    "mac_address",
+    ("aa:bb:cc:dd:ee:ff", "aa:bb:cc:dd:ee:gg"),
+)
+@pytest.mark.parametrize(
+    ("position", "expected_payload"), [(0, b"0"), (100, b"100"), (42, b"42")]
+)
+def test__report_position(
+    caplog, mac_address: str, position: int, expected_payload: bytes
+):
+    with unittest.mock.patch(
+        "switchbot.SwitchbotCurtain.__init__", return_value=None
+    ) as device_init_mock, caplog.at_level(logging.DEBUG):
+        actor = switchbot_mqtt._CurtainMotor(
+            mac_address=mac_address, retry_count=7, password=None
+        )
+    device_init_mock.assert_called_once_with(
+        mac=mac_address,
+        retry_count=7,
+        password=None,
+        # > The position of the curtain is saved in self._pos with 0 = open and 100 = closed.
+        # > [...] The parameter 'reverse_mode' reverse these values, [...]
+        # > The parameter is default set to True so that the definition of position
+        # > is the same as in Home Assistant.
+        # https://github.com/Danielhiversen/pySwitchbot/blob/0.10.0/switchbot/__init__.py#L150
+        reverse_mode=True,
+    )
+    with unittest.mock.patch.object(
+        actor, "_mqtt_publish"
+    ) as publish_mock, unittest.mock.patch(
+        "switchbot.SwitchbotCurtain.get_position", return_value=position
+    ):
+        actor._report_position(mqtt_client="dummy")
+    publish_mock.assert_called_once_with(
+        topic_levels=[
+            "homeassistant",
+            "cover",
+            "switchbot-curtain",
+            switchbot_mqtt._MQTTTopicPlaceholder.MAC_ADDRESS,
+            "position",
+        ],
+        payload=expected_payload,
+        mqtt_client="dummy",
+    )
+    assert not caplog.record_tuples
+
+
+@pytest.mark.parametrize("position", ("", 'lambda: print("")'))
+def test__report_position_invalid(caplog, position):
+    with unittest.mock.patch(
+        "switchbot.SwitchbotCurtain.__init__", return_value=None
+    ), caplog.at_level(logging.DEBUG):
+        actor = switchbot_mqtt._CurtainMotor(
+            mac_address="aa:bb:cc:dd:ee:ff", retry_count=3, password=None
+        )
+    with unittest.mock.patch.object(
+        actor, "_mqtt_publish"
+    ) as publish_mock, unittest.mock.patch(
+        "switchbot.SwitchbotCurtain.get_position", return_value=position
+    ), pytest.raises(
+        ValueError
+    ):
+        actor._report_position(mqtt_client="dummy")
+    publish_mock.assert_not_called()
+
+
 @pytest.mark.parametrize("mac_address", ["aa:bb:cc:dd:ee:ff", "aa:bb:cc:11:22:33"])
 @pytest.mark.parametrize("password", ["pa$$word", None])
 @pytest.mark.parametrize("retry_count", (2, 3))
@@ -70,9 +136,7 @@ def test_execute_command(
                 mqtt_client="dummy", mqtt_message_payload=message_payload
             )
     device_init_mock.assert_called_once_with(
-        mac=mac_address,
-        password=password,
-        retry_count=retry_count,
+        mac=mac_address, password=password, retry_count=retry_count, reverse_mode=True
     )
     action_mock.assert_called_once_with()
     if command_successful:
@@ -125,7 +189,7 @@ def test_execute_command_invalid_payload(
                 mqtt_client="dummy", mqtt_message_payload=message_payload
             )
     device_mock.assert_called_once_with(
-        mac=mac_address, password=password, retry_count=7
+        mac=mac_address, password=password, retry_count=7, reverse_mode=True
     )
     assert not device_mock().mock_calls  # no methods called
     report_mock.assert_not_called()