1
0

__init__.py 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341
  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 abc
  19. import argparse
  20. import enum
  21. import logging
  22. import pathlib
  23. import re
  24. import typing
  25. import paho.mqtt.client
  26. import switchbot
  27. _LOGGER = logging.getLogger(__name__)
  28. _MAC_ADDRESS_REGEX = re.compile(r"^[0-9a-f]{2}(:[0-9a-f]{2}){5}$")
  29. class _MQTTTopicPlaceholder(enum.Enum):
  30. MAC_ADDRESS = "MAC_ADDRESS"
  31. _MQTTTopicLevel = typing.Union[str, _MQTTTopicPlaceholder]
  32. # "homeassistant" for historic reason, may be parametrized in future
  33. _MQTT_TOPIC_LEVELS_PREFIX = ["homeassistant"] # type: typing.List[_MQTTTopicLevel]
  34. def _mac_address_valid(mac_address: str) -> bool:
  35. return _MAC_ADDRESS_REGEX.match(mac_address.lower()) is not None
  36. class _MQTTControlledActor(abc.ABC):
  37. MQTT_COMMAND_TOPIC_LEVELS = NotImplemented # type: typing.List[_MQTTTopicLevel]
  38. MQTT_STATE_TOPIC_LEVELS = NotImplemented # type: typing.List[_MQTTTopicLevel]
  39. def __init__(self, mac_address: str) -> None:
  40. # alternative: pySwitchbot >=0.10.0 provides SwitchbotDevice.get_mac()
  41. self._mac_address = mac_address
  42. @abc.abstractmethod
  43. def execute_command(
  44. self, mqtt_message_payload: bytes, mqtt_client: paho.mqtt.client.Client
  45. ) -> None:
  46. raise NotImplementedError()
  47. @classmethod
  48. def _mqtt_command_callback(
  49. cls,
  50. mqtt_client: paho.mqtt.client.Client,
  51. userdata: None,
  52. message: paho.mqtt.client.MQTTMessage,
  53. ) -> None:
  54. # pylint: disable=unused-argument; callback
  55. # https://github.com/eclipse/paho.mqtt.python/blob/v1.5.0/src/paho/mqtt/client.py#L469
  56. _LOGGER.debug("received topic=%s payload=%r", message.topic, message.payload)
  57. if message.retain:
  58. _LOGGER.info("ignoring retained message")
  59. return
  60. topic_split = message.topic.split("/")
  61. if len(topic_split) != len(cls.MQTT_COMMAND_TOPIC_LEVELS):
  62. _LOGGER.warning("unexpected topic %s", message.topic)
  63. return
  64. mac_address = None
  65. for given_part, expected_part in zip(
  66. topic_split, cls.MQTT_COMMAND_TOPIC_LEVELS
  67. ):
  68. if expected_part == _MQTTTopicPlaceholder.MAC_ADDRESS:
  69. mac_address = given_part
  70. elif expected_part != given_part:
  71. _LOGGER.warning("unexpected topic %s", message.topic)
  72. return
  73. assert mac_address
  74. if not _mac_address_valid(mac_address):
  75. _LOGGER.warning("invalid mac address %s", mac_address)
  76. return
  77. cls(mac_address=mac_address).execute_command(
  78. mqtt_message_payload=message.payload, mqtt_client=mqtt_client
  79. )
  80. @classmethod
  81. def mqtt_subscribe(cls, mqtt_client: paho.mqtt.client.Client) -> None:
  82. command_topic = "/".join(
  83. "+" if isinstance(l, _MQTTTopicPlaceholder) else l
  84. for l in cls.MQTT_COMMAND_TOPIC_LEVELS
  85. )
  86. _LOGGER.info("subscribing to MQTT topic %r", command_topic)
  87. mqtt_client.subscribe(command_topic)
  88. mqtt_client.message_callback_add(
  89. sub=command_topic, callback=cls._mqtt_command_callback
  90. )
  91. def _mqtt_publish(
  92. self,
  93. topic_levels: typing.List[_MQTTTopicLevel],
  94. payload: bytes,
  95. mqtt_client: paho.mqtt.client.Client,
  96. ) -> None:
  97. topic = "/".join(
  98. self._mac_address
  99. if l == _MQTTTopicPlaceholder.MAC_ADDRESS
  100. else typing.cast(str, l)
  101. for l in topic_levels
  102. )
  103. # https://pypi.org/project/paho-mqtt/#publishing
  104. _LOGGER.debug("publishing topic=%s payload=%r", topic, payload)
  105. message_info = mqtt_client.publish(
  106. topic=topic, payload=payload, retain=True
  107. ) # type: paho.mqtt.client.MQTTMessageInfo
  108. # wait before checking status?
  109. if message_info.rc != paho.mqtt.client.MQTT_ERR_SUCCESS:
  110. _LOGGER.error(
  111. "Failed to publish MQTT message on topic %s (rc=%d)",
  112. topic,
  113. message_info.rc,
  114. )
  115. def report_state(self, state: bytes, mqtt_client: paho.mqtt.client.Client) -> None:
  116. self._mqtt_publish(
  117. topic_levels=self.MQTT_STATE_TOPIC_LEVELS,
  118. payload=state,
  119. mqtt_client=mqtt_client,
  120. )
  121. class _ButtonAutomator(_MQTTControlledActor):
  122. # https://www.home-assistant.io/integrations/switch.mqtt/
  123. MQTT_COMMAND_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
  124. "switch",
  125. "switchbot",
  126. _MQTTTopicPlaceholder.MAC_ADDRESS,
  127. "set",
  128. ]
  129. MQTT_STATE_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
  130. "switch",
  131. "switchbot",
  132. _MQTTTopicPlaceholder.MAC_ADDRESS,
  133. "state",
  134. ]
  135. def __init__(self, mac_address) -> None:
  136. self._device = switchbot.Switchbot(mac=mac_address)
  137. super().__init__(mac_address=mac_address)
  138. def execute_command(
  139. self, mqtt_message_payload: bytes, mqtt_client: paho.mqtt.client.Client
  140. ) -> None:
  141. # https://www.home-assistant.io/integrations/switch.mqtt/#payload_on
  142. if mqtt_message_payload.lower() == b"on":
  143. if not self._device.turn_on():
  144. _LOGGER.error("failed to turn on switchbot %s", self._mac_address)
  145. else:
  146. _LOGGER.info("switchbot %s turned on", self._mac_address)
  147. # https://www.home-assistant.io/integrations/switch.mqtt/#state_on
  148. self.report_state(mqtt_client=mqtt_client, state=b"ON")
  149. # https://www.home-assistant.io/integrations/switch.mqtt/#payload_off
  150. elif mqtt_message_payload.lower() == b"off":
  151. if not self._device.turn_off():
  152. _LOGGER.error("failed to turn off switchbot %s", self._mac_address)
  153. else:
  154. _LOGGER.info("switchbot %s turned off", self._mac_address)
  155. self.report_state(mqtt_client=mqtt_client, state=b"OFF")
  156. else:
  157. _LOGGER.warning(
  158. "unexpected payload %r (expected 'ON' or 'OFF')", mqtt_message_payload
  159. )
  160. class _CurtainMotor(_MQTTControlledActor):
  161. # https://www.home-assistant.io/integrations/cover.mqtt/
  162. MQTT_COMMAND_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
  163. "cover",
  164. "switchbot-curtain",
  165. _MQTTTopicPlaceholder.MAC_ADDRESS,
  166. "set",
  167. ]
  168. MQTT_STATE_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
  169. "cover",
  170. "switchbot-curtain",
  171. _MQTTTopicPlaceholder.MAC_ADDRESS,
  172. "state",
  173. ]
  174. _MQTT_POSITION_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
  175. "cover",
  176. "switchbot-curtain",
  177. _MQTTTopicPlaceholder.MAC_ADDRESS,
  178. "position",
  179. ]
  180. def __init__(self, mac_address) -> None:
  181. # > The position of the curtain is saved in self._pos with 0 = open and 100 = closed.
  182. # https://github.com/Danielhiversen/pySwitchbot/blob/0.10.0/switchbot/__init__.py#L150
  183. self._device = switchbot.SwitchbotCurtain(mac=mac_address, reverse_mode=True)
  184. super().__init__(mac_address=mac_address)
  185. def _report_position(self, mqtt_client: paho.mqtt.client.Client) -> None:
  186. # > position_closed integer (Optional, default: 0)
  187. # > position_open integer (Optional, default: 100)
  188. # https://www.home-assistant.io/integrations/cover.mqtt/#position_closed
  189. # SwitchbotCurtain.get_position() returns a cached value within [0, 100].
  190. # SwitchbotCurtain.open() and .close() update the position optimistically,
  191. # SwitchbotCurtain.update() fetches the real position via bluetooth.
  192. # https://github.com/Danielhiversen/pySwitchbot/blob/0.10.0/switchbot/__init__.py#L202
  193. self._mqtt_publish(
  194. topic_levels=self._MQTT_POSITION_TOPIC_LEVELS,
  195. payload=str(int(self._device.get_position())).encode(),
  196. mqtt_client=mqtt_client,
  197. )
  198. def _update_position(self, mqtt_client: paho.mqtt.client.Client) -> None:
  199. self._device.update()
  200. self._report_position(mqtt_client=mqtt_client)
  201. def execute_command(
  202. self, mqtt_message_payload: bytes, mqtt_client: paho.mqtt.client.Client
  203. ) -> None:
  204. # https://www.home-assistant.io/integrations/cover.mqtt/#payload_open
  205. if mqtt_message_payload.lower() == b"open":
  206. if not self._device.open():
  207. _LOGGER.error("failed to open switchbot curtain %s", self._mac_address)
  208. else:
  209. _LOGGER.info("switchbot curtain %s opening", self._mac_address)
  210. # > state_opening string (Optional, default: opening)
  211. # https://www.home-assistant.io/integrations/cover.mqtt/#state_opening
  212. self.report_state(mqtt_client=mqtt_client, state=b"opening")
  213. elif mqtt_message_payload.lower() == b"close":
  214. if not self._device.close():
  215. _LOGGER.error("failed to close switchbot curtain %s", self._mac_address)
  216. else:
  217. _LOGGER.info("switchbot curtain %s closing", self._mac_address)
  218. # https://www.home-assistant.io/integrations/cover.mqtt/#state_closing
  219. self.report_state(mqtt_client=mqtt_client, state=b"closing")
  220. elif mqtt_message_payload.lower() == b"stop":
  221. if not self._device.stop():
  222. _LOGGER.error("failed to stop switchbot curtain %s", self._mac_address)
  223. else:
  224. _LOGGER.info("switchbot curtain %s stopped", self._mac_address)
  225. # no "stopped" state mentioned at
  226. # https://www.home-assistant.io/integrations/cover.mqtt/#configuration-variables
  227. # https://community.home-assistant.io/t/mqtt-how-to-remove-retained-messages/79029/2
  228. self.report_state(mqtt_client=mqtt_client, state=b"")
  229. self._update_position(mqtt_client=mqtt_client)
  230. else:
  231. _LOGGER.warning(
  232. "unexpected payload %r (expected 'OPEN', 'CLOSE', or 'STOP')",
  233. mqtt_message_payload,
  234. )
  235. def _mqtt_on_connect(
  236. mqtt_client: paho.mqtt.client.Client,
  237. user_data: typing.Any,
  238. flags: typing.Dict,
  239. return_code: int,
  240. ) -> None:
  241. # pylint: disable=unused-argument; callback
  242. # https://github.com/eclipse/paho.mqtt.python/blob/v1.5.0/src/paho/mqtt/client.py#L441
  243. assert return_code == 0, return_code # connection accepted
  244. mqtt_broker_host, mqtt_broker_port = mqtt_client.socket().getpeername()
  245. _LOGGER.debug("connected to MQTT broker %s:%d", mqtt_broker_host, mqtt_broker_port)
  246. _ButtonAutomator.mqtt_subscribe(mqtt_client=mqtt_client)
  247. _CurtainMotor.mqtt_subscribe(mqtt_client=mqtt_client)
  248. def _run(
  249. mqtt_host: str,
  250. mqtt_port: int,
  251. mqtt_username: typing.Optional[str],
  252. mqtt_password: typing.Optional[str],
  253. ) -> None:
  254. # https://pypi.org/project/paho-mqtt/
  255. mqtt_client = paho.mqtt.client.Client()
  256. mqtt_client.on_connect = _mqtt_on_connect
  257. _LOGGER.info("connecting to MQTT broker %s:%d", mqtt_host, mqtt_port)
  258. if mqtt_username:
  259. mqtt_client.username_pw_set(username=mqtt_username, password=mqtt_password)
  260. elif mqtt_password:
  261. raise ValueError("Missing MQTT username")
  262. mqtt_client.connect(host=mqtt_host, port=mqtt_port)
  263. # https://github.com/eclipse/paho.mqtt.python/blob/master/src/paho/mqtt/client.py#L1740
  264. mqtt_client.loop_forever()
  265. def _main() -> None:
  266. logging.basicConfig(
  267. level=logging.DEBUG,
  268. format="%(asctime)s:%(levelname)s:%(name)s:%(message)s",
  269. datefmt="%Y-%m-%dT%H:%M:%S%z",
  270. )
  271. argparser = argparse.ArgumentParser(
  272. description="MQTT client controlling SwitchBot button automators, "
  273. "compatible with home-assistant.io's MQTT Switch platform"
  274. )
  275. argparser.add_argument("--mqtt-host", type=str, required=True)
  276. argparser.add_argument("--mqtt-port", type=int, default=1883)
  277. argparser.add_argument("--mqtt-username", type=str)
  278. password_argument_group = argparser.add_mutually_exclusive_group()
  279. password_argument_group.add_argument("--mqtt-password", type=str)
  280. password_argument_group.add_argument(
  281. "--mqtt-password-file",
  282. type=pathlib.Path,
  283. metavar="PATH",
  284. dest="mqtt_password_path",
  285. help="stripping trailing newline",
  286. )
  287. args = argparser.parse_args()
  288. if args.mqtt_password_path:
  289. # .read_text() replaces \r\n with \n
  290. mqtt_password = args.mqtt_password_path.read_bytes().decode()
  291. if mqtt_password.endswith("\r\n"):
  292. mqtt_password = mqtt_password[:-2]
  293. elif mqtt_password.endswith("\n"):
  294. mqtt_password = mqtt_password[:-1]
  295. else:
  296. mqtt_password = args.mqtt_password
  297. _run(
  298. mqtt_host=args.mqtt_host,
  299. mqtt_port=args.mqtt_port,
  300. mqtt_username=args.mqtt_username,
  301. mqtt_password=mqtt_password,
  302. )