__init__.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299
  1. # switchbot-mqtt - MQTT client controlling SwitchBot button & curtain automators,
  2. # compatible with home-assistant.io's MQTT Switch & Cover platform
  3. #
  4. # Copyright (C) 2020 Fabian Peter Hammerle <fabian@hammerle.me>
  5. #
  6. # This program is free software: you can redistribute it and/or modify
  7. # it under the terms of the GNU General Public License as published by
  8. # the Free Software Foundation, either version 3 of the License, or
  9. # any later version.
  10. #
  11. # This program is distributed in the hope that it will be useful,
  12. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  13. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  14. # GNU General Public License for more details.
  15. #
  16. # You should have received a copy of the GNU General Public License
  17. # along with this program. If not, see <https://www.gnu.org/licenses/>.
  18. import logging
  19. import typing
  20. import bluepy.btle
  21. import paho.mqtt.client
  22. import switchbot
  23. from switchbot_mqtt._actors.base import _MQTTCallbackUserdata, _MQTTControlledActor
  24. from switchbot_mqtt._utils import (
  25. _join_mqtt_topic_levels,
  26. _MQTTTopicLevel,
  27. _MQTTTopicPlaceholder,
  28. )
  29. _LOGGER = logging.getLogger(__name__)
  30. _BUTTON_TOPIC_LEVELS_PREFIX = (
  31. "switch",
  32. "switchbot",
  33. _MQTTTopicPlaceholder.MAC_ADDRESS,
  34. )
  35. _CURTAIN_TOPIC_LEVELS_PREFIX = (
  36. "cover",
  37. "switchbot-curtain",
  38. _MQTTTopicPlaceholder.MAC_ADDRESS,
  39. )
  40. class _ButtonAutomator(_MQTTControlledActor):
  41. # https://www.home-assistant.io/integrations/switch.mqtt/
  42. MQTT_COMMAND_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + ("set",)
  43. _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + (
  44. "request-device-info",
  45. )
  46. MQTT_STATE_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + ("state",)
  47. _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + (
  48. "battery-percentage",
  49. )
  50. def __init__(
  51. self, *, mac_address: str, retry_count: int, password: typing.Optional[str]
  52. ) -> None:
  53. self.__device = switchbot.Switchbot(
  54. mac=mac_address, password=password, retry_count=retry_count
  55. )
  56. super().__init__(
  57. mac_address=mac_address, retry_count=retry_count, password=password
  58. )
  59. def _get_device(self) -> switchbot.SwitchbotDevice:
  60. return self.__device
  61. def execute_command(
  62. self,
  63. *,
  64. mqtt_message_payload: bytes,
  65. mqtt_client: paho.mqtt.client.Client,
  66. update_device_info: bool,
  67. mqtt_topic_prefix: str,
  68. ) -> None:
  69. # https://www.home-assistant.io/integrations/switch.mqtt/#payload_on
  70. if mqtt_message_payload.lower() == b"on":
  71. if not self.__device.turn_on():
  72. _LOGGER.error("failed to turn on switchbot %s", self._mac_address)
  73. else:
  74. _LOGGER.info("switchbot %s turned on", self._mac_address)
  75. # https://www.home-assistant.io/integrations/switch.mqtt/#state_on
  76. self.report_state(
  77. mqtt_client=mqtt_client,
  78. mqtt_topic_prefix=mqtt_topic_prefix,
  79. state=b"ON",
  80. )
  81. if update_device_info:
  82. self._update_and_report_device_info(mqtt_client, mqtt_topic_prefix)
  83. # https://www.home-assistant.io/integrations/switch.mqtt/#payload_off
  84. elif mqtt_message_payload.lower() == b"off":
  85. if not self.__device.turn_off():
  86. _LOGGER.error("failed to turn off switchbot %s", self._mac_address)
  87. else:
  88. _LOGGER.info("switchbot %s turned off", self._mac_address)
  89. self.report_state(
  90. mqtt_client=mqtt_client,
  91. mqtt_topic_prefix=mqtt_topic_prefix,
  92. state=b"OFF",
  93. )
  94. if update_device_info:
  95. self._update_and_report_device_info(mqtt_client, mqtt_topic_prefix)
  96. else:
  97. _LOGGER.warning(
  98. "unexpected payload %r (expected 'ON' or 'OFF')", mqtt_message_payload
  99. )
  100. class _CurtainMotor(_MQTTControlledActor):
  101. # https://www.home-assistant.io/integrations/cover.mqtt/
  102. MQTT_COMMAND_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + ("set",)
  103. _MQTT_SET_POSITION_TOPIC_LEVELS: typing.Tuple[
  104. _MQTTTopicLevel, ...
  105. ] = _CURTAIN_TOPIC_LEVELS_PREFIX + (
  106. "position",
  107. "set-percent",
  108. )
  109. _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + (
  110. "request-device-info",
  111. )
  112. MQTT_STATE_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + ("state",)
  113. _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + (
  114. "battery-percentage",
  115. )
  116. _MQTT_POSITION_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + ("position",)
  117. @classmethod
  118. def get_mqtt_position_topic(cls, prefix: str, mac_address: str) -> str:
  119. return _join_mqtt_topic_levels(
  120. topic_prefix=prefix,
  121. topic_levels=cls._MQTT_POSITION_TOPIC_LEVELS,
  122. mac_address=mac_address,
  123. )
  124. def __init__(
  125. self, *, mac_address: str, retry_count: int, password: typing.Optional[str]
  126. ) -> None:
  127. # > The position of the curtain is saved in self._pos with 0 = open and 100 = closed.
  128. # https://github.com/Danielhiversen/pySwitchbot/blob/0.10.0/switchbot/__init__.py#L150
  129. self.__device = switchbot.SwitchbotCurtain(
  130. mac=mac_address,
  131. password=password,
  132. retry_count=retry_count,
  133. reverse_mode=True,
  134. )
  135. super().__init__(
  136. mac_address=mac_address, retry_count=retry_count, password=password
  137. )
  138. def _get_device(self) -> switchbot.SwitchbotDevice:
  139. return self.__device
  140. def _report_position(
  141. self,
  142. mqtt_client: paho.mqtt.client.Client, # pylint: disable=duplicate-code; similar param list
  143. mqtt_topic_prefix: str,
  144. ) -> None:
  145. # > position_closed integer (Optional, default: 0)
  146. # > position_open integer (Optional, default: 100)
  147. # https://www.home-assistant.io/integrations/cover.mqtt/#position_closed
  148. # SwitchbotCurtain.get_position() returns a cached value within [0, 100].
  149. # SwitchbotCurtain.open() and .close() update the position optimistically,
  150. # SwitchbotCurtain.update() fetches the real position via bluetooth.
  151. # https://github.com/Danielhiversen/pySwitchbot/blob/0.10.0/switchbot/__init__.py#L202
  152. self._mqtt_publish(
  153. topic_prefix=mqtt_topic_prefix,
  154. topic_levels=self._MQTT_POSITION_TOPIC_LEVELS,
  155. payload=str(int(self.__device.get_position())).encode(),
  156. mqtt_client=mqtt_client,
  157. )
  158. def _update_and_report_device_info( # pylint: disable=arguments-differ; report_position is optional
  159. self,
  160. mqtt_client: paho.mqtt.client.Client,
  161. mqtt_topic_prefix: str,
  162. *,
  163. report_position: bool = True,
  164. ) -> None:
  165. super()._update_and_report_device_info(mqtt_client, mqtt_topic_prefix)
  166. if report_position:
  167. self._report_position(
  168. mqtt_client=mqtt_client, mqtt_topic_prefix=mqtt_topic_prefix
  169. )
  170. def execute_command(
  171. self,
  172. *,
  173. mqtt_message_payload: bytes,
  174. mqtt_client: paho.mqtt.client.Client,
  175. update_device_info: bool,
  176. mqtt_topic_prefix: str,
  177. ) -> None:
  178. # https://www.home-assistant.io/integrations/cover.mqtt/#payload_open
  179. report_device_info, report_position = False, False
  180. if mqtt_message_payload.lower() == b"open":
  181. if not self.__device.open():
  182. _LOGGER.error("failed to open switchbot curtain %s", self._mac_address)
  183. else:
  184. _LOGGER.info("switchbot curtain %s opening", self._mac_address)
  185. # > state_opening string (Optional, default: opening)
  186. # https://www.home-assistant.io/integrations/cover.mqtt/#state_opening
  187. self.report_state(
  188. mqtt_client=mqtt_client,
  189. mqtt_topic_prefix=mqtt_topic_prefix,
  190. state=b"opening",
  191. )
  192. report_device_info = update_device_info
  193. elif mqtt_message_payload.lower() == b"close":
  194. if not self.__device.close():
  195. _LOGGER.error("failed to close switchbot curtain %s", self._mac_address)
  196. else:
  197. _LOGGER.info("switchbot curtain %s closing", self._mac_address)
  198. # https://www.home-assistant.io/integrations/cover.mqtt/#state_closing
  199. self.report_state(
  200. mqtt_client=mqtt_client,
  201. mqtt_topic_prefix=mqtt_topic_prefix,
  202. state=b"closing",
  203. )
  204. report_device_info = update_device_info
  205. elif mqtt_message_payload.lower() == b"stop":
  206. if not self.__device.stop():
  207. _LOGGER.error("failed to stop switchbot curtain %s", self._mac_address)
  208. else:
  209. _LOGGER.info("switchbot curtain %s stopped", self._mac_address)
  210. # no "stopped" state mentioned at
  211. # https://www.home-assistant.io/integrations/cover.mqtt/#configuration-variables
  212. # https://community.home-assistant.io/t/mqtt-how-to-remove-retained-messages/79029/2
  213. self.report_state(
  214. mqtt_client=mqtt_client,
  215. mqtt_topic_prefix=mqtt_topic_prefix,
  216. state=b"",
  217. )
  218. report_device_info = update_device_info
  219. report_position = True
  220. else:
  221. _LOGGER.warning(
  222. "unexpected payload %r (expected 'OPEN', 'CLOSE', or 'STOP')",
  223. mqtt_message_payload,
  224. )
  225. if report_device_info:
  226. self._update_and_report_device_info(
  227. mqtt_client=mqtt_client,
  228. mqtt_topic_prefix=mqtt_topic_prefix,
  229. report_position=report_position,
  230. )
  231. @classmethod
  232. def _mqtt_set_position_callback(
  233. cls,
  234. mqtt_client: paho.mqtt.client.Client,
  235. userdata: _MQTTCallbackUserdata,
  236. message: paho.mqtt.client.MQTTMessage,
  237. ) -> None:
  238. # pylint: disable=unused-argument; callback
  239. # https://github.com/eclipse/paho.mqtt.python/blob/v1.6.1/src/paho/mqtt/client.py#L3556
  240. _LOGGER.debug("received topic=%s payload=%r", message.topic, message.payload)
  241. if message.retain:
  242. _LOGGER.info("ignoring retained message on topic %s", message.topic)
  243. return
  244. actor = cls._init_from_topic(
  245. topic=message.topic,
  246. expected_topic_levels=cls._MQTT_SET_POSITION_TOPIC_LEVELS,
  247. settings=userdata,
  248. )
  249. if not actor:
  250. return # warning in _init_from_topic
  251. position_percent = int(message.payload.decode(), 10)
  252. if position_percent < 0 or position_percent > 100:
  253. _LOGGER.warning("invalid position %u%%, ignoring message", position_percent)
  254. return
  255. # pylint: disable=protected-access; own instance
  256. if actor._get_device().set_position(position_percent):
  257. _LOGGER.info(
  258. "set position of switchbot curtain %s to %u%%",
  259. actor._mac_address,
  260. position_percent,
  261. )
  262. else:
  263. _LOGGER.error(
  264. "failed to set position of switchbot curtain %s", actor._mac_address
  265. )
  266. @classmethod
  267. def _get_mqtt_message_callbacks(
  268. # pylint: disable=duplicate-code; param list in parent class
  269. cls,
  270. *,
  271. enable_device_info_update_topic: bool,
  272. ) -> typing.Dict[typing.Tuple[_MQTTTopicLevel, ...], typing.Callable]:
  273. callbacks = super()._get_mqtt_message_callbacks(
  274. enable_device_info_update_topic=enable_device_info_update_topic
  275. )
  276. callbacks[cls._MQTT_SET_POSITION_TOPIC_LEVELS] = cls._mqtt_set_position_callback
  277. return callbacks