diff --git a/apps/vertex2.py b/apps/vertex2.py index 2cc5cda..f83a931 100644 --- a/apps/vertex2.py +++ b/apps/vertex2.py @@ -17,6 +17,7 @@ class Vertex2(SmartObject): self._interrupt_stop = threading.Event() self._status_requested = threading.Event() self._last_logged_status = None + self._input_5v_states = {} self._spam_lock = threading.RLock() self._spam_last_interrupt = None self._spam_snapshot = None @@ -187,10 +188,77 @@ class Vertex2(SmartObject): name=f'{self.name}_interrupt_listener', daemon=True, ) + self._initialize_input_5v_sensors() self._initialize_input_select() self.update_vertex2_states() self._interrupt_thread.start() + def _initialize_input_5v_sensors(self): + self.input_5v_sensors = {} + prefix = self.args['input_selected_sensor'].split('.', 1)[1].removesuffix('_input_selected') + used_ids = set() + for name, index in self.input_names.items(): + if index not in range(4): + continue + slug = re.sub(r'[^a-z0-9]+', '_', str(name).lower()).strip('_') or f'input_{index}' + entity_id = f'binary_sensor.{prefix}_{slug}_5v' + if entity_id in used_ids: + entity_id = f'binary_sensor.{prefix}_{slug}_{index}_5v' + used_ids.add(entity_id) + config = self.get_input_config(index) + self.input_5v_sensors[index] = self.create_entity( + entity_id, state='unavailable', friendly_name=f'{name} +5 V', + icon=config.get('icon'), device_class='power', + ) + + def _invalidate_input_5v_sensors(self): + self._input_5v_states = {} + for sensor in getattr(self, 'input_5v_sensors', {}).values(): + if sensor.get_state() != 'unavailable': + sensor.set_state(state='unavailable') + + def _schedule_input_5v_refresh(self): + # Device monitoring must run even when no HA input sensors are exposed. + self.run_in(self._refresh_input_5v_sensors, 0.1) + + def _refresh_input_5v_sensors(self, kwargs=None): + if not self._vertex2_ready(): + self._invalidate_input_5v_sensors() + return + try: + response = get(f'{self.args["vertex2_address"].rstrip("/")}/ssi/infopage.ssi', timeout=5) + response.raise_for_status() + status = response.json() + if not isinstance(status, dict): + raise ValueError('expected an information object') + except (RequestException, ValueError) as error: + self._invalidate_input_5v_sensors() + self.log_warning(f'Vertex2 input +5V readback failed: {error}') + return + if not self._vertex2_ready(): + self._invalidate_input_5v_sensors() + return + states = {} + for index in range(4): + # Firmware uses 2 for present and 1 for absent, not boolean 1/0. + states[index] = {'2': 'on', '1': 'off'}.get( + str(status.get(f'rx{index}in5v')), 'unavailable') + previous_states = self._input_5v_states + self._input_5v_states = states + input_changed = any( + previous_states.get(index) in ('on', 'off') + and state in ('on', 'off') + and previous_states[index] != state + for index, state in states.items() + ) + if input_changed: + self._start_command_processing() + # HA sensors only expose the monitored state; they do not drive detection. + for index, sensor in getattr(self, 'input_5v_sensors', {}).items(): + state = states[index] + if sensor.get_state() != state: + sensor.set_state(state=state) + def _initialize_input_select(self): """Expose a real HA select, with device-confirmed state, through MQTT.""" self._input_select_config = self.args.get('input_select') @@ -296,7 +364,7 @@ class Vertex2(SmartObject): self._spam_last_interrupt = None self._spam_snapshot = None self._spam_count = 0 - if self.no_signal_spam_sensor.get_state() != state: + if state is not None and self.no_signal_spam_sensor.get_state() != state: self.no_signal_spam_sensor.set_state(state=state) def _observe_no_signal_spam(self, status): @@ -438,6 +506,7 @@ class Vertex2(SmartObject): continue self._reset_no_signal_spam('unavailable') self._invalidate_plus5_sensors() + self._invalidate_input_5v_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(): @@ -466,10 +535,11 @@ class Vertex2(SmartObject): 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() + # Cancel spam tracking without emitting state changes during app teardown. + self._reset_no_signal_spam(state=None) self._close_interrupt_connection() interrupt_thread = getattr(self, '_interrupt_thread', None) if interrupt_thread is not None: @@ -651,6 +721,7 @@ class Vertex2(SmartObject): ) def _suspend_vertex2(self): + self._invalidate_input_5v_sensors() self._reset_no_signal_spam() self._invalidate_plus5_sensors() self._close_interrupt_connection() @@ -689,7 +760,7 @@ class Vertex2(SmartObject): self.turn_off(self.power_plug_entity) def _start_command_processing(self, only_if_active=False): - # Only commands start a window; changed status may extend an active one. + # Commands and input +5V transitions start a window; status may extend it. if self.command_processing_sensor is None: return with self._command_processing_lock: @@ -1003,3 +1074,5 @@ class Vertex2(SmartObject): if kwargs.get('reconfigure', False): self._reconfigure_vertex2() self._refresh_plus5_sensors() + # Let the firmware settle before HTTP readback, including initial connection. + self._schedule_input_5v_refresh() diff --git a/tests/test_vertex2.py b/tests/test_vertex2.py index 290c6a3..52b26f4 100644 --- a/tests/test_vertex2.py +++ b/tests/test_vertex2.py @@ -19,6 +19,8 @@ class Harness(Vertex2): self._interrupt_stop = threading.Event() self._status_requested = threading.Event() self._last_logged_status = None + self._input_5v_states = {} + self._schedule_input_5v_refresh = Mock() self._connection_lock = threading.Lock() self._interrupt_connection = None self.startup_delay = 30 @@ -348,6 +350,21 @@ class StatusChangeTests(unittest.TestCase): self.status = dict(rx0='no signal', rx1='no signal', tx0='', tx1='', audout='', insel='0 4') + def test_input_5v_refreshes_on_interrupt_even_when_video_status_is_unchanged(self): + app = self.app + app.input_5v_sensors = {} + app._schedule_input_5v_refresh = lambda: Vertex2._schedule_input_5v_refresh(app) + app._refresh_input_5v_sensors = Mock() + for metadata in ({'reconfigure': True}, {'interrupt_at': 100}): + app.run_in.reset_mock() + app._refresh_input_5v_sensors.reset_mock() + app._apply_vertex2_status(dict(self.status, **metadata)) + app._refresh_input_5v_sensors.assert_not_called() + app.run_in.assert_called_once_with(app._refresh_input_5v_sensors, 0.1) + callback = app.run_in.call_args.args[0] + callback({}) + app._refresh_input_5v_sensors.assert_called_once_with({}) + def test_identical_status_is_silent_but_still_applied(self): app = self.app app._apply_vertex2_status(self.status) @@ -454,6 +471,32 @@ class SpamSensorTests(unittest.TestCase): with patch('vertex2.time.monotonic', return_value=at): self.app._observe_no_signal_spam(status) + def test_stop_cancels_timers_without_publishing_spam_state(self): + app = self.app + self.observe(100) + self.observe(103) + self.observe(106) + self.assertEqual(self.state, 'on') + app._spam_timer = 'spam_timer' + app.no_signal_spam_sensor.reset_mock() + app.cancel_timer.reset_mock() + + app._stop_interrupt_listener() + + self.assertTrue(app._interrupt_stop.is_set()) + self.assertIsNone(app._spam_timer) + self.assertIsNone(app._spam_deadline) + app.cancel_timer.assert_any_call('spam_timer') + app.cancel_timer.assert_any_call('timer') + app.no_signal_spam_sensor.set_state.assert_not_called() + # An already queued expiry callback must also leave the state alone. + app._expire_no_signal_spam({'deadline': 110}) + app.no_signal_spam_sensor.set_state.assert_not_called() + + def test_connection_loss_still_marks_spam_sensor_unavailable(self): + self.app._reset_no_signal_spam('unavailable') + self.assertEqual(self.state, 'unavailable') + def test_three_identical_interrupts_detect_and_silence_clears(self): self.observe(100) self.observe(103) @@ -622,3 +665,136 @@ class InputSelectTests(unittest.TestCase): config = json.loads(discovery.kwargs['payload']) self.assertEqual(config['availability'][1], { 'topic': 'transport/connection', 'payload_available': 'ready', 'payload_not_available': 'gone'}) + + +class Input5VTests(unittest.TestCase): + def setUp(self): + self.app = app = SelectHarness() + app.input_names = {'AppleTV': 0, 'Xbox': 1} + app.create_entity = Mock(side_effect=lambda *a, **kw: Mock()) + app.run_every = Mock() + app._initialize_input_5v_sensors() + + def test_names_and_icons_come_from_input_configuration(self): + calls = self.app.create_entity.call_args_list + self.assertEqual(calls[0].args[0], 'binary_sensor.input_appletv_5v') + self.assertEqual(calls[0].kwargs['friendly_name'], 'AppleTV +5 V') + self.assertEqual(calls[0].kwargs['icon'], 'mdi:apple') + self.assertEqual(calls[1].kwargs['icon'], 'mdi:microsoft-xbox') + self.app.run_every.assert_not_called() + + def test_reads_each_input_even_without_video_signal(self): + with patch('vertex2.get') as get: + get.return_value.json.return_value = {'RX0': 'no signal', 'rx0in5v': '2', 'rx1in5v': '1'} + self.app._refresh_input_5v_sensors() + self.app.input_5v_sensors[0].set_state.assert_called_once_with(state='on') + self.app.input_5v_sensors[1].set_state.assert_called_once_with(state='off') + + def test_invalid_or_missing_values_are_unavailable(self): + with patch('vertex2.get') as get: + get.return_value.json.return_value = {'rx0in5v': 'unexpected'} + self.app._refresh_input_5v_sensors() + for sensor in self.app.input_5v_sensors.values(): + sensor.set_state.assert_called_once_with(state='unavailable') + + def test_http_failure_invalidates_old_values(self): + from requests.exceptions import Timeout + with patch('vertex2.get', side_effect=Timeout): + self.app._refresh_input_5v_sensors() + for sensor in self.app.input_5v_sensors.values(): + sensor.set_state.assert_called_once_with(state='unavailable') + + def test_no_read_while_off_or_booting(self): + for plugged, deadline in ((False, 0), (True, float('inf'))): + self.app.plugged = plugged + self.app._startup_deadline = deadline + with patch('vertex2.get') as get: + self.app._refresh_input_5v_sensors() + get.assert_not_called() + + def test_power_cut_during_request_discards_result(self): + def result(): + self.app.plugged = False + return {'rx0in5v': '2', 'rx1in5v': '2'} + with patch('vertex2.get') as get: + get.return_value.json.side_effect = result + self.app._refresh_input_5v_sensors() + for sensor in self.app.input_5v_sensors.values(): + sensor.set_state.assert_called_once_with(state='unavailable') + + def test_5v_edges_start_or_extend_command_processing_once_per_read(self): + app = self.app + for previous, raw in (('off', '2'), ('on', '1')): + app._input_5v_states = {index: previous for index in range(4)} + for sensor in app.input_5v_sensors.values(): + sensor.get_state.return_value = previous + app.command_processing_sensor.get_state.return_value = 'off' + app.command_processing_sensor.set_state.reset_mock() + app.run_in.reset_mock() + with patch('vertex2.time.monotonic', return_value=200), patch('vertex2.get') as get: + get.return_value.json.return_value = {'rx0in5v': raw, 'rx1in5v': raw} + app._refresh_input_5v_sensors() + app.command_processing_sensor.set_state.assert_called_once_with(state='on') + self.assertEqual(app._command_processing_deadline, 230) + app.run_in.assert_called_once() + app._on_command_processing_timeout({'deadline': 230}) + app.command_processing_sensor.set_state.assert_called_with(state='off') + + def test_initial_unavailable_and_unchanged_5v_do_not_start_processing(self): + app = self.app + app._start_command_processing = Mock() + for previous, raw in (('unavailable', '2'), ('unknown', '1'), ('on', '2'), + ('off', '1'), ('on', None)): + app._input_5v_states = {index: previous for index in range(4)} + for sensor in app.input_5v_sensors.values(): + sensor.get_state.return_value = previous + with patch('vertex2.get') as get: + get.return_value.json.return_value = {'rx0in5v': raw, 'rx1in5v': raw} + app._refresh_input_5v_sensors() + app._start_command_processing.assert_not_called() + + def test_processing_is_identical_with_all_some_or_no_input_sensors(self): + app = self.app + sensors = app.input_5v_sensors.copy() + for exposed in (sensors, {0: sensors[0]}, {}): + with self.subTest(exposed=list(exposed)): + app.input_5v_sensors = exposed + app._input_5v_states = {} + app._start_command_processing = Mock() + with patch('vertex2.get') as get: + # All physical ports are tracked even without input_config entries. + for values in (('1', '1', '1', '1'), ('1', '1', '1', '2'), + ('1', '1', '1', '2'), ('1', '1', '1', '1')): + get.return_value.json.return_value = { + f'rx{index}in5v': value for index, value in enumerate(values)} + app._refresh_input_5v_sensors() + self.assertEqual(app._start_command_processing.call_count, 2) + self.assertEqual(app._input_5v_states[3], 'off') + + def test_ha_sensor_state_cannot_create_or_hide_a_transition(self): + app = self.app + app._start_command_processing = Mock() + app._input_5v_states = {0: 'off', 1: 'off'} + for sensor in app.input_5v_sensors.values(): + sensor.get_state.return_value = 'on' + with patch('vertex2.get') as get: + get.return_value.json.return_value = {'rx0in5v': '1', 'rx1in5v': '1'} + app._refresh_input_5v_sensors() + app._start_command_processing.assert_not_called() + get.return_value.json.return_value = {'rx0in5v': '2', 'rx1in5v': '2'} + app._refresh_input_5v_sensors() + app._start_command_processing.assert_called_once() + + def test_read_failure_resets_internal_baseline_without_processing(self): + from requests.exceptions import Timeout + app = self.app + app.input_5v_sensors = {} + app._input_5v_states = {0: 'on'} + app._start_command_processing = Mock() + with patch('vertex2.get', side_effect=Timeout): + app._refresh_input_5v_sensors() + self.assertEqual(app._input_5v_states, {}) + with patch('vertex2.get') as get: + get.return_value.json.return_value = {'rx0in5v': '1'} + app._refresh_input_5v_sensors() + app._start_command_processing.assert_not_called()