123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137 |
- import argparse
- import logging
- import pathlib
- import typing
- import paho.mqtt.client
- import intertechno_cc1101
- _LOGGER = logging.getLogger(__name__)
- def _mqtt_on_message(
- mqtt_client: paho.mqtt.client.Client,
- userdata: None,
- message: paho.mqtt.client.MQTTMessage,
- ):
- # 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.warning("ignoring retained message")
- return
- topic_split = message.topic.split("/")
- try:
- address = int(topic_split[1])
- except ValueError:
- _LOGGER.warning(
- "failed to parse address %r, expected integer; ignoring message",
- topic_split[1],
- )
- return
- try:
- button_index = int(topic_split[2])
- except ValueError:
- _LOGGER.warning(
- "failed to parse button index %r, expected integer; ignoring message",
- topic_split[2],
- )
- return
- try:
- remote_control = intertechno_cc1101.RemoteControl(address=address)
- except AssertionError:
- _LOGGER.warning(
- "failed to initialize remote control, invalid address? ignoring message",
- exc_info=True,
- )
- return
- # https://www.home-assistant.io/integrations/switch.mqtt/#payload_on
- try:
- if message.payload.upper() == b"ON":
- remote_control.turn_on(button_index=button_index)
- elif message.payload.upper() == b"OFF":
- remote_control.turn_off(button_index=button_index)
- else:
- _LOGGER.warning(
- "unexpected payload %r; expected 'ON' or 'OFF'", message.payload
- )
- except Exception: # pylint: disable=broad-except; invalid perms? spi error? invalid button index?
- _LOGGER.error("failed to send signal", exc_info=True)
- 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)
- set_topic = "intertechno-cc1101/+/+/set"
- _LOGGER.info("subscribing to MQTT topic %r", set_topic)
- # alternative: .message_callback_add
- mqtt_client.on_message = _mqtt_on_message
- mqtt_client.subscribe(set_topic)
- 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
- _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:%(name)s:%(message)s",
- datefmt="%Y-%m-%dT%H:%M:%S%z",
- )
- logging.getLogger("cc1101").setLevel(logging.INFO)
- argparser = argparse.ArgumentParser(
- description="MQTT client controlling Intertechno smart outlets via a CC1101 transceiver, "
- "compatible with home-assistant.io's MQTT Switch platform",
- allow_abbrev=False,
- )
- 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)
- password_argument_group = argparser.add_mutually_exclusive_group()
- password_argument_group.add_argument("--mqtt-password", type=str)
- password_argument_group.add_argument(
- "--mqtt-password-file",
- type=pathlib.Path,
- metavar="PATH",
- dest="mqtt_password_path",
- help="stripping trailing newline",
- )
- args = argparser.parse_args()
- if args.mqtt_password_path:
- # .read_text() replaces \r\n with \n
- mqtt_password = args.mqtt_password_path.read_bytes().decode()
- if mqtt_password.endswith("\r\n"):
- mqtt_password = mqtt_password[:-2]
- elif mqtt_password.endswith("\n"):
- mqtt_password = mqtt_password[:-1]
- else:
- mqtt_password = args.mqtt_password
- _run(
- mqtt_host=args.mqtt_host,
- mqtt_port=args.mqtt_port,
- mqtt_username=args.mqtt_username,
- mqtt_password=mqtt_password,
- )
|