report curtain motor's battery level in topic `homeassistant/cover/sw…

…itchbot-curtain/MAC_ADDRESS/battery-percentage` after every command

CHANGELOG.md

@@ -5,6 +5,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+### Added
+- command-line option `--fetch-device-info` enables reporting of curtain motors'
+  battery level on topic `homeassistant/cover/switchbot-curtain/MAC_ADDRESS/battery-percentage`
+  after executing commands (open, close, stop).
+
 ### Removed
 - compatibility with `python3.5`
 

README.md

@@ -47,7 +47,9 @@ $ mosquitto_pub -h MQTT_BROKER -t homeassistant/cover/switchbot-curtain/aa:bb:cc
 ```
 
 The command-line option `--fetch-device-info` enables position reports on topic
-`homeassistant/cover/switchbot-curtain/MAC_ADDRESS/position` after `STOP` commands.
+`homeassistant/cover/switchbot-curtain/MAC_ADDRESS/position` after `STOP` commands
+and battery level reports on topic `homeassistant/cover/switchbot-curtain/MAC_ADDRESS/battery-percentage`
+after every command.
 
 ### Device Passwords
 

switchbot_mqtt/__init__.py

@@ -304,21 +304,32 @@ class _CurtainMotor(_MQTTControlledActor):
         _MQTTTopicPlaceholder.MAC_ADDRESS,
         "set",
     ]
-
     MQTT_STATE_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
         "cover",
         "switchbot-curtain",
         _MQTTTopicPlaceholder.MAC_ADDRESS,
         "state",
     ]
-
+    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
+        "cover",
+        "switchbot-curtain",
+        _MQTTTopicPlaceholder.MAC_ADDRESS,
+        "battery-percentage",
+    ]
     _MQTT_POSITION_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
         "cover",
         "switchbot-curtain",
         _MQTTTopicPlaceholder.MAC_ADDRESS,
         "position",
     ]
 
