src/citadel/scene/__init__.py

@@ -1,18 +1,193 @@
-from .action import Action, Publish, TasmotaShutter
+import time
-from .scene import Scene
+import threading
+import queue
+import logging
+
+import paho.mqtt.client as mqtt
+
+from citadel.mqtt import Configuration
+
+class Scene(threading.Thread, mqtt.Client):
+    """ A scene run as a thread, has his own mqtt session
+
+    Can be activated by topic:
+
+        cmnd/scene/{mqtt_name}/activate ON|OFF|null
+
+    And return status on topic:
+
+        stat/scene/{mqtt_name}/state  ON|OFF
+
+    Special class attributes:
+
+    mqtt_name: define topic name, default class name in lower case
+
+    """
+    TOPIC_CMND = "cmnd/scene/{mqtt_name}/activate"
+    TOPIC_STAT = "stat/scene/{mqtt_name}/activate"
+
+    def __init__(self, config: Configuration, logger: logging.Logger=None):
+        threading.Thread.__init__(self)
+        mqtt.Client.__init__(self)
+
+        self.configuration = config
+        self._scene_state = False
+
+        if not hasattr(self, 'mqtt_name'):
+            self.mqtt_name = self.__class__.__name__.lower()
+
+        if not logger:
+            logger = logging.getLogger('scene.%s' % self.mqtt_name)
+        self.logger = logger
+
+        self._pub_queue = queue.Queue()
+
+    @property
+    def scene_state(self):
+        return self._scene_state
+
+    def run(self):
+        self.username_pw_set(self.configuration.user,
+                self.configuration.password)
+        self.connect(self.configuration.host,
+                self.configuration.port)
+
+        self.subscribe(self.TOPIC_CMND.format(mqtt_name=self.mqtt_name), 2)
+
+        for watch in self.WATCH:
+            self.subscribe(watch[0], 1)
+
+        run = True
+        while run:
+            self.loop()
+            try:
+                (args, kwargs) = self._pub_queue.get(block=False)
+                self.publish(*args, **kwargs)
+            except queue.Empty:
+                continue
+
+    def threaded_publish(self, *args, **kwargs):
+        self._pub_queue.put((args, kwargs))
+
+    def on_message(self, mqttc, obj, msg):
+        payload = msg.payload.decode().lower()
+        topic = msg.topic
+
+        # on CMND
+        if topic == self.TOPIC_CMND.format(mqtt_name=self.mqtt_name):
+            if payload == 'on':
+                self.logger.info('CMND activate ON')
+                self.__activate()
+            elif payload == 'off':
+                self.logger.info('CMND activate OFF')
+                self.__deactivate()
+            elif payload == '':
+                self.logger.info('CMND activate state')
+                self.__send_state()
+            else:
+                self.logger.warn('CMND activate invalide')
+
+        elif self.scene_state and self.__check_watched(topic, payload):
+            self.logger.info('Watched topic %s with invalid payload %s deactivating', topic, payload)
+            self.__deactivate()
+
+    def __check_watched(self, topic, payload):
+        """ return True if is a watched topic with an invalid payload """
+        for watch in self.WATCH:
+            if topic == watch[0]:
+                if payload != watch[1].lower():
+                    return True
+                return False
+        return False
+
+    def __activate(self):
+        #activate scene
+        # lock for activate again or deactivate
+        # start all activate action concurently
+        # start result catcher thread
+        # result catcher set active if ok else log error
+
+        results = queue.Queue()
+
+        for a in self.ON:
+            a.setup(self, results).start()
+
+        self._scene_state = True
+        self.publish(self.TOPIC_STAT.format(mqtt_name=self.mqtt_name),
+                payload='ON',
+                qos=2,
+                retain=False)
+
+    def __deactivate(self):
+        #deactivate scene
+        self._scene_state = False
+        self.publish(self.TOPIC_STAT.format(mqtt_name=self.mqtt_name),
+                payload='OFF',
+                qos=2,
+                retain=False)
+
+    def __send_state(self):
+        print(self._scene_state)
+        self.publish(self.TOPIC_STAT.format(mqtt_name=self.mqtt_name),
+                payload='ON' if self.scene_state else 'OFF',
+                qos=2,
+                retain=False)
+
+class Action(threading.Thread):
+    OK = 0
+    FAILED = 1
+
+    def __init__(self):
+        threading.Thread.__init__(self)
+
+    def setup(self, proxy: Scene, result: queue.Queue):
+        """
+        proxy: a thread safe proxy object with publish, subscribe method
+        result: a queue for put result
+        """
+        self.scene = proxy
+        self.result = result
+        return self
+
+    def ok(self):
+        self.result.put(Action.OK)
+
+    def failed(self):
+        self.result.put(Action.FAILED)
+
+class Publish(Action):
+
+    def __init__(self, topic: str):
+        Action.__init__(self)
+        self.topic = topic
+        self.payload = ''
+        self.qos = 0
+
+    def Payload(self, payload):
+        self.payload = payload
+        return self
+
+    def Qos(self, qos):
+        self.qos = qos
+        return self
+
+    def run(self):
+        self.scene.threaded_publish(self.topic,
+                payload=self.payload, qos=self.qos, retain=False)
+        self.ok()
+
 
 class Sleep(Scene):
 
     ON = [
-        Publish('cmnd/tasmota/screen/Screen/POWER').Payload('OFF').Qos(2)\
+        Publish('cmnd/tasmota/screen/Screen/POWER').Payload('OFF').Qos(2),
-                .CheckResponse('stat/tasmota/screen/Screen/POWER', 'OFF', 3),
         Publish('cmnd/tasmota/light/LivingroomFireplace/POWER').Payload('OFF').Qos(2),
-        TasmotaShutter('Bedroom1x1').Close(),
-        TasmotaShutter('Bedroom1x2').Close(),
-        TasmotaShutter('Staircase1').Close(),
-        TasmotaShutter('Bathroom').Close(),
     ]
 
     WATCH = [
             ('stat/tasmota/screen/Screen/POWER', 'OFF')
+            # ('cmnd/tasmota/shutter/Bedroom1x1/ShutterClose', ''),
+            # ('cmnd/tasmota/shutter/Bedroom1x2/ShutterClose', ''),
+            # ('cmnd/tasmota/shutter/Bathroom/ShutterClose', ''),
+            # ('cmnd/tasmota/shutter/Staicase1/ShutterClose', ''),
     ]

