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

import intertechno_cc1101_mqtt._homeassistant

_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:
            # accepting strings to support `{{ aliases_in_ansible_vault | to_nice_json }}`
            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)
    address, button_index = _parse_topic(topic=message.topic, aliases=aliases)
    if not address:
        return
    try:
        remote_control = intertechno_cc1101.RemoteControl(address=address)
    except ValueError:
        _LOGGER.warning(
            "failed to initialize remote control, invalid address? ignoring message",
            exc_info=True,
        )
        return
    power_setting = intertechno_cc1101.DEFAULT_POWER_SETTING
    # 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, power_setting=power_setting
            )
        elif message.payload.upper() == b"OFF":
            remote_control.turn_off(
                button_index=button_index, power_setting=power_setting
            )
        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 _publish_homeassistant_discovery_configs(
    mqtt_client: paho.mqtt.client.Client, aliases: Aliases
) -> None:
    # <discovery_prefix>/<component>/[<node_id>/]<object_id>/config
    # https://www.home-assistant.io/docs/mqtt/discovery/
    # https://www.home-assistant.io/integrations/switch.mqtt/#configuration-variables
    # https://github.com/fphammerle/systemctl-mqtt/blob/v0.5.0/systemctl_mqtt/__init__.py#L163
    # https://github.com/fphammerle/wireless-sensor-mqtt/blob/v0.3.0/wireless_sensor_mqtt/__init__.py#L153
    for alias in aliases.keys():
        # pylint: disable=protected-access; internal
        if not intertechno_cc1101_mqtt._homeassistant.validate_object_id(alias):
            if len(alias) == 0:
                _LOGGER.warning(
                    "empty alias; skipping publishing of discovery config for home assistant"
                )
                continue
            _LOGGER.warning(
                "alias %r contains characters unsupported by home assistant"
                " (supported characters: %s); skipping publishing of discovery config",
                alias,
                intertechno_cc1101_mqtt._homeassistant.OBJECT_ID_ALLOWED_CHARS,
            )
            continue
        discovery_topic = "homeassistant/switch/{}/config".format(alias)
        _LOGGER.debug(
            "publishing home assistant discovery config on topic %s", discovery_topic
        )
        mqtt_client.publish(
            topic="homeassistant/switch/{}/config".format(alias),
            payload=json.dumps(
                {
                    "unique_id": "intertechno-cc1101-mqtt/aliases/{}".format(alias),
                    "command_topic": "intertechno-cc1101/{}/set".format(alias),
                    "payload_on": "ON",  # default
                    "payload_off": "OFF",  # default
                    "retain": "true",
                    # friendly_name & template for default entity_id
                    "name": alias,
                }
            ),
            retain=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)
        _publish_homeassistant_discovery_configs(
            mqtt_client=mqtt_client, aliases=aliases
        )


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,
    )