_base.py 10.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263
  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. from __future__ import annotations # PEP563 (default in python>=3.10)
  19. import abc
  20. import dataclasses
  21. import logging
  22. import queue
  23. import shlex
  24. import typing
  25. import bluepy.btle
  26. import paho.mqtt.client
  27. import switchbot
  28. from switchbot_mqtt._utils import (
  29. _join_mqtt_topic_levels,
  30. _mac_address_valid,
  31. _MQTTTopicLevel,
  32. _MQTTTopicPlaceholder,
  33. _parse_mqtt_topic,
  34. _QueueLogHandler,
  35. )
  36. _LOGGER = logging.getLogger(__name__)
  37. @dataclasses.dataclass
  38. class _MQTTCallbackUserdata:
  39. retry_count: int
  40. device_passwords: typing.Dict[str, str]
  41. fetch_device_info: bool
  42. class _MQTTControlledActor(abc.ABC):
  43. MQTT_COMMAND_TOPIC_LEVELS: typing.List[_MQTTTopicLevel] = NotImplemented
  44. _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS: typing.List[_MQTTTopicLevel] = NotImplemented
  45. MQTT_STATE_TOPIC_LEVELS: typing.List[_MQTTTopicLevel] = NotImplemented
  46. _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS: typing.List[_MQTTTopicLevel] = NotImplemented
  47. @classmethod
  48. def get_mqtt_update_device_info_topic(cls, mac_address: str) -> str:
  49. return _join_mqtt_topic_levels(
  50. topic_levels=cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS,
  51. mac_address=mac_address,
  52. )
  53. @classmethod
  54. def get_mqtt_battery_percentage_topic(cls, mac_address: str) -> str:
  55. return _join_mqtt_topic_levels(
  56. topic_levels=cls._MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS,
  57. mac_address=mac_address,
  58. )
  59. @abc.abstractmethod
  60. def __init__(
  61. self, *, mac_address: str, retry_count: int, password: typing.Optional[str]
  62. ) -> None:
  63. # alternative: pySwitchbot >=0.10.0 provides SwitchbotDevice.get_mac()
  64. self._mac_address = mac_address
  65. @abc.abstractmethod
  66. def _get_device(self) -> switchbot.SwitchbotDevice:
  67. raise NotImplementedError()
  68. def _update_device_info(self) -> None:
  69. log_queue: queue.Queue[logging.LogRecord] = queue.Queue(maxsize=0)
  70. logging.getLogger("switchbot").addHandler(_QueueLogHandler(log_queue))
  71. try:
  72. self._get_device().update()
  73. # pySwitchbot>=v0.10.1 catches bluepy.btle.BTLEManagementError :(
  74. # https://github.com/Danielhiversen/pySwitchbot/blob/0.10.1/switchbot/__init__.py#L141
  75. # pySwitchbot<0.11.0 WARNING, >=0.11.0 ERROR
  76. while not log_queue.empty():
  77. log_record = log_queue.get()
  78. if log_record.exc_info:
  79. exc: typing.Optional[BaseException] = log_record.exc_info[1]
  80. if (
  81. isinstance(exc, bluepy.btle.BTLEManagementError)
  82. and exc.emsg == "Permission Denied"
  83. ):
  84. raise exc
  85. except bluepy.btle.BTLEManagementError as exc:
  86. if (
  87. exc.emsg == "Permission Denied"
  88. and exc.message == "Failed to execute management command 'le on'"
  89. ):
  90. raise PermissionError(
  91. "bluepy-helper failed to enable low energy mode"
  92. " due to insufficient permissions."
  93. "\nSee https://github.com/IanHarvey/bluepy/issues/313#issuecomment-428324639"
  94. ", https://github.com/fphammerle/switchbot-mqtt/pull/31#issuecomment-846383603"
  95. ", and https://github.com/IanHarvey/bluepy/blob/v/1.3.0/bluepy"
  96. "/bluepy-helper.c#L1260."
  97. "\nInsecure workaround:"
  98. "\n1. sudo apt-get install --no-install-recommends libcap2-bin"
  99. f"\n2. sudo setcap cap_net_admin+ep {shlex.quote(bluepy.btle.helperExe)}"
  100. "\n3. restart switchbot-mqtt"
  101. "\nIn docker-based setups, you could use"
  102. " `sudo docker run --cap-drop ALL --cap-add NET_ADMIN --user 0 …`"
  103. " (seriously insecure)."
  104. ) from exc
  105. raise
  106. def _report_battery_level(self, mqtt_client: paho.mqtt.client.Client) -> None:
  107. # > battery: Percentage of battery that is left.
  108. # https://www.home-assistant.io/integrations/sensor/#device-class
  109. self._mqtt_publish(
  110. topic_levels=self._MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS,
  111. payload=str(self._get_device().get_battery_percent()).encode(),
  112. mqtt_client=mqtt_client,
  113. )
  114. def _update_and_report_device_info(
  115. self, mqtt_client: paho.mqtt.client.Client
  116. ) -> None:
  117. self._update_device_info()
  118. self._report_battery_level(mqtt_client=mqtt_client)
  119. @classmethod
  120. def _init_from_topic(
  121. cls,
  122. userdata: _MQTTCallbackUserdata,
  123. topic: str,
  124. expected_topic_levels: typing.List[_MQTTTopicLevel],
  125. ) -> typing.Optional[_MQTTControlledActor]:
  126. try:
  127. mac_address = _parse_mqtt_topic(
  128. topic=topic, expected_levels=expected_topic_levels
  129. )[_MQTTTopicPlaceholder.MAC_ADDRESS]
  130. except ValueError as exc:
  131. _LOGGER.warning(str(exc), exc_info=False)
  132. return None
  133. if not _mac_address_valid(mac_address):
  134. _LOGGER.warning("invalid mac address %s", mac_address)
  135. return None
  136. return cls(
  137. mac_address=mac_address,
  138. retry_count=userdata.retry_count,
  139. password=userdata.device_passwords.get(mac_address, None),
  140. )
  141. @classmethod
  142. def _mqtt_update_device_info_callback(
  143. cls,
  144. mqtt_client: paho.mqtt.client.Client,
  145. userdata: _MQTTCallbackUserdata,
  146. message: paho.mqtt.client.MQTTMessage,
  147. ) -> None:
  148. # pylint: disable=unused-argument; callback
  149. # https://github.com/eclipse/paho.mqtt.python/blob/v1.5.0/src/paho/mqtt/client.py#L469
  150. _LOGGER.debug("received topic=%s payload=%r", message.topic, message.payload)
  151. if message.retain:
  152. _LOGGER.info("ignoring retained message")
  153. return
  154. actor = cls._init_from_topic(
  155. userdata=userdata,
  156. topic=message.topic,
  157. expected_topic_levels=cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS,
  158. )
  159. if actor:
  160. # pylint: disable=protected-access; own instance
  161. actor._update_and_report_device_info(mqtt_client)
  162. @abc.abstractmethod
  163. def execute_command(
  164. self,
  165. mqtt_message_payload: bytes,
  166. mqtt_client: paho.mqtt.client.Client,
  167. update_device_info: bool,
  168. ) -> None:
  169. raise NotImplementedError()
  170. @classmethod
  171. def _mqtt_command_callback(
  172. cls,
  173. mqtt_client: paho.mqtt.client.Client,
  174. userdata: _MQTTCallbackUserdata,
  175. message: paho.mqtt.client.MQTTMessage,
  176. ) -> None:
  177. # pylint: disable=unused-argument; callback
  178. # https://github.com/eclipse/paho.mqtt.python/blob/v1.5.0/src/paho/mqtt/client.py#L469
  179. _LOGGER.debug("received topic=%s payload=%r", message.topic, message.payload)
  180. if message.retain:
  181. _LOGGER.info("ignoring retained message")
  182. return
  183. actor = cls._init_from_topic(
  184. userdata=userdata,
  185. topic=message.topic,
  186. expected_topic_levels=cls.MQTT_COMMAND_TOPIC_LEVELS,
  187. )
  188. if actor:
  189. actor.execute_command(
  190. mqtt_message_payload=message.payload,
  191. mqtt_client=mqtt_client,
  192. update_device_info=userdata.fetch_device_info,
  193. )
  194. @classmethod
  195. def mqtt_subscribe(
  196. cls,
  197. mqtt_client: paho.mqtt.client.Client,
  198. *,
  199. enable_device_info_update_topic: bool,
  200. ) -> None:
  201. topics = [(cls.MQTT_COMMAND_TOPIC_LEVELS, cls._mqtt_command_callback)]
  202. if enable_device_info_update_topic:
  203. topics.append(
  204. (
  205. cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS,
  206. cls._mqtt_update_device_info_callback,
  207. )
  208. )
  209. for topic_levels, callback in topics:
  210. topic = _join_mqtt_topic_levels(topic_levels, mac_address="+")
  211. _LOGGER.info("subscribing to MQTT topic %r", topic)
  212. mqtt_client.subscribe(topic)
  213. mqtt_client.message_callback_add(sub=topic, callback=callback)
  214. def _mqtt_publish(
  215. self,
  216. *,
  217. topic_levels: typing.List[_MQTTTopicLevel],
  218. payload: bytes,
  219. mqtt_client: paho.mqtt.client.Client,
  220. ) -> None:
  221. topic = _join_mqtt_topic_levels(
  222. topic_levels=topic_levels, mac_address=self._mac_address
  223. )
  224. # https://pypi.org/project/paho-mqtt/#publishing
  225. _LOGGER.debug("publishing topic=%s payload=%r", topic, payload)
  226. message_info: paho.mqtt.client.MQTTMessageInfo = mqtt_client.publish(
  227. topic=topic, payload=payload, retain=True
  228. )
  229. # wait before checking status?
  230. if message_info.rc != paho.mqtt.client.MQTT_ERR_SUCCESS:
  231. _LOGGER.error(
  232. "Failed to publish MQTT message on topic %s (rc=%d)",
  233. topic,
  234. message_info.rc,
  235. )
  236. def report_state(self, state: bytes, mqtt_client: paho.mqtt.client.Client) -> None:
  237. self._mqtt_publish(
  238. topic_levels=self.MQTT_STATE_TOPIC_LEVELS,
  239. payload=state,
  240. mqtt_client=mqtt_client,
  241. )