_cli.py 5.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120
  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 switchbot
  24. import switchbot_mqtt
  25. from switchbot_mqtt._actors import _ButtonAutomator, _CurtainMotor
  26. _LOGGER = logging.getLogger(__name__)
  27. def _main() -> None:
  28. argparser = argparse.ArgumentParser(
  29. description="MQTT client controlling SwitchBot button automators, "
  30. "compatible with home-assistant.io's MQTT Switch platform"
  31. )
  32. argparser.add_argument("--mqtt-host", type=str, required=True)
  33. argparser.add_argument("--mqtt-port", type=int, default=1883)
  34. argparser.add_argument("--mqtt-username", type=str)
  35. password_argument_group = argparser.add_mutually_exclusive_group()
  36. password_argument_group.add_argument("--mqtt-password", type=str)
  37. password_argument_group.add_argument(
  38. "--mqtt-password-file",
  39. type=pathlib.Path,
  40. metavar="PATH",
  41. dest="mqtt_password_path",
  42. help="stripping trailing newline",
  43. )
  44. argparser.add_argument(
  45. "--device-password-file",
  46. type=pathlib.Path,
  47. metavar="PATH",
  48. dest="device_password_path",
  49. help="path to json file mapping mac addresses of switchbot devices to passwords, e.g. "
  50. + json.dumps({"11:22:33:44:55:66": "password", "aa:bb:cc:dd:ee:ff": "secret"}),
  51. )
  52. argparser.add_argument(
  53. "--retries",
  54. dest="retry_count",
  55. type=int,
  56. default=switchbot.DEFAULT_RETRY_COUNT,
  57. help="Maximum number of attempts to send a command to a SwitchBot device"
  58. " (default: %(default)d)",
  59. )
  60. argparser.add_argument(
  61. "--fetch-device-info",
  62. action="store_true",
  63. help="Report devices' battery level on topic"
  64. # pylint: disable=protected-access; internal
  65. f" {_ButtonAutomator.get_mqtt_battery_percentage_topic(mac_address='MAC_ADDRESS')}"
  66. " or, respectively,"
  67. f" {_CurtainMotor.get_mqtt_battery_percentage_topic(mac_address='MAC_ADDRESS')}"
  68. " after every command. Additionally report curtain motors' position on"
  69. f" topic {_CurtainMotor.get_mqtt_position_topic(mac_address='MAC_ADDRESS')}"
  70. " after executing stop commands."
  71. " When this option is enabled, the mentioned reports may also be requested"
  72. " by sending a MQTT message to the topic"
  73. f" {_ButtonAutomator.get_mqtt_update_device_info_topic(mac_address='MAC_ADDRESS')}"
  74. f" or {_CurtainMotor.get_mqtt_update_device_info_topic(mac_address='MAC_ADDRESS')}."
  75. " This option can also be enabled by assigning a non-empty value to the"
  76. " environment variable FETCH_DEVICE_INFO.",
  77. )
  78. argparser.add_argument("--debug", action="store_true")
  79. args = argparser.parse_args()
  80. # https://github.com/fphammerle/python-cc1101/blob/26d8122661fc4587ecc7c73df55b92d05cf98fe8/cc1101/_cli.py#L51
  81. logging.basicConfig(
  82. level=logging.DEBUG if args.debug else logging.INFO,
  83. format="%(asctime)s:%(levelname)s:%(name)s:%(funcName)s:%(message)s"
  84. if args.debug
  85. else "%(message)s",
  86. datefmt="%Y-%m-%dT%H:%M:%S%z",
  87. )
  88. _LOGGER.debug("args=%r", args)
  89. if args.mqtt_password_path:
  90. # .read_text() replaces \r\n with \n
  91. mqtt_password = args.mqtt_password_path.read_bytes().decode()
  92. if mqtt_password.endswith("\r\n"):
  93. mqtt_password = mqtt_password[:-2]
  94. elif mqtt_password.endswith("\n"):
  95. mqtt_password = mqtt_password[:-1]
  96. else:
  97. mqtt_password = args.mqtt_password
  98. if args.device_password_path:
  99. device_passwords = json.loads(args.device_password_path.read_text())
  100. else:
  101. device_passwords = {}
  102. switchbot_mqtt._run( # pylint: disable=protected-access; internal
  103. mqtt_host=args.mqtt_host,
  104. mqtt_port=args.mqtt_port,
  105. mqtt_username=args.mqtt_username,
  106. mqtt_password=mqtt_password,
  107. mqtt_disable_tls=True,
  108. retry_count=args.retry_count,
  109. device_passwords=device_passwords,
  110. fetch_device_info=args.fetch_device_info
  111. # > In formal language theory, the empty string, [...], is the unique string of length zero.
  112. # https://en.wikipedia.org/wiki/Empty_string
  113. or bool(os.environ.get("FETCH_DEVICE_INFO")),
  114. )