Added warning to information collector

APPDAEMON_RELOAD_BUG_FIX.md

@@ -0,0 +1,114 @@
+# Bug AppDaemon 4.5.10 : KeyError au rechargement des apps
+
+## Symptôme
+Lors d'une modification de configuration YAML, le rechargement des apps
+échoue dans la boucle utilitaire avec cette erreur :
+
+```text
+check_app_config_files → utils.deep_compare(current_app, prev_app)
+  if isinstance(v, dict) and isinstance(data[k], dict):
+KeyError: 'input_selection'
+```
+
+Le nom de la clé dépend de la configuration modifiée.
+
+## Cause
+`deep_compare()` accède à `data[k]` sans vérifier que la clé existe dans
+l'ancienne configuration. L'ajout d'une clé contenant un dictionnaire peut
+ainsi provoquer un `KeyError`. La fonction peut aussi ignorer les clés
+supprimées, car elle parcourt uniquement la nouvelle configuration.
+
+Le même problème a été signalé avec la clé `mutes` dans les commentaires de
+[AppDaemon/appdaemon#2396](https://github.com/AppDaemon/appdaemon/issues/2396).
+Le correctif du problème initial de ce ticket concerne les plugins personnalisés,
+pas nécessairement ce problème de comparaison signalé en commentaire.
+
+## Statut
+- Patch local appliqué dans le conteneur sur AppDaemon **4.5.10**.
+- Le code défectueux et son appel lors du reload ont également été vérifiés
+  dans la version **4.5.13** : cette mise à jour seule ne corrige pas ce bug.
+- Source de référence :
+  [utils.py en 4.5.13](https://github.com/AppDaemon/appdaemon/blob/4.5.13/appdaemon/utils.py).
+- La correction compare les ensembles de clés avant de comparer les valeurs,
+  y compris dans les dictionnaires imbriqués.
+
+## Comment réappliquer le patch
+
+Fichier concerné : `/usr/local/lib/python3.12/site-packages/appdaemon/utils.py`
+
+Dans `deep_compare()`, juste après `data = data or {}`, ajouter :
+
+```python
+    if check.keys() != data.keys():
+        return False
+```
+
+Le début de la fonction devient :
+
+```python
+def deep_compare(check: dict, data: dict) -> bool:
+    """Compares 2 nested dictionaries of values"""
+    data = data or {}  # Replaces a None value with an empty dict
+
+    if check.keys() != data.keys():
+        return False
+
+    for k, v in tuple(check.items()):
+        if isinstance(v, dict) and isinstance(data[k], dict):
+```
+
+La suite de la fonction reste inchangée. Avant de réappliquer sur une autre
+version, vérifier son implémentation : ne pas remplacer aveuglément une
+fonction qui aurait été corrigée en amont.
+
+**Redémarrer AppDaemon après application.** Modifier le fichier sur disque
+ne remplace pas la fonction déjà importée par le processus en cours ; un
+simple reload d'app ne suffit pas.
+
+Ce patch réside dans `site-packages`, hors de `/conf`. Il est perdu lors
+d'une recréation du conteneur ou peut être écrasé par une réinstallation
+ou mise à jour d'AppDaemon.
+
+## Comment vérifier le patch
+
+Exécuter dans le conteneur :
+
+```sh
+python - <<'PY'
+from appdaemon.utils import deep_compare
+
+assert deep_compare({'input_selection': {}}, {}) is False
+assert deep_compare({}, {'input_selection': {}}) is False
+assert deep_compare({'x': {'a': 1}}, {'x': {}}) is False
+assert deep_compare({'x': {}}, {'x': {'a': 1}}) is False
+assert deep_compare({'x': None}, {}) is False
+assert deep_compare({'x': [1, 2]}, {'x': [1, 2]}) is True
+assert deep_compare({'x': [1, 2]}, {'x': [2, 1]}) is False
+assert deep_compare({'x': {'a': 1}}, {'x': {'a': 1}}) is True
+assert deep_compare({'x': {}}, {'x': None}) is False
+assert deep_compare({}, None) is True
+assert deep_compare({'input_selection': {}}, None) is False
+print('OK : 11 vérifications')
+PY
+```
+
+Ces 11 vérifications ont réussi après application sur le module installé.
+Elles vérifient la fonction dans un nouveau processus Python, pas le reload
+complet dans le service en cours. Le service n'a pas été redémarré lors de
+l'application de ce patch.
+
+## Retour arrière
+
+Une sauvegarde du fichier original a été créée dans le même répertoire :
+`utils.py.before-deep-compare-fix.bak`.
+
+Pour annuler le patch sur cette installation, avant toute autre modification
+ou mise à jour de `utils.py` :
+
+```sh
+cp /usr/local/lib/python3.12/site-packages/appdaemon/utils.py.before-deep-compare-fix.bak \
+   /usr/local/lib/python3.12/site-packages/appdaemon/utils.py
+```
+
+Redémarrer ensuite AppDaemon. Ne pas restaurer cette sauvegarde sur une
+version différente d'AppDaemon.

appdaemon.yaml

@@ -6,6 +6,8 @@ appdaemon:
   internal_function_timeout: 00:20:00
   exclude_dirs:
     - unit_tests
+    - vscode-appdaemon
+    - node_modules
   plugins:
     HASS:
       type: hass

apps/apps.yaml

@@ -18,6 +18,10 @@ informations_collector:
     reset_error_count_button: input_button.reset_error_count
     output_sensor: sensor.ad_errors
 
+  warning_collector:
+    reset_warning_count_button: input_button.reset_error_count
+    output_sensor: sensor.ad_warnings
+
   min_max_temp_sensors:
     temperature_sensors:
       - sensor.ext_front_temperature

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
@@ -182,6 +188,13 @@ vertex_2:
       name: Switch
       icon: mdi:nintendo-switch
 
+  # input_selection:
+  #   None: binary_sensor.vertex2_has_input_signal # this will prevent input change if the current input is still active
+  #   Switch: binary_sensor.vertex2_switch_5v
+  #   Playstation: binary_sensor.vertex2_playstation_5v
+  #   Xbox: binary_sensor.vertex2_xbox_5v
+  #   AppleTV: true
+
   output_config:
     default:
       name: None
@@ -211,13 +224,6 @@ vertex_2:
     Projector: binary_sensor.projector_screen_down  
     TV: True  
 
-  # input_selection:
-  #   None: binary_sensor.vertex2_has_input_signal # this will prevent input change if the current input is still active
-  #   Switch: binary_sensor.is_playing_switch
-  #   AppleTV: not binary_sensor.is_appletv_standy
-  #   Playstation: switch.ps5_136_power
-  #   Xbox: True
-
   vertex2_address: http://10.0.0.33
   vertex2_tcp_port: 2220
   startup_delay: 30 # seconds after the power plug turns on
@@ -298,3 +304,66 @@ homecinema_volume:
           event_condition: not (media_player.bravia_theatre == 'on' or media_player.bravia_theatre == 'playing' or media_player.bravia_theatre == 'paused' or media_player.bravia_theatre == 'idle') and binary_sensor.is_tv_on != 'on'
           data:
             entity_id: media_player.living_room_sonos
+
+vertex_content:
+  module: vertexcontent
+  class: VertexContent
+
+  output: sensor.vertex2_current_content
+  input_selected: sensor.vertex2_input_selected
+  has_signal: binary_sensor.vertex2_has_input_signal
+
+  sources:
+    AppleTV:
+      label: Apple TV
+      platform: Apple TV
+      icon: mdi:apple
+      entity: media_player.apple_tv
+      title_attributes:
+        - media_title
+      series_title_attributes:
+        - media_series_title
+      application_attributes:
+        - app_name
+      content_type_attributes:
+        - media_content_type
+      artwork_attributes:
+        - entity_picture
+      pause_states:
+        - paused
+      # is_paused is meaningful only for apps known to report playback state.
+      pause_supported_apps:
+        - CANAL+
+        - Disney+
+        - HBO Max
+        - Netflix
+        - Plex
+        - Prime Video
+        - TV
+        - VLC
+        - YouTube
+
+    Xbox:
+      label: Xbox
+      platform: Xbox
+      content_type: game
+      icon: mdi:microsoft-xbox
+      entity: sensor.yetanotherpedro_now_playing
+      state_as_title: true
+      artwork_entity: image.yetanotherpedro_now_playing
+
+    Playstation:
+      label: PlayStation 5
+      platform: PlayStation 5
+      content_type: game
+      icon: mdi:sony-playstation
+      entity: sensor.ps5_136_activity
+      state_as_title: true
+
+    Switch:
+      label: Nintendo Switch
+      platform: Nintendo Switch
+      content_type: game
+      icon: mdi:nintendo-switch
+
+

apps/global/virtualsensors.yaml

@@ -110,7 +110,7 @@ virtual_sensors:
     binary_sensor.is_playstation_on: switch.ps5_136_power
     sensor.vertex2_input_selected_icon: sensor.vertex2_input_selected.icon
     retain_condition.vertex2_has_input_signal: 
-      retain_time: 20
+      retain_time: 20 #20 seems like the sweet spot to handle transition between source (AppleTV -> Xbox on projector)
       conditions: 
         trigger_conditions: (sensor.vertex2_video_signal and sensor.vertex2_video_signal != 'no signal') or sensor.vertex2_audio_signal
         disable_conditions: self and (binary_sensor.vertex2_unreachable or binary_sensor.vertex2_command_processing)

apps/informationscollector.py

@@ -51,6 +51,24 @@ class InformationsCollector(SmartObject):
             else:
                 self.log_error("No output_sensor defined for error_collector")
 
+        if 'warning_collector' in self.args:
+            if 'output_sensor' in self.args['warning_collector']:
+                sensor_name = self.args['warning_collector']['output_sensor']
+                self.warning_list = []
+                self.warning_log_handle = self.listen_log(self.on_log_warning, "WARNING", log="main_log")
+                self.warning_count_sensor = self.create_entity(
+                    sensor_name, state=0,
+                    attributes={"last_warning": "", "warning_list": []},
+                    icon="mdi:alert-outline"
+                )
+                if 'reset_warning_count_button' in self.args['warning_collector']:
+                    self.listen_state(
+                        self.on_reset_warning_count,
+                        self.args['warning_collector']['reset_warning_count_button']
+                    )
+            else:
+                self.log_error("No output_sensor defined for warning_collector")
+
         if 'min_max_temp_sensors' in self.args:
             for sensor in self.args['min_max_temp_sensors']['temperature_sensors']:
                 self.listen_state(self.on_temperature_change, sensor)
@@ -218,6 +236,21 @@ class InformationsCollector(SmartObject):
         self.error_list.clear()
         self.error_count_sensor.set_state(state=0, attributes={"last_error": "", "error_list": self.error_list})
 
+    def on_log_warning(self, name, ts, level, type, message, kwargs):
+        warning_count = int(self.warning_count_sensor.get_state()) + 1
+        self.warning_list.append(message)
+        self.warning_count_sensor.set_state(
+            state=warning_count,
+            attributes={"last_warning": message, "warning_list": self.warning_list}
+        )
+
+    def on_reset_warning_count(self, entity, attribute, old, new, kwargs):
+        self.warning_list.clear()
+        self.warning_count_sensor.set_state(
+            state=0,
+            attributes={"last_warning": "", "warning_list": self.warning_list}
+        )
+
     def _on_group_condition_changed(self, prev_result,result):
         self.update_group_collectors()
 

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#" ]
-
-        should_retry = False
-        with telnetlib.Telnet(self.telnet_ip,self.telnet_port) as session:
-            if log_command: self.log_info(f'Sending command "{cmd}"')
+    def _process_commands(self):
+        while not self._command_stop.is_set():
+            operation = self._command_queue.get()
             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)
+                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()
+
+    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
-
-
-        # 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}")
+            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
-        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

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
@@ -16,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
@@ -186,9 +188,171 @@ 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')
+        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
@@ -200,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):
@@ -342,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():
@@ -370,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:
@@ -391,6 +557,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()
 
