Make command timeout for workers configurable (#62)

config.yaml.example

@@ -12,8 +12,10 @@ manager:
     update_all:
       topic: homeassistant/status
       payload: online
+  command_timeout: 35           # Timeout for worker operations. Can be removed if the default of 35 seconds is sufficient.
   workers:
     mysensors:
+      command_timeout: 35       # Optional override of globally set command_timeout.
       args:
         port: /dev/ttyUSB0
         baudrate: 9600

gateway.py

@@ -39,8 +39,8 @@ logger.suppress_update_failures(parsed.suppress)
 _LOGGER.info('Starting')
 
 mqtt = MqttClient(settings['mqtt'])
-manager = WorkersManager()
+manager = WorkersManager(settings['manager'])
-manager.register_workers(settings['manager']).start(mqtt)
+manager.register_workers().start(mqtt)
 
 running = True
 
@@ -49,8 +49,8 @@ while running:
     mqtt.publish(_WORKERS_QUEUE.get(timeout=10).execute())
   except queue.Empty: # Allow for SIGINT processing
     pass
-  except TimeoutError:
+  except TimeoutError as e:
-    logger.log_exception(_LOGGER, "Timeout while executing worker command", suppress=True)
+    logger.log_exception(_LOGGER, str(e) if str(e) else 'Timeout while executing worker command', suppress=True)
   except (KeyboardInterrupt, SystemExit):
     running = False
     _LOGGER.info('Finish current jobs and shut down. If you need force exit use kill')

workers/base.py

@@ -1,5 +1,6 @@
 class BaseWorker:
-  def __init__(self, **args):
+  def __init__(self, command_timeout, **args):
+    self.command_timeout = command_timeout
     for arg, value in args.items():
       setattr(self, arg, value)
     self._setup()

workers/blescanmulti.py

@@ -84,8 +84,8 @@ class BlescanmultiWorker(BaseWorker):
   scan_timeout = 10.  # type: float
   scan_passive = True  # type: str or bool
 
-  def __init__(self, **kwargs):
+  def __init__(self, command_timeout, **kwargs):
-    super(BlescanmultiWorker, self).__init__(**kwargs)
+    super(BlescanmultiWorker, self).__init__(command_timeout, **kwargs)
     self.scanner = Scanner().withDelegate(ScanDelegate())
     self.last_status = [
       BleDeviceStatus(self, mac, name) for name, mac in self.devices.items()

workers/linakdesk.py

@@ -6,6 +6,9 @@ from workers.base import BaseWorker
 REQUIREMENTS = ['git+https://github.com/zewelor/linak_bt_desk.git@aa9412f98b3044be34c70e89d02721e6813ea731#egg=linak_bt_desk']
 
 class LinakdeskWorker(BaseWorker):
+
+  SCAN_TIMEOUT = 20
+
   def _setup(self):
     from linak_dpg_bt import LinakDesk
 
@@ -15,7 +18,7 @@ class LinakdeskWorker(BaseWorker):
     return [MqttMessage(topic=self.format_topic('height/cm'), payload=self._get_height())]
 
   def _get_height(self):
-    with timeout(20, exception=TimeoutError):
+    with timeout(SCAN_TIMEOUT, exception=TimeoutError('Retrieving the height from {} device {} timed out after {} seconds'.format(repr(self), self.mac, SCAN_TIMEOUT))):
       self.desk.read_dpg_data()
       return self.desk.current_height_with_offset.cm
 

workers/miscale.py

@@ -11,6 +11,9 @@ REQUIREMENTS = ['bluepy']
 # sudo setcap 'cap_net_raw,cap_net_admin+eip' /usr/local/lib/python3.6/dist-packages/bluepy/bluepy-helper
 
 class MiscaleWorker(BaseWorker):
+
+  SCAN_TIMEOUT = 5
+
   def status_update(self):
     return [MqttMessage(topic=self.format_topic('weight/kg'), payload=self._get_weight())]
 
@@ -19,9 +22,9 @@ class MiscaleWorker(BaseWorker):
 
     scan_processor = ScanProcessor(self.mac)
     scanner = btle.Scanner().withDelegate(scan_processor)
-    scanner.scan(5, passive=True)
+    scanner.scan(SCAN_TIMEOUT, passive=True)
 
-    with timeout(5, exception=TimeoutError):
+    with timeout(SCAN_TIMEOUT, exception=TimeoutError('Retrieving the weight from {} device {} timed out after {} seconds'.format(repr(self), self.mac, SCAN_TIMEOUT))):
       while scan_processor.weight is None:
         time.sleep(1)
       return scan_processor.weight

workers_manager.py

@@ -21,38 +21,43 @@ _LOGGER = logger.get(__name__)
 
 class WorkersManager:
   class Command:
-    def __init__(self, callback, args=(), options=dict()):
+    def __init__(self, callback, timeout, args=(), options=dict()):
       self._callback = callback
+      self._timeout = timeout
       self._args = args
       self._options = options
+      self._source = '{}.{}'.format(callback.__self__.__class__.__name__ if hasattr(callback, '__self__') else callback.__module__, callback.__name__)
 
     def execute(self):
       messages = []
-      with timeout(35, exception=TimeoutError):
+      with timeout(self._timeout, exception=TimeoutError('Execution of command {} timed out after {} seconds'.format(self._source, self._timeout))):
         messages = self._callback(*self._args)
 
-      _LOGGER.debug("Command execution result: %s", messages)
+      _LOGGER.debug('Execution result of command %s: %s', self._source, messages)
       return messages
 
-  def __init__(self):
+  def __init__(self, config):
     self._mqtt_callbacks = []
     self._update_commands = []
     self._scheduler = BackgroundScheduler(timezone=utc)
     self._daemons = []
+    self._config = config
+    self._command_timeout = config.get('command_timeout', 35)
 
-  def register_workers(self, config):
+  def register_workers(self):
-    for (worker_name, worker_config) in config['workers'].items():
+    for (worker_name, worker_config) in self._config['workers'].items():
       module_obj = importlib.import_module("workers.%s" % worker_name)
       klass = getattr(module_obj, "%sWorker" % worker_name.title())
 
       if module_obj.REQUIREMENTS is not None:
         self._pip_install_helper(module_obj.REQUIREMENTS)
 
-      worker_obj = klass(**worker_config['args'])
+      command_timeout = worker_config.get('command_timeout', self._command_timeout)
+      worker_obj = klass(command_timeout, **worker_config['args'])
 
       if hasattr(worker_obj, 'status_update'):
-        _LOGGER.debug("Added %s worker with %d seconds interval", repr(worker_obj), worker_config['update_interval'])
+        _LOGGER.debug("Added %s worker with %d seconds interval and a %d seconds timeout", repr(worker_obj), worker_config['update_interval'], worker_obj.command_timeout)
-        command = self.Command(worker_obj.status_update, [])
+        command = self.Command(worker_obj.status_update, worker_obj.command_timeout, [])
         self._update_commands.append(command)
 
         if 'update_interval' in worker_config:
@@ -77,11 +82,11 @@ class WorkersManager:
           partial(self._on_command_wrapper, worker_obj)
         ))
 
-    if 'topic_subscription' in config:
+    if 'topic_subscription' in self._config:
-      for (callback_name, options) in config['topic_subscription'].items():
+      for (callback_name, options) in self._config['topic_subscription'].items():
         self._mqtt_callbacks.append((
           options['topic'],
-          lambda client, _ , c: self._queue_if_matching_payload(self.Command(getattr(self, callback_name)), c.payload, options['payload']))
+          lambda client, _ , c: self._queue_if_matching_payload(self.Command(getattr(self, callback_name), self._command_timeout), c.payload, options['payload']))
         )
 
     return self
@@ -127,4 +132,4 @@ class WorkersManager:
     _LOGGER.debug("Received command for %s on %s: %s", repr(worker_obj), c.topic, c.payload)
     global_topic_prefix = userdata['global_topic_prefix']
     topic = c.topic[len(global_topic_prefix+'/'):] if global_topic_prefix is not None else c.topic
-    self._queue_command(self.Command(worker_obj.on_command, [topic, c.payload]))
+    self._queue_command(self.Command(worker_obj.on_command, worker_obj.command_timeout, [topic, c.payload]))