Removed the timeout warning of the smartprojector and pusshed all the command into a queue.

apps/smartprojector_benq_over_telnet.py

@@ -3,15 +3,35 @@ from smartswitch import SmartSwitch
 from ad_toolbox.eventhandler import EventHandler
 import ad_toolbox.smartcondition as SmartCondition
 import telnetlib
-import time
+import queue
+import threading
+from functools import wraps
 import datetime
 
+def queued_operation(method):
+    """Keep compound operations and their retries together in the FIFO worker."""
+    @wraps(method)
+    def enqueue(self, *args, **kwargs):
+        if self._command_stop.is_set():
+            return None
+        if threading.current_thread() is self._command_worker:
+            return method(self, *args, **kwargs)
+        self._command_queue.put((method, args, kwargs))
+        return None
+    return enqueue
+
+
 class SmartProjector(SmartSwitch): 
 
     WARM_UP_TIME = 35
     def get_default_entity_state(self): return 'off'
 
     def on_initialize_smart_object(self):
+        # Queue callbacks during initialization; start only once setup is complete.
+        self._command_queue = queue.Queue()
+        self._command_stop = threading.Event()
+        self._command_worker = threading.Thread(
+            target=self._process_commands, name=f"{self.name}-telnet", daemon=True)
         #those variable are use withing get_default_entity_state()
         self.telnet_ip = self.args['telnet_connection']['ip']
         self.telnet_port = self.args['telnet_connection']['port']
@@ -63,6 +83,7 @@ class SmartProjector(SmartSwitch):
         self.reset_update_state_cb()
         #self.entity.listen_state(self.on_state_change) 
         self.update_state()
+        self._command_worker.start()
 
     def on_power_change(self, entity, attribute, old, new, kwargs):
         self.update_state()
@@ -91,6 +112,7 @@ class SmartProjector(SmartSwitch):
 
     def is_available(self): return self.entity.get_state() != 'unavailable'
 
+    @queued_operation
     def switch_on(self):
         if self.is_available():
             self.log(f"Turn on {self.entity_id}")
@@ -111,6 +133,7 @@ class SmartProjector(SmartSwitch):
         else:
             self.log_warning("Turn On ignored as the projector is unavailable")
     
+    @queued_operation
     def switch_off(self):
         if self.is_available():
             self.log(f"Turn off {self.entity_id}")
@@ -140,6 +163,7 @@ class SmartProjector(SmartSwitch):
                     else: raise
         else: return 'off'
 
+    @queued_operation
     def on_update_picture_mode(self):
         for evaluator in self.picture_mode_evaluators:
             if evaluator[0].evaluate(False) == SmartCondition.Result.Succeeded:
@@ -155,6 +179,7 @@ class SmartProjector(SmartSwitch):
                     self.entity.set_state(attributes = self.attributes)
                 break
 
+    @queued_operation
     def on_update_lamp_mode(self): 
         for evaluator in self.lamp_mode_evaluators:
             if evaluator[0].evaluate(False) == SmartCondition.Result.Succeeded:
@@ -163,6 +188,7 @@ class SmartProjector(SmartSwitch):
                     result = self.send_cmd(f"lampm={evaluator[1]}",True)
                 break
 
+    @queued_operation
     def update_state(self,kwargs = None):
         new_state = self.get_projector_state()
         
@@ -171,7 +197,7 @@ class SmartProjector(SmartSwitch):
         
         if new_state == 'on':
             # the projector is not responding during power on sequence
-            if self.attributes['warming_up'] == 'off':
+            if not self.attributes['warming_up']:
                 cmd_result = self.send_cmd("appmod=?")
                 if cmd_result: self.attributes['picture_mode'] = cmd_result.split("=")[1]
                 else: self.attributes['picture_mode'] = "unavailable"
@@ -193,6 +219,7 @@ class SmartProjector(SmartSwitch):
 
             self.attributes['warming_up'] = new_value
 
+    @queued_operation
     def on_warm_up_done(self,*kwargs):
         self.warming_up_done_cb_handle = None
         self.attributes['warming_up'] = False
