diff --git a/apps/global/homecinema.yaml b/apps/global/homecinema.yaml index 8f410a3..bdf0389 100644 --- a/apps/global/homecinema.yaml +++ b/apps/global/homecinema.yaml @@ -165,6 +165,12 @@ vertex_2: sensors: binary_sensor.projector_screen_down: input_boolean.projection_screen + input_select: + entity: select.vertex2_input + mqtt_namespace: mqtt + mqtt_topic: appdaemon/vertex2/input + unique_id: vertex2_input + input_config: default: name: None diff --git a/apps/vertex2.py b/apps/vertex2.py index 690b831..2cc5cda 100644 --- a/apps/vertex2.py +++ b/apps/vertex2.py @@ -4,6 +4,7 @@ from requests.exceptions import Timeout,ConnectionError,RequestException from ad_toolbox.smartobject import SmartObject import ad_toolbox.smartcondition as SmartCondition import time +import json import socket import threading import re @@ -186,9 +187,104 @@ class Vertex2(SmartObject): name=f'{self.name}_interrupt_listener', daemon=True, ) + self._initialize_input_select() self.update_vertex2_states() self._interrupt_thread.start() + 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') + if not self._input_select_config: + return + config = self._input_select_config + self._select_namespace = config.get('mqtt_namespace', 'mqtt') + self._select_topic = config.get('mqtt_topic', f'appdaemon/{self.name}/input').rstrip('/') + self._select_uid = config.get('unique_id', f'{self.name}_input') + self.listen_event(self._on_input_select_mqtt, 'MQTT_MESSAGE', + namespace=self._select_namespace) + for entity in (self.args['input_selected_sensor'], self.args['unreachable_sensor']): + self.listen_state(self._on_input_select_state, entity, attribute='all') + if self.power_plug_entity: + self.listen_state(self._on_input_select_state, self.power_plug_entity) + self._discover_input_select() + + def _publish_input_select(self, suffix, payload): + if not isinstance(payload, str): + payload = json.dumps(payload, ensure_ascii=False) + self.call_service('mqtt/publish', topic=self._select_topic + suffix, + payload=payload, retain=True, namespace=self._select_namespace) + + def _discover_input_select(self): + config = self._input_select_config + for topic in (self._select_topic + '/set', 'homeassistant/status'): + self.call_service('mqtt/subscribe', topic=topic, namespace=self._select_namespace) + availability = [{'topic': self._select_topic + '/availability'}] + # The shared plugin's connection topic belongs to the MQTT transport, + # regardless of the name originally chosen for that plugin. + mqtt_config = self.get_plugin_config(namespace=self._select_namespace) or {} + connection_topic = config.get('connection_topic') or mqtt_config.get('will_topic') + if connection_topic: + availability.append(dict( + topic=connection_topic, + payload_available=mqtt_config.get('birth_payload', 'online'), + payload_not_available=mqtt_config.get('will_payload', 'offline'), + )) + payload = dict( + name='Input', unique_id=self._select_uid, + default_entity_id=config.get('entity', 'select.vertex2_input'), + device=dict(identifiers=[f'appdaemon_{self.name}'], name='Vertex2', + manufacturer='HDFury', model='Vertex2'), + options=list(self.input_names), optimistic=False, + command_topic=self._select_topic + '/set', + state_topic=self._select_topic + '/state', + json_attributes_topic=self._select_topic + '/attributes', + availability=availability, availability_mode='all', + ) + self._select_discovery_payload = payload + self._sync_input_select(force_discovery=True) + self.log_info(f'Published MQTT discovery for {payload["default_entity_id"]}') + + def _on_input_select_state(self, entity, attribute, old, new, kwargs): + self._sync_input_select() + + def _sync_input_select(self, force_discovery=False): + if not getattr(self, '_input_select_config', None): + return + name = self.input_selected_sensor.get_state() + valid = name in self.input_names + config = self.get_input_config(self.input_names[name] if valid else None) + icon = config.get('icon') or 'mdi:video-input-hdmi' + # HA's MQTT JSON attributes do not override the entity's icon property. + # Update discovery only when the icon changes (or on reconnect). + discovery = self._select_discovery_payload + if force_discovery or discovery.get('icon') != icon: + discovery['icon'] = icon + self.call_service('mqtt/publish', + topic=f'homeassistant/select/{self._select_uid}/config', + payload=json.dumps(discovery), retain=True, + namespace=self._select_namespace) + self._publish_input_select('/attributes', {'icon': icon}) + self._publish_input_select('/state', name if valid else 'None') + available = (valid and self._vertex2_powered() + and self.unreachable_sensor.get_state() == 'off') + self._publish_input_select('/availability', 'online' if available else 'offline') + + def _on_input_select_mqtt(self, event, data, kwargs): + topic, payload = data.get('topic'), data.get('payload') + if ((topic is None and data.get('state') == 'Connected') + or (topic == 'homeassistant/status' and payload == 'online')): + self._discover_input_select() + elif topic == self._select_topic + '/set': + if payload not in self.input_names: + self.log_warning(f'Rejected unknown Vertex2 input: {payload!r}') + return + if not self._vertex2_ready(): + self.log_warning('Vertex2 input command ignored: device is not ready') + return + # insel sets both TX ports; only a subsequent device status updates HA. + self.select_input(payload) + self.update_vertex2_states() + def _reset_no_signal_spam(self, state='off'): if getattr(self, 'no_signal_spam_sensor', None) is None: return @@ -391,6 +487,8 @@ class Vertex2(SmartObject): self._command_processing_handle = None def terminate(self): + if getattr(self, '_input_select_config', None): + self._publish_input_select('/availability', 'offline') self._stop_interrupt_listener() super().terminate() diff --git a/tests/test_vertex2.py b/tests/test_vertex2.py index a6ed04d..290c6a3 100644 --- a/tests/test_vertex2.py +++ b/tests/test_vertex2.py @@ -504,3 +504,121 @@ class SpamSensorTests(unittest.TestCase): if __name__ == '__main__': unittest.main() + + +class SelectHarness(Harness): + @property + def args(self): + return dict(super().args, + input_selected_sensor='sensor.input', unreachable_sensor='binary_sensor.unreachable', + input_select={'entity': 'select.vertex2_input', 'unique_id': 'vertex2_input', + 'connection_topic': 'appdaemon/connection'}, + input_config={0: {'name': 'AppleTV', 'icon': 'mdi:apple'}, + 1: {'name': 'Xbox', 'icon': 'mdi:microsoft-xbox'}, + 'default': {'name': 'None', 'icon': 'mdi:television-off'}}) + + +class InputSelectTests(unittest.TestCase): + def setUp(self): + self.app = app = SelectHarness() + app.input_names = {'AppleTV': 0, 'Xbox': 1} + app.input_selected_sensor = Mock() + app.input_selected_sensor.get_state.return_value = 'AppleTV' + app.unreachable_sensor.get_state.return_value = 'off' + app.call_service = Mock() + app.get_plugin_config = Mock(return_value={'will_topic': 'transport/connection'}) + app.listen_state = Mock() + app.listen_event = Mock() + app.send_cmd = Mock() + app._initialize_input_select() + + def published(self, suffix): + return [c.kwargs['payload'] for c in self.app.call_service.call_args_list + if c.args[0] == 'mqtt/publish' + and c.kwargs['topic'] == self.app._select_topic + suffix][-1] + + def test_discovery_options_and_initial_icon(self): + import json + discovery = next(c for c in self.app.call_service.call_args_list + if c.kwargs.get('topic') == 'homeassistant/select/vertex2_input/config') + config = json.loads(discovery.kwargs['payload']) + self.assertEqual(config['options'], ['AppleTV', 'Xbox']) + self.assertFalse(config['optimistic']) + self.assertEqual(config['availability'][1]['topic'], 'appdaemon/connection') + self.assertEqual(json.loads(self.published('/attributes'))['icon'], 'mdi:apple') + + def test_user_command_sets_both_tx_without_changing_confirmed_state(self): + app = self.app + app.call_service.reset_mock() + app._on_input_select_mqtt('MQTT_MESSAGE', + {'topic': app._select_topic + '/set', 'payload': 'Xbox'}, {}) + app.send_cmd.assert_called_once_with('insel', '1 1') + self.assertTrue(app._status_requested.is_set()) + app.input_selected_sensor.set_state.assert_not_called() + app.call_service.assert_not_called() + + def test_unknown_or_powered_off_command_is_rejected(self): + app = self.app + for plugged, value in ((True, 'invalid'), (False, 'Xbox')): + app.plugged = plugged + app._on_input_select_mqtt('MQTT_MESSAGE', + {'topic': app._select_topic + '/set', 'payload': value}, {}) + app.send_cmd.assert_not_called() + + def test_state_and_icon_follow_input_on_active_output(self): + import json + app = self.app + state = {'name': 'AppleTV'} + app.input_selected_sensor.get_state.side_effect = lambda: state['name'] + app.input_selected_sensor.set_state.side_effect = lambda **kw: state.update(name=kw.get('state', state['name'])) + app.input_selected_sensors = { + app.generate_input_selected_sensor_name(0): Mock(), + app.generate_input_selected_sensor_name(1): Mock(), + } + for index, name in enumerate(('AppleTV', 'Xbox')): + app.input_selected_sensors[app.generate_input_selected_sensor_name(index)].get_state.return_value = name + for index, expected, icon in ((1, 'Xbox', 'mdi:microsoft-xbox'), (0, 'AppleTV', 'mdi:apple')): + app._refresh_selected_input_sensor(index) + app._on_input_select_state(None, None, None, None, {}) + self.assertEqual(self.published('/state'), expected) + self.assertEqual(json.loads(self.published('/attributes'))['icon'], icon) + self.assertEqual(app._select_discovery_payload['icon'], icon) + app._refresh_selected_input_sensor(None) + app._sync_input_select() + self.assertEqual(self.published('/availability'), 'offline') + self.assertEqual(self.published('/state'), 'None') + + def test_unreachable_and_power_off_mark_select_unavailable(self): + app = self.app + app.unreachable_sensor.get_state.return_value = 'on' + app._sync_input_select() + self.assertEqual(self.published('/availability'), 'offline') + app.unreachable_sensor.get_state.return_value = 'off' + app.plugged = False + app._sync_input_select() + self.assertEqual(self.published('/availability'), 'offline') + + def test_reconnect_resubscribes_and_republishes_discovery(self): + app = self.app + for data in ({'state': 'Connected'}, {'topic': 'homeassistant/status', 'payload': 'online'}): + app.call_service.reset_mock() + app._on_input_select_mqtt('MQTT_MESSAGE', data, {}) + subscriptions = [c.kwargs['topic'] for c in app.call_service.call_args_list + if c.args[0] == 'mqtt/subscribe'] + self.assertIn(app._select_topic + '/set', subscriptions) + self.assertTrue(any(c.kwargs.get('topic', '').endswith('/config') + for c in app.call_service.call_args_list)) + + def test_connection_topic_is_derived_from_shared_plugin(self): + import json + app = self.app + app._input_select_config.pop('connection_topic') + app.get_plugin_config.return_value = { + 'will_topic': 'transport/connection', 'birth_payload': 'ready', 'will_payload': 'gone'} + app.call_service.reset_mock() + app._discover_input_select() + discovery = next(c for c in app.call_service.call_args_list + if c.kwargs.get('topic') == 'homeassistant/select/vertex2_input/config') + config = json.loads(discovery.kwargs['payload']) + self.assertEqual(config['availability'][1], { + 'topic': 'transport/connection', 'payload_available': 'ready', 'payload_not_available': 'gone'})