MQTT connection options

README.md

@@ -31,7 +31,7 @@ or other intellectual property remain the property of their respective owners.
 
 There are many!
 
-* Not currently possible to set a username or password for MQTT!
+* ~~Not currently possible to set a username or password for MQTT!~~
 * Currently only the Filter, Aux 1, Aux 2, and Super Chlorinate controls are exposed
 * Currently only Air/Pool/Spa Temperature, Pool/Spa Chlorinator (%), Salt Level, and Check System sensors are exposed (need to add especially system messages!)
 * Serial failures may result in hanging—the process may not exit nor recover, and may have to be killed manually
@@ -62,9 +62,7 @@ This venv should remain activated when you run the module as described below.
 
 ## Running
 
-**THE COMMAND LINE INTERFACE WILL CHANGE SOON!**
-
-Currently the module can be started like so:
+The module can be minimally started like so:
 
 ```
 python -m aqualogic_mqtt.client \
@@ -78,12 +76,8 @@ E.g.
 ```console
 (venv-pool)$ python -m aqualogic_mqtt.client -s /dev/ttyUSB0 -m localhost:1883 -p homeassistant
 ```
-
-The MQTT Discovery Prefix determines the "path" on the MQTT broker where
-the interface is exposed. For Home Assistant, the default is
-`homeassistant` unless you have changed it in your configuration.
-
-> **NOTE:** While the topic cannot be covered in depth here, be aware that using multiple USB serial devices (including for example a mix of a USB RS485 interface and Z-Wave or Zigbee stick) may result in unpredictable paths for the serial devices--you may need to set up udev rules to make the correct devices show up at the configured path(s).
+> [!NOTE]
+> While the topic cannot be covered in depth here, be aware that using multiple USB serial devices (including for example a mix of a USB RS485 interface and Z-Wave or Zigbee stick) may result in unpredictable paths for the serial devices--you may need to set up udev rules to make the correct devices show up at the configured path(s).
 
 It is also possible to use the `-t` option (in lieu of `-s`) to connect to a Serial/TCP converter (e.g. a USR-N510) with a host:port, like so
 ```console
@@ -93,6 +87,37 @@ Note, however, that using a network converter such as this has
 been found to be unreliable for device control (reading values
 usually works well enough).
 
+### MQTT Connection Options
+
+Besides just the MQTT broker's host and port, there are a number of other options that you can specify regarding the connection:
+
+* `--mqtt-username MQTT_USERNAME`
+  * username for the MQTT broker
+* `--mqtt-password MQTT_PASSWORD`
+  * password for MQTT broker 
+    > [!CAUTION]
+    > Generally, specifying passwords on the command line is an insecure practice. See below for a better option.
+* `--mqtt-clientid MQTT_CLIENTID`
+  * client ID provided to the MQTT broker
+* `--mqtt-insecure`
+    * ignore certificate validation errors for the MQTT broker
+      > [!CAUTION]
+      > Using this option exposes you to potential impersonation/MITM attacks.
+* `--mqtt-version {3,5}`
+  * MQTT protocol major version number (default is 5)
+* `--mqtt-transport {tcp,websockets}`
+  * MQTT transport mode (default is tcp unless dest port is 9001 or 443)
+
+#### `AQUALOGIC_MQTT_PASSWORD` environment variable
+
+To avoid specifying the MQTT client password on the command line (where it may be visible in history and process listings), you should instead store such a password in the environment variable `AQUALOGIC_MQTT_PASSWORD`. This variable will be checked if you specify
+`--mqtt-username`. If `--mqtt-password` is also specified, the command line option overrides the environment variable.
+
+### Home Assistant related options
+
+* `-p DISCOVER_PREFIX` or `--discover-prefix DISCOVER_PREFIX`
+  * The MQTT Discovery Prefix determines the "path" on the MQTT broker where the interface is exposed. For Home Assistant discovery to work, you should use `homeassistant`, which is the default unless you have changed it in your configuration.
+
 ## Running in a container
 
 **COMING SOON!**
@@ -107,7 +132,8 @@ configuration in your Home Assistant instance and make sure that
 discovery is enabled and that the discovery prefix matches what you are
 providing to this module.
 
-> **NOTE:** It's not yet possible to customize the birth message location and the birth message is used—so this must be set to `[prefix]/status` for now!
+> [!IMPORTANT]
+> It's not yet possible to customize the birth message location and the birth message is used—so this must be set to `[prefix]/status` for now!
 
 ## Design Goals
 

aqualogic_mqtt/client.py

