refactor to remove duplicate definition of _CurtainMotor._MQTT_POSITION_TOPIC_LEVELS

switchbot_mqtt/__init__.py

@@ -46,6 +46,15 @@ _MQTTTopicLevel = typing.Union[str, _MQTTTopicPlaceholder]
 _MQTT_TOPIC_LEVELS_PREFIX = ["homeassistant"]  # type: typing.List[_MQTTTopicLevel]
 
 
+def _join_mqtt_topic_levels(
+    topic_levels: typing.List[_MQTTTopicLevel], mac_address: str
+) -> str:
+    return "/".join(
+        mac_address if l == _MQTTTopicPlaceholder.MAC_ADDRESS else typing.cast(str, l)
+        for l in topic_levels
+    )
+
+
 def _mac_address_valid(mac_address: str) -> bool:
     return _MAC_ADDRESS_REGEX.match(mac_address.lower()) is not None
 
@@ -161,11 +170,8 @@ class _MQTTControlledActor(abc.ABC):
         payload: bytes,
         mqtt_client: paho.mqtt.client.Client,
     ) -> None:
-        topic = "/".join(
+        topic = _join_mqtt_topic_levels(
-            self._mac_address
+            topic_levels=topic_levels, mac_address=self._mac_address
-            if l == _MQTTTopicPlaceholder.MAC_ADDRESS
-            else typing.cast(str, l)
-            for l in topic_levels
         )
         # https://pypi.org/project/paho-mqtt/#publishing
         _LOGGER.debug("publishing topic=%s payload=%r", topic, payload)
@@ -266,6 +272,12 @@ class _CurtainMotor(_MQTTControlledActor):
         "position",
     ]
 
+    @classmethod
+    def get_mqtt_position_topic(cls, mac_address: str) -> str:
+        return _join_mqtt_topic_levels(
+            topic_levels=cls._MQTT_POSITION_TOPIC_LEVELS, mac_address=mac_address
+        )
+
     def __init__(
         self, mac_address: str, retry_count: int, password: typing.Optional[str]
     ) -> None:
@@ -463,8 +475,9 @@ def _main() -> None:
     argparser.add_argument(
         "--fetch-device-info",  # generic name to cover future addition of battery level etc.
         action="store_true",
-        help="Report curtain motors' position on topic"
+        help="Report curtain motors' position on topic {} after sending stop command.".format(
-        " homeassistant/cover/switchbot-curtain/MAC_ADDRESS/position after sending stop command.",
+            _CurtainMotor.get_mqtt_position_topic(mac_address="MAC_ADDRESS")
+        ),
     )
     args = argparser.parse_args()
     if args.mqtt_password_path:

tests/test_switchbot_curtain_motor.py

@@ -28,6 +28,13 @@ import switchbot_mqtt
 # 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_position_topic(mac_address):
+    assert switchbot_mqtt._CurtainMotor.get_mqtt_position_topic(
+        mac_address=mac_address
+    ) == "homeassistant/cover/switchbot-curtain/{}/position".format(mac_address)
+
+
 @pytest.mark.parametrize(
     "mac_address",
     ("aa:bb:cc:dd:ee:ff", "aa:bb:cc:dd:ee:gg"),