From 6a1c7b0181cc89899264afcccd944b82766bc865 Mon Sep 17 00:00:00 2001 From: zewelor Date: Sun, 26 Nov 2017 21:57:12 +0100 Subject: [PATCH] Support for commands handling --- config.yaml.example | 3 +- gateway.py | 23 +++-------- mqtt.py | 14 ++++++- workers/base.py | 17 ++++++++ workers/miscale.py | 10 ++--- workers/thermostat.py | 95 ++++++++++++++++++++++++++++++++++--------- workers_manager.py | 64 +++++++++++++++++------------ workers_queue.py | 4 ++ 8 files changed, 160 insertions(+), 70 deletions(-) create mode 100644 workers/base.py create mode 100644 workers_queue.py diff --git a/config.yaml.example b/config.yaml.example index 2b34739..d44e7f4 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -7,5 +7,6 @@ workers: args: devices: bedroom: 00:11:22:33:44:55 - topic_prefix: thermostats + topic_prefix: thermostat + topic_subscription: thermostat update_interval: 60 diff --git a/gateway.py b/gateway.py index 3378a50..1f27c7d 100644 --- a/gateway.py +++ b/gateway.py @@ -6,11 +6,10 @@ if sys.version_info < (3,4): from logger import _LOGGER import argparse -from queue import Queue +from workers_queue import _WORKERS_QUEUE from config import settings from mqtt import MqttClient from workers_manager import WorkersManager -from apscheduler.schedulers.background import BackgroundScheduler parser = argparse.ArgumentParser() parser.add_argument('-d', '--debug', action='store_true', default=False) @@ -21,31 +20,21 @@ if parsed.debug: else: _LOGGER.setLevel(logging.INFO) -queue = Queue() -scheduler = BackgroundScheduler() -mqtt = MqttClient(settings['mqtt']) -manager = WorkersManager(settings['workers']) - _LOGGER.info('Starting') -for worker_name, interval in manager.interval_enabled_workers.items(): - _LOGGER.debug("Added: %s with %d seconds interval" % (worker_name, interval)) - queue.put([worker_name]) - scheduler.add_job(lambda x: queue.put([x]), 'interval', seconds=interval, args=[worker_name]) - -scheduler.start() +mqtt = MqttClient(settings['mqtt']) +manager = WorkersManager(settings['workers'], mqtt) running = True try: while running: - for worker_name in queue.get(block=True): try: - mqtt.publish(manager.update(worker_name)) + mqtt.publish(_WORKERS_QUEUE.get(block=True).execute()) except (KeyboardInterrupt, SystemExit): raise - except: - _LOGGER.info('Some error') + except Exception as e: + _LOGGER.exception(e) except KeyboardInterrupt: running = False _LOGGER.info('Exiting allowing jobs to finish. If you need force exit use kill') diff --git a/mqtt.py b/mqtt.py index eedd83d..04bf241 100644 --- a/mqtt.py +++ b/mqtt.py @@ -1,12 +1,11 @@ -from logger import _LOGGER 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): - _LOGGER.debug(messages) publish.multiple(list(map(lambda m: m.as_dict, messages)), hostname=self.hostname, auth={'username': self.username, 'password': self.password}) @@ -23,6 +22,17 @@ class MqttClient: def password(self): return self._config['password'] + def callbacks_subscription(self, callbacks): + mqttc = mqtt.Client() + + mqttc.username_pw_set(self.username, self.password) + mqttc.connect(self.hostname) + + for topic, callback in callbacks: + mqttc.message_callback_add(topic, callback) + mqttc.subscribe(topic) + + mqttc.loop_start() class MqttMessage: def __init__(self, topic=None, payload=None): diff --git a/workers/base.py b/workers/base.py new file mode 100644 index 0000000..c1bd851 --- /dev/null +++ b/workers/base.py @@ -0,0 +1,17 @@ +from abc import abstractmethod + +class BaseWorker: + def __init__(self, **args): + for arg, value in args.items(): + setattr(self, arg, value) + self._setup() + + def _setup(self): + return + + @abstractmethod + def status_update(self, **args): + ... + + def format_topic(self, *args): + return '/'.join([self.topic_prefix, *args]) diff --git a/workers/miscale.py b/workers/miscale.py index 94192e3..c7518bf 100644 --- a/workers/miscale.py +++ b/workers/miscale.py @@ -2,23 +2,22 @@ import time from interruptingcow import timeout from mqtt import MqttMessage +from workers.base import BaseWorker REQUIREMENTS = ['bluepy'] + # Bluepy might need special settings # sudo setcap 'cap_net_raw,cap_net_admin+eip' /usr/local/lib/python3.6/dist-packages/bluepy/bluepy-helper -class MiscaleWorker(): - def __init__(self, mac): - self._mac = mac - +class MiscaleWorker(BaseWorker): def status_update(self): return [MqttMessage(topic='weight/kg', payload=self._get_weight())] def _get_weight(self): from bluepy import btle - scan_processor = ScanProcessor(self._mac) + scan_processor = ScanProcessor(self.mac) scanner = btle.Scanner().withDelegate(scan_processor) scanner.scan(5, passive=True) @@ -29,6 +28,7 @@ class MiscaleWorker(): return -1 + class ScanProcessor(): def __init__(self, mac): self._mac = mac diff --git a/workers/thermostat.py b/workers/thermostat.py index fea79c3..0cd8266 100644 --- a/workers/thermostat.py +++ b/workers/thermostat.py @@ -1,41 +1,98 @@ +from builtins import staticmethod from mqtt import MqttMessage +from workers.base import BaseWorker REQUIREMENTS = ['python-eq3bt'] monitoredAttrs = ["low_battery", "valve_state", "target_temperature"] -class ThermostatWorker: - def __init__(self, devices): - self._devices = devices +STATE_BOOST = 'boost' +STATE_AWAY = 'away' +STATE_MANUAL = 'manual' - def status_update(self): - from eq3bt import Thermostat, Mode +class ThermostatWorker(BaseWorker): + class ModesMapper(): + from eq3bt import Mode - haModesMapping = { + MAPPED_MODES = { Mode.Closed: 'off', Mode.Open: 'on', Mode.Auto: 'auto', - Mode.Manual: 'manual', - Mode.Away: 'away', - Mode.Boost: 'boost', + Mode.Manual: STATE_MANUAL, + Mode.Away: STATE_AWAY, + Mode.Boost: STATE_BOOST, } - ret = [] + REVERSED_MODES = {v: k for k, v in MAPPED_MODES.items()} + + @classmethod + def get_mapping(cls, mode): + return cls.MAPPED_MODES[mode] + + @classmethod + def get_reverse_mapping(cls, mode): + return cls.REVERSED_MODES[mode] + + @staticmethod + def away_mode_on_off(mode): + if mode == STATE_AWAY: + return 'ON' + else: + return 'OFF' + + @staticmethod + def on_off_to_mode(on_off): + if on_off == 'ON': + return STATE_AWAY + else: + return 'auto' + + + def _setup(self): + from eq3bt import Thermostat + for name, mac in self.devices.items(): - thermostat = Thermostat(mac) - thermostat.update() + self.devices[name] = Thermostat(mac) - for attr in monitoredAttrs: - ret.append(MqttMessage(topic=self.format_topic(name, attr), payload=getattr(thermostat, attr))) - ret.append(MqttMessage(topic=self.format_topic(name, 'mode'), payload=haModesMapping[thermostat.mode])) - ret.append(MqttMessage(topic=self.format_topic(name, 'away'), payload='ON' if thermostat.mode == Mode.Away else 'OFF')) + def status_update(self): + ret = [] + for name, thermostat in self.devices.items(): + ret += self.update_device_state(name, thermostat) return ret - @property - def devices(self): - return self._devices + def on_command(self, topic, value): + _, device_name, method = topic.split('/') + thermostat = self.devices[device_name] + + if method == STATE_AWAY: + method = "mode" + value = ModesMapper.on_off_to_mode(value) + + # It needs to be on separate if because first if can change method + if method == "mode": + value = ModesMapper.get_reverse_mapping(value) + elif method == "target_temperature": + value = float(value) + + setattr(thermostat, method, value) + return self.update_device_state(device_name, thermostat) + + def update_device_state(self, name, thermostat): + thermostat.update() + + ret = [] + for attr in monitoredAttrs: + ret.append(MqttMessage(topic=self.format_topic(name, attr), payload=getattr(thermostat, attr))) + + ret.append(MqttMessage(topic=self.format_topic(name, 'mode'), payload=self.ModesMapper.get_mapping(thermostat.mode))) + ret.append(MqttMessage(topic=self.format_topic(name, 'away'), payload=self.ModesMapper.away_mode_on_off(thermostat.mode))) + + return ret + + def device_for(self, mac): + return eq3bt.Thermostat(mac) def format_topic(self, device, attr): return '/'.join([device, attr]) diff --git a/workers_manager.py b/workers_manager.py index dd7b2a1..6e65261 100644 --- a/workers_manager.py +++ b/workers_manager.py @@ -1,50 +1,62 @@ import importlib import pip +from apscheduler.schedulers.background import BackgroundScheduler from interruptingcow import timeout +from logger import _LOGGER +from workers_queue import _WORKERS_QUEUE class WorkersManager: - def __init__(self, config): - self._config = config - self._interval_enabled_workers = {} - self._workers = self._register_klasses() + class Command: + def __init__(self, callback, args=[], options={}): + self._callback = callback + self._args = args + self._options = options - def update(self, worker_name): - messages = [] - with timeout(30): - messages = self.workers[worker_name].status_update() - for message in messages: - message.topic = '/'.join([self._config[worker_name]['topic_prefix'], message.topic]) + def execute(self): + messages = [] + with timeout(30): + messages = self._callback(*self._args) - return messages + _LOGGER.debug(messages) + return messages + + def __init__(self, config, mqtt): + self._mqtt = mqtt + self._workers = self._register_klasses(config) @property def workers(self): return self._workers - @property - def interval_enabled_workers(self): - return self._interval_enabled_workers - - def _register_klasses(self): + def _register_klasses(self, config): arr = {} + scheduler = BackgroundScheduler() + mqtt_callbacks = [] - for worker_klass in self._config.keys(): - moduleObj = importlib.import_module("workers.%s" % worker_klass) - klass = getattr(moduleObj, "%sWorker" % worker_klass.title()) + for (worker_name, worker_config) in config.items(): + moduleObj = importlib.import_module("workers.%s" % worker_name) + klass = getattr(moduleObj, "%sWorker" % worker_name.title()) + worker_obj = klass(**worker_config['args']) if moduleObj.REQUIREMENTS is not None: self._pip_install_helper(moduleObj.REQUIREMENTS) - if self._config[worker_klass]['update_interval'] is not None: - self._interval_enabled_workers[worker_klass] = self._config[worker_klass]['update_interval'] + if 'update_interval' in worker_config: + _LOGGER.debug("Added: %s with %d seconds interval" % (worker_name, worker_config['update_interval'])) + _WORKERS_QUEUE.put(self.Command(worker_obj.status_update, [])) + scheduler.add_job(lambda x: _WORKERS_QUEUE.put(self.Command(worker_obj.status_update, [])), 'interval', seconds=worker_config['update_interval'], args=[worker_name]) - arr[worker_klass] = klass(**self._args_for(worker_klass)) + if 'topic_subscription' in worker_config: + _LOGGER.debug("Subscribing to: %s" % worker_config['topic_subscription'] + '/#') + mqtt_callbacks.append((worker_config['topic_subscription'] + '/#', lambda client, _ , c: _WORKERS_QUEUE.put(self.Command(worker_obj.on_command, [c.topic, c.payload])))) + + arr[worker_name] = worker_obj + + scheduler.start() + self._mqtt.callbacks_subscription(mqtt_callbacks) return arr - def _args_for(self, klass): - return self._config[klass]['args'] - def _pip_install_helper(self, package_names): for package in package_names: - pip.main(['install','-q', package]) + pip.main(['install', '-q', package]) diff --git a/workers_queue.py b/workers_queue.py new file mode 100644 index 0000000..cb58542 --- /dev/null +++ b/workers_queue.py @@ -0,0 +1,4 @@ +from queue import Queue + +_WORKERS_QUEUE = Queue() +