update & report device info when receiving msg on `homeassistant/{swi…

…tch/switchbot,cover/switchbot-curtain}/MAC_ADDRESS/request-device-info` (requires `--fetch-device-info`)

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
+- MQTT messages on topic `homeassistant/switch/switchbot/MAC_ADDRESS/request-device-info`
+  and `homeassistant/cover/switchbot-curtain/MAC_ADDRESS/request-device-info` trigger
+  update and reporting of device information (battery level, and curtains' position).
+  Requires `--fetch-device-info`.
 
 ## [2.1.0] - 2021-10-19
 ### Added

README.md

@@ -40,6 +40,8 @@ $ mosquitto_pub -h MQTT_BROKER -t homeassistant/switch/switchbot/aa:bb:cc:dd:ee:
 
 The command-line option `--fetch-device-info` enables battery level reports on topic
 `homeassistant/switch/switchbot/MAC_ADDRESS/battery-percentage` after every command.
+The report may be requested manually by sending a MQTT message to the topic
+`homeassistant/switch/switchbot/MAC_ADDRESS/request-device-info` (requires `--fetch-device-info`)
 
 ### Curtain Motor
 
@@ -53,6 +55,8 @@ 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`
 after every command.
+These reports may be requested manually by sending a MQTT message to the topic
+`homeassistant/cover/switchbot-curtain/MAC_ADDRESS/request-device-info` (requires `--fetch-device-info`)
 
 ### Device Passwords
 

switchbot_mqtt/__init__.py

@@ -38,8 +38,14 @@ def _mqtt_on_connect(
     assert return_code == 0, return_code  # connection accepted
     mqtt_broker_host, mqtt_broker_port = mqtt_client.socket().getpeername()
     _LOGGER.debug("connected to MQTT broker %s:%d", mqtt_broker_host, mqtt_broker_port)
-    _ButtonAutomator.mqtt_subscribe(mqtt_client=mqtt_client)
-    _CurtainMotor.mqtt_subscribe(mqtt_client=mqtt_client)
+    _ButtonAutomator.mqtt_subscribe(
+        mqtt_client=mqtt_client,
+        enable_device_info_update_topic=userdata.fetch_device_info,
+    )
+    _CurtainMotor.mqtt_subscribe(
+        mqtt_client=mqtt_client,
+        enable_device_info_update_topic=userdata.fetch_device_info,
+    )
 
 
 def _run(

switchbot_mqtt/_actors/__init__.py

@@ -33,32 +33,32 @@
 _LOGGER = logging.getLogger(__name__)
 
 # "homeassistant" for historic reason, may be parametrized in future
-_MQTT_TOPIC_LEVELS_PREFIX: typing.List[_MQTTTopicLevel] = ["homeassistant"]
+_TOPIC_LEVELS_PREFIX: typing.List[_MQTTTopicLevel] = ["homeassistant"]
+_BUTTON_TOPIC_LEVELS_PREFIX = _TOPIC_LEVELS_PREFIX + [
+    "switch",
+    "switchbot",
+    _MQTTTopicPlaceholder.MAC_ADDRESS,
+]
+_CURTAIN_TOPIC_LEVELS_PREFIX = _TOPIC_LEVELS_PREFIX + [
+    "cover",
+    "switchbot-curtain",
+    _MQTTTopicPlaceholder.MAC_ADDRESS,
+]
 
 
 class _ButtonAutomator(_MQTTControlledActor):
     # https://www.home-assistant.io/integrations/switch.mqtt/
 
-    MQTT_COMMAND_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
-        "switch",
-        "switchbot",
-        _MQTTTopicPlaceholder.MAC_ADDRESS,
-        "set",
-    ]
-    MQTT_STATE_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
-        "switch",
-        "switchbot",
-        _MQTTTopicPlaceholder.MAC_ADDRESS,
-        "state",
+    MQTT_COMMAND_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + ["set"]
+    _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + [
+        "request-device-info"
     ]
-    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
-        "switch",
-        "switchbot",
-        _MQTTTopicPlaceholder.MAC_ADDRESS,
-        "battery-percentage",
+    MQTT_STATE_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + ["state"]
+    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + [
+        "battery-percentage"
     ]
     # for downward compatibility (will be removed in v3):
-    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS_LEGACY = _MQTT_TOPIC_LEVELS_PREFIX + [
+    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS_LEGACY = _TOPIC_LEVELS_PREFIX + [
         "cover",
         "switchbot",
         _MQTTTopicPlaceholder.MAC_ADDRESS,
@@ -121,30 +121,15 @@ def execute_command(
 class _CurtainMotor(_MQTTControlledActor):
     # https://www.home-assistant.io/integrations/cover.mqtt/
 
-    MQTT_COMMAND_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
-        "cover",
-        "switchbot-curtain",
-        _MQTTTopicPlaceholder.MAC_ADDRESS,
-        "set",
+    MQTT_COMMAND_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + ["set"]
+    _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + [
+        "request-device-info"
     ]
-    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",
+    MQTT_STATE_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + ["state"]
+    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + [
+        "battery-percentage"
     ]
+    _MQTT_POSITION_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + ["position"]
 
     @classmethod
     def get_mqtt_position_topic(cls, mac_address: str) -> str:

switchbot_mqtt/_actors/_base.py

@@ -56,9 +56,17 @@ def __eq__(self, other: object) -> bool:
 
 class _MQTTControlledActor(abc.ABC):
     MQTT_COMMAND_TOPIC_LEVELS: typing.List[_MQTTTopicLevel] = NotImplemented
+    _MQTT_UPDATE_DEVICE_INFO_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_update_device_info_topic(cls, mac_address: str) -> str:
+        return _join_mqtt_topic_levels(
+            topic_levels=cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS,
+            mac_address=mac_address,
+        )
+
     @classmethod
     def get_mqtt_battery_percentage_topic(cls, mac_address: str) -> str:
         return _join_mqtt_topic_levels(
@@ -130,6 +138,51 @@ def _update_and_report_device_info(
         self._update_device_info()
         self._report_battery_level(mqtt_client=mqtt_client)
 
+    @classmethod
+    def _init_from_topic(
+        cls,
+        userdata: _MQTTCallbackUserdata,
+        topic: str,
+        expected_topic_levels: typing.List[_MQTTTopicLevel],
+    ) -> typing.Optional["_MQTTControlledActor"]:
+        try:
+            mac_address = _parse_mqtt_topic(
+                topic=topic, expected_levels=expected_topic_levels
+            )[_MQTTTopicPlaceholder.MAC_ADDRESS]
+        except ValueError as exc:
+            _LOGGER.warning(str(exc), exc_info=False)
+            return None
+        if not _mac_address_valid(mac_address):
+            _LOGGER.warning("invalid mac address %s", mac_address)
+            return None
+        return cls(
+            mac_address=mac_address,
+            retry_count=userdata.retry_count,
+            password=userdata.device_passwords.get(mac_address, None),
+        )
+
+    @classmethod
+    def _mqtt_update_device_info_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.5.0/src/paho/mqtt/client.py#L469
+        _LOGGER.debug("received topic=%s payload=%r", message.topic, message.payload)
+        if message.retain:
+            _LOGGER.info("ignoring retained message")
+            return
+        actor = cls._init_from_topic(
+            userdata=userdata,
+            topic=message.topic,
+            expected_topic_levels=cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS,
+        )
+        if actor:
+            # pylint: disable=protected-access; own instance
+            actor._update_and_report_device_info(mqtt_client)
+
     @abc.abstractmethod
     def execute_command(
         self,
@@ -152,40 +205,38 @@ def _mqtt_command_callback(
         if message.retain:
             _LOGGER.info("ignoring retained message")
             return
-        try:
-            mac_address = _parse_mqtt_topic(
-                topic=message.topic, expected_levels=cls.MQTT_COMMAND_TOPIC_LEVELS
-            )[_MQTTTopicPlaceholder.MAC_ADDRESS]
-        except ValueError as exc:
-            _LOGGER.warning(str(exc), exc_info=False)
-            return
-        if not _mac_address_valid(mac_address):
-            _LOGGER.warning("invalid mac address %s", mac_address)
-            return
-        actor = cls(
-            mac_address=mac_address,
-            retry_count=userdata.retry_count,
-            password=userdata.device_passwords.get(mac_address, None),
-        )
-        actor.execute_command(
-            mqtt_message_payload=message.payload,
-            mqtt_client=mqtt_client,
-            # consider calling update+report method directly when adding support for battery levels
-            update_device_info=userdata.fetch_device_info,
+        actor = cls._init_from_topic(
+            userdata=userdata,
+            topic=message.topic,
+            expected_topic_levels=cls.MQTT_COMMAND_TOPIC_LEVELS,
         )
+        if actor:
+            actor.execute_command(
+                mqtt_message_payload=message.payload,
+                mqtt_client=mqtt_client,
+                update_device_info=userdata.fetch_device_info,
+            )
 
     @classmethod
-    def mqtt_subscribe(cls, mqtt_client: paho.mqtt.client.Client) -> None:
-        command_topic = "/".join(
-            "+" if isinstance(l, _MQTTTopicPlaceholder) else l
-            for l in cls.MQTT_COMMAND_TOPIC_LEVELS
-        )
-        _LOGGER.info("subscribing to MQTT topic %r", command_topic)
-        mqtt_client.subscribe(command_topic)
-        mqtt_client.message_callback_add(
-            sub=command_topic,
-            callback=cls._mqtt_command_callback,
-        )
+    def mqtt_subscribe(
+        cls,
+        mqtt_client: paho.mqtt.client.Client,
+        *,
+        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:
+            topic = _join_mqtt_topic_levels(topic_levels, mac_address="+")
+            _LOGGER.info("subscribing to MQTT topic %r", topic)
+            mqtt_client.subscribe(topic)
+            mqtt_client.message_callback_add(sub=topic, callback=callback)
 
     def _mqtt_publish(
         self,

switchbot_mqtt/_cli.py

@@ -74,6 +74,10 @@ def _main() -> None:
         " after every command. Additionally report curtain motors' position on"
         f" topic {_CurtainMotor.get_mqtt_position_topic(mac_address='MAC_ADDRESS')}"
         " after executing stop commands."
+        " When this option is enabled, the mentioned reports may also be requested"
+        " by sending a MQTT message to the topic"
+        f" {_ButtonAutomator.get_mqtt_update_device_info_topic(mac_address='MAC_ADDRESS')}"
+        f" or {_CurtainMotor.get_mqtt_update_device_info_topic(mac_address='MAC_ADDRESS')}."
         " This option can also be enabled by assigning a non-empty value to the"
         " environment variable FETCH_DEVICE_INFO.",
     )