Exit all threads when exceptions occur

README.md

@@ -33,9 +33,10 @@ There are several!
 
 * ~~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~~ 
+* ~~Currently only Air/Pool/Spa Temperature, Pool/Spa Chlorinator (%), Salt Level, and Check System sensors are exposed~~
+* Pool/Spa button and Service button not yet supported
 * ~~System Messages are not yet supported~~
-* Serial failures may result in hanging—the process may not exit nor recover, and may have to be killed manually
+* ~~Serial failures may result in hanging—the process may not exit nor recover, and may have to be killed manually~~
 * Metric unit configured systems are not yet supported
 * Not yet possible to use a customized Home Assistant MQTT birth message topic or payload
 * Only one pool controller is supported per MQTT broker (please describe your setup in an issue if this affects you 🤨)
@@ -188,6 +189,14 @@ To avoid specifying the MQTT client password on the command line (where it may b
 * `-p DISCOVER_PREFIX` or `--discover-prefix DISCOVER_PREFIX`
   * The MQTT Discovery Prefix determines the "path" on the MQTT broker where the interface is exposed. The default for this option is `homeassistant`, which matches the default in Home Assistant. If you have changed it in your Home Assistant configuration, you should specify a different value here.
 
+### Other options
+
+#### Source Timeout
+
+When running, the module keeps track of how long it has been since the last update was received from the pool controller. If no message has been received within the timeout period, the process _exits_. This is designed to allow some other managing process (e.g. container orchestrator, systemd, supervisor, etc.) to restart the module process, hopefully fixing any connection issue. In practice, this has worked to solve serial port "timeout" errors if the cable is disconnected and reconnected or if a power outage takes the pool controller offline.
+
+The default timeout before exit is 10 seconds. Use the `-T`/`--source-timeout` option to change this value, providing some number of seconds as an argument. Using an arbitrarily high number for this value effectively disables the exiting behavior, but this is not recommended.
+
 ## Running in a container
 
 **COMING SOON!**

aqualogic_mqtt/client.py

@@ -20,11 +20,6 @@ def _patched_write_to_serial(self, data):
     self._serial.write(data)
     self._serial.flush()
 AquaLogic._write_to_serial = _patched_write_to_serial
-# Monkey-patch class property into _web so we can avoid running the web server without an error
-class _WebDummy:
-    def text_updated(self, str):
-        return
-AquaLogic._web = _WebDummy()
 
 logging.basicConfig(level=logging.DEBUG)
 
@@ -142,14 +137,19 @@ def mqtt_connect(self, dest:(str), port:(int)=1883, keepalive=60):
 
     def loop_forever(self):
         try:
-            #self._paho_client.loop_start()
+            self._paho_client.loop_start()
             self._panel_thread = threading.Thread(target=self._panel.process, args=[self._panel_changed])
+            self._panel_thread.daemon = True # https://stackoverflow.com/a/50788759/489116 ?
             self._panel_thread.start()
-            self._paho_client.loop_forever()
-            #while True:
-            #    sleep(1)
+            #self._paho_client.loop_forever()
+            while True:
+                logging.debug(f"Update age: {self._pman.get_last_update_age()}")
+                if not self._pman.is_updating():
+                    logging.critical("Panel not updated in "+str(self._pman.get_last_update_age())+"s, exiting!")
+                    raise RuntimeError("Panel stopped updating!")
+                sleep(1)
         finally:
-            #self._paho_client.loop_stop()
+            self._paho_client.loop_stop()
             pass
 
 
@@ -181,6 +181,8 @@ def loop_forever(self):
         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")
+    source_group.add_argument('-T', '--source-timeout', nargs=1, type=int, default=10, metavar="SECONDS",
+        help="seconds after which the source connection is deemed to be lost if no updates have been seen--the program will exit if the timeout is reached")
 
     mqtt_group = parser.add_argument_group('MQTT destination options')
     mqtt_group.add_argument('-m', '--mqtt-dest', required=True, type=str, metavar="mqtthost:port",
@@ -205,7 +207,9 @@ def loop_forever(self):
     source = args.serial if args.serial is not None else args.tcp
     dest = args.mqtt_dest
 
-    pman = PanelManager(args.system_message_expiration)
+    pman = PanelManager(args.source_timeout, args.system_message_expiration)
+    # Monkey-patch PanelManager into _web so we can avoid running the web server without an error
+    AquaLogic._web = pman
 
     formatter = Messages(identifier="aqualogic", discover_prefix=args.discover_prefix,
                          enable=args.enable if args.enable is not None else [], 

aqualogic_mqtt/panelmanager.py

@@ -1,12 +1,17 @@
 import time
+import logging
 
 # At present PanelManager only keeps track of system messages, though
 # it may expand to handle all aqualogic.panel concerns in the future.
 class PanelManager:
+    _timeout = None
     _exp_s = None
+    _last_text_update = None
 
-    def __init__(self, exp_seconds:(int)):
-        self._exp_s = exp_seconds
+    def __init__(self, connect_timeout:(int), message_exp_seconds:(int)):
+        self._last_text_update = time.time()
+        self._timeout = connect_timeout
+        self._exp_s = message_exp_seconds
         self._registry = {}
 
     def observe_system_message(self, message:(str)):
@@ -19,4 +24,19 @@ def observe_system_message(self, message:(str)):
         self._registry = { k:v for k,v in self._registry.items() if v > exp }
 
     def get_system_messages(self):
-        return sorted(self._registry.keys())
+        return sorted(self._registry.keys())
+
+    def get_last_update_age(self):
+        return time.time() - self._last_text_update
+
+    def is_updating(self):
+        return (time.time() - self._last_text_update) < self._timeout
+
+    # This is a method with the same name/sig as one in aqualogic.web.WebServer. This
+    # allows 1: monkey-patching this class into aqualogic to allow the process loop to
+    # function without its web server running, 2: us to pick up activity and screen
+    # updates from the panel (e.g. to determine if the connection is lost).
+    def text_updated(self, str):
+        self._last_text_update = time.time()
+        logging.debug(f"text_updated: {str}")
+        return