From 4feeb7a56d99d2a3b3649c027a150157d2678b2c Mon Sep 17 00:00:00 2001 From: Fabian Peter Hammerle Date: Sat, 16 Oct 2021 13:21:02 +0200 Subject: [PATCH] report button automator's battery level in topic `homeassistant/cover/switchbot/MAC_ADDRESS/battery-percentage` after every command https://github.com/fphammerle/switchbot-mqtt/issues/41 --- CHANGELOG.md | 6 +- README.md | 3 + switchbot_mqtt/__init__.py | 85 ++++++++++++++---------- tests/test_switchbot_button_automator.py | 40 ++++++++++- 4 files changed, 95 insertions(+), 39 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2198731..b167dcc 100644 --- a/CHANGELOG.md +++ b/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` diff --git a/README.md b/README.md index fc8ede8..aaa36d7 100644 --- a/README.md +++ b/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`. diff --git a/switchbot_mqtt/__init__.py b/switchbot_mqtt/__init__.py index 26eeae9..0fa3f87 100644 --- a/switchbot_mqtt/__init__.py +++ b/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,6 +310,8 @@ 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(): @@ -289,6 +319,8 @@ def execute_command( 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: diff --git a/tests/test_switchbot_button_automator.py b/tests/test_switchbot_button_automator.py index b105b34..0a5b387 100644 --- a/tests/test_switchbot_button_automator.py +++ b/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"])