diff --git a/config.yaml.example b/config.yaml.example index 690c978..83a21f7 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -2,11 +2,17 @@ mqtt: host: 192.168.1.1 username: user password: password -workers: - thermostat: - args: - devices: - bedroom: 00:11:22:33:44:55 - topic_prefix: thermostat - topic_subscription: thermostat/+/+/set - update_interval: 60 + +manager: + topic_subcription: + update_all: + topic: homeassistant/status + payload: online + workers: + thermostat: + args: + devices: + bedroom: 00:11:22:33:44:55 + topic_prefix: thermostat + topic_subscription: thermostat/+/+/set + update_interval: 60 diff --git a/gateway.py b/gateway.py index 4b27e7d..442e547 100755 --- a/gateway.py +++ b/gateway.py @@ -26,7 +26,7 @@ else: _LOGGER.debug('Starting') mqtt = MqttClient(settings['mqtt']) -WorkersManager.start(mqtt, settings['workers']) +WorkersManager().register_workers(settings['manager']).start(mqtt) running = True diff --git a/mqtt.py b/mqtt.py index 04bf241..6db7288 100644 --- a/mqtt.py +++ b/mqtt.py @@ -1,11 +1,15 @@ import paho.mqtt.publish as publish import paho.mqtt.client as mqtt + class MqttClient: def __init__(self, config): self._config = config def publish(self, messages): + if messages is None: + return + publish.multiple(list(map(lambda m: m.as_dict, messages)), hostname=self.hostname, auth={'username': self.username, 'password': self.password}) @@ -34,6 +38,7 @@ class MqttClient: mqttc.loop_start() + class MqttMessage: def __init__(self, topic=None, payload=None): self._topic = topic @@ -59,7 +64,7 @@ class MqttMessage: } def __repr__(self): - return self.as_dict.__str__() + return self.as_dict.__str__() def __str__(self): return self.__repr__() diff --git a/workers/thermostat.py b/workers/thermostat.py index fb8142c..fd8a28b 100644 --- a/workers/thermostat.py +++ b/workers/thermostat.py @@ -28,6 +28,8 @@ class ThermostatWorker(BaseWorker): self._reverse_modes = {v: k for k, v in self._mapped_modes.items()} def get_mapping(self, mode): + if mode < 0: + return None return self._mapped_modes[mode] def get_reverse_mapping(self, mode): @@ -67,6 +69,8 @@ class ThermostatWorker(BaseWorker): _, device_name, method, _ = topic.split('/') thermostat = self.devices[device_name] + value = value.decode('utf-8') + if method == STATE_AWAY: method = "mode" value = ModesMapper.on_off_to_mode(value) diff --git a/workers_manager.py b/workers_manager.py index 308ab29..0cbdc07 100644 --- a/workers_manager.py +++ b/workers_manager.py @@ -20,38 +20,70 @@ class WorkersManager: _LOGGER.debug(messages) return messages - @classmethod - def start(cls, mqtt, config): - scheduler = BackgroundScheduler() - mqtt_callbacks = [] + def __init__(self): + self._mqtt_callbacks = [] + self._update_commands = [] + self._scheduler = BackgroundScheduler() - for (worker_name, worker_config) in config.items(): + def register_workers(self, config): + for (worker_name, worker_config) in 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: - cls._pip_install_helper(module_obj.REQUIREMENTS) + self._pip_install_helper(module_obj.REQUIREMENTS) worker_obj = klass(**worker_config['args']) - if 'update_interval' in worker_config: + if hasattr(worker_obj, 'status_update'): _LOGGER.debug("Added: %s with %d seconds interval" % (worker_name, worker_config['update_interval'])) - _WORKERS_QUEUE.put(cls.Command(worker_obj.status_update, [])) - scheduler.add_job( - lambda x: _WORKERS_QUEUE.put(cls.Command(worker_obj.status_update, [])), 'interval', - seconds=worker_config['update_interval'], - args=[worker_name] - ) + command = self.Command(worker_obj.status_update, []) + self._update_commands.append(command) + + if 'update_interval' in worker_config: + self._scheduler.add_job( + lambda x: self._queue_command(command), 'interval', + seconds=worker_config['update_interval'], + args=[worker_name] + ) if 'topic_subscription' in worker_config: _LOGGER.debug("Subscribing to: %s" % worker_config['topic_subscription']) - mqtt_callbacks.append(( + self._mqtt_callbacks.append(( worker_config['topic_subscription'], - lambda client, _ , c: _WORKERS_QUEUE.put(cls.Command(worker_obj.on_command, [c.topic, c.payload])) + lambda client, _ , c: self._queue_command(self.Command(worker_obj.on_command, [c.topic, c.payload])) )) - scheduler.start() - mqtt.callbacks_subscription(mqtt_callbacks) + for (callback_name, options) in config['topic_subcription'].items(): + _LOGGER.debug("Subscribing to: %s with command: %s" % (options['topic'], callback_name)) + self._mqtt_callbacks.append(( + options['topic'], + lambda client, _ , c: self._queue_if_matching_payload(self.Command(getattr(self, callback_name)), c.payload, options['payload'])) + ) + + return self + + def start(self, mqtt): + self._scheduler.start() + mqtt.callbacks_subscription(self._mqtt_callbacks) + self.update_all() + + def _queue_if_matching_payload(self, command, payload, expected_payload): + if payload.decode('utf-8') == expected_payload: + self._queue_command(command) + + @property + def mqtt_callbacks(self): + return self._mqtt_callbacks + + def update_all(self): + _LOGGER.debug("Updating all workers") + for command in self._update_commands: + self._queue_command(command) + + @staticmethod + def _queue_command(command): + _WORKERS_QUEUE.put(command) @staticmethod def _pip_install_helper(package_names):