__init__.py 2.8 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677
  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 logging
  19. import typing
  20. import paho.mqtt.client
  21. from switchbot_mqtt._actors import _ButtonAutomator, _CurtainMotor
  22. from switchbot_mqtt._actors._base import _MQTTCallbackUserdata
  23. _LOGGER = logging.getLogger(__name__)
  24. def _mqtt_on_connect(
  25. mqtt_client: paho.mqtt.client.Client,
  26. userdata: _MQTTCallbackUserdata,
  27. flags: typing.Dict[str, int],
  28. return_code: int,
  29. ) -> None:
  30. # pylint: disable=unused-argument; callback
  31. # https://github.com/eclipse/paho.mqtt.python/blob/v1.5.0/src/paho/mqtt/client.py#L441
  32. assert return_code == 0, return_code # connection accepted
  33. mqtt_broker_host, mqtt_broker_port = mqtt_client.socket().getpeername()
  34. _LOGGER.debug("connected to MQTT broker %s:%d", mqtt_broker_host, mqtt_broker_port)
  35. _ButtonAutomator.mqtt_subscribe(
  36. mqtt_client=mqtt_client,
  37. enable_device_info_update_topic=userdata.fetch_device_info,
  38. )
  39. _CurtainMotor.mqtt_subscribe(
  40. mqtt_client=mqtt_client,
  41. enable_device_info_update_topic=userdata.fetch_device_info,
  42. )
  43. def _run(
  44. *,
  45. mqtt_host: str,
  46. mqtt_port: int,
  47. mqtt_username: typing.Optional[str],
  48. mqtt_password: typing.Optional[str],
  49. retry_count: int,
  50. device_passwords: typing.Dict[str, str],
  51. fetch_device_info: bool,
  52. ) -> None:
  53. # https://pypi.org/project/paho-mqtt/
  54. mqtt_client = paho.mqtt.client.Client(
  55. userdata=_MQTTCallbackUserdata(
  56. retry_count=retry_count,
  57. device_passwords=device_passwords,
  58. fetch_device_info=fetch_device_info,
  59. )
  60. )
  61. mqtt_client.on_connect = _mqtt_on_connect
  62. _LOGGER.info("connecting to MQTT broker %s:%d", mqtt_host, mqtt_port)
  63. if mqtt_username:
  64. mqtt_client.username_pw_set(username=mqtt_username, password=mqtt_password)
  65. elif mqtt_password:
  66. raise ValueError("Missing MQTT username")
  67. mqtt_client.connect(host=mqtt_host, port=mqtt_port)
  68. # https://github.com/eclipse/paho.mqtt.python/blob/master/src/paho/mqtt/client.py#L1740
  69. mqtt_client.loop_forever()