@@ -7,6 +7,7 @@
 import argparse
 
 import paho.mqtt.client as mqtt
+from paho.mqtt.reasoncodes import ReasonCode
 
 from aqualogic.core import AquaLogic
 from aqualogic.states import States
@@ -33,17 +34,28 @@ class Client:
     _identifier = None
     _discover_prefix = None
     _formatter = None
+    _disconnect_retries = 3
+    _disconnect_retry_wait_max = 30
+    _disconnect_retry_wait = 1
+    _disconnect_retry_num = 0
 
-    def __init__(self, identifier="aqualogic", discover_prefix="homeassistant"):
+    def __init__(self, identifier="aqualogic", discover_prefix="homeassistant", 
+                 client_id=None, transport='tcp', protocol_num=5):
         self._identifier = identifier
         self._discover_prefix = discover_prefix
 
         self._formatter = Messages(identifier, discover_prefix)
 
         self._panel = AquaLogic(web_port=0)
-        self._paho_client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
+
+        protocol = mqtt.MQTTv311 if protocol_num == 3 else mqtt.MQTTv5
+        self._paho_client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
+                                        client_id=client_id, transport=transport,
+                                        protocol=protocol)
         self._paho_client.on_message = self._on_message
         self._paho_client.on_connect = self._on_connect
+        self._paho_client.on_disconnect = self._on_disconnect
+        self._paho_client.on_connect_fail = self._on_connect_fail
 
     # Respond to panel events
     def _panel_changed(self, panel):
@@ -61,31 +73,73 @@ def _on_message(self, client, userdata, msg):
 
     def _on_connect(self, client, userdata, flags, reason_code, properties):
         logging.debug("_on_connect called")
+        if isinstance(reason_code, ReasonCode):
+            if reason_code.is_failure:
+                logging.critical(f"Got failure when connecting MQTT: {reason_code.getName()}! Exiting!")
+                raise RuntimeError(reason_code)
+            #elif : #FIXME: elif what?
+            #    logging.debug(f"Got unexpected reason_code when connecting MQTT: {reason_code.getName()}")
+            #    logging.debug(reason_code)
+        self._disconnect_retry_num = 0
+        self._disconnect_retry_wait = 1
+
         sub_topics = self._formatter.get_subscription_topics()
         for topic in sub_topics:
             self._paho_client.subscribe(topic)
         logging.debug(f"Publishing to {self._formatter.get_discovery_topic()}...")
         logging.debug(self._formatter.get_discovery_message())
         self._paho_client.publish(self._formatter.get_discovery_topic(), self._formatter.get_discovery_message())
         ...
+
+    def _on_connect_fail(self, userdata, reason_code):
+        #TODO: Have not been able to reach here, needs testing!
+        logging.debug("_on_connect_fail called")
+
+    def _on_disconnect(self, client, userdata, flags, reason_code, properties):
+        if isinstance(reason_code, ReasonCode):
+            if reason_code.is_failure:
+                logging.error(f"MQTT Disconnected: {reason_code.getName()}!")
+                #NOTE: Paho documentation is confusing about loop_forever and reconnection. Will
+                # this ever be called when loop_forever "automatically handles reconnecting"? If not, it 
+                # seems this callback is really only hit on initial connect failures?
+                if self._disconnect_retry_num < self._disconnect_retries:
+                    self._disconnect_retry_num += 1
+                    self._disconnect_retry_wait = min(self._disconnect_retry_wait*2, self._disconnect_retry_wait_max)
+                    logging.info(f"Retrying ({self._disconnect_retry_num}) after {self._disconnect_retry_wait}s...")
+                    sleep(self._disconnect_retry_wait)
+                    self._paho_client.reconnect()
+                else:
+                    logging.critical("MQTT connection failed!")
+                    self._paho_client.disconnect()
+                    raise RuntimeError(reason_code)
+            else:
+                logging.debug(f"MQTT Disconnected: {reason_code.getName()}")
+        elif isinstance(reason_code, int):
+            if reason_code > 0:
+                logging.error(f"MQTT Disconnected: {reason_code}")
 
-    def connect_panel(self, source):
+    def panel_connect(self, source):
         if ':' in source:
             s_host, s_port = source.split(':')
             self._panel.connect(s_host, int(s_port))
         else:
             self._panel.connect_serial(source)
         ...
 
