report button automator's battery level in topic `homeassistant/cover…

…/switchbot/MAC_ADDRESS/battery-percentage` after every command

#41

CHANGELOG.md

@@ -6,9 +6,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [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).
+- command-line option `--fetch-device-info` enables battery level reports on topics
+  `homeassistant/cover/{switchbot,switchbot-curtain}/MAC_ADDRESS/battery-percentage`
+  after every command.
 
 ### Removed
 - compatibility with `python3.5`

README.md

@@ -38,6 +38,9 @@ Send `ON` or `OFF` to topic `homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/se
 $ mosquitto_pub -h MQTT_BROKER -t homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/set -m ON
 ```
 
+The command-line option `--fetch-device-info` enables battery level reports on topic
+`homeassistant/cover/switchbot-curtain/MAC_ADDRESS/battery-percentage` after every command.
+
 ### Curtain Motor
 
 Send `OPEN`, `CLOSE`, or `STOP` to topic `homeassistant/cover/switchbot-curtain/aa:bb:cc:dd:ee:ff/set`.

switchbot_mqtt/__init__.py

@@ -93,6 +93,14 @@ def __eq__(self, other: object) -> bool:
 class _MQTTControlledActor(abc.ABC):
     MQTT_COMMAND_TOPIC_LEVELS: typing.List[_MQTTTopicLevel] = NotImplemented
     MQTT_STATE_TOPIC_LEVELS: typing.List[_MQTTTopicLevel] = NotImplemented
+    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS: typing.List[_MQTTTopicLevel] = NotImplemented
+
+    @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,
+        )
 
     @abc.abstractmethod
     def __init__(
@@ -101,15 +109,6 @@ def __init__(
         # alternative: pySwitchbot >=0.10.0 provides SwitchbotDevice.get_mac()
         self._mac_address = mac_address
 
-    @abc.abstractmethod
-    def execute_command(
-        self,
-        mqtt_message_payload: bytes,
-        mqtt_client: paho.mqtt.client.Client,
-        update_device_info: bool,
-    ) -> None:
-        raise NotImplementedError()
-
     @abc.abstractmethod
     def _get_device(self) -> switchbot.SwitchbotDevice:
         raise NotImplementedError()
@@ -152,6 +151,30 @@ def _update_device_info(self) -> None:
                 ) from exc
             raise
 
+    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._get_device().get_battery_percent()).encode(),
+            mqtt_client=mqtt_client,
+        )
+
+    def _update_and_report_device_info(
+        self, mqtt_client: paho.mqtt.client.Client
+    ) -> None:
+        self._update_device_info()
+        self._report_battery_level(mqtt_client=mqtt_client)
+
+    @abc.abstractmethod
+    def execute_command(
+        self,
+        mqtt_message_payload: bytes,
+        mqtt_client: paho.mqtt.client.Client,
+        update_device_info: bool,
+    ) -> None:
+        raise NotImplementedError()
+
     @classmethod
     def _mqtt_command_callback(
         cls,
@@ -247,13 +270,18 @@ class _ButtonAutomator(_MQTTControlledActor):
         _MQTTTopicPlaceholder.MAC_ADDRESS,
         "set",
     ]
-
     MQTT_STATE_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
         "switch",
         "switchbot",
         _MQTTTopicPlaceholder.MAC_ADDRESS,
         "state",
     ]
+    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
+        "cover",
+        "switchbot",
+        _MQTTTopicPlaceholder.MAC_ADDRESS,
+        "battery-percentage",
+    ]
 
     def __init__(
         self, *, mac_address: str, retry_count: int, password: typing.Optional[str]
@@ -282,13 +310,17 @@ def execute_command(
                 _LOGGER.info("switchbot %s turned on", self._mac_address)
                 # https://www.home-assistant.io/integrations/switch.mqtt/#state_on
                 self.report_state(mqtt_client=mqtt_client, state=b"ON")
+                if update_device_info:
+                    self._update_and_report_device_info(mqtt_client)
         # https://www.home-assistant.io/integrations/switch.mqtt/#payload_off
         elif mqtt_message_payload.lower() == b"off":
             if not self.__device.turn_off():
                 _LOGGER.error("failed to turn off switchbot %s", self._mac_address)
             else:
                 _LOGGER.info("switchbot %s turned off", self._mac_address)
                 self.report_state(mqtt_client=mqtt_client, state=b"OFF")
+                if update_device_info:
+                    self._update_and_report_device_info(mqtt_client)
         else:
             _LOGGER.warning(
                 "unexpected payload %r (expected 'ON' or 'OFF')", mqtt_message_payload
@@ -323,13 +355,6 @@ class _CurtainMotor(_MQTTControlledActor):
         "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(
@@ -354,15 +379,6 @@ 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)
@@ -377,11 +393,10 @@ def _report_position(self, mqtt_client: paho.mqtt.client.Client) -> None:
             mqtt_client=mqtt_client,
         )
 
-    def _update_and_report_device_info(
-        self, mqtt_client: paho.mqtt.client.Client, *, report_position: bool
+    def _update_and_report_device_info(  # pylint: disable=arguments-differ; report_position is optional
+        self, mqtt_client: paho.mqtt.client.Client, *, report_position: bool = True
     ) -> None:
-        self._update_device_info()
-        self._report_battery_level(mqtt_client=mqtt_client)
+        super()._update_and_report_device_info(mqtt_client)
         if report_position:
             self._report_position(mqtt_client=mqtt_client)
 
@@ -517,11 +532,13 @@ def _main() -> None:
     argparser.add_argument(
         "--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 and battery level on topic"
+        help="Report devices' battery level on topic"
+        f" {_ButtonAutomator.get_mqtt_battery_percentage_topic(mac_address='MAC_ADDRESS')}"
+        " or, respectively,"
         f" {_CurtainMotor.get_mqtt_battery_percentage_topic(mac_address='MAC_ADDRESS')}"
-        " after every commands.",
+        " after every command. Additionally report curtain motors' position on"
+        f" topic {_CurtainMotor.get_mqtt_position_topic(mac_address='MAC_ADDRESS')}"
+        " after executing stop commands.",
     )
     args = argparser.parse_args()
     if args.mqtt_password_path:

tests/test_switchbot_button_automator.py

@@ -28,6 +28,36 @@
 # 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(("battery_percent", "battery_percent_encoded"), [(42, b"42")])
+def test__update_and_report_device_info(
+    battery_percent: int, battery_percent_encoded: bytes
+):
+    with unittest.mock.patch("switchbot.SwitchbotCurtain.__init__", return_value=None):
+        actor = switchbot_mqtt._ButtonAutomator(
+            mac_address="dummy", retry_count=21, password=None
+        )
+    actor._get_device()._battery_percent = battery_percent
+    mqtt_client_mock = unittest.mock.MagicMock()
+    with unittest.mock.patch("switchbot.Switchbot.update") as update_mock:
+        actor._update_and_report_device_info(mqtt_client=mqtt_client_mock)
+    update_mock.assert_called_once_with()
+    mqtt_client_mock.publish.assert_called_once_with(
+        topic="homeassistant/cover/switchbot/dummy/battery-percentage",
+        payload=battery_percent_encoded,
+        retain=True,
+    )
+
+
 @pytest.mark.parametrize("mac_address", ["aa:bb:cc:dd:ee:ff", "aa:bb:cc:11:22:33"])
 @pytest.mark.parametrize("password", (None, "secret"))
 @pytest.mark.parametrize("retry_count", (3, 21))
@@ -42,6 +72,7 @@
         (b"Off", "switchbot.Switchbot.turn_off"),
     ],
 )
+@pytest.mark.parametrize("update_device_info", [True, False])
 @pytest.mark.parametrize("command_successful", [True, False])
 def test_execute_command(
     caplog,
@@ -50,6 +81,7 @@ def test_execute_command(
     retry_count,
     message_payload,
     action_name,
+    update_device_info,
     command_successful,
 ):
     with unittest.mock.patch(
@@ -62,11 +94,13 @@ def test_execute_command(
             actor, "report_state"
         ) as report_mock, unittest.mock.patch(
             action_name, return_value=command_successful
-        ) as action_mock:
+        ) as action_mock, unittest.mock.patch.object(
+            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=True,
+                update_device_info=update_device_info,
             )
     device_init_mock.assert_called_once_with(
         mac=mac_address, password=password, retry_count=retry_count
@@ -83,6 +117,7 @@ def test_execute_command(
         report_mock.assert_called_once_with(
             mqtt_client="dummy", state=message_payload.upper()
         )
+        assert update_device_info_mock.call_count == (1 if update_device_info else 0)
     else:
         assert caplog.record_tuples == [
             (
@@ -92,6 +127,7 @@ def test_execute_command(
             )
         ]
         report_mock.assert_not_called()
+        update_device_info_mock.assert_not_called()
 
 
 @pytest.mark.parametrize("mac_address", ["aa:bb:cc:dd:ee:ff"])