new vertex2 content sensor

apps/global/homecinema.yaml

@@ -304,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/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_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()