+    @classmethod
+    def get_mqtt_battery_percentage_topic(cls, mac_address: str) -> str:
+        return _join_mqtt_topic_levels(
+            topic_levels=cls._MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS,
+            mac_address=mac_address,
+        )
+
     @classmethod
     def get_mqtt_position_topic(cls, mac_address: str) -> str:
         return _join_mqtt_topic_levels(
@@ -343,6 +354,15 @@ def __init__(
     def _get_device(self) -> switchbot.SwitchbotDevice:
         return self.__device
 
+    def _report_battery_level(self, mqtt_client: paho.mqtt.client.Client) -> None:
+        # > battery: Percentage of battery that is left.
+        # https://www.home-assistant.io/integrations/sensor/#device-class
+        self._mqtt_publish(
+            topic_levels=self._MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS,
+            payload=str(self.__device.get_battery_percent()).encode(),
+            mqtt_client=mqtt_client,
+        )
+
     def _report_position(self, mqtt_client: paho.mqtt.client.Client) -> None:
         # > position_closed integer (Optional, default: 0)
         # > position_open integer (Optional, default: 100)
@@ -357,9 +377,13 @@ def _report_position(self, mqtt_client: paho.mqtt.client.Client) -> None:
             mqtt_client=mqtt_client,
         )
 
-    def _update_position(self, mqtt_client: paho.mqtt.client.Client) -> None:
+    def _update_and_report_device_info(
+        self, mqtt_client: paho.mqtt.client.Client, *, report_position: bool
+    ) -> None:
         self._update_device_info()
-        self._report_position(mqtt_client=mqtt_client)
+        self._report_battery_level(mqtt_client=mqtt_client)
+        if report_position:
+            self._report_position(mqtt_client=mqtt_client)
 
     def execute_command(
         self,
@@ -368,6 +392,7 @@ def execute_command(
         update_device_info: bool,
     ) -> None:
         # https://www.home-assistant.io/integrations/cover.mqtt/#payload_open
+        report_device_info, report_position = False, False
         if mqtt_message_payload.lower() == b"open":
             if not self.__device.open():
                 _LOGGER.error("failed to open switchbot curtain %s", self._mac_address)
@@ -376,13 +401,15 @@ def execute_command(
                 # > state_opening string (Optional, default: opening)
                 # https://www.home-assistant.io/integrations/cover.mqtt/#state_opening
                 self.report_state(mqtt_client=mqtt_client, state=b"opening")
+                report_device_info = update_device_info
         elif mqtt_message_payload.lower() == b"close":
             if not self.__device.close():
                 _LOGGER.error("failed to close switchbot curtain %s", self._mac_address)
             else:
                 _LOGGER.info("switchbot curtain %s closing", self._mac_address)
                 # https://www.home-assistant.io/integrations/cover.mqtt/#state_closing
                 self.report_state(mqtt_client=mqtt_client, state=b"closing")
+                report_device_info = update_device_info
         elif mqtt_message_payload.lower() == b"stop":
             if not self.__device.stop():
                 _LOGGER.error("failed to stop switchbot curtain %s", self._mac_address)
@@ -392,13 +419,17 @@ def execute_command(
                 # https://www.home-assistant.io/integrations/cover.mqtt/#configuration-variables
                 # https://community.home-assistant.io/t/mqtt-how-to-remove-retained-messages/79029/2
                 self.report_state(mqtt_client=mqtt_client, state=b"")
-                if update_device_info:
-                    self._update_position(mqtt_client=mqtt_client)
+                report_device_info = update_device_info
+                report_position = True
         else:
             _LOGGER.warning(
                 "unexpected payload %r (expected 'OPEN', 'CLOSE', or 'STOP')",
                 mqtt_message_payload,
             )
+        if report_device_info:
+            self._update_and_report_device_info(
+                mqtt_client=mqtt_client, report_position=report_position
+            )
 
 
 def _mqtt_on_connect(
@@ -484,11 +515,13 @@ def _main() -> None:
         " (default: %(default)d)",
     )
     argparser.add_argument(
-        "--fetch-device-info",  # generic name to cover future addition of battery level etc.
+        "--fetch-device-info",
         action="store_true",
         help="Report curtain motors' position on"
         f" topic {_CurtainMotor.get_mqtt_position_topic(mac_address='MAC_ADDRESS')}"
-        " after sending stop command.",
+        " after sending stop command and battery level on topic"
+        f" {_CurtainMotor.get_mqtt_battery_percentage_topic(mac_address='MAC_ADDRESS')}"
+        " after every commands.",
     )
     args = argparser.parse_args()
     if args.mqtt_password_path:

tests/test_switchbot_curtain_motor.py

@@ -28,6 +28,16 @@
 # pylint: disable=too-many-arguments; these are tests, no API
 
 
+@pytest.mark.parametrize("mac_address", ["{MAC_ADDRESS}", "aa:bb:cc:dd:ee:ff"])
+def test_get_mqtt_battery_percentage_topic(mac_address):
+    assert (
+        switchbot_mqtt._CurtainMotor.get_mqtt_battery_percentage_topic(
+            mac_address=mac_address
+        )
+        == f"homeassistant/cover/switchbot-curtain/{mac_address}/battery-percentage"
+    )
+
+
 @pytest.mark.parametrize("mac_address", ["{MAC_ADDRESS}", "aa:bb:cc:dd:ee:ff"])
 def test_get_mqtt_position_topic(mac_address):
     assert (
@@ -102,19 +112,65 @@ def test__report_position_invalid(caplog, position):
     publish_mock.assert_not_called()
 
 
-def test__update_position():
+@pytest.mark.parametrize(("battery_percent", "battery_percent_encoded"), [(42, b"42")])
+@pytest.mark.parametrize("report_position", [True, False])
+@pytest.mark.parametrize(("position", "position_encoded"), [(21, b"21")])
+def test__update_and_report_device_info(
+    report_position: bool,
+    battery_percent: int,
+    battery_percent_encoded: bytes,
+    position: int,
+    position_encoded: bytes,
+):
     with unittest.mock.patch("switchbot.SwitchbotCurtain.__init__", return_value=None):
         actor = switchbot_mqtt._CurtainMotor(
             mac_address="dummy", retry_count=21, password=None
         )
-    with unittest.mock.patch(
-        "switchbot.SwitchbotCurtain.update"
-    ) as update_mock, unittest.mock.patch.object(
-        actor, "_report_position"
-    ) as report_position_mock:
-        actor._update_position(mqtt_client="client")
+    actor._get_device()._battery_percent = battery_percent
+    actor._get_device()._pos = position
+    mqtt_client_mock = unittest.mock.MagicMock()
+    with unittest.mock.patch("switchbot.SwitchbotCurtain.update") as update_mock:
+        actor._update_and_report_device_info(
+            mqtt_client=mqtt_client_mock, report_position=report_position
+        )
     update_mock.assert_called_once_with()
-    report_position_mock.assert_called_once_with(mqtt_client="client")
+    assert mqtt_client_mock.publish.call_count == (1 + report_position)
+    assert (
+        unittest.mock.call(
+            topic="homeassistant/cover/switchbot-curtain/dummy/battery-percentage",
+            payload=battery_percent_encoded,
+            retain=True,
+        )
+        in mqtt_client_mock.publish.call_args_list
+    )
+    if report_position:
+        assert (
+            unittest.mock.call(
+                topic="homeassistant/cover/switchbot-curtain/dummy/position",
+                payload=position_encoded,
+                retain=True,
+            )
+            in mqtt_client_mock.publish.call_args_list
+        )
+
+
+@pytest.mark.parametrize(
+    "exception",
+    [
+        PermissionError("bluepy-helper failed to enable low energy mode..."),
+        bluepy.btle.BTLEManagementError("test"),
+    ],
+)
+def test__update_and_report_device_info_update_error(exception):
+    actor = switchbot_mqtt._CurtainMotor(
+        mac_address="dummy", retry_count=21, password=None
+    )
+    mqtt_client_mock = unittest.mock.MagicMock()
+    with unittest.mock.patch.object(
+        actor._get_device(), "update", side_effect=exception
+    ), pytest.raises(type(exception)):
+        actor._update_and_report_device_info(mqtt_client_mock, report_position=True)
+    mqtt_client_mock.publish.assert_not_called()
 
 
 @pytest.mark.parametrize("mac_address", ["aa:bb:cc:dd:ee:ff", "aa:bb:cc:11:22:33"])
@@ -134,7 +190,7 @@ def test__update_position():
         (b"Stop", "switchbot.SwitchbotCurtain.stop"),
     ],
 )
-@pytest.mark.parametrize("report_position_upon_stop", [True, False])
+@pytest.mark.parametrize("update_device_info", [True, False])
 @pytest.mark.parametrize("command_successful", [True, False])
 def test_execute_command(
     caplog,
@@ -143,7 +199,7 @@ def test_execute_command(
     retry_count,
     message_payload,
     action_name,
-    report_position_upon_stop,
+    update_device_info,
     command_successful,
 ):
     with unittest.mock.patch(
@@ -157,12 +213,12 @@ def test_execute_command(
         ) as report_mock, unittest.mock.patch(
             action_name, return_value=command_successful
         ) as action_mock, unittest.mock.patch.object(
-            actor, "_update_position"
-        ) as update_position_mock:
+            actor, "_update_and_report_device_info"
+        ) as update_device_info_mock:
             actor.execute_command(
                 mqtt_client="dummy",
                 mqtt_message_payload=message_payload,
-                update_device_info=report_position_upon_stop,
+                update_device_info=update_device_info,
             )
     device_init_mock.assert_called_once_with(
         mac=mac_address, password=password, retry_count=retry_count, reverse_mode=True
@@ -195,14 +251,13 @@ def test_execute_command(
             )
         ]
         report_mock.assert_not_called()
-    if (
-        report_position_upon_stop
-        and action_name == "switchbot.SwitchbotCurtain.stop"
-        and command_successful
-    ):
-        update_position_mock.assert_called_once_with(mqtt_client="dummy")
+    if update_device_info and command_successful:
+        update_device_info_mock.assert_called_once_with(
+            mqtt_client="dummy",
+            report_position=(action_name == "switchbot.SwitchbotCurtain.stop"),
+        )
     else:
-        update_position_mock.assert_not_called()
+        update_device_info_mock.assert_not_called()
 
 
 @pytest.mark.parametrize("mac_address", ["aa:bb:cc:dd:ee:ff"])
@@ -271,25 +326,3 @@ def test_execute_command_bluetooth_error(caplog, mac_address, message_payload):
             f"failed to {message_payload.decode().lower()} switchbot curtain {mac_address}",
         ),
     ]
-
-
-@pytest.mark.parametrize(
-    "exception",
-    [
-        PermissionError("bluepy-helper failed to enable low energy mode..."),
-        bluepy.btle.BTLEManagementError("test"),
-    ],
-)
-def test__update_position_update_error(exception):
-    actor = switchbot_mqtt._CurtainMotor(
-        mac_address="dummy", retry_count=21, password=None
-    )
-    with unittest.mock.patch.object(
-        actor._get_device(), "update", side_effect=exception
-    ), unittest.mock.patch.object(
-        actor, "_report_position"
-    ) as report_position_mock, pytest.raises(
-        type(exception)
-    ):
-        actor._update_position(mqtt_client="client")
-    report_position_mock.assert_not_called()