1
0

_cli.py 6.1 KB

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