diff --git a/config.yaml.example b/config.yaml.example index 83a21f7..9ef669b 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -9,6 +9,11 @@ manager: topic: homeassistant/status payload: online workers: + mysensors: + args: + port: /dev/ttyUSB0 + baudrate: 9600 + topic_prefix: mysensors/out thermostat: args: devices: diff --git a/workers/base.py b/workers/base.py index c1bd851..e88f5a8 100644 --- a/workers/base.py +++ b/workers/base.py @@ -1,5 +1,3 @@ -from abc import abstractmethod - class BaseWorker: def __init__(self, **args): for arg, value in args.items(): @@ -9,9 +7,5 @@ class BaseWorker: 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/mysensors.py b/workers/mysensors.py new file mode 100644 index 0000000..799b05b --- /dev/null +++ b/workers/mysensors.py @@ -0,0 +1,20 @@ +from mqtt import MqttMessage +from workers.base import BaseWorker +from logger import _LOGGER +import serial + +REQUIREMENTS = ['pyserial'] + + +class MysensorsWorker(BaseWorker): + def run(self, mqtt): + with serial.Serial(self.port, self.baudrate, timeout=10) as ser: + _LOGGER.debug("Starting mysensors at: %s" % ser.name) + while True: + line = ser.readline() + if not line: + continue + splited_line = self.format_topic(line.decode('utf-8').strip()).split(';') + topic = '/'.join(splited_line[0:-1]) + payload = ''.join(splited_line[-1]) + mqtt.publish([MqttMessage(topic=topic, payload=payload)]) diff --git a/workers_manager.py b/workers_manager.py index cc9a41b..09080d4 100644 --- a/workers_manager.py +++ b/workers_manager.py @@ -1,4 +1,6 @@ import importlib +import threading + import pip from apscheduler.schedulers.background import BackgroundScheduler from interruptingcow import timeout @@ -24,6 +26,7 @@ class WorkersManager: self._mqtt_callbacks = [] self._update_commands = [] self._scheduler = BackgroundScheduler() + self._daemons = [] def register_workers(self, config): for (worker_name, worker_config) in config['workers'].items(): @@ -46,6 +49,11 @@ class WorkersManager: seconds=worker_config['update_interval'], args=[worker_name] ) + elif hasattr(worker_obj, 'run'): + _LOGGER.debug("Registered: %s as daemon" % (worker_name)) + self._daemons.append(worker_obj) + else: + raise "%s cannot be initialized, it has to define run or status_update method" % worker_name if 'topic_subscription' in worker_config: _LOGGER.debug("Subscribing to: %s" % worker_config['topic_subscription']) @@ -68,15 +76,13 @@ class WorkersManager: mqtt.callbacks_subscription(self._mqtt_callbacks) self._scheduler.start() self.update_all() + for daemon in self._daemons: + threading.Thread(target=daemon.run, args=[mqtt], daemon=True).start() 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: