From e089d53d1630dfa1815a3fa64e7797f727b5d80d Mon Sep 17 00:00:00 2001 From: Ben Dews Date: Sat, 9 Feb 2019 19:17:25 +1100 Subject: [PATCH] Feature: dynamic update intervals (#49) * Updated workers_manager with ability to dynamically set update_interval * Updated README file with documentation for dynamic update intervals --- README.md | 10 ++++++++++ workers_manager.py | 21 +++++++++++++++++++-- 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index dfd71c2..ad9339e 100644 --- a/README.md +++ b/README.md @@ -100,6 +100,16 @@ Use mosquitto_sub to print all messages mosquitto_sub -h localhost -d -t # command also help for me to test MQTT messages ``` +**Dynamically Changing the Update Interval** +To dynamically change the `update_interval` of a worker, publish a message containing the new interval in seconds at the `update_interval` topic. Note that the `update_interval` will revert back to the value in `config.yaml` when the gateway is restarted. +I.E: +``` +# Set a new update interval of 3 minutes +mosquitto_pub -h localhost -t 'miflora/update_interval' -m '150' +# Set a new update interval of 30 seconds +mosquitto_pub -h localhost -t 'mithermometer/update_interval' -m '30' +``` + ## Custom worker development diff --git a/workers_manager.py b/workers_manager.py index 6328ba3..ee73f5b 100644 --- a/workers_manager.py +++ b/workers_manager.py @@ -50,10 +50,15 @@ class WorkersManager: self._update_commands.append(command) if 'update_interval' in worker_config: - self._scheduler.add_job( + job_id = '{}_interval_job'.format(worker_name) + interval_job = self._scheduler.add_job( partial(self._queue_command, command), 'interval', - seconds=worker_config['update_interval'], + seconds=worker_config['update_interval'], id=job_id ) + self._mqtt_callbacks.append(( + worker_obj.format_topic('update_interval'), + partial(self._update_interval_wrapper, command, job_id) + )) elif hasattr(worker_obj, 'run'): _LOGGER.debug("Registered: %s as daemon" % (worker_name)) self._daemons.append(worker_obj) @@ -100,6 +105,18 @@ class WorkersManager: for package in package_names: pip_main(['install', '-q', package]) + def _update_interval_wrapper(self, command, job_id, client, userdata, c): + _LOGGER.debug("Recieved updated interval for %s with: %s", c.topic, c.payload) + try: + new_interval = int(c.payload) + self._scheduler.remove_job(job_id) + self._scheduler.add_job( + partial(self._queue_command, command), 'interval', + seconds=new_interval, id=job_id + ) + except ValueError: + _LOGGER.info("New interval invalid, recieved: %s", c.payload) + def _on_command_wrapper(self, worker_obj, client, userdata, c): _LOGGER.debug("on command wrapper for with %s: %s", c.topic, c.payload) global_topic_prefix = userdata['global_topic_prefix']