1
0

base.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302
  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. # > Even with __all__ set appropriately, internal interfaces (packages,
  19. # > modules, classes, functions, attributes or other names) should still be
  20. # > prefixed with a single leading underscore. An interface is also considered
  21. # > internal if any containing namespace (package, module or class) is
  22. # > considered internal.
  23. # https://peps.python.org/pep-0008/#public-and-internal-interfaces
  24. from __future__ import annotations # PEP563 (default in python>=3.10)
  25. import abc
  26. import logging
  27. import typing
  28. import aiomqtt
  29. import bleak
  30. import bleak.backends.device
  31. import switchbot
  32. from switchbot_mqtt._utils import (
  33. _join_mqtt_topic_levels,
  34. _mac_address_valid,
  35. _MQTTTopicLevel,
  36. _MQTTTopicPlaceholder,
  37. _parse_mqtt_topic,
  38. )
  39. _LOGGER = logging.getLogger(__name__)
  40. class _MQTTControlledActor(abc.ABC):
  41. MQTT_COMMAND_TOPIC_LEVELS: typing.Tuple[_MQTTTopicLevel, ...] = NotImplemented
  42. _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS: typing.Tuple[_MQTTTopicLevel, ...] = (
  43. NotImplemented
  44. )
  45. MQTT_STATE_TOPIC_LEVELS: typing.Tuple[_MQTTTopicLevel, ...] = NotImplemented
  46. _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS: typing.Tuple[_MQTTTopicLevel, ...] = (
  47. NotImplemented
  48. )
  49. @classmethod
  50. def get_mqtt_update_device_info_topic(cls, *, prefix: str, mac_address: str) -> str:
  51. return _join_mqtt_topic_levels(
  52. topic_prefix=prefix,
  53. topic_levels=cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS,
  54. mac_address=mac_address,
  55. )
  56. @classmethod
  57. def get_mqtt_battery_percentage_topic(cls, *, prefix: str, mac_address: str) -> str:
  58. return _join_mqtt_topic_levels(
  59. topic_prefix=prefix,
  60. topic_levels=cls._MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS,
  61. mac_address=mac_address,
  62. )
  63. @abc.abstractmethod
  64. def __init__(
  65. self,
  66. *,
  67. device: bleak.backends.device.BLEDevice,
  68. retry_count: int,
  69. password: typing.Optional[str],
  70. ) -> None:
  71. # alternative: pySwitchbot >=0.10.0 provides SwitchbotDevice.get_mac()
  72. self._mac_address = device.address
  73. self._basic_device_info: typing.Optional[typing.Dict[str, typing.Any]] = None
  74. @abc.abstractmethod
  75. def _get_device(self) -> switchbot.SwitchbotDevice:
  76. raise NotImplementedError()
  77. async def _report_battery_level(
  78. self, mqtt_client: aiomqtt.Client, mqtt_topic_prefix: str
  79. ) -> None:
  80. assert self._basic_device_info is not None
  81. # > battery: Percentage of battery that is left.
  82. # https://www.home-assistant.io/integrations/sensor/#device-class
  83. await self._mqtt_publish(
  84. topic_prefix=mqtt_topic_prefix,
  85. topic_levels=self._MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS,
  86. payload=str(self._basic_device_info["battery"]).encode(),
  87. mqtt_client=mqtt_client,
  88. )
  89. async def _update_and_report_device_info(
  90. self, mqtt_client: aiomqtt.Client, mqtt_topic_prefix: str
  91. ) -> None:
  92. self._basic_device_info = await self._get_device().get_basic_info()
  93. if self._basic_device_info is None:
  94. _LOGGER.error(
  95. "failed to retrieve basic device info from %s", self._mac_address
  96. )
  97. else:
  98. await self._report_battery_level(
  99. mqtt_client=mqtt_client, mqtt_topic_prefix=mqtt_topic_prefix
  100. )
  101. @classmethod
  102. async def _init_from_topic(
  103. cls,
  104. *,
  105. topic: aiomqtt.Topic,
  106. mqtt_topic_prefix: str,
  107. expected_topic_levels: typing.Collection[_MQTTTopicLevel],
  108. retry_count: int,
  109. device_passwords: typing.Dict[str, str],
  110. ) -> typing.Optional[_MQTTControlledActor]:
  111. try:
  112. mac_address = _parse_mqtt_topic(
  113. topic=topic.value,
  114. expected_prefix=mqtt_topic_prefix,
  115. expected_levels=expected_topic_levels,
  116. )[_MQTTTopicPlaceholder.MAC_ADDRESS]
  117. except ValueError as exc:
  118. _LOGGER.warning(str(exc), exc_info=False)
  119. return None
  120. if not _mac_address_valid(mac_address):
  121. _LOGGER.warning("invalid mac address %s", mac_address)
  122. return None
  123. # SwitchbotBaseDevice.__init__ expects BLEDevice
  124. # disabling mypy to workaround false-positive:
  125. # > error: Missing named argument "service_uuids" for
  126. # . "find_device_by_address" of "BleakScanner" [call-arg]
  127. device = await bleak.BleakScanner.find_device_by_address(mac_address) # type: ignore
  128. if device is None:
  129. _LOGGER.error(
  130. "failed to find bluetooth low energy device with mac address %s",
  131. mac_address,
  132. )
  133. return None
  134. return cls(
  135. device=device,
  136. retry_count=retry_count,
  137. password=device_passwords.get(mac_address, None),
  138. )
  139. @classmethod
  140. async def _mqtt_update_device_info_callback(
  141. # pylint: disable=duplicate-code; other callbacks with same params
  142. cls,
  143. *,
  144. mqtt_client: aiomqtt.Client,
  145. message: aiomqtt.Message,
  146. mqtt_topic_prefix: str,
  147. retry_count: int,
  148. device_passwords: typing.Dict[str, str],
  149. fetch_device_info: bool,
  150. ) -> None:
  151. # pylint: disable=unused-argument; callback
  152. # https://github.com/eclipse/paho.mqtt.python/blob/v1.5.0/src/paho/mqtt/client.py#L469
  153. _LOGGER.debug("received topic=%s payload=%r", message.topic, message.payload)
  154. if message.retain:
  155. _LOGGER.info("ignoring retained message")
  156. return
  157. actor = await cls._init_from_topic(
  158. topic=message.topic,
  159. mqtt_topic_prefix=mqtt_topic_prefix,
  160. expected_topic_levels=cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS,
  161. retry_count=retry_count,
  162. device_passwords=device_passwords,
  163. )
  164. if actor:
  165. # pylint: disable=protected-access; own instance
  166. await actor._update_and_report_device_info(
  167. mqtt_client=mqtt_client, mqtt_topic_prefix=mqtt_topic_prefix
  168. )
  169. @abc.abstractmethod
  170. async def execute_command( # pylint: disable=duplicate-code; implementations
  171. self,
  172. *,
  173. mqtt_message_payload: bytes,
  174. mqtt_client: aiomqtt.Client,
  175. update_device_info: bool,
  176. mqtt_topic_prefix: str,
  177. ) -> None:
  178. raise NotImplementedError()
  179. @classmethod
  180. async def _mqtt_command_callback(
  181. # pylint: disable=duplicate-code; other callbacks with same params
  182. cls,
  183. *,
  184. mqtt_client: aiomqtt.Client,
  185. message: aiomqtt.Message,
  186. mqtt_topic_prefix: str,
  187. retry_count: int,
  188. device_passwords: typing.Dict[str, str],
  189. fetch_device_info: bool,
  190. ) -> None:
  191. # pylint: disable=unused-argument; callback
  192. # https://github.com/eclipse/paho.mqtt.python/blob/v1.5.0/src/paho/mqtt/client.py#L469
  193. _LOGGER.debug("received topic=%s payload=%r", message.topic, message.payload)
  194. if message.retain:
  195. _LOGGER.info("ignoring retained message")
  196. return
  197. actor = await cls._init_from_topic(
  198. topic=message.topic,
  199. mqtt_topic_prefix=mqtt_topic_prefix,
  200. expected_topic_levels=cls.MQTT_COMMAND_TOPIC_LEVELS,
  201. retry_count=retry_count,
  202. device_passwords=device_passwords,
  203. )
  204. if actor:
  205. assert isinstance(message.payload, bytes), message.payload
  206. await actor.execute_command(
  207. mqtt_message_payload=message.payload,
  208. mqtt_client=mqtt_client,
  209. update_device_info=fetch_device_info,
  210. mqtt_topic_prefix=mqtt_topic_prefix,
  211. )
  212. @classmethod
  213. def _get_mqtt_message_callbacks(
  214. cls,
  215. *,
  216. enable_device_info_update_topic: bool,
  217. ) -> typing.Dict[typing.Tuple[_MQTTTopicLevel, ...], typing.Callable]:
  218. # returning dict because `paho.mqtt.client.Client.message_callback_add` overwrites
  219. # callbacks with same topic pattern
  220. # https://github.com/eclipse/paho.mqtt.python/blob/v1.6.1/src/paho/mqtt/client.py#L2304
  221. # https://github.com/eclipse/paho.mqtt.python/blob/v1.6.1/src/paho/mqtt/matcher.py#L19
  222. callbacks = {cls.MQTT_COMMAND_TOPIC_LEVELS: cls._mqtt_command_callback}
  223. if enable_device_info_update_topic:
  224. callbacks[cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS] = (
  225. cls._mqtt_update_device_info_callback
  226. )
  227. return callbacks
  228. @classmethod
  229. async def mqtt_subscribe(
  230. cls,
  231. *,
  232. mqtt_client: aiomqtt.Client,
  233. mqtt_topic_prefix: str,
  234. fetch_device_info: bool,
  235. ) -> typing.AsyncIterator[typing.Tuple[str, typing.Callable]]:
  236. for topic_levels, callback in cls._get_mqtt_message_callbacks(
  237. enable_device_info_update_topic=fetch_device_info
  238. ).items():
  239. topic = _join_mqtt_topic_levels(
  240. topic_prefix=mqtt_topic_prefix,
  241. topic_levels=topic_levels,
  242. mac_address="+",
  243. )
  244. _LOGGER.info("subscribing to MQTT topic %r", topic)
  245. await mqtt_client.subscribe(topic)
  246. yield (topic, callback)
  247. async def _mqtt_publish(
  248. self,
  249. *,
  250. topic_prefix: str,
  251. topic_levels: typing.Iterable[_MQTTTopicLevel],
  252. payload: bytes,
  253. mqtt_client: aiomqtt.Client,
  254. ) -> None:
  255. topic = _join_mqtt_topic_levels(
  256. topic_prefix=topic_prefix,
  257. topic_levels=topic_levels,
  258. mac_address=self._mac_address,
  259. )
  260. # https://pypi.org/project/paho-mqtt/#publishing
  261. _LOGGER.debug("publishing topic=%s payload=%r", topic, payload)
  262. try:
  263. await mqtt_client.publish(topic=topic, payload=payload, retain=True)
  264. except aiomqtt.MqttCodeError as exc:
  265. _LOGGER.error(
  266. "Failed to publish MQTT message on topic %s: aiomqtt.MqttCodeError %s",
  267. topic,
  268. exc,
  269. )
  270. async def report_state(
  271. self,
  272. state: bytes,
  273. mqtt_client: aiomqtt.Client,
  274. mqtt_topic_prefix: str,
  275. ) -> None:
  276. await self._mqtt_publish(
  277. topic_prefix=mqtt_topic_prefix,
  278. topic_levels=self.MQTT_STATE_TOPIC_LEVELS,
  279. payload=state,
  280. mqtt_client=mqtt_client,
  281. )