# switchbot-mqtt - MQTT client controlling SwitchBot button automators, # compatible with home-assistant.io's MQTT Switch platform # # Copyright (C) 2020 Fabian Peter Hammerle # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . import argparse import enum import logging import re import typing import paho.mqtt.client import switchbot _LOGGER = logging.getLogger(__name__) class _SwitchbotAction(enum.Enum): ON = 1 OFF = 2 class _SwitchbotState(enum.Enum): ON = 1 OFF = 2 # https://www.home-assistant.io/docs/mqtt/discovery/#switches _MQTT_TOPIC_PREFIX_LEVELS = ["homeassistant", "switch", "switchbot"] _MQTT_TOPIC_MAC_ADDRESS_PLACEHOLDER = "{mac_address}" _MQTT_SET_TOPIC_LEVELS = _MQTT_TOPIC_PREFIX_LEVELS + [ _MQTT_TOPIC_MAC_ADDRESS_PLACEHOLDER, "set", ] _MQTT_SET_TOPIC = "/".join(_MQTT_SET_TOPIC_LEVELS).replace( _MQTT_TOPIC_MAC_ADDRESS_PLACEHOLDER, "+" ) _MQTT_STATE_TOPIC = "/".join( _MQTT_TOPIC_PREFIX_LEVELS + [_MQTT_TOPIC_MAC_ADDRESS_PLACEHOLDER, "state"] ) # https://www.home-assistant.io/integrations/switch.mqtt/#state_off _MQTT_STATE_PAYLOAD_MAPPING = {_SwitchbotState.ON: b"ON", _SwitchbotState.OFF: b"OFF"} _MAC_ADDRESS_REGEX = re.compile(r"^[0-9a-f]{2}(:[0-9a-f]{2}){5}$") def _mac_address_valid(mac_address: str) -> bool: return _MAC_ADDRESS_REGEX.match(mac_address.lower()) is not None def _mqtt_on_connect( mqtt_client: paho.mqtt.client.Client, user_data: typing.Any, flags: typing.Dict, return_code: int, ) -> None: # pylint: disable=unused-argument; callback # https://github.com/eclipse/paho.mqtt.python/blob/v1.5.0/src/paho/mqtt/client.py#L441 assert return_code == 0, return_code # connection accepted mqtt_broker_host, mqtt_broker_port = mqtt_client.socket().getpeername() _LOGGER.debug("connected to MQTT broker %s:%d", mqtt_broker_host, mqtt_broker_port) # https://www.home-assistant.io/docs/mqtt/discovery/#discovery_prefix mqtt_client.subscribe(_MQTT_SET_TOPIC) def _report_state( mqtt_client: paho.mqtt.client.Client, switchbot_mac_address: str, switchbot_state: _SwitchbotState, ) -> None: # https://pypi.org/project/paho-mqtt/#publishing topic = _MQTT_STATE_TOPIC.replace( _MQTT_TOPIC_MAC_ADDRESS_PLACEHOLDER, switchbot_mac_address ) payload = _MQTT_STATE_PAYLOAD_MAPPING[switchbot_state] _LOGGER.debug("publishing topic=%s payload=%r", topic, payload) message_info: paho.mqtt.client.MQTTMessageInfo = mqtt_client.publish( topic=topic, payload=payload, retain=True, ) if message_info.rc != paho.mqtt.client.MQTT_ERR_SUCCESS: _LOGGER.error("failed to publish state (rc=%d)", message_info.rc) def _send_command( mqtt_client: paho.mqtt.client.Client, switchbot_mac_address: str, action: _SwitchbotAction, ) -> None: switchbot_device = switchbot.Switchbot(mac=switchbot_mac_address) if action == _SwitchbotAction.ON: if not switchbot_device.turn_on(): _LOGGER.error("failed to turn on switchbot %s", switchbot_mac_address) else: _LOGGER.info("switchbot %s turned on", switchbot_mac_address) _report_state( mqtt_client=mqtt_client, switchbot_mac_address=switchbot_mac_address, switchbot_state=_SwitchbotState.ON, ) else: assert action == _SwitchbotAction.OFF, action if not switchbot_device.turn_off(): _LOGGER.error("failed to turn off switchbot %s", switchbot_mac_address) else: _LOGGER.info("switchbot %s turned off", switchbot_mac_address) _report_state( mqtt_client=mqtt_client, switchbot_mac_address=switchbot_mac_address, switchbot_state=_SwitchbotState.OFF, ) def _mqtt_on_message( mqtt_client: paho.mqtt.client.Client, user_data: typing.Any, message: paho.mqtt.client.MQTTMessage, ) -> None: # pylint: disable=unused-argument; callback # https://github.com/eclipse/paho.mqtt.python/blob/v1.5.0/src/paho/mqtt/client.py#L469 _LOGGER.debug("received topic=%s payload=%r", message.topic, message.payload) if message.retain: _LOGGER.info("ignoring retained message") return topic_split = message.topic.split("/") if len(topic_split) != len(_MQTT_SET_TOPIC_LEVELS): _LOGGER.warning("unexpected topic %s", message.topic) return switchbot_mac_address = None for given_part, expected_part in zip(topic_split, _MQTT_SET_TOPIC_LEVELS): if expected_part == _MQTT_TOPIC_MAC_ADDRESS_PLACEHOLDER: switchbot_mac_address = given_part elif expected_part != given_part: _LOGGER.warning("unexpected topic %s", message.topic) return assert switchbot_mac_address if not _mac_address_valid(switchbot_mac_address): _LOGGER.warning("invalid mac address %s", switchbot_mac_address) return # https://www.home-assistant.io/integrations/switch.mqtt/#payload_off if message.payload.lower() == b"on": action = _SwitchbotAction.ON elif message.payload.lower() == b"off": action = _SwitchbotAction.OFF else: _LOGGER.warning("unexpected payload %r", message.payload) return _send_command( mqtt_client=mqtt_client, switchbot_mac_address=switchbot_mac_address, action=action, ) def _run( mqtt_host: str, mqtt_port: int, mqtt_username: typing.Optional[str], mqtt_password: typing.Optional[str], ) -> None: # https://pypi.org/project/paho-mqtt/ mqtt_client = paho.mqtt.client.Client() mqtt_client.on_connect = _mqtt_on_connect mqtt_client.on_message = _mqtt_on_message _LOGGER.info( "connecting to MQTT broker %s:%d", mqtt_host, mqtt_port, ) if mqtt_username: mqtt_client.username_pw_set(username=mqtt_username, password=mqtt_password) elif mqtt_password: raise ValueError("Missing MQTT username") mqtt_client.connect(host=mqtt_host, port=mqtt_port) mqtt_client.loop_forever() def _main() -> None: logging.basicConfig( level=logging.DEBUG, format="%(asctime)s:%(levelname)s:%(message)s", datefmt="%Y-%m-%dT%H:%M:%S%z", ) argparser = argparse.ArgumentParser( description="MQTT client controlling SwitchBot button automators, " "compatible with home-assistant.io's MQTT Switch platform" ) argparser.add_argument("--mqtt-host", type=str, required=True) argparser.add_argument("--mqtt-port", type=int, default=1883) argparser.add_argument("--mqtt-username", type=str) argparser.add_argument("--mqtt-password", type=str) args = argparser.parse_args() _run( mqtt_host=args.mqtt_host, mqtt_port=args.mqtt_port, mqtt_username=args.mqtt_username, mqtt_password=args.mqtt_password, )