@@ -201,48 +228,70 @@ class SmartProjector(SmartSwitch):
     def on_press_enter_event(self, event_name, data, event_context):
         self.press_enter()
 
+    @queued_operation
     def press_enter(self):
         if self.entity.get_state() == 'on':
             self.send_cmd("enter")
 
+    @queued_operation
     def send_cmd(self,cmd, log_command = False): 
         cmd = f"*{cmd}#"
         return self._send_cmd(cmd,log_command,3)
 
-    def _send_cmd(self,cmd,log_command,try_count):
+    def _process_commands(self):
-        invalid_result = [ "*Unsupported Item#", "*Block Item#", "*Illegal format#" ]
+        while not self._command_stop.is_set():
-
+            operation = self._command_queue.get()
-        should_retry = False
-        with telnetlib.Telnet(self.telnet_ip,self.telnet_port) as session:
-            if log_command: self.log_info(f'Sending command "{cmd}"')
             try:
-                session.write(bytes(cmd,'utf-8'))
+                if operation is None or self._command_stop.is_set():
-                output = session.read_until(b"#",timeout = 5).decode('ascii')
+                    return
-            except (ConnectionResetError,EOFError): 
+                method, args, kwargs = operation
-                self.log_warning(f"Connection Reset While sending command {cmd}. {try_count} try left")
+                try:
-                if try_count <= 0: 
+                    method(self, *args, **kwargs)
-                    self.log_error(f"Could not send command {cmd}",dump_stack = True)
+                except Exception as error:
+                    self.log_error(f"Projector operation {method.__name__} failed: {error}", dump_stack=True)
+            finally:
+                self._command_queue.task_done()
+
+    def terminate(self):
+        # Discard pending work and interrupt retry delays. An active exchange is
+        # bounded by the connection and read timeouts (2 + 5 seconds).
+        self._command_stop.set()
+        self._command_queue.put(None)
+        if self._command_worker.is_alive():
+            self._command_worker.join(timeout=8)
+        super().terminate()
+
+    def _send_cmd(self, cmd, log_command, try_count):
+        invalid_result = {"*Unsupported Item#", "*Block Item#", "*Illegal format#"}
+        for attempt in range(try_count + 1):
+            if self._command_stop.is_set():
                 return None
-                else: should_retry = True
+            try:
-
+                with telnetlib.Telnet(self.telnet_ip, self.telnet_port, timeout=2) as session:
-
+                    if log_command:
-        # opening session recursivly inside the with doesn't sound like a good idea
+                        self.log_info(f'Sending command "{cmd}"')
-        if should_retry: 
+                    session.write(cmd.encode('utf-8'))
-            if try_count == 1: 
+                    output = session.read_until(b"#", timeout=5).decode('ascii').strip()
-                # I know calling sleep() is in callback is evil 
+                if self._command_stop.is_set():
-                # but this is a very small sleep time and working with event would make the code much much more complexe
+                    return None
-                time.sleep(1) 
+                if log_command:
-            return self._send_cmd(cmd,log_command,try_count - 1)
+                    self.log_info(f"Result = {output}")
-            
-        if log_command: self.log_info(f"Result = {output}")
                 if output in invalid_result:
-            self.log_error(f"Command rejected. {cmd} returned {output}",dump_stack = True)
+                    self.log_error(f"Command rejected. {cmd} returned {output}", dump_stack=True)
-            return None
-        if not output:
-            if try_count > 0: 
-                self.log_warning(f"Command output is empty. {cmd}. {try_count} try left")
-                return self._send_cmd(cmd,log_command,try_count - 1)
-            else:
-                self.log_error(f"Command output is empty. {cmd}. {try_count} try left",dump_stack = True)
                     return None
+                # read_until may return a partial response when its timeout expires.
+                if output.endswith("#"):
                     return output.strip("*").rstrip("#")
