__init__.py 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570
  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 collections
  21. import enum
  22. import json
  23. import logging
  24. import pathlib
  25. import queue
  26. import re
  27. import shlex
  28. import typing
  29. import bluepy.btle
  30. import paho.mqtt.client
  31. import switchbot
  32. _LOGGER = logging.getLogger(__name__)
  33. _MAC_ADDRESS_REGEX = re.compile(r"^[0-9a-f]{2}(:[0-9a-f]{2}){5}$")
  34. class _MQTTTopicPlaceholder(enum.Enum):
  35. MAC_ADDRESS = "MAC_ADDRESS"
  36. _MQTTTopicLevel = typing.Union[str, _MQTTTopicPlaceholder]
  37. # "homeassistant" for historic reason, may be parametrized in future
  38. _MQTT_TOPIC_LEVELS_PREFIX: typing.List[_MQTTTopicLevel] = ["homeassistant"]
  39. def _join_mqtt_topic_levels(
  40. topic_levels: typing.List[_MQTTTopicLevel], mac_address: str
  41. ) -> str:
  42. return "/".join(
  43. mac_address if l == _MQTTTopicPlaceholder.MAC_ADDRESS else typing.cast(str, l)
  44. for l in topic_levels
  45. )
  46. def _mac_address_valid(mac_address: str) -> bool:
  47. return _MAC_ADDRESS_REGEX.match(mac_address.lower()) is not None
  48. class _QueueLogHandler(logging.Handler):
  49. """
  50. logging.handlers.QueueHandler drops exc_info
  51. """
  52. # TypeError: 'type' object is not subscriptable
  53. def __init__(self, log_queue: "queue.Queue[logging.LogRecord]") -> None:
  54. self.log_queue = log_queue
  55. super().__init__()
  56. def emit(self, record: logging.LogRecord) -> None:
  57. self.log_queue.put(record)
  58. class _MQTTCallbackUserdata:
  59. # pylint: disable=too-few-public-methods; @dataclasses.dataclass when python_requires>=3.7
  60. def __init__(
  61. self,
  62. *,
  63. retry_count: int,
  64. device_passwords: typing.Dict[str, str],
  65. fetch_device_info: bool,
  66. ) -> None:
  67. self.retry_count = retry_count
  68. self.device_passwords = device_passwords
  69. self.fetch_device_info = fetch_device_info
  70. def __eq__(self, other: object) -> bool:
  71. return isinstance(other, type(self)) and vars(self) == vars(other)
  72. class _MQTTControlledActor(abc.ABC):
  73. MQTT_COMMAND_TOPIC_LEVELS: typing.List[_MQTTTopicLevel] = NotImplemented
  74. MQTT_STATE_TOPIC_LEVELS: typing.List[_MQTTTopicLevel] = NotImplemented
  75. _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS: typing.List[_MQTTTopicLevel] = NotImplemented
  76. @classmethod
  77. def get_mqtt_battery_percentage_topic(cls, mac_address: str) -> str:
  78. return _join_mqtt_topic_levels(
  79. topic_levels=cls._MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS,
  80. mac_address=mac_address,
  81. )
  82. @abc.abstractmethod
  83. def __init__(
  84. self, *, mac_address: str, retry_count: int, password: typing.Optional[str]
  85. ) -> None:
  86. # alternative: pySwitchbot >=0.10.0 provides SwitchbotDevice.get_mac()
  87. self._mac_address = mac_address
  88. @abc.abstractmethod
  89. def _get_device(self) -> switchbot.SwitchbotDevice:
  90. raise NotImplementedError()
  91. def _update_device_info(self) -> None:
  92. log_queue: queue.Queue[logging.LogRecord] = queue.Queue(maxsize=0)
  93. logging.getLogger("switchbot").addHandler(_QueueLogHandler(log_queue))
  94. try:
  95. self._get_device().update()
  96. # pySwitchbot>=v0.10.1 catches bluepy.btle.BTLEManagementError :(
  97. # https://github.com/Danielhiversen/pySwitchbot/blob/0.10.1/switchbot/__init__.py#L141
  98. while not log_queue.empty():
  99. log_record = log_queue.get()
  100. if log_record.exc_info:
  101. exc: typing.Optional[BaseException] = log_record.exc_info[1]
  102. if (
  103. isinstance(exc, bluepy.btle.BTLEManagementError)
  104. and exc.emsg == "Permission Denied"
  105. ):
  106. raise exc
  107. except bluepy.btle.BTLEManagementError as exc:
  108. if (
  109. exc.emsg == "Permission Denied"
  110. and exc.message == "Failed to execute management command 'le on'"
  111. ):
  112. raise PermissionError(
  113. "bluepy-helper failed to enable low energy mode"
  114. " due to insufficient permissions."
  115. "\nSee https://github.com/IanHarvey/bluepy/issues/313#issuecomment-428324639"
  116. ", https://github.com/fphammerle/switchbot-mqtt/pull/31#issuecomment-846383603"
  117. ", and https://github.com/IanHarvey/bluepy/blob/v/1.3.0/bluepy"
  118. "/bluepy-helper.c#L1260."
  119. "\nInsecure workaround:"
  120. "\n1. sudo apt-get install --no-install-recommends libcap2-bin"
  121. f"\n2. sudo setcap cap_net_admin+ep {shlex.quote(bluepy.btle.helperExe)}"
  122. "\n3. restart switchbot-mqtt"
  123. "\nIn docker-based setups, you could use"
  124. " `sudo docker run --cap-drop ALL --cap-add NET_ADMIN --user 0 …`"
  125. " (seriously insecure)."
  126. ) from exc
  127. raise
  128. def _report_battery_level(self, mqtt_client: paho.mqtt.client.Client) -> None:
  129. # > battery: Percentage of battery that is left.
  130. # https://www.home-assistant.io/integrations/sensor/#device-class
  131. self._mqtt_publish(
  132. topic_levels=self._MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS,
  133. payload=str(self._get_device().get_battery_percent()).encode(),
  134. mqtt_client=mqtt_client,
  135. )
  136. def _update_and_report_device_info(
  137. self, mqtt_client: paho.mqtt.client.Client
  138. ) -> None:
  139. self._update_device_info()
  140. self._report_battery_level(mqtt_client=mqtt_client)
  141. @abc.abstractmethod
  142. def execute_command(
  143. self,
  144. mqtt_message_payload: bytes,
  145. mqtt_client: paho.mqtt.client.Client,
  146. update_device_info: bool,
  147. ) -> None:
  148. raise NotImplementedError()
  149. @classmethod
  150. def _mqtt_command_callback(
  151. cls,
  152. mqtt_client: paho.mqtt.client.Client,
  153. userdata: _MQTTCallbackUserdata,
  154. message: paho.mqtt.client.MQTTMessage,
  155. ) -> None:
  156. # pylint: disable=unused-argument; callback
  157. # https://github.com/eclipse/paho.mqtt.python/blob/v1.5.0/src/paho/mqtt/client.py#L469
  158. _LOGGER.debug("received topic=%s payload=%r", message.topic, message.payload)
  159. if message.retain:
  160. _LOGGER.info("ignoring retained message")
  161. return
  162. topic_split = message.topic.split("/")
  163. if len(topic_split) != len(cls.MQTT_COMMAND_TOPIC_LEVELS):
  164. _LOGGER.warning("unexpected topic %s", message.topic)
  165. return
  166. mac_address = None
  167. for given_part, expected_part in zip(
  168. topic_split, cls.MQTT_COMMAND_TOPIC_LEVELS
  169. ):
  170. if expected_part == _MQTTTopicPlaceholder.MAC_ADDRESS:
  171. mac_address = given_part
  172. elif expected_part != given_part:
  173. _LOGGER.warning("unexpected topic %s", message.topic)
  174. return
  175. assert mac_address
  176. if not _mac_address_valid(mac_address):
  177. _LOGGER.warning("invalid mac address %s", mac_address)
  178. return
  179. actor = cls(
  180. mac_address=mac_address,
  181. retry_count=userdata.retry_count,
  182. password=userdata.device_passwords.get(mac_address, None),
  183. )
  184. actor.execute_command(
  185. mqtt_message_payload=message.payload,
  186. mqtt_client=mqtt_client,
  187. # consider calling update+report method directly when adding support for battery levels
  188. update_device_info=userdata.fetch_device_info,
  189. )
  190. @classmethod
  191. def mqtt_subscribe(cls, mqtt_client: paho.mqtt.client.Client) -> None:
  192. command_topic = "/".join(
  193. "+" if isinstance(l, _MQTTTopicPlaceholder) else l
  194. for l in cls.MQTT_COMMAND_TOPIC_LEVELS
  195. )
  196. _LOGGER.info("subscribing to MQTT topic %r", command_topic)
  197. mqtt_client.subscribe(command_topic)
  198. mqtt_client.message_callback_add(
  199. sub=command_topic,
  200. callback=cls._mqtt_command_callback,
  201. )
  202. def _mqtt_publish(
  203. self,
  204. *,
  205. topic_levels: typing.List[_MQTTTopicLevel],
  206. payload: bytes,
  207. mqtt_client: paho.mqtt.client.Client,
  208. ) -> None:
  209. topic = _join_mqtt_topic_levels(
  210. topic_levels=topic_levels, mac_address=self._mac_address
  211. )
  212. # https://pypi.org/project/paho-mqtt/#publishing
  213. _LOGGER.debug("publishing topic=%s payload=%r", topic, payload)
  214. message_info: paho.mqtt.client.MQTTMessageInfo = mqtt_client.publish(
  215. topic=topic, payload=payload, retain=True
  216. )
  217. # wait before checking status?
  218. if message_info.rc != paho.mqtt.client.MQTT_ERR_SUCCESS:
  219. _LOGGER.error(
  220. "Failed to publish MQTT message on topic %s (rc=%d)",
  221. topic,
  222. message_info.rc,
  223. )
  224. def report_state(self, state: bytes, mqtt_client: paho.mqtt.client.Client) -> None:
  225. self._mqtt_publish(
  226. topic_levels=self.MQTT_STATE_TOPIC_LEVELS,
  227. payload=state,
  228. mqtt_client=mqtt_client,
  229. )
  230. class _ButtonAutomator(_MQTTControlledActor):
  231. # https://www.home-assistant.io/integrations/switch.mqtt/
  232. MQTT_COMMAND_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
  233. "switch",
  234. "switchbot",
  235. _MQTTTopicPlaceholder.MAC_ADDRESS,
  236. "set",
  237. ]
  238. MQTT_STATE_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
  239. "switch",
  240. "switchbot",
  241. _MQTTTopicPlaceholder.MAC_ADDRESS,
  242. "state",
  243. ]
  244. _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
  245. "cover",
  246. "switchbot",
  247. _MQTTTopicPlaceholder.MAC_ADDRESS,
  248. "battery-percentage",
  249. ]
  250. def __init__(
  251. self, *, mac_address: str, retry_count: int, password: typing.Optional[str]
  252. ) -> None:
  253. self.__device = switchbot.Switchbot(
  254. mac=mac_address, password=password, retry_count=retry_count
  255. )
  256. super().__init__(
  257. mac_address=mac_address, retry_count=retry_count, password=password
  258. )
  259. def _get_device(self) -> switchbot.SwitchbotDevice:
  260. return self.__device
  261. def execute_command(
  262. self,
  263. mqtt_message_payload: bytes,
  264. mqtt_client: paho.mqtt.client.Client,
  265. update_device_info: bool,
  266. ) -> None:
  267. # https://www.home-assistant.io/integrations/switch.mqtt/#payload_on
  268. if mqtt_message_payload.lower() == b"on":
  269. if not self.__device.turn_on():
  270. _LOGGER.error("failed to turn on switchbot %s", self._mac_address)
  271. else:
  272. _LOGGER.info("switchbot %s turned on", self._mac_address)
  273. # https://www.home-assistant.io/integrations/switch.mqtt/#state_on
  274. self.report_state(mqtt_client=mqtt_client, state=b"ON")
  275. if update_device_info:
  276. self._update_and_report_device_info(mqtt_client)
  277. # https://www.home-assistant.io/integrations/switch.mqtt/#payload_off
  278. elif mqtt_message_payload.lower() == b"off":
  279. if not self.__device.turn_off():
  280. _LOGGER.error("failed to turn off switchbot %s", self._mac_address)
  281. else:
  282. _LOGGER.info("switchbot %s turned off", self._mac_address)
  283. self.report_state(mqtt_client=mqtt_client, state=b"OFF")
  284. if update_device_info:
  285. self._update_and_report_device_info(mqtt_client)
  286. else:
  287. _LOGGER.warning(
  288. "unexpected payload %r (expected 'ON' or 'OFF')", mqtt_message_payload
  289. )
  290. class _CurtainMotor(_MQTTControlledActor):
  291. # https://www.home-assistant.io/integrations/cover.mqtt/
  292. MQTT_COMMAND_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
  293. "cover",
  294. "switchbot-curtain",
  295. _MQTTTopicPlaceholder.MAC_ADDRESS,
  296. "set",
  297. ]
  298. MQTT_STATE_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
  299. "cover",
  300. "switchbot-curtain",
  301. _MQTTTopicPlaceholder.MAC_ADDRESS,
  302. "state",
  303. ]
  304. _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
  305. "cover",
  306. "switchbot-curtain",
  307. _MQTTTopicPlaceholder.MAC_ADDRESS,
  308. "battery-percentage",
  309. ]
  310. _MQTT_POSITION_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
  311. "cover",
  312. "switchbot-curtain",
  313. _MQTTTopicPlaceholder.MAC_ADDRESS,
  314. "position",
  315. ]
  316. @classmethod
  317. def get_mqtt_position_topic(cls, mac_address: str) -> str:
  318. return _join_mqtt_topic_levels(
  319. topic_levels=cls._MQTT_POSITION_TOPIC_LEVELS, mac_address=mac_address
  320. )
  321. def __init__(
  322. self, *, mac_address: str, retry_count: int, password: typing.Optional[str]
  323. ) -> None:
  324. # > The position of the curtain is saved in self._pos with 0 = open and 100 = closed.
  325. # https://github.com/Danielhiversen/pySwitchbot/blob/0.10.0/switchbot/__init__.py#L150
  326. self.__device = switchbot.SwitchbotCurtain(
  327. mac=mac_address,
  328. password=password,
  329. retry_count=retry_count,
  330. reverse_mode=True,
  331. )
  332. super().__init__(
  333. mac_address=mac_address, retry_count=retry_count, password=password
  334. )
  335. def _get_device(self) -> switchbot.SwitchbotDevice:
  336. return self.__device
  337. def _report_position(self, mqtt_client: paho.mqtt.client.Client) -> None:
  338. # > position_closed integer (Optional, default: 0)
  339. # > position_open integer (Optional, default: 100)
  340. # https://www.home-assistant.io/integrations/cover.mqtt/#position_closed
  341. # SwitchbotCurtain.get_position() returns a cached value within [0, 100].
  342. # SwitchbotCurtain.open() and .close() update the position optimistically,
  343. # SwitchbotCurtain.update() fetches the real position via bluetooth.
  344. # https://github.com/Danielhiversen/pySwitchbot/blob/0.10.0/switchbot/__init__.py#L202
  345. self._mqtt_publish(
  346. topic_levels=self._MQTT_POSITION_TOPIC_LEVELS,
  347. payload=str(int(self.__device.get_position())).encode(),
  348. mqtt_client=mqtt_client,
  349. )
  350. def _update_and_report_device_info( # pylint: disable=arguments-differ; report_position is optional
  351. self, mqtt_client: paho.mqtt.client.Client, *, report_position: bool = True
  352. ) -> None:
  353. super()._update_and_report_device_info(mqtt_client)
  354. if report_position:
  355. self._report_position(mqtt_client=mqtt_client)
  356. def execute_command(
  357. self,
  358. mqtt_message_payload: bytes,
  359. mqtt_client: paho.mqtt.client.Client,
  360. update_device_info: bool,
  361. ) -> None:
  362. # https://www.home-assistant.io/integrations/cover.mqtt/#payload_open
  363. report_device_info, report_position = False, False
  364. if mqtt_message_payload.lower() == b"open":
  365. if not self.__device.open():
  366. _LOGGER.error("failed to open switchbot curtain %s", self._mac_address)
  367. else:
  368. _LOGGER.info("switchbot curtain %s opening", self._mac_address)
  369. # > state_opening string (Optional, default: opening)
  370. # https://www.home-assistant.io/integrations/cover.mqtt/#state_opening
  371. self.report_state(mqtt_client=mqtt_client, state=b"opening")
  372. report_device_info = update_device_info
  373. elif mqtt_message_payload.lower() == b"close":
  374. if not self.__device.close():
  375. _LOGGER.error("failed to close switchbot curtain %s", self._mac_address)
  376. else:
  377. _LOGGER.info("switchbot curtain %s closing", self._mac_address)
  378. # https://www.home-assistant.io/integrations/cover.mqtt/#state_closing
  379. self.report_state(mqtt_client=mqtt_client, state=b"closing")
  380. report_device_info = update_device_info
  381. elif mqtt_message_payload.lower() == b"stop":
  382. if not self.__device.stop():
  383. _LOGGER.error("failed to stop switchbot curtain %s", self._mac_address)
  384. else:
  385. _LOGGER.info("switchbot curtain %s stopped", self._mac_address)
  386. # no "stopped" state mentioned at
  387. # https://www.home-assistant.io/integrations/cover.mqtt/#configuration-variables
  388. # https://community.home-assistant.io/t/mqtt-how-to-remove-retained-messages/79029/2
  389. self.report_state(mqtt_client=mqtt_client, state=b"")
  390. report_device_info = update_device_info
  391. report_position = True
  392. else:
  393. _LOGGER.warning(
  394. "unexpected payload %r (expected 'OPEN', 'CLOSE', or 'STOP')",
  395. mqtt_message_payload,
  396. )
  397. if report_device_info:
  398. self._update_and_report_device_info(
  399. mqtt_client=mqtt_client, report_position=report_position
  400. )
  401. def _mqtt_on_connect(
  402. mqtt_client: paho.mqtt.client.Client,
  403. userdata: _MQTTCallbackUserdata,
  404. flags: typing.Dict,
  405. return_code: int,
  406. ) -> None:
  407. # pylint: disable=unused-argument; callback
  408. # https://github.com/eclipse/paho.mqtt.python/blob/v1.5.0/src/paho/mqtt/client.py#L441
  409. assert return_code == 0, return_code # connection accepted
  410. mqtt_broker_host, mqtt_broker_port = mqtt_client.socket().getpeername()
  411. _LOGGER.debug("connected to MQTT broker %s:%d", mqtt_broker_host, mqtt_broker_port)
  412. _ButtonAutomator.mqtt_subscribe(mqtt_client=mqtt_client)
  413. _CurtainMotor.mqtt_subscribe(mqtt_client=mqtt_client)
  414. def _run(
  415. *,
  416. mqtt_host: str,
  417. mqtt_port: int,
  418. mqtt_username: typing.Optional[str],
  419. mqtt_password: typing.Optional[str],
  420. retry_count: int,
  421. device_passwords: typing.Dict[str, str],
  422. fetch_device_info: bool,
  423. ) -> None:
  424. # https://pypi.org/project/paho-mqtt/
  425. mqtt_client = paho.mqtt.client.Client(
  426. userdata=_MQTTCallbackUserdata(
  427. retry_count=retry_count,
  428. device_passwords=device_passwords,
  429. fetch_device_info=fetch_device_info,
  430. )
  431. )
  432. mqtt_client.on_connect = _mqtt_on_connect
  433. _LOGGER.info("connecting to MQTT broker %s:%d", mqtt_host, mqtt_port)
  434. if mqtt_username:
  435. mqtt_client.username_pw_set(username=mqtt_username, password=mqtt_password)
  436. elif mqtt_password:
  437. raise ValueError("Missing MQTT username")
  438. mqtt_client.connect(host=mqtt_host, port=mqtt_port)
  439. # https://github.com/eclipse/paho.mqtt.python/blob/master/src/paho/mqtt/client.py#L1740
  440. mqtt_client.loop_forever()
  441. def _main() -> None:
  442. argparser = argparse.ArgumentParser(
  443. description="MQTT client controlling SwitchBot button automators, "
  444. "compatible with home-assistant.io's MQTT Switch platform"
  445. )
  446. argparser.add_argument("--mqtt-host", type=str, required=True)
  447. argparser.add_argument("--mqtt-port", type=int, default=1883)
  448. argparser.add_argument("--mqtt-username", type=str)
  449. password_argument_group = argparser.add_mutually_exclusive_group()
  450. password_argument_group.add_argument("--mqtt-password", type=str)
  451. password_argument_group.add_argument(
  452. "--mqtt-password-file",
  453. type=pathlib.Path,
  454. metavar="PATH",
  455. dest="mqtt_password_path",
  456. help="stripping trailing newline",
  457. )
  458. argparser.add_argument(
  459. "--device-password-file",
  460. type=pathlib.Path,
  461. metavar="PATH",
  462. dest="device_password_path",
  463. help="path to json file mapping mac addresses of switchbot devices to passwords, e.g. "
  464. + json.dumps({"11:22:33:44:55:66": "password", "aa:bb:cc:dd:ee:ff": "secret"}),
  465. )
  466. argparser.add_argument(
  467. "--retries",
  468. dest="retry_count",
  469. type=int,
  470. default=switchbot.DEFAULT_RETRY_COUNT,
  471. help="Maximum number of attempts to send a command to a SwitchBot device"
  472. " (default: %(default)d)",
  473. )
  474. argparser.add_argument(
  475. "--fetch-device-info",
  476. action="store_true",
  477. help="Report devices' battery level on topic"
  478. f" {_ButtonAutomator.get_mqtt_battery_percentage_topic(mac_address='MAC_ADDRESS')}"
  479. " or, respectively,"
  480. f" {_CurtainMotor.get_mqtt_battery_percentage_topic(mac_address='MAC_ADDRESS')}"
  481. " after every command. Additionally report curtain motors' position on"
  482. f" topic {_CurtainMotor.get_mqtt_position_topic(mac_address='MAC_ADDRESS')}"
  483. " after executing stop commands.",
  484. )
  485. argparser.add_argument("--debug", action="store_true")
  486. args = argparser.parse_args()
  487. # https://github.com/fphammerle/python-cc1101/blob/26d8122661fc4587ecc7c73df55b92d05cf98fe8/cc1101/_cli.py#L51
  488. logging.basicConfig(
  489. level=logging.DEBUG if args.debug else logging.INFO,
  490. format="%(asctime)s:%(levelname)s:%(name)s:%(funcName)s:%(message)s"
  491. if args.debug
  492. else "%(message)s",
  493. datefmt="%Y-%m-%dT%H:%M:%S%z",
  494. )
  495. _LOGGER.debug("args=%r", args)
  496. if args.mqtt_password_path:
  497. # .read_text() replaces \r\n with \n
  498. mqtt_password = args.mqtt_password_path.read_bytes().decode()
  499. if mqtt_password.endswith("\r\n"):
  500. mqtt_password = mqtt_password[:-2]
  501. elif mqtt_password.endswith("\n"):
  502. mqtt_password = mqtt_password[:-1]
  503. else:
  504. mqtt_password = args.mqtt_password
  505. if args.device_password_path:
  506. device_passwords = json.loads(args.device_password_path.read_text())
  507. else:
  508. device_passwords = {}
  509. _run(
  510. mqtt_host=args.mqtt_host,
  511. mqtt_port=args.mqtt_port,
  512. mqtt_username=args.mqtt_username,
  513. mqtt_password=mqtt_password,
  514. retry_count=args.retry_count,
  515. device_passwords=device_passwords,
  516. fetch_device_info=args.fetch_device_info,
  517. )