Cleanups and add support for mysensors

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:

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])

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)])

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: