intertechno-cc1101-mqtt/intertechno_cc1101_mqtt/__init__.py

import argparse
import json
import logging
import pathlib
import typing

import paho.mqtt.client
import intertechno_cc1101

_LOGGER = logging.getLogger(__name__)

Aliases = typing.Dict[str, typing.Dict[str, int]]


def _parse_topic(
    topic: str, aliases: Aliases
) -> typing.Tuple[typing.Optional[int], typing.Optional[int]]:
    topic_split = topic.split("/")
    if len(topic_split) == 3:
        try:
            alias_attrs = aliases[topic_split[1]]
        except KeyError:
            _LOGGER.warning("unknown alias %r; ignoring message", topic_split[1])
            return None, None
        try:
            return int(alias_attrs["address"]), int(alias_attrs["button-index"])
        except KeyError:
            _LOGGER.error(
                "alias file must provide fields 'address' and 'button-index' for each alias"
            )
            return None, None
    try:
        address = int(topic_split[1])
    except ValueError:
        _LOGGER.warning(
            "failed to parse address %r, expected integer; ignoring message",
            topic_split[1],
        )
        return None, None
    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 None, None
    return address, button_index


def _mqtt_on_message(
    mqtt_client: paho.mqtt.client.Client,
    aliases: Aliases,
    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:  # TODO remove
        _LOGGER.warning("ignoring retained message")
        return
    address, button_index = _parse_topic(topic=message.topic, aliases=aliases)
    if not address:
        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,
    aliases: Aliases,
    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)
    # alternative: .message_callback_add
    mqtt_client.on_message = _mqtt_on_message
    set_topic = "intertechno-cc1101/+/+/set"
    _LOGGER.info("subscribing to MQTT topic %r (address & button index)", set_topic)
    mqtt_client.subscribe(set_topic)
    if aliases:
        set_alias_topic = "intertechno-cc1101/+/set"
        _LOGGER.info("subscribing to MQTT topic %r (alias)", set_alias_topic)
        mqtt_client.subscribe(set_alias_topic)


def _run(
    mqtt_host: str,
    mqtt_port: int,
    mqtt_username: typing.Optional[str],
    mqtt_password: typing.Optional[str],
    alias_file_path: typing.Optional[pathlib.Path],
) -> None:
    if alias_file_path:
        with alias_file_path.open("r") as alias_file:
            aliases = json.load(alias_file)
    else:
        aliases = {}
    # https://pypi.org/project/paho-mqtt/
    mqtt_client = paho.mqtt.client.Client(userdata=aliases)
    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",
    )
    argparser.add_argument(
        "--alias-file",
        metavar="PATH",
        dest="alias_file_path",
        type=pathlib.Path,
        help="json: {}".format(
            json.dumps(
                {
                    "some-alias": {"address": 12345678, "button-index": 0},
                    "another-alias": {"address": 12345678, "button-index": 0},
                }
            )
        ),
    )
    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,
        alias_file_path=args.alias_file_path,
    )