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]] class _MQTTEventUserData: # pylint: disable=too-few-public-methods; @dataclass eventually def __init__(self, aliases: Aliases, power_setting: int): self.aliases = aliases self.power_setting = power_setting 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, userdata: _MQTTEventUserData, 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=userdata.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 # 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=userdata.power_setting ) elif message.payload.upper() == b"OFF": remote_control.turn_off( button_index=button_index, power_setting=userdata.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: # //[/]/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, userdata: _MQTTEventUserData, 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 userdata.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=userdata.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], power_setting: int, ) -> 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=_MQTTEventUserData(aliases=aliases, power_setting=power_setting) ) 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}, } ) ), ) argparser.add_argument( "--power-setting", type=int, default=intertechno_cc1101.DEFAULT_POWER_SETTING, help='see "Table 39: Optimum PATABLE Settings for Various Output Power Levels […]"' " in CC1101's docs (default: %(default)d / 0x%(default)X)", ) 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, power_setting=args.power_setting, )