diff --git a/apps/global/homecinema.yaml b/apps/global/homecinema.yaml index ecc47f5..56258f2 100644 --- a/apps/global/homecinema.yaml +++ b/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 + + diff --git a/apps/vertexcontent.py b/apps/vertexcontent.py new file mode 100644 index 0000000..b0d3d5d --- /dev/null +++ b/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) + diff --git a/tests/test_vertexcontent.py b/tests/test_vertexcontent.py new file mode 100644 index 0000000..2c3dfb1 --- /dev/null +++ b/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()