@@ -553,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()
@@ -591,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:
@@ -905,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()

apps/vertexcontent.py

@@ -0,0 +1,166 @@
+from ad_toolbox.smartobject import SmartObject
+
+
+EMPTY_VALUES = {None, "", "unknown", "unavailable", "none", "null"}
+
+
+def useful(value):
+    """Return a displayable Home Assistant value, or None."""
+    if isinstance(value, str):
+        value = value.strip()
+        if value.lower() in EMPTY_VALUES:
+            return None
+    elif value in EMPTY_VALUES:
+        return None
+    return value
+
+
+class VertexContent(SmartObject):
+    """Expose normalized metadata for the source currently selected on a Vertex2."""
+
+    def on_initialize_smart_object(self):
+        super().on_initialize_smart_object()
+        self.output = self.args.get("output", "sensor.vertex2_current_content")
+        self.input_selected = self.args.get(
+            "input_selected", "sensor.vertex2_input_selected"
+        )
+        self.has_signal = self.args.get(
+            "has_signal", "binary_sensor.vertex2_has_input_signal"
+        )
+        self.sources = self.args.get("sources", {})
+
+        state, attributes = self.build_content()
+        self.output_entity = self.create_entity(
+            self.output, state=state, attributes=attributes
+        )
+
+        self.listen_state(self._state_changed, self.input_selected)
+        self.listen_state(self._state_changed, self.has_signal)
+
+        # attribute="all" is important: a media title can change while the
+        # media_player state remains "playing".
+        listened = set()
+        for config in self.sources.values():
+            for entity in (config.get("entity"), config.get("artwork_entity")):
+                if entity and entity not in listened:
+                    self.listen_state(self._state_changed, entity, attribute="all")
+                    listened.add(entity)
+
+    def _state_changed(self, entity, attribute, old, new, kwargs):
+        self.update_content()
+
+    def _entity_data(self, entity):
+        if not entity:
+            return None, {}
+        data = self.get_state(entity, attribute="all") or {}
+        return data.get("state"), data.get("attributes", {}) or {}
+
+    @staticmethod
+    def _first_attribute(attributes, names):
+        for name in names:
+            value = useful(attributes.get(name))
+            if value is not None:
+                return value
+        return None
+
+    def _artwork(self, config, attributes):
+        artwork = self._first_attribute(
+            attributes, config.get("artwork_attributes", ["entity_picture"])
+        )
+        if artwork is not None:
+            return artwork
+        _, artwork_attributes = self._entity_data(config.get("artwork_entity"))
+        return useful(artwork_attributes.get("entity_picture"))
+
+    @staticmethod
+    def _pause_supported(config, application, source_state):
+        pause_states = config.get("pause_states", [])
+        if not pause_states or useful(source_state) is None:
+            return False
+        supported_apps = config.get("pause_supported_apps")
+        if supported_apps is None:
+            return True
+        return application in supported_apps
+
+    def build_content(self):
+        selected = useful(self.get_state(self.input_selected))
+        has_signal = self.get_state(self.has_signal) == "on"
+
+        base = {
+            "friendly_name": "Contenu Vertex2",
+            "icon": "mdi:television-play",
+            "source": selected,
+            "platform": None,
+            "content_type": None,
+            "title": None,
+            "series_title": None,
+            "application": None,
+            "status": "idle",
+            "artwork": None,
+            "entity_picture": None,
+            "available": False,
+            "is_paused": False,
+            "is_pause_supported": False,
+        }
+
+        if not has_signal or selected is None:
+            base["source"] = selected
+            base["icon"] = "mdi:television-off"
+            return "idle", base
+
+        config = self.sources.get(str(selected), {})
+        label = config.get("label", selected)
+        source_state, attributes = self._entity_data(config.get("entity"))
+        application = self._first_attribute(
+            attributes, config.get("application_attributes", ["app_name"])
+        )
+        title = self._first_attribute(
+            attributes, config.get("title_attributes", ["media_title"])
+        )
+        series_title = self._first_attribute(
+            attributes, config.get("series_title_attributes", ["media_series_title"])
+        )
+
+        if title is None and config.get("state_as_title"):
+            title = useful(source_state)
+
+        artwork = self._artwork(config, attributes)
+        pause_supported = self._pause_supported(config, application, source_state)
+        pause_states = config.get("pause_states", [])
+        is_paused = pause_supported and source_state in pause_states
+        content_type = self._first_attribute(
+            attributes, config.get("content_type_attributes", ["media_content_type"])
+        )
+        if content_type is None:
+            content_type = config.get("content_type")
+        if content_type is None and application and title is None:
+            content_type = "application"
+
+        display = title or application or label
+        base.update({
+            "icon": config.get("icon", base["icon"]),
+            "platform": config.get("platform", label),
+            "content_type": content_type,
+            "title": title,
+            "series_title": series_title,
+            "application": application,
+            "status": useful(source_state) or "active",
+            "artwork": artwork,
+            "entity_picture": artwork,
+            "available": True,
+            "is_paused": is_paused,
+            "is_pause_supported": pause_supported,
+        })
+        return str(display)[:255], base
+
+    def update_content(self):
+        state, attributes = self.build_content()
+        current_state = self.output_entity.get_state()
+        current_attributes = self.output_entity.get_state(attribute="all") or {}
+        if isinstance(current_attributes, dict) and "attributes" in current_attributes:
+            current_attributes = current_attributes["attributes"]
+        expected_attributes = dict(attributes)
+        expected_attributes["ad_app"] = self.name
+        if current_state != state or current_attributes != expected_attributes:
+            self.output_entity.set_state(state=state, attributes=expected_attributes)
+