-    def connect_mqtt(self, dest:(str), port:(int)=1883, keepalive=60):
+    def mqtt_username_pw_set(self, username:(str), password:(str)):
+        return self._paho_client.username_pw_set(username=username, password=password)
+
+    def mqtt_tls_set(self, certfile=None, keyfile=None, cert_reqs=ssl.CERT_REQUIRED):
+        return self._paho_client.tls_set(certfile=certfile, keyfile=keyfile, cert_reqs=cert_reqs)
+
+    def mqtt_connect(self, dest:(str), port:(int)=1883, keepalive=60):
         host = dest
         if dest is not None:
             if ':' in dest:
                 host, port = dest.split(':')
                 port = int(port)
             else:
                 host = dest
-        self._paho_client.tls_set(cert_reqs=ssl.CERT_NONE)
         r = self._paho_client.connect(host, port, keepalive)
         logging.debug(f"Connected to {host}:{port} with result {r}")
 
@@ -107,24 +161,61 @@ def loop_forever(self):
     autodisc_prefix = None
     source = None
     dest = None
+    mqtt_password = os.environ.get('AQUALOGIC_MQTT_PASSWORD')
 
     parser = argparse.ArgumentParser(
                     prog='aqualogic_mqtt',
                     description='MQTT adapter for pool controllers',
                     )
+    source_group = parser.add_argument_group("source options")
+    source_group_mex = source_group.add_mutually_exclusive_group(required=True)
+    source_group_mex.add_argument('-s', '--serial', type=str, metavar="/dev/path",
+        help="serial device source (path)")
+    source_group_mex.add_argument('-t', '--tcp', type=str, metavar="tcpserialhost:port",
+        help="network serial adapter source in the format host:port")
+    mqtt_group = parser.add_argument_group('MQTT destination options')
+    mqtt_group.add_argument('-m', '--mqtt-dest', required=True, type=str, metavar="mqtthost:port",
+        help="MQTT broker destination in the format host:port")
+    mqtt_group.add_argument('--mqtt-username', type=str, help="username for the MQTT broker")
+    mqtt_group.add_argument('--mqtt-password', type=str, 
+        help="password for MQTT broker (recommend set the environment variable AQUALOGIC_MQTT_PASSWORD instead!)")
+    mqtt_group.add_argument('--mqtt-clientid', type=str, help="client ID provided to the MQTT broker")
+    mqtt_group.add_argument('--mqtt-insecure', action='store_true', 
+        help="ignore certificate validation errors for the MQTT broker (dangerous!)")
+    mqtt_group.add_argument('--mqtt-version', type=int, choices=[3,5], default=5, 
+        help="MQTT protocol major version number (default is 5)")
+    mqtt_group.add_argument('--mqtt-transport', type=str, choices=["tcp","websockets"], default="tcp",
+        help="MQTT transport mode (default is tcp unless dest port is 9001 or 443)")
+    ha_group = parser.add_argument_group("Home Assistant options")
+    ha_group.add_argument('-p', '--discover-prefix', default="homeassistant", type=str, 
+        help="MQTT prefix path (default is \"homeassistant\")")
+
     parser.add_argument('-m', '--mqtt-dest', required=True, type=str, help="MQTT broker destination in the format host:port")
     source_group = parser.add_mutually_exclusive_group(required=True)
     source_group.add_argument('-s', '--serial', type=str, help="serial device source (path)")
     source_group.add_argument('-t', '--tcp', type=str, help="network serial adapter source in the format host:port")
     parser.add_argument('-p', '--discover-prefix', default="homeassistant", type=str, help="MQTT prefix path (default is \"homeassistant\")")
     args = parser.parse_args()
-
+    
     source = args.serial if args.serial is not None else args.tcp
     dest = args.mqtt_dest
 
     mqtt_client = Client(discover_prefix=args.discover_prefix)
     mqtt_client.connect_mqtt(dest=dest)
     mqtt_client.connect_panel(source)
+
+    mqtt_client = Client(discover_prefix=args.discover_prefix, 
+                         client_id=args.mqtt_clientid, transport=args.mqtt_transport, 
+                         protocol_num=args.mqtt_version
+                         )
+    if args.mqtt_username is not None:
+        mqtt_password = args.mqtt_password if args.mqtt_password is not None else mqtt_password
+        mqtt_client.mqtt_username_pw_set(args.mqtt_username, mqtt_password)
+    #TODO Broker client cert
+    if args.mqtt_insecure:
+        mqtt_client.mqtt_tls_set(cert_reqs=ssl.CERT_NONE)
+    mqtt_client.mqtt_connect(dest=dest)
+    mqtt_client.panel_connect(source)
     mqtt_client.loop_forever()