+                reason = f"empty or incomplete response: {output!r}"
+            except (OSError, EOFError) as error:
+                if self._command_stop.is_set():
+                    return None
+                if isinstance(error, OSError) and error.errno == 113:
+                    raise
+                reason = str(error)
+            remaining = try_count - attempt
+            self.log_info(f"Command {cmd} failed ({reason}); {remaining} retries left")
+            if remaining and self._command_stop.wait(1):
+                return None
+        self.log_error(f"Could not send command {cmd}", dump_stack=True)
+        return None

tests/test_smartprojector_benq_over_telnet.py

@@ -0,0 +1,81 @@
+"""Exercise the real FIFO and Telnet code without an AppDaemon installation."""
+import ast
+from functools import wraps
+from pathlib import Path
+import queue
+import threading
+import types
+import unittest
+
+
+source = Path(__file__).resolve().parents[1] / 'apps/smartprojector_benq_over_telnet.py'
+tree = ast.parse(source.read_text())
+tree.body = [node for node in tree.body if isinstance(node, (ast.ClassDef, ast.FunctionDef))]
+namespace = dict(SmartSwitch=object, queue=queue, threading=threading, wraps=wraps)
+exec(compile(tree, str(source), 'exec'), namespace)
+SmartProjector = namespace['SmartProjector']
+
+
+class ProjectorQueueTests(unittest.TestCase):
+    def setUp(self):
+        self.app = SmartProjector()
+        self.app._command_queue = queue.Queue()
+        self.app._command_stop = threading.Event()
+        self.app._command_worker = threading.Thread(target=self.app._process_commands)
+        self.app.telnet_ip, self.app.telnet_port = 'projector', 8000
+        self.app.log_info = self.app.log_warning = self.app.log_error = lambda *a, **k: None
+        self.app._command_worker.start()
+
+    def tearDown(self):
+        self.app._command_stop.set()
+        self.app._command_queue.put(None)
+        self.app._command_worker.join(2)
+        self.assertFalse(self.app._command_worker.is_alive())
+
+    def test_enqueue_returns_while_network_blocked_and_retries_stay_before_off(self):
+        entered, release = threading.Event(), threading.Event()
+        sent = []
+
+        class Session:
+            def __init__(self, *args, **kwargs):
+                assert kwargs['timeout'] == 2
+            def __enter__(self): return self
+            def __exit__(self, *args): pass
+            def write(self, data): sent.append(data)
+            def read_until(self, *args, **kwargs):
+                if len(sent) == 1:
+                    entered.set()
+                    if not release.wait(2): raise AssertionError('test gate timed out')
+                    return b''
+                return b'*POW=ON#' if sent[-1] == b'*pow=on#' else b'*POW=OFF#'
+
+        namespace['telnetlib'] = types.SimpleNamespace(Telnet=Session)
+        self.assertIsNone(self.app.send_cmd('pow=on'))
+        self.assertTrue(entered.wait(2))
+        self.assertIsNone(self.app.send_cmd('pow=off'))
+        self.assertEqual(sent, [b'*pow=on#'])
+        release.set()
+        self.app._command_queue.join()
+        self.assertEqual(sent, [b'*pow=on#', b'*pow=on#', b'*pow=off#'])
+
+    def test_failed_operation_does_not_stop_queue(self):
+        completed = threading.Event()
+        def fail(app): raise RuntimeError('test failure')
+        self.app._command_queue.put((fail, (), {}))
+        self.app._command_queue.put((lambda app: completed.set(), (), {}))
+        self.assertTrue(completed.wait(2))
+
+    def test_nested_commands_execute_within_current_operation(self):
+        order = []
+        self.app._send_cmd = lambda cmd, *args: order.append(cmd)
+        def compound(app):
+            app.send_cmd('appmod=?')
+            app.send_cmd('appmod=cinema')
+        self.app._command_queue.put((compound, (), {}))
+        self.app.send_cmd('pow=off')
+        self.app._command_queue.join()
+        self.assertEqual(order, ['*appmod=?#', '*appmod=cinema#', '*pow=off#'])
+
+
+if __name__ == '__main__':
+    unittest.main()