From 1fd4dab91697ae28779a2a7e6aef261d7cd1567a Mon Sep 17 00:00:00 2001 From: Pierre Date: Sat, 17 Oct 2026 22:59:17 +0200 Subject: [PATCH] Added support for power plug and spam sensor --- apps/vertex2.py | 312 +++++++++++++++++++++++--- tests/test_vertex2.py | 506 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 793 insertions(+), 25 deletions(-) create mode 100644 tests/test_vertex2.py diff --git a/apps/vertex2.py b/apps/vertex2.py index db4c199..690b831 100644 --- a/apps/vertex2.py +++ b/apps/vertex2.py @@ -1,6 +1,6 @@ import appdaemon.plugins.hass.hassapi as hass from requests import get -from requests.exceptions import Timeout,ConnectionError +from requests.exceptions import Timeout,ConnectionError,RequestException from ad_toolbox.smartobject import SmartObject import ad_toolbox.smartcondition as SmartCondition import time @@ -15,6 +15,24 @@ class Vertex2(SmartObject): self._stop_interrupt_listener() self._interrupt_stop = threading.Event() self._status_requested = threading.Event() + self._last_logged_status = None + self._spam_lock = threading.RLock() + self._spam_last_interrupt = None + self._spam_snapshot = None + self._spam_count = 0 + self._spam_timer = None + self._spam_deadline = None + self.no_signal_spam_sensor = None + if self.args.get('no_signal_spam_sensor'): + self.no_signal_spam_sensor = self.create_entity( + self.args['no_signal_spam_sensor'], state='off') + self._connection_lock = threading.Lock() + self._interrupt_connection = None + self.startup_delay = max(0, float(self.args.get('startup_delay', 30))) + self._startup_deadline = 0 + self._startup_commands = [] + self.power_plug_entity = self.args.get("vertex2_power_plug", {}).get("entity") + self._power_allowed = self.power_plug_entity is None self._command_processing_lock = threading.Lock() self._command_processing_handle = None self._command_processing_deadline = None @@ -44,6 +62,11 @@ class Vertex2(SmartObject): ) for index in range(2) } + self.plus5_sensors = { + index: self.create_entity(self.args[key], state='unavailable') + for index, key in enumerate(('tx0_plus5_sensor', 'tx1_plus5_sensor')) + if self.args.get(key) + } self._tx_video_signal_values = {0: 'unavailable', 1: 'unavailable'} self.audio_out_signal_sensor = self.create_entity( @@ -99,7 +122,11 @@ class Vertex2(SmartObject): state='off', ) - self.power_plug_entity = None + if "vertex2_power_plug" in self.args: + self.power_plug_entity = self.args["vertex2_power_plug"]["entity"] + self.power_plug_conditions_evaluator = SmartCondition.Evaluator(self,self.args["vertex2_power_plug"]["conditions"],condition_name = 'vertex2_power_plug',on_change_cb = self.on_change_power_plug_conditions) + if self.power_plug_entity is not None: + self.listen_state(self.on_power_plug_state_change, self.power_plug_entity) if "filmmaker_mode_conditions" in self.args: self.filmmaker_mode_conditions = SmartCondition.Evaluator(self,self.args['filmmaker_mode_conditions'], condition_name = "filmmaker_mode_conditions",on_change_cb = self.on_filmmaker_mode_conditions_change) @@ -152,9 +179,6 @@ class Vertex2(SmartObject): )) self.on_update_output_selection() - if "vertex2_power_plug" in self.args: - self.power_plug_entity = self.args["vertex2_power_plug"]["entity"] - self.power_plug_conditions_evaluator = SmartCondition.Evaluator(self,self.args["vertex2_power_plug"]["conditions"],condition_name = 'vertex2_power_plug',on_change_cb = self.on_change_power_plug_conditions) self.vertex2_tcp_port = int(self.args.get('vertex2_tcp_port', 2220)) self._interrupt_thread = threading.Thread( target=self._interrupt_listener, @@ -165,6 +189,65 @@ class Vertex2(SmartObject): self.update_vertex2_states() self._interrupt_thread.start() + def _reset_no_signal_spam(self, state='off'): + if getattr(self, 'no_signal_spam_sensor', None) is None: + return + with self._spam_lock: + if self._spam_timer is not None: + self.cancel_timer(self._spam_timer) + self._spam_timer = None + self._spam_deadline = None + self._spam_last_interrupt = None + self._spam_snapshot = None + self._spam_count = 0 + if self.no_signal_spam_sensor.get_state() != state: + self.no_signal_spam_sensor.set_state(state=state) + + def _observe_no_signal_spam(self, status): + if getattr(self, 'no_signal_spam_sensor', None) is None: + return + with self._spam_lock: + missing = ('', 'no signal') + if any(status[field].strip().lower() not in missing for field in ('rx0', 'tx0')): + self._reset_no_signal_spam() + return + snapshot = tuple(status[field].strip() for field in + ('rx0', 'rx1', 'tx0', 'tx1', 'audout', 'insel')) + if self._spam_snapshot is not None and snapshot != self._spam_snapshot: + self._reset_no_signal_spam() + at = status.get('interrupt_at') + if at is None: + if self.no_signal_spam_sensor.get_state() == 'unavailable': + self.no_signal_spam_sensor.set_state(state='off') + return # Polls and commands do not count as device interrupts. + now = time.monotonic() + if now - at >= 4: + return # Do not revive a streak with a delayed callback. + previous = self._spam_last_interrupt + self._spam_last_interrupt = at + snapshot = tuple(status[field].strip() for field in + ('rx0', 'rx1', 'tx0', 'tx1', 'audout', 'insel')) + self._spam_count = (self._spam_count + 1 + if previous is not None and 2 <= at - previous <= 4 + and snapshot == self._spam_snapshot else 1) + self._spam_snapshot = snapshot + state = 'on' if self._spam_count >= 3 else 'off' + if self.no_signal_spam_sensor.get_state() != state: + self.no_signal_spam_sensor.set_state(state=state) + if self._spam_timer is not None: + self.cancel_timer(self._spam_timer) + self._spam_deadline = at + 4 + self._spam_timer = self.run_in( + self._expire_no_signal_spam, max(0, self._spam_deadline - now), + deadline=self._spam_deadline) + + def _expire_no_signal_spam(self, kwargs): + with self._spam_lock: + if kwargs['deadline'] != self._spam_deadline: + return + self._spam_timer = None + self._reset_no_signal_spam() + def _interrupt_listener(self, interrupt_stop): address = urlparse(self.args['vertex2_address']) host = address.hostname @@ -173,10 +256,15 @@ class Vertex2(SmartObject): expected = set(status_fields) | {'insel'} while not interrupt_stop.is_set(): + if not self._vertex2_ready(): + interrupt_stop.wait(0.5) + continue try: with socket.create_connection((host, self.vertex2_tcp_port), timeout=10) as connection: - if interrupt_stop.is_set(): - break + with self._connection_lock: + self._interrupt_connection = connection + if interrupt_stop.is_set() or not self._vertex2_powered(): + continue # Detect a silent reboot/network loss without polling HDMI status. connection.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 10) @@ -188,12 +276,16 @@ class Vertex2(SmartObject): buffer = '' status = {} status_started = None + reconfigure = True + pending_interrupt_at = None - while not interrupt_stop.is_set(): + while not interrupt_stop.is_set() and self._vertex2_powered(): if status_started is None and self._status_requested.is_set(): self._status_requested.clear() status = {} status_started = time.monotonic() + request_interrupt_at = pending_interrupt_at + pending_interrupt_at = None for command in commands: connection.sendall((command + '\r').encode('ascii')) @@ -204,7 +296,7 @@ class Vertex2(SmartObject): raise OSError('Vertex2 TCP status response timed out') continue - if interrupt_stop.is_set(): + if interrupt_stop.is_set() or not self._vertex2_powered(): break if not data: raise OSError('Vertex2 closed the IP INTERRUPT connection') @@ -217,8 +309,7 @@ class Vertex2(SmartObject): line, buffer = buffer.split('\n', 1) line = line.strip() if line == 'int': - self._refresh_command_processing() - self.log_info('Vertex2 IP INTERRUPT received') + pending_interrupt_at = time.monotonic() self._status_requested.set() elif status_started is not None: if line.startswith('insel '): @@ -235,7 +326,10 @@ class Vertex2(SmartObject): self.run_in(self._apply_vertex2_status, 0, rx0=snapshot['rx0'], rx1=snapshot['rx1'], tx0=snapshot['tx0'], tx1=snapshot['tx1'], - audout=snapshot['audout'], insel=snapshot['insel']) + audout=snapshot['audout'], insel=snapshot['insel'], + reconfigure=reconfigure, + interrupt_at=request_interrupt_at) + reconfigure = False status_started = None if status_started is not None and time.monotonic() - status_started > 5: @@ -244,6 +338,10 @@ class Vertex2(SmartObject): except (OSError, ValueError) as error: if interrupt_stop.is_set(): break + if not self._vertex2_powered(): + continue + self._reset_no_signal_spam('unavailable') + self._invalidate_plus5_sensors() self.unreachable_sensor.set_state(state='on') self.video_tx_signal_sensor.set_state(state='unavailable') for sensor in self.video_tx_signal_sensors.values(): @@ -254,11 +352,29 @@ class Vertex2(SmartObject): self.audio_out_signal_sensor.set_state(state='unavailable') self.log_warning(f'Vertex2 IP INTERRUPT unavailable: {error}') interrupt_stop.wait(5) + finally: + with self._connection_lock: + self._interrupt_connection = None + + def _close_interrupt_connection(self): + connection_lock = getattr(self, "_connection_lock", None) + if connection_lock is None: + return + with connection_lock: + connection = self._interrupt_connection + if connection is not None: + try: + connection.shutdown(socket.SHUT_RDWR) + except OSError: + pass + connection.close() def _stop_interrupt_listener(self): + self._reset_no_signal_spam('unavailable') interrupt_stop = getattr(self, '_interrupt_stop', None) if interrupt_stop is not None: interrupt_stop.set() + self._close_interrupt_connection() interrupt_thread = getattr(self, '_interrupt_thread', None) if interrupt_thread is not None: # Allow the 10s connect timeout or 1s recv timeout to finish. @@ -296,6 +412,7 @@ class Vertex2(SmartObject): except KeyError: pass def on_update_input_selection(self): + input_selected = None for evaluator in self.input_selection_evaluators: if evaluator[0].evaluate(False) == SmartCondition.Result.Succeeded: evaluator[0].log_evaluation_result() @@ -304,7 +421,7 @@ class Vertex2(SmartObject): self.log_info("Input selection conditions changed") - if input_selected != "None": + if input_selected is not None and str(input_selected) != "None": self.select_input(input_selected) self.update_vertex2_states() @@ -384,17 +501,101 @@ class Vertex2(SmartObject): else: self.toggle_filmmaker_mode(False) + def _vertex2_powered(self): + return self._power_allowed and ( + self.power_plug_entity is None + or self.get_state(self.power_plug_entity) == 'on' + ) + + def _vertex2_ready(self): + return self._vertex2_powered() and time.monotonic() >= self._startup_deadline + + def _begin_startup(self): + self._startup_deadline = time.monotonic() + self.startup_delay + self.log_info(f'Waiting {self.startup_delay:g}s for Vertex2 to start') + + def _flush_startup_commands(self): + with self._connection_lock: + commands = self._startup_commands + self._startup_commands = [] + for cmd, value in commands: + self.send_cmd(cmd, value) + + def _reconfigure_vertex2(self): + # Conditions may not have changed while the device was powered off. + # Evaluate them again instead of relying on change callbacks or old commands. + if self.power_plug_entity is not None: + result = self.power_plug_conditions_evaluator.evaluate(False) + self.on_change_power_plug_conditions(None, result) + if not self._vertex2_ready(): + return + + self.log_info('Reapplying Vertex2 configuration after connection restored') + if self.input_selection_evaluators: + self.on_update_input_selection() + if self.output_selection_evaluators: + # Also reapplies audio/video EDID and audio mute for the selected output. + self.on_update_output_selection() + else: + selected_output = self.output_names.get(self.output_selected_sensor.get_state()) + if selected_output in (0, 1): + self.apply_output_audio_settings(selected_output) + + for output_index, conditions in self.output_config_conditions.items(): + evaluator = conditions.get('force_plus5') + if evaluator is not None: + self.on_output_plus5_condition_change( + output_index, None, evaluator.evaluate(False), + ) + if 'filmmaker_mode_conditions' in self.args: + self.on_filmmaker_mode_conditions_change( + None, self.filmmaker_mode_conditions.evaluate(False), + ) + + def _suspend_vertex2(self): + self._reset_no_signal_spam() + self._invalidate_plus5_sensors() + self._close_interrupt_connection() + with self._connection_lock: + self._startup_commands.clear() + self._startup_deadline = 0 + self._status_requested.clear() + self._update_output_video_sensors('', '') + self.audio_out_signal_sensor.set_state(state='unavailable') + self.unreachable_sensor.set_state(state='off') + with self._command_processing_lock: + self._command_processing_deadline = None + if self._command_processing_handle is not None: + self.cancel_timer(self._command_processing_handle) + self._command_processing_handle = None + if self.command_processing_sensor is not None: + self.command_processing_sensor.set_state(state='off') + + def on_power_plug_state_change(self, entity, attribute, old, new, kwargs): + if self._vertex2_powered(): + if old != 'on' and new == 'on': + self._begin_startup() + self.update_vertex2_states() + else: + self._suspend_vertex2() + def on_change_power_plug_conditions(self,prev_result,result): - if result == SmartCondition.Result.Succeeded: + self._power_allowed = result == SmartCondition.Result.Succeeded + if self._power_allowed: + if self.get_state(self.power_plug_entity) != 'on': + self._begin_startup() self.turn_on(self.power_plug_entity) else: + # Close TCP before asking Home Assistant to cut the power. + self._suspend_vertex2() self.turn_off(self.power_plug_entity) - def _refresh_command_processing(self, start=False): + def _start_command_processing(self, only_if_active=False): + # Only commands start a window; changed status may extend an active one. if self.command_processing_sensor is None: return with self._command_processing_lock: - if not start and self.command_processing_sensor.get_state() != 'on': + if only_if_active and self._command_processing_deadline is None: return if self._command_processing_handle is not None: self.cancel_timer(self._command_processing_handle) @@ -417,8 +618,16 @@ class Vertex2(SmartObject): self.command_processing_sensor.set_state(state='off') def send_cmd(self,cmd,value = ''): + if not self._vertex2_powered(): + return + if not self._vertex2_ready(): + with self._connection_lock: + self._startup_commands.append((cmd, value)) + return request = f'{self.args["vertex2_address"]}/cmd?{cmd}={value}' - self._refresh_command_processing(start=True) + if cmd in ('tx0plus5', 'tx1plus5'): + self._invalidate_plus5_sensors() + self._start_command_processing() try: r = get(request, timeout=5) if cmd == 'edid' and value.startswith(('audiotx', 'audioout')): @@ -429,6 +638,43 @@ class Vertex2(SmartObject): except (Timeout,ConnectionError): self.log_warning(f"{request} is unreachable") + finally: + if cmd in ('tx0plus5', 'tx1plus5'): + # Read back the device setting even if no IP interrupt follows. + self._status_requested.set() + + def _invalidate_plus5_sensors(self): + for sensor in self.plus5_sensors.values(): + if sensor.get_state() != 'unavailable': + sensor.set_state(state='unavailable') + + def _refresh_plus5_sensors(self): + if not self.plus5_sensors: + return + if not self._vertex2_ready(): + self._invalidate_plus5_sensors() + return + try: + # Same read-only endpoint used by the Vertex2 configuration page. + response = get(f'{self.args["vertex2_address"].rstrip("/")}/ssi/confpage.ssi', timeout=5) + response.raise_for_status() + config = response.json() + if not isinstance(config, dict): + raise ValueError('expected a configuration object') + except (RequestException, ValueError) as error: + self._invalidate_plus5_sensors() + self.log_warning(f'Vertex2 +5V readback failed: {error}') + return + if not self._vertex2_ready(): + self._invalidate_plus5_sensors() + return + for index in range(2): + value = config.get(f'tx{index}plus5') + state = {'0': 'off', '1': 'on'}.get(str(value), 'unavailable') + sensor = self.plus5_sensors.get(index) + if sensor is not None and sensor.get_state() != state: + sensor.set_state(state=state) + def mute_tx0(self, mute): self.send_cmd("mutetx0audio","on" if mute else "off") @@ -535,9 +781,8 @@ class Vertex2(SmartObject): return {"name": config, "icon": None} def update_vertex2_states(self): - if self.power_plug_entity is not None and self.get_state(self.power_plug_entity) != 'on': - self.run_in(self._apply_vertex2_status, 0, - rx0='', rx1='', tx0='', tx1='', audout='', insel='0 0') + if not self._vertex2_powered(): + self._suspend_vertex2() return self._status_requested.set() @@ -597,11 +842,8 @@ class Vertex2(SmartObject): self.allm_sensor.set_state(state=selected_allm) def _apply_vertex2_status(self, kwargs): - self.log_info( - f"Vertex2 TCP status: RX0={kwargs['rx0']!r}, RX1={kwargs['rx1']!r}, " - f"TX0={kwargs['tx0']!r}, TX1={kwargs['tx1']!r}, " - f"AUDOUT={kwargs['audout']!r}, insel={kwargs['insel']!r}" - ) + if not self._vertex2_powered(): + return try: portselect = [int(value) for value in kwargs['insel'].split()] if len(portselect) != 2: @@ -611,6 +853,22 @@ class Vertex2(SmartObject): self.unreachable_sensor.set_state(state='on') return + self._observe_no_signal_spam(kwargs) + + # Ignore callback metadata (e.g. reconfigure) when comparing device state. + snapshot = tuple(kwargs[field].strip() for field in + ('rx0', 'rx1', 'tx0', 'tx1', 'audout', 'insel')) + previous = self._last_logged_status + if snapshot != previous: + self._last_logged_status = snapshot + self.log_info( + f"Vertex2 TCP status: RX0={kwargs['rx0']!r}, RX1={kwargs['rx1']!r}, " + f"TX0={kwargs['tx0']!r}, TX1={kwargs['tx1']!r}, " + f"AUDOUT={kwargs['audout']!r}, insel={kwargs['insel']!r}" + ) + if previous is not None: + self._start_command_processing(only_if_active=True) + if portselect[0] == 4: # 4 = copy TX1 portselect[0] = portselect[1] if portselect[1] == 4: # 4 = copy TX0 @@ -643,3 +901,7 @@ class Vertex2(SmartObject): if self.audio_out_signal_sensor.get_state() != audio_out_signal: self.audio_out_signal_sensor.set_state(state=audio_out_signal) self.unreachable_sensor.set_state(state='off') + self._flush_startup_commands() + if kwargs.get('reconfigure', False): + self._reconfigure_vertex2() + self._refresh_plus5_sensors() diff --git a/tests/test_vertex2.py b/tests/test_vertex2.py new file mode 100644 index 0000000..a6ed04d --- /dev/null +++ b/tests/test_vertex2.py @@ -0,0 +1,506 @@ +from pathlib import Path +import socket +import sys +import threading +import unittest +from unittest.mock import Mock, patch + +sys.path[:0] = [str(Path(__file__).resolve().parents[1] / 'apps'), + str(Path(__file__).resolve().parents[1] / 'apps/ad_toolbox')] +from vertex2 import Vertex2 +import ad_toolbox.smartcondition as SmartCondition + + +class Harness(Vertex2): + def __init__(self, plugged=True): + self.plugged = plugged + self._power_allowed = True + self.power_plug_entity = 'switch.vertex2' + self._interrupt_stop = threading.Event() + self._status_requested = threading.Event() + self._last_logged_status = None + self._connection_lock = threading.Lock() + self._interrupt_connection = None + self.startup_delay = 30 + self._startup_deadline = 0 + self._startup_commands = [] + self._command_processing_lock = threading.Lock() + self._command_processing_deadline = 123 + self._command_processing_handle = 'timer' + self.command_processing_timeout = 30 + self.command_processing_sensor = Mock() + self.plus5_sensors = {} + self.audio_out_signal_sensor = Mock() + self.unreachable_sensor = Mock() + self._update_output_video_sensors = Mock() + self.turn_on = Mock() + self.turn_off = Mock() + self.cancel_timer = Mock() + self.log_warning = Mock() + self.log_info = Mock() + self.run_in = Mock() + self.vertex2_tcp_port = 2220 + + @property + def args(self): + return {'vertex2_address': 'http://192.0.2.1'} + + @property + def name(self): + return 'test_vertex' + + def get_state(self, entity): + return 'on' if self.plugged else 'off' + + +class PowerTests(unittest.TestCase): + def test_condition_closes_socket_before_cutting_power(self): + app = Harness() + connection = Mock() + app._interrupt_connection = connection + app.turn_off.side_effect = lambda entity: connection.close.assert_called_once() + app.on_change_power_plug_conditions(None, SmartCondition.Result.Failed) + connection.shutdown.assert_called_once_with(socket.SHUT_RDWR) + app.turn_off.assert_called_once_with('switch.vertex2') + self.assertFalse(app._vertex2_powered()) + app.cancel_timer.assert_called_once_with('timer') + self.assertIsNone(app._command_processing_deadline) + app.command_processing_sensor.set_state.assert_called_once_with(state='off') + app.unreachable_sensor.set_state.assert_called_once_with(state='off') + + def test_off_blocks_http_commands(self): + for plugged, allowed in [(False, True), (True, False)]: + app = Harness(plugged) + app._power_allowed = allowed + with patch('vertex2.get') as get: + app.send_cmd('reboot') + get.assert_not_called() + + def test_no_plug_preserves_connectivity(self): + app = Harness(False) + app.power_plug_entity = None + self.assertTrue(app._vertex2_powered()) + + def test_delayed_status_is_ignored_after_power_off(self): + app = Harness(False) + app._apply_vertex2_status({}) + app.log_info.assert_not_called() + app.run_in.assert_not_called() + + def test_external_plug_off_suspends_connection(self): + app = Harness(False) + connection = Mock() + app._interrupt_connection = connection + app.on_power_plug_state_change('switch.vertex2', 'state', 'on', 'off', {}) + connection.close.assert_called_once() + app._update_output_video_sensors.assert_called_once_with('', '') + + def test_plug_on_requests_fresh_status(self): + app = Harness() + app.on_power_plug_state_change('switch.vertex2', 'state', 'off', 'on', {}) + self.assertTrue(app._status_requested.is_set()) + self.assertFalse(app._vertex2_ready()) + + def test_boot_commands_wait_and_are_replayed(self): + app = Harness() + with patch('vertex2.time.monotonic', return_value=100): + app._begin_startup() + with patch('vertex2.get') as get: + app.send_cmd('tx1plus5', 'on') + app.send_cmd('edid', 'audiotx1 on') + get.assert_not_called() + self.assertFalse(app._vertex2_ready()) + with patch('vertex2.time.monotonic', return_value=130): + self.assertTrue(app._vertex2_ready()) + with patch('vertex2.get') as get: + app._flush_startup_commands() + self.assertEqual(get.call_count, 2) + self.assertEqual(app._startup_commands, []) + + def test_http_failure_after_boot_still_warns(self): + from requests.exceptions import ConnectionError + app = Harness() + app._startup_deadline = 130 + with patch('vertex2.time.monotonic', return_value=130): + with patch('vertex2.get', side_effect=ConnectionError): + app.send_cmd('tx1plus5', 'on') + app.log_warning.assert_called_once() + + def test_power_cut_discards_queued_commands(self): + app = Harness() + app._begin_startup() + app.send_cmd('tx1plus5', 'on') + app._suspend_vertex2() + self.assertEqual(app._startup_commands, []) + + def test_listener_does_not_connect_during_boot(self): + app = Harness() + app._begin_startup() + worker = threading.Thread(target=app._interrupt_listener, args=(app._interrupt_stop,)) + with patch('vertex2.socket.create_connection') as connect: + worker.start() + try: + self.assertFalse(app._interrupt_stop.wait(.1)) + connect.assert_not_called() + finally: + app._interrupt_stop.set() + worker.join(2) + self.assertFalse(worker.is_alive()) + + def test_cut_power_unblocks_recv_without_warning(self): + app = Harness() + receiving = threading.Event() + closed = threading.Event() + connection = Mock() + connection.shutdown.side_effect = lambda how: closed.set() + def recv(size): + receiving.set() + closed.wait(2) + return b'' + connection.recv.side_effect = recv + context = Mock(__enter__=Mock(return_value=connection), __exit__=Mock(return_value=False)) + worker = threading.Thread(target=app._interrupt_listener, args=(app._interrupt_stop,)) + with patch('vertex2.socket.create_connection', return_value=context): + worker.start() + try: + self.assertTrue(receiving.wait(2)) + app.on_change_power_plug_conditions(None, SmartCondition.Result.Failed) + self.assertTrue(closed.is_set()) + finally: + app._interrupt_stop.set() + closed.set() + worker.join(2) + self.assertFalse(worker.is_alive()) + app.log_warning.assert_not_called() + + def test_listener_waits_off_then_reconnects_when_on(self): + app = Harness(False) + attempted = threading.Event() + def connect(*args, **kwargs): + attempted.set() + app._interrupt_stop.set() + raise OSError('test stop') + worker = threading.Thread(target=app._interrupt_listener, args=(app._interrupt_stop,)) + with patch('vertex2.socket.create_connection', side_effect=connect) as create: + worker.start() + try: + self.assertFalse(attempted.wait(.1)) + create.assert_not_called() + app.plugged = True + self.assertTrue(attempted.wait(2)) + finally: + app._interrupt_stop.set() + worker.join(2) + self.assertFalse(worker.is_alive()) + app.log_warning.assert_not_called() + + +class ReconfigurationTests(unittest.TestCase): + def test_reconnect_reapplies_current_plus5_and_selections(self): + app = Harness() + app.power_plug_conditions_evaluator = Mock() + app.power_plug_conditions_evaluator.evaluate.return_value = SmartCondition.Result.Succeeded + app.input_selection_evaluators = [Mock()] + app.output_selection_evaluators = [Mock()] + app.on_update_input_selection = Mock() + app.on_update_output_selection = Mock() + tx0, tx1 = Mock(), Mock() + tx0.evaluate.return_value = SmartCondition.Result.Failed + tx1.evaluate.return_value = SmartCondition.Result.Succeeded + app.output_config_conditions = {0: {'force_plus5': tx0}, 1: {'force_plus5': tx1}} + app.force_tx0_plus5 = Mock() + app.force_tx1_plus5 = Mock() + app._reconfigure_vertex2() + app.power_plug_conditions_evaluator.evaluate.assert_called_once_with(False) + app.on_update_input_selection.assert_called_once_with() + app.on_update_output_selection.assert_called_once_with() + app.force_tx0_plus5.assert_called_once_with(False) + app.force_tx1_plus5.assert_called_once_with(True) + for evaluator in (tx0, tx1): + evaluator.evaluate.assert_called_once_with(False) + + def test_reconnect_stops_if_power_condition_no_longer_allows_device(self): + app = Harness() + app.power_plug_conditions_evaluator = Mock() + app.power_plug_conditions_evaluator.evaluate.return_value = SmartCondition.Result.Failed + app.on_update_output_selection = Mock() + app._reconfigure_vertex2() + app.turn_off.assert_called_once_with('switch.vertex2') + app.on_update_output_selection.assert_not_called() + + def test_input_selection_without_matching_condition_does_not_send_command(self): + app = Harness() + evaluator = Mock() + evaluator.evaluate.return_value = SmartCondition.Result.Failed + app.input_selection_evaluators = [(evaluator, 'AppleTV')] + app.select_input = Mock() + app.on_update_input_selection() + app.select_input.assert_not_called() + + def test_only_first_complete_status_requests_reconfiguration(self): + app = Harness() + fields = ('rx0', 'rx1', 'tx0', 'tx1', 'tx0sink', 'tx1sink', + 'aud0', 'aud1', 'audout', 'spd0', 'spd1') + response = ('\n'.join(f'{field.upper()}: no signal' for field in fields) + + '\ninsel 0 0\n').encode('ascii') + connection = Mock() + responses = iter([response, response]) + def recv(size): + try: + return next(responses) + except StopIteration: + app._interrupt_stop.set() + return b'' + connection.recv.side_effect = recv + app.run_in.side_effect = lambda *args, **kwargs: app._status_requested.set() + context = Mock(__enter__=Mock(return_value=connection), __exit__=Mock(return_value=False)) + with patch('vertex2.socket.create_connection', return_value=context): + app._interrupt_listener(app._interrupt_stop) + self.assertEqual([call.kwargs['reconfigure'] for call in app.run_in.call_args_list], + [True, False]) + + def test_status_reconfigures_before_readback(self): + app = Harness() + app.input_selected_sensors = {i: Mock() for i in range(2)} + app.generate_input_selected_sensor_name = lambda i: i + app.get_input_config = Mock(return_value={'name': 'AppleTV'}) + app.output_names = {} + app.output_selected_sensor = Mock() + app._refresh_selected_input_sensor = Mock() + sequence = Mock() + app._flush_startup_commands = sequence.flush + app._reconfigure_vertex2 = sequence.reconfigure + app._refresh_plus5_sensors = sequence.readback + app._apply_vertex2_status(dict(rx0='', rx1='', tx0='', tx1='', audout='', + insel='0 0', reconfigure=True)) + self.assertEqual([call[0] for call in sequence.mock_calls], + ['flush', 'reconfigure', 'readback']) + sequence.reset_mock() + app._apply_vertex2_status(dict(rx0='', rx1='', tx0='', tx1='', audout='', insel='0 0')) + sequence.reconfigure.assert_not_called() + + +class Plus5Tests(unittest.TestCase): + def setUp(self): + self.app = Harness() + self.app.plus5_sensors = {i: Mock() for i in range(2)} + + def test_reads_actual_settings_independently(self): + with patch('vertex2.get') as get: + get.return_value.json.return_value = {'tx0plus5': '0', 'tx1plus5': '1'} + self.app._refresh_plus5_sensors() + self.app.plus5_sensors[0].set_state.assert_called_once_with(state='off') + self.app.plus5_sensors[1].set_state.assert_called_once_with(state='on') + + def test_unknown_value_does_not_confirm_off(self): + with patch('vertex2.get') as get: + get.return_value.json.return_value = {'tx0plus5': 'unexpected'} + self.app._refresh_plus5_sensors() + for sensor in self.app.plus5_sensors.values(): + sensor.set_state.assert_called_once_with(state='unavailable') + + def test_failed_or_malformed_read_invalidates_old_settings(self): + from requests.exceptions import HTTPError, Timeout + for error in (HTTPError('503'), Timeout(), ValueError('invalid JSON')): + with self.subTest(error=error): + for sensor in self.app.plus5_sensors.values(): + sensor.reset_mock() + with patch('vertex2.get', side_effect=error): + self.app._refresh_plus5_sensors() + for sensor in self.app.plus5_sensors.values(): + sensor.set_state.assert_called_once_with(state='unavailable') + + def test_power_off_during_read_discards_response(self): + with patch('vertex2.get') as get: + def result(): + self.app.plugged = False + return {'tx0plus5': '1', 'tx1plus5': '1'} + get.return_value.json.side_effect = result + self.app._refresh_plus5_sensors() + for sensor in self.app.plus5_sensors.values(): + sensor.set_state.assert_called_once_with(state='unavailable') + + def test_command_requests_readback_without_optimistic_confirmation(self): + with patch('vertex2.get'): + self.app.send_cmd('tx0plus5', 'on') + self.assertTrue(self.app._status_requested.is_set()) + for sensor in self.app.plus5_sensors.values(): + sensor.set_state.assert_called_once_with(state='unavailable') + + def test_power_cut_invalidates_settings(self): + self.app._suspend_vertex2() + for sensor in self.app.plus5_sensors.values(): + sensor.set_state.assert_called_once_with(state='unavailable') + + +class StatusChangeTests(unittest.TestCase): + def setUp(self): + self.app = app = Harness() + app.input_selected_sensors = {i: Mock() for i in range(2)} + app.generate_input_selected_sensor_name = lambda i: i + app.get_input_config = Mock(return_value={'name': 'AppleTV'}) + app.output_names = {} + app.output_selected_sensor = Mock() + app._refresh_selected_input_sensor = Mock() + app._flush_startup_commands = Mock() + app._reconfigure_vertex2 = Mock() + app._refresh_plus5_sensors = Mock() + self.status = dict(rx0='no signal', rx1='no signal', tx0='', tx1='', + audout='', insel='0 4') + + def test_identical_status_is_silent_but_still_applied(self): + app = self.app + app._apply_vertex2_status(self.status) + app._apply_vertex2_status(dict(self.status, reconfigure=True)) + app.log_info.assert_called_once() + app.run_in.assert_not_called() + self.assertEqual(app._update_output_video_sensors.call_count, 2) + app._reconfigure_vertex2.assert_called_once() + + def test_changed_status_extends_active_window_only_once(self): + app = self.app + app._apply_vertex2_status(self.status) + changed = dict(self.status, tx0='4K50') + with patch('vertex2.time.monotonic', return_value=200): + app._apply_vertex2_status(changed) + self.assertEqual(app._command_processing_deadline, 230) + app._apply_vertex2_status(changed) + self.assertEqual(app.log_info.call_count, 2) + app.run_in.assert_called_once() + app._on_command_processing_timeout({'deadline': 123}) + self.assertEqual(app._command_processing_deadline, 230) + app._on_command_processing_timeout({'deadline': 230}) + app.command_processing_sensor.set_state.assert_called_with(state='off') + + def test_change_without_active_command_logs_without_starting_window(self): + app = self.app + app._command_processing_deadline = None + app._apply_vertex2_status(self.status) + app._apply_vertex2_status(dict(self.status, audout='L-PCM')) + self.assertEqual(app.log_info.call_count, 2) + app.run_in.assert_not_called() + app.command_processing_sensor.set_state.assert_not_called() + + +class CommandProcessingTests(unittest.TestCase): + def test_interrupt_spam_does_not_extend_command_deadline(self): + app = Harness() + app.command_processing_sensor.get_state.return_value = 'off' + with patch('vertex2.time.monotonic', return_value=100), patch('vertex2.get'): + app.send_cmd('mutetx0audio', 'on') + self.assertEqual(app._command_processing_deadline, 130) + app.run_in.reset_mock() + app.cancel_timer.reset_mock() + connection = Mock() + chunks = iter([b'int\n'] * 20) + def recv(size): + try: + return next(chunks) + except StopIteration: + app._interrupt_stop.set() + return b'' + connection.recv.side_effect = recv + context = Mock(__enter__=Mock(return_value=connection), __exit__=Mock(return_value=False)) + with patch('vertex2.socket.create_connection', return_value=context): + app._interrupt_listener(app._interrupt_stop) + self.assertEqual(app._command_processing_deadline, 130) + app.cancel_timer.assert_not_called() + app.run_in.assert_not_called() + app._on_command_processing_timeout({'deadline': 130}) + app.command_processing_sensor.set_state.assert_called_with(state='off') + + def test_new_command_extends_window_and_old_callback_is_ignored(self): + app = Harness() + with patch('vertex2.time.monotonic', return_value=100), patch('vertex2.get'): + app.send_cmd('mutetx0audio', 'on') + with patch('vertex2.time.monotonic', return_value=110), patch('vertex2.get'): + app.send_cmd('mutetx0audio', 'off') + self.assertEqual(app._command_processing_deadline, 140) + app.command_processing_sensor.set_state.reset_mock() + app._on_command_processing_timeout({'deadline': 130}) + app.command_processing_sensor.set_state.assert_not_called() + app._on_command_processing_timeout({'deadline': 140}) + app.command_processing_sensor.set_state.assert_called_once_with(state='off') + + def test_queued_command_does_not_start_window_before_boot(self): + app = Harness() + app._begin_startup() + with patch('vertex2.get') as get: + app.send_cmd('mutetx0audio', 'on') + get.assert_not_called() + app.command_processing_sensor.set_state.assert_not_called() + app.run_in.assert_not_called() + + +class SpamSensorTests(unittest.TestCase): + def setUp(self): + self.app = app = Harness() + app.no_signal_spam_sensor = Mock() + self.state = 'off' + app.no_signal_spam_sensor.get_state.side_effect = lambda: self.state + app.no_signal_spam_sensor.set_state.side_effect = lambda **kw: setattr(self, 'state', kw['state']) + app._spam_lock = threading.RLock() + app._spam_last_interrupt = None + app._spam_snapshot = None + app._spam_count = 0 + app._spam_timer = None + app._spam_deadline = None + self.status = dict(rx0='no signal', rx1='no signal', tx0='', tx1='', audout='', insel='0 4') + + def observe(self, at, interrupt=True, **changes): + status = dict(self.status, **changes) + if interrupt: + status['interrupt_at'] = at + with patch('vertex2.time.monotonic', return_value=at): + self.app._observe_no_signal_spam(status) + + def test_three_identical_interrupts_detect_and_silence_clears(self): + self.observe(100) + self.observe(103) + self.assertEqual(self.state, 'off') + self.observe(106) + self.assertEqual(self.state, 'on') + self.app._expire_no_signal_spam({'deadline': 107}) + self.assertEqual(self.state, 'on') + self.app._expire_no_signal_spam({'deadline': 110}) + self.assertEqual(self.state, 'off') + + def test_polls_do_not_count_and_changed_poll_breaks_streak(self): + for at in range(100, 110): + self.observe(at, interrupt=False) + self.assertEqual(self.state, 'off') + self.observe(112) + self.observe(115) + self.observe(116, interrupt=False, insel='1 4') + self.observe(118) + self.assertEqual(self.state, 'off') + + def test_normal_transitions_and_nonperiodic_events_do_not_trigger(self): + self.observe(100) + self.observe(103, audout='L-PCM') + self.observe(106) + self.assertEqual(self.state, 'off') + self.observe(107) + self.observe(113) + self.assertEqual(self.state, 'off') + + def test_signal_return_or_power_loss_clears(self): + for at in (100, 103, 106): + self.observe(at) + self.observe(107, tx0='4K50') + self.assertEqual(self.state, 'off') + for at in (110, 113, 116): + self.observe(at) + self.app._suspend_vertex2() + self.assertEqual(self.state, 'off') + self.app._expire_no_signal_spam({'deadline': 120}) + self.assertEqual(self.state, 'off') + + def test_unknown_input_is_not_no_signal(self): + for at in (100, 103, 106): + self.observe(at, rx0='unavailable') + self.assertEqual(self.state, 'off') + + +if __name__ == '__main__': + unittest.main()