diff --git a/config.yaml.example b/config.yaml.example index c34ada7..599f102 100644 --- a/config.yaml.example +++ b/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 diff --git a/gateway.py b/gateway.py index 619338c..1ce2e27 100755 --- a/gateway.py +++ b/gateway.py @@ -39,8 +39,8 @@ logger.suppress_update_failures(parsed.suppress) _LOGGER.info('Starting') mqtt = MqttClient(settings['mqtt']) -manager = WorkersManager() -manager.register_workers(settings['manager']).start(mqtt) +manager = WorkersManager(settings['manager']) +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: - logger.log_exception(_LOGGER, "Timeout while executing worker command", suppress=True) + except TimeoutError as e: + 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') diff --git a/workers/base.py b/workers/base.py index 222ee94..24173dd 100644 --- a/workers/base.py +++ b/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() diff --git a/workers/blescanmulti.py b/workers/blescanmulti.py index dbb0c23..efdc897 100644 --- a/workers/blescanmulti.py +++ b/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): - super(BlescanmultiWorker, self).__init__(**kwargs) + def __init__(self, command_timeout, **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() diff --git a/workers/linakdesk.py b/workers/linakdesk.py index 21a2cc3..cd0849d 100644 --- a/workers/linakdesk.py +++ b/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 diff --git a/workers/miscale.py b/workers/miscale.py index 72017d9..5d8c00a 100644 --- a/workers/miscale.py +++ b/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 diff --git a/workers_manager.py b/workers_manager.py index b168ae5..3803890 100644 --- a/workers_manager.py +++ b/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): - for (worker_name, worker_config) in config['workers'].items(): + def register_workers(self): + 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']) - command = self.Command(worker_obj.status_update, []) + _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, 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: - for (callback_name, options) in config['topic_subscription'].items(): + if 'topic_subscription' in self._config: + 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]))