tests/test_informationscollector.py

@@ -0,0 +1,45 @@
+from pathlib import Path
+import sys
+import unittest
+from unittest.mock import Mock
+
+sys.path[:0] = [str(Path(__file__).resolve().parents[1] / 'apps'),
+                str(Path(__file__).resolve().parents[1] / 'apps/ad_toolbox')]
+
+from informationscollector import InformationsCollector
+
+
+class WarningCollectorTests(unittest.TestCase):
+    def setUp(self):
+        self.app = object.__new__(InformationsCollector)
+        self.app.warning_list = []
+        self.app.warning_count_sensor = Mock()
+        self.app.warning_count_sensor.get_state.return_value = '2'
+
+    def test_warning_is_counted_and_exposed_in_attributes(self):
+        self.app.on_log_warning(
+            'example', None, 'WARNING', None, 'Something happened', {}
+        )
+
+        self.app.warning_count_sensor.set_state.assert_called_once_with(
+            state=3,
+            attributes={
+                'last_warning': 'Something happened',
+                'warning_list': ['Something happened'],
+            },
+        )
+
+    def test_warning_reset_clears_count_and_attributes(self):
+        self.app.warning_list = ['Old warning']
+
+        self.app.on_reset_warning_count(None, None, None, None, {})
+
+        self.assertEqual(self.app.warning_list, [])
+        self.app.warning_count_sensor.set_state.assert_called_once_with(
+            state=0,
+            attributes={'last_warning': '', 'warning_list': []},
+        )
+
+
+if __name__ == '__main__':
+    unittest.main()

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()

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)
@@ -504,3 +547,254 @@ 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'})
+
+
+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()

