From 896fa2aa6d26c0fdb1995e4755d4760e896702f2 Mon Sep 17 00:00:00 2001 From: Pierre Date: Thu, 22 Oct 2026 22:07:48 +0200 Subject: [PATCH] Removed the timeout warning of the smartprojector and pusshed all the command into a queue. --- apps/smartprojector_benq_over_telnet.py | 121 ++++++++++++------ tests/test_smartprojector_benq_over_telnet.py | 81 ++++++++++++ 2 files changed, 166 insertions(+), 36 deletions(-) create mode 100644 tests/test_smartprojector_benq_over_telnet.py diff --git a/apps/smartprojector_benq_over_telnet.py b/apps/smartprojector_benq_over_telnet.py index bed92ad..fe1adac 100644 --- a/apps/smartprojector_benq_over_telnet.py +++ b/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): - invalid_result = [ "*Unsupported Item#", "*Block Item#", "*Illegal format#" ] + def _process_commands(self): + while not self._command_stop.is_set(): + operation = self._command_queue.get() + try: + if operation is None or self._command_stop.is_set(): + return + method, args, kwargs = operation + try: + method(self, *args, **kwargs) + except Exception as error: + self.log_error(f"Projector operation {method.__name__} failed: {error}", dump_stack=True) + finally: + self._command_queue.task_done() - 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')) - output = session.read_until(b"#",timeout = 5).decode('ascii') - except (ConnectionResetError,EOFError): - self.log_warning(f"Connection Reset While sending command {cmd}. {try_count} try left") - if try_count <= 0: - self.log_error(f"Could not send command {cmd}",dump_stack = True) - return None - else: should_retry = True + 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() - - # opening session recursivly inside the with doesn't sound like a good idea - if should_retry: - if try_count == 1: - # I know calling sleep() is in callback is evil - # but this is a very small sleep time and working with event would make the code much much more complexe - time.sleep(1) - return self._send_cmd(cmd,log_command,try_count - 1) - - 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) - 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) + 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 - return output.strip("*").rstrip("#") \ No newline at end of file + try: + with telnetlib.Telnet(self.telnet_ip, self.telnet_port, timeout=2) as session: + if log_command: + self.log_info(f'Sending command "{cmd}"') + session.write(cmd.encode('utf-8')) + output = session.read_until(b"#", timeout=5).decode('ascii').strip() + if self._command_stop.is_set(): + return None + 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) + 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 diff --git a/tests/test_smartprojector_benq_over_telnet.py b/tests/test_smartprojector_benq_over_telnet.py new file mode 100644 index 0000000..b5e6b5e --- /dev/null +++ b/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()