1
0

__init__.py 11 KB

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