tests/test_vertexcontent.py

@@ -0,0 +1,116 @@
+from pathlib import Path
+import sys
+import unittest
+from unittest.mock import Mock
+
+
+APPS = Path(__file__).resolve().parents[1] / "apps"
+sys.path[:0] = [str(APPS), str(APPS / "ad_toolbox")]
+from vertexcontent import VertexContent
+
+
+class Harness(VertexContent):
+    def __init__(self, states, sources):
+        self.states = states
+        self.sources = sources
+        self.input_selected = "sensor.vertex2_input_selected"
+        self.has_signal = "binary_sensor.vertex2_has_input_signal"
+        self.output = "sensor.vertex2_current_content"
+
+    @property
+    def name(self):
+        return "vertex_content"
+
+    def get_state(self, entity, attribute=None):
+        data = self.states.get(entity)
+        if attribute == "all":
+            return data
+        return data.get("state") if data else None
+
+
+APPLE = {
+    "AppleTV": {
+        "label": "Apple TV",
+        "platform": "Apple TV",
+        "entity": "media_player.apple_tv",
+        "pause_states": ["paused"],
+        "pause_supported_apps": ["Netflix"],
+    }
+}
+
+
+class VertexContentTests(unittest.TestCase):
+    def states(self, player_state="playing", attributes=None, signal="on"):
+        return {
+            "sensor.vertex2_input_selected": {"state": "AppleTV", "attributes": {}},
+            "binary_sensor.vertex2_has_input_signal": {"state": signal, "attributes": {}},
+            "media_player.apple_tv": {
+                "state": player_state,
+                "attributes": attributes or {},
+            },
+        }
+
+    def test_uses_title_before_application(self):
+        app = Harness(self.states(attributes={
+            "media_title": "Dune",
+            "app_name": "Netflix",
+            "media_content_type": "movie",
+        }), APPLE)
+        state, attributes = app.build_content()
+        self.assertEqual(state, "Dune")
+        self.assertEqual(attributes["application"], "Netflix")
+        self.assertEqual(attributes["content_type"], "movie")
+        self.assertFalse(attributes["is_paused"])
+        self.assertTrue(attributes["is_pause_supported"])
+
+    def test_falls_back_to_application(self):
+        app = Harness(self.states(attributes={"app_name": "Netflix"}), APPLE)
+        state, attributes = app.build_content()
+        self.assertEqual(state, "Netflix")
+        self.assertIsNone(attributes["title"])
+
+    def test_pause_requires_paused_state_and_supported_app(self):
+        supported = Harness(self.states("paused", {"app_name": "Netflix"}), APPLE)
+        unsupported = Harness(self.states("paused", {"app_name": "Photos"}), APPLE)
+        self.assertTrue(supported.build_content()[1]["is_paused"])
+        self.assertFalse(unsupported.build_content()[1]["is_paused"])
+        self.assertFalse(unsupported.build_content()[1]["is_pause_supported"])
+
+    def test_no_signal_is_idle(self):
+        app = Harness(self.states(signal="off"), APPLE)
+        state, attributes = app.build_content()
+        self.assertEqual(state, "idle")
+        self.assertFalse(attributes["available"])
+        self.assertFalse(attributes["is_paused"])
+
+    def test_console_uses_activity_state(self):
+        states = {
+            "sensor.vertex2_input_selected": {"state": "Playstation", "attributes": {}},
+            "binary_sensor.vertex2_has_input_signal": {"state": "on", "attributes": {}},
+            "sensor.ps5_136_activity": {"state": "Astro Bot", "attributes": {}},
+        }
+        sources = {"Playstation": {
+            "label": "PlayStation 5",
+            "platform": "PlayStation 5",
+            "entity": "sensor.ps5_136_activity",
+            "state_as_title": True,
+            "content_type": "game",
+        }}
+        state, attributes = Harness(states, sources).build_content()
+        self.assertEqual(state, "Astro Bot")
+        self.assertEqual(attributes["content_type"], "game")
+        self.assertFalse(attributes["is_pause_supported"])
+
+    def test_updates_through_created_entity_handle(self):
+        app = Harness(self.states(attributes={"app_name": "Netflix"}), APPLE)
+        app.output_entity = Mock()
+        app.output_entity.get_state.side_effect = ["old", {"attributes": {}}]
+        app.update_content()
+        app.output_entity.set_state.assert_called_once()
+        update = app.output_entity.set_state.call_args.kwargs
+        self.assertEqual(update["state"], "Netflix")
+        self.assertEqual(update["attributes"]["ad_app"], "vertex_content")
+
+
+if __name__ == "__main__":
+    unittest.main()