__init__.py 13 KB

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