1
0

_cli.py 6.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170
  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 argparse
  19. import json
  20. import logging
  21. import os
  22. import pathlib
  23. import warnings
  24. import switchbot
  25. import switchbot_mqtt
  26. from switchbot_mqtt._actors import _ButtonAutomator, _CurtainMotor
  27. _MQTT_DEFAULT_PORT = 1883
  28. _MQTT_DEFAULT_TLS_PORT = 8883
  29. _LOGGER = logging.getLogger(__name__)
  30. def _main() -> None:
  31. argparser = argparse.ArgumentParser(
  32. description="MQTT client controlling SwitchBot button automators, "
  33. "compatible with home-assistant.io's MQTT Switch platform"
  34. )
  35. argparser.add_argument("--mqtt-host", type=str, required=True)
  36. argparser.add_argument(
  37. "--mqtt-port",
  38. type=int,
  39. help=f"default {_MQTT_DEFAULT_PORT} ({_MQTT_DEFAULT_TLS_PORT} with --mqtt-enable-tls)",
  40. )
  41. mqtt_tls_argument_group = argparser.add_mutually_exclusive_group()
  42. mqtt_tls_argument_group.add_argument(
  43. "--mqtt-enable-tls",
  44. action="store_true",
  45. help="TLS will be enabled by default in the next major release",
  46. )
  47. mqtt_tls_argument_group.add_argument( # for upward compatibility
  48. "--mqtt-disable-tls", action="store_true", help="Currently enabled by default"
  49. )
  50. argparser.add_argument("--mqtt-username", type=str)
  51. password_argument_group = argparser.add_mutually_exclusive_group()
  52. password_argument_group.add_argument("--mqtt-password", type=str)
  53. password_argument_group.add_argument(
  54. "--mqtt-password-file",
  55. type=pathlib.Path,
  56. metavar="PATH",
  57. dest="mqtt_password_path",
  58. help="Stripping trailing newline",
  59. )
  60. argparser.add_argument(
  61. "--mqtt-topic-prefix",
  62. metavar="PREFIX",
  63. default="homeassistant/", # for historic reasons (change to empty string?)
  64. help="Default: %(default)s",
  65. )
  66. argparser.add_argument(
  67. "--device-password-file",
  68. type=pathlib.Path,
  69. metavar="PATH",
  70. dest="device_password_path",
  71. help="Path to json file mapping mac addresses of switchbot devices to passwords, e.g. "
  72. + json.dumps({"11:22:33:44:55:66": "password", "aa:bb:cc:dd:ee:ff": "secret"}),
  73. )
  74. argparser.add_argument(
  75. "--retries",
  76. dest="retry_count",
  77. type=int,
  78. default=switchbot.DEFAULT_RETRY_COUNT,
  79. help="Maximum number of attempts to send a command to a SwitchBot device"
  80. " (default: %(default)d)",
  81. )
  82. argparser.add_argument(
  83. "--fetch-device-info",
  84. action="store_true",
  85. help="Report devices' battery level on topic "
  86. + _ButtonAutomator.get_mqtt_battery_percentage_topic(
  87. prefix="[PREFIX]", mac_address="MAC_ADDRESS"
  88. )
  89. + " or, respectively, "
  90. + _CurtainMotor.get_mqtt_battery_percentage_topic(
  91. prefix="[PREFIX]", mac_address="MAC_ADDRESS"
  92. )
  93. + " after every command. Additionally report curtain motors' position on topic "
  94. + _CurtainMotor.get_mqtt_position_topic(
  95. prefix="[PREFIX]", mac_address="MAC_ADDRESS"
  96. )
  97. + " after executing stop commands."
  98. " When this option is enabled, the mentioned reports may also be requested"
  99. " by sending a MQTT message to the topic "
  100. + _ButtonAutomator.get_mqtt_update_device_info_topic(
  101. prefix="[PREFIX]", mac_address="MAC_ADDRESS"
  102. )
  103. + " or "
  104. + _CurtainMotor.get_mqtt_update_device_info_topic(
  105. prefix="[PREFIX]", mac_address="MAC_ADDRESS"
  106. )
  107. + ". This option can also be enabled by assigning a non-empty value to the"
  108. " environment variable FETCH_DEVICE_INFO."
  109. " [PREFIX] can be set via --mqtt-topic-prefix.",
  110. )
  111. argparser.add_argument("--debug", action="store_true")
  112. args = argparser.parse_args()
  113. # https://github.com/fphammerle/python-cc1101/blob/26d8122661fc4587ecc7c73df55b92d05cf98fe8/cc1101/_cli.py#L51
  114. logging.basicConfig(
  115. level=logging.DEBUG if args.debug else logging.INFO,
  116. format="%(asctime)s:%(levelname)s:%(name)s:%(funcName)s:%(message)s"
  117. if args.debug
  118. else "%(message)s",
  119. datefmt="%Y-%m-%dT%H:%M:%S%z",
  120. )
  121. _LOGGER.debug("args=%r", args)
  122. if args.mqtt_port:
  123. mqtt_port = args.mqtt_port
  124. elif args.mqtt_enable_tls:
  125. mqtt_port = _MQTT_DEFAULT_TLS_PORT
  126. else:
  127. mqtt_port = _MQTT_DEFAULT_PORT
  128. if not args.mqtt_enable_tls and not args.mqtt_disable_tls:
  129. warnings.warn(
  130. "In switchbot-mqtt's next major release, TLS will be enabled by default"
  131. " (--mqtt-enable-tls)."
  132. " Please add --mqtt-disable-tls to your command for upward compatibility.",
  133. UserWarning, # DeprecationWarning ignored by default
  134. )
  135. if args.mqtt_password_path:
  136. # .read_text() replaces \r\n with \n
  137. mqtt_password = args.mqtt_password_path.read_bytes().decode()
  138. if mqtt_password.endswith("\r\n"):
  139. mqtt_password = mqtt_password[:-2]
  140. elif mqtt_password.endswith("\n"):
  141. mqtt_password = mqtt_password[:-1]
  142. else:
  143. mqtt_password = args.mqtt_password
  144. if ( # pylint: disable=consider-ternary-expression; bulky with black's wraps
  145. args.device_password_path
  146. ):
  147. device_passwords = json.loads(args.device_password_path.read_text())
  148. else:
  149. device_passwords = {}
  150. switchbot_mqtt._run( # pylint: disable=protected-access; internal
  151. mqtt_host=args.mqtt_host,
  152. mqtt_port=mqtt_port,
  153. mqtt_disable_tls=not args.mqtt_enable_tls,
  154. mqtt_username=args.mqtt_username,
  155. mqtt_password=mqtt_password,
  156. mqtt_topic_prefix=args.mqtt_topic_prefix,
  157. retry_count=args.retry_count,
  158. device_passwords=device_passwords,
  159. fetch_device_info=args.fetch_device_info
  160. # > In formal language theory, the empty string, [...], is the unique string of length zero.
  161. # https://en.wikipedia.org/wiki/Empty_string
  162. or bool(os.environ.get("FETCH_DEVICE_INFO")),
  163. )