src/citadel/scene/__main__.py

@@ -4,7 +4,6 @@ import typer
 import systemd.journal
 
 from citadel.mqtt import Configuration
-
 from . import Sleep
 
 def main(\

src/citadel/scene/action.py

@@ -1,108 +0,0 @@
-import time
-
-from .helper import WaitForTopic
-
-class Action(object):
-    SUCCESS = 0
-    FAILED = 1
-
-    def setup(self, proxy):
-        """
-        proxy: a thread safe proxy object with publish, subscribe method
-        """
-        self.proxy = proxy
-        return self
-
-    def ok(self):
-        return Action.SUCCESS
-
-    def failed(self):
-        return Action.FAILED
-
-    @property
-    def name(self):
-        return self.__class__.__name__
-
-
-class Publish(Action):
-
-    def __init__(self, topic: str):
-        Action.__init__(self)
-        self.topic = topic
-        self.payload = ''
-        self.qos = 0
-
-        self.check_response = None
-
-    def Payload(self, payload):
-        self.payload = payload
-        return self
-
-    def Qos(self, qos):
-        self.qos = qos
-        return self
-
-    def CheckResponse(self, topic: str, payload: str=None, timeout: int=None):
-        self.check_response= {'topic': topic, 'payload':payload, 'timeout': timeout}
-        return self
-
-    def run(self):
-        checker = None
-        if self.check_response:
-            checker = WaitForTopic(self.proxy.configuration,
-                    self.check_response['topic'],
-                    self.check_response['payload'])
-
-        self.proxy.threaded_publish(self.topic,
-                payload=self.payload, qos=self.qos, retain=False)
-
-        if self.check_response:
-            checker.join(timeout=self.check_response['timeout'])
-            if checker.is_alive(): #timeout occur
-                checker.stop()
-                return self.failed()
-
-        return self.ok()
-
-class TasmotaShutter(Action):
-
-    TOPIC_FORMAT = "cmnd/tasmota/shutter/{shutter_name}/{action}"
-    TOPIC_CHECK_FORMAT = "stat/tasmota/shutter/{shutter_name}/RESULT"
-    ACTION_OPEN = 'ShutterOpen'
-    ACTION_CLOSE = 'ShutterClose'
-
-    def __init__(self, shutter_name: str):
-        Action.__init__(self)
-        self.shutter_name = shutter_name
-        self.action = None
-
-    def Close(self):
-        self.action = self.ACTION_CLOSE
-        return self
-
-    def Open(self):
-        self.action = self.ACTION_OPEN
-        return self
-
-    def run(self):
-        if self.action is None:
-            raise RuntimeError('No action specified')
-
-        topic = self.TOPIC_FORMAT.format(
-            shutter_name = self.shutter_name,
-            action = self.action
-            )
-
-        checker = WaitForTopic(self.proxy.configuration,
-                self.TOPIC_CHECK_FORMAT.format(shutter_name = self.shutter_name),
-                payload= '{"Shutter1":{"Position":100,"Direction":0,"Target":100}}' \
-                        if self.action == self.ACTION_OPEN \
-                        else '{"Shutter1":{"Position":0,"Direction":0,"Target":0}}')
-        time.sleep(1)
-        self.proxy.threaded_publish(topic, qos=1, retain=False)
-
-        checker.join(timeout=35)
-        if checker.is_alive():
-            return self.failed()
-
-        return self.ok()

src/citadel/scene/helper.py

@@ -1,65 +0,0 @@
-import threading
-import logging
-from typing import Union, Callable
-
-import paho.mqtt.client as mqtt
-
-from citadel.mqtt import Configuration
-
-class WaitForTopic(threading.Thread, mqtt.Client):
-
-    def __init__(self,
-            config: Configuration,
-            topic: str,
-            payload: Union[str, Callable[[str],bool]],
-            logger: logging.Logger=None):
-        """ payload: str or function stop if payload is received or function return True, the function must have this signature func(payload: str) -> bool """
-
-        threading.Thread.__init__(self)
-        mqtt.Client.__init__(self)
-
-        self.configuration = config
-
-        self.topic = topic
-
-        if callable(payload):
-            self.check_payload = payload
-        else:
-            self.check_payload = lambda x: x == payload
-
-        self.daemon = True
-
-        if not logger:
-            logger = logging.getLogger('watcher')
-        self.logger = logger
-
-        self.__run = True
-
-        self.username_pw_set(self.configuration.user,
-                self.configuration.password)
-        self.connect(self.configuration.host,
-                self.configuration.port)
-
-        self.subscribe(self.topic, 1)
-
-        self.start()
-
-    def on_message(self, mqttc, obj, msg):
-        payload = msg.payload.decode()
-        topic = msg.topic
-        self.logger.debug('Topic: %s, Payload: %s', topic, payload)
-
-        if topic != self.topic:
-            self.logger.error('Invaliad topic: %s', topic)
-            return
-
-        if self.check_payload(payload):
-            self.__run = False
-
-    def stop(self):
-        self.__run = False
-
-    def run(self):
-        while self.__run:
-            self.loop()
-        self.disconnect()

src/citadel/scene/scene.py

@@ -1,169 +0,0 @@
-import threading
-import logging
-import queue
-import concurrent.futures
-
-import pebble
-
-import paho.mqtt.client as mqtt
-
-from citadel.mqtt import Configuration
-
-from .action import Action
-
-class Scene(threading.Thread, mqtt.Client):
-    """ A scene run as a thread, has his own mqtt session
-
-    Can be activated by topic:
-
-        cmnd/scene/{mqtt_name}/activate ON|OFF|null
-
-    And return status on topic:
-
-        stat/scene/{mqtt_name}/state  ON|OFF
-
-    Special class attributes:
-
-    mqtt_name: define topic name, default class name in lower case
-
-    """
-    TOPIC_CMND = "cmnd/scene/{mqtt_name}/activate"
-    TOPIC_STAT = "stat/scene/{mqtt_name}/activate"
-
-    STATE_DEACTIVATE = 0
-    STATE_ACTIVATE = 1
-    STATE_ACTIVATING = 2
-
-    def __init__(self, config: Configuration, logger: logging.Logger=None):
-        threading.Thread.__init__(self)
-        mqtt.Client.__init__(self)
-
-        self.configuration = config
-
-        if not hasattr(self, 'mqtt_name'):
-            self.mqtt_name = self.__class__.__name__.lower()
-
-        if not logger:
-            logger = logging.getLogger('scene.%s' % self.mqtt_name)
-        self.logger = logger
-
-        self.__pub_queue = queue.Queue()
-        self.__state = Scene.STATE_DEACTIVATE
-
-        self.__executor = None
-        self.__activating_r_futures = []
-
-    @property
-    def scene_state(self):
-        return self.__state
-
-    def run(self):
-        self.username_pw_set(self.configuration.user,
-                self.configuration.password)
-        self.connect(self.configuration.host,
-                self.configuration.port)
-
-        self.subscribe(self.TOPIC_CMND.format(mqtt_name=self.mqtt_name), 2)
-
-        for watch in self.WATCH:
-            self.subscribe(watch[0], 1)
-
-        run = True
-        while run:
-            #Mqtt client loop
-            self.loop()
-
-            #Check publish queue
-            try:
-                (args, kwargs) = self.__pub_queue.get(block=False)
-                self.publish(*args, **kwargs)
-            except queue.Empty:
-                pass
-
-            # Check for futures if activating
-            if self.scene_state == Scene.STATE_ACTIVATING:
-                for (action, futures) in self.__activating_r_futures:
-                    try:
-                        r = futures.result(timeout=0)
-                        if r != Action.SUCCESS:
-                            self.logger.error("Action: '%s' error", action.name)
-                            self.logger.info('Activating error')
-                            self.__executor.join()
-                            self.__executor = None
-                            self.__activating_r_futures = []
-                            self.__state = Scene.STATE_DEACTIVATE
-                            self.__send_state()
-                            break
-                    except concurrent.futures.TimeoutError:
-                        break
-                else:
-                    self.logger.info('Activating terminate')
-                    self.__executor.join()
-                    self.__executor = None
-                    self.__activating_r_futures = []
-                    self.__state = Scene.STATE_ACTIVATE
-                    self.__send_state()
-
-
-    def threaded_publish(self, *args, **kwargs):
-        self.__pub_queue.put((args, kwargs))
-
-    def on_message(self, mqttc, obj, msg):
-        payload = msg.payload.decode().lower()
-        topic = msg.topic
-
-        # on CMND
-        if topic == self.TOPIC_CMND.format(mqtt_name=self.mqtt_name):
-            if payload == 'on':
-                self.logger.info('CMND activate ON')
-                self.__activate()
-            elif payload == 'off':
-                self.logger.info('CMND activate OFF')
-                self.__deactivate()
-            elif payload == '':
-                self.logger.info('CMND activate state')
-                self.__send_state()
-            else:
-                self.logger.warn('CMND activate invalide')
-
-        elif self.scene_state and self.__check_watched(topic, payload):
-            self.logger.info('Watched topic %s with invalid payload %s deactivating', topic, payload)
-            self.__deactivate()
-
-    def __check_watched(self, topic, payload):
-        """ return True if is a watched topic with an invalid payload """
-        for watch in self.WATCH:
-            if topic == watch[0]:
-                if payload != watch[1].lower():
-                    return True
-                return False
-        return False
-
-    def __activate(self):
-        if self.scene_state:
-            self.logger.info('Already activate/activating ignore cmnd')
-            return
-        if self.__executor != None:
-            self.logger.error('Trying to activate but __executor not None')
-            return
-
-        self.__state = Scene.STATE_ACTIVATING
-        self.__executor = pebble.ThreadPool()
-
-        for a in self.ON:
-            self.__activating_r_futures.append(
-                    (a, self.__executor.schedule(a.setup(self).run))
-            )
-
-        self.__executor.close()
-        self.__send_state()
-
-    def __deactivate(self):
-        self.__state = Scene.STATE_DEACTIVATE
-        self.__send_state()
-
-    def __send_state(self):
-        self.publish(self.TOPIC_STAT.format(mqtt_name=self.mqtt_name),
-                payload='ON' if self.scene_state else 'OFF',
-                qos=2,
-                retain=False)