home-assistant · balloob · Feb 18, 2022 · Feb 17, 2022 · Feb 17, 2022 · Feb 17, 2022
@@ -121,6 +121,10 @@ def device_info(self) -> dict[str, Any] | None:
     def mac_from_device(self) -> str | None:
         """Try to fetch the mac address of the TV."""
 
+    @abstractmethod
+    def get_app_list(self) -> dict[str, str] | None:
+        """Get installed app list."""
+
     def is_on(self) -> bool:
         """Tells if the TV is on."""
         if self._remote is not None:
@@ -139,14 +143,14 @@ def is_on(self) -> bool:
             # Different reasons, e.g. hostname not resolveable
             return False
 
-    def send_key(self, key: str) -> None:
+    def send_key(self, key: str, key_type: str | None = None) -> None:
         """Send a key to the tv and handles exceptions."""
         try:
             # recreate connection if connection was dead
             retry_count = 1
             for _ in range(retry_count + 1):
                 try:
-                    self._send_key(key)
+                    self._send_key(key, key_type)
                     break
                 except (
                     ConnectionClosed,
@@ -164,7 +168,7 @@ def send_key(self, key: str) -> None:
             pass
 
     @abstractmethod
-    def _send_key(self, key: str) -> None:
+    def _send_key(self, key: str, key_type: str | None = None) -> None:
         """Send the key."""
 
     @abstractmethod
@@ -212,6 +216,10 @@ def mac_from_device(self) -> None:
         """Try to fetch the mac address of the TV."""
         return None
 
+    def get_app_list(self) -> dict[str, str]:
+        """Get installed app list."""
+        return {}
+
     def try_connect(self) -> str:
         """Try to connect to the Legacy TV."""
         config = {
@@ -261,7 +269,7 @@ def _get_remote(self, avoid_open: bool = False) -> Remote:
                 pass
         return self._remote
 
-    def _send_key(self, key: str) -> None:
+    def _send_key(self, key: str, key_type: str | None = None) -> None:
         """Send the key using legacy protocol."""
         if remote := self._get_remote():
             remote.control(key)
@@ -281,12 +289,25 @@ def __init__(
         """Initialize Bridge."""
         super().__init__(method, host, port)
         self.token = token
+        self._app_list: dict[str, str] | None = None
 
     def mac_from_device(self) -> str | None:
         """Try to fetch the mac address of the TV."""
         info = self.device_info()
         return mac_from_device_info(info) if info else None
 
+    def get_app_list(self) -> dict[str, str] | None:
+        """Get installed app list."""
+        if self._app_list is None:
+            if remote := self._get_remote():
+                raw_app_list: list[dict[str, str]] = remote.app_list()
+                self._app_list = {
+                    app["name"]: app["appId"]
+                    for app in sorted(raw_app_list, key=lambda app: app["name"])
+                }
+
+        return self._app_list
+
     def try_connect(self) -> str:
         """Try to connect to the Websocket TV."""
         for self.port in WEBSOCKET_PORTS:
@@ -338,12 +359,15 @@ def device_info(self) -> dict[str, Any] | None:
 
         return None
 
-    def _send_key(self, key: str) -> None:
+    def _send_key(self, key: str, key_type: str | None = None) -> None:
         """Send the key using websocket protocol."""
         if key == "KEY_POWEROFF":
             key = "KEY_POWER"
         if remote := self._get_remote():
-            remote.send_key(key)
+            if key_type == "run_app":
 retry_count = 1 
 for _ in range(retry_count + 1): 
     try: 
         self._send_key(key) 
         break 
 retry_count = 1 
 for _ in range(retry_count + 1): 
     try: 
         self._send_key(key) 
         break 
+                remote.run_app(key)
+            else:
+                remote.send_key(key)
 
     def _get_remote(self, avoid_open: bool = False) -> Remote:
         """Create or return a remote control instance."""

@@ -13,6 +13,7 @@
     MediaPlayerEntity,
 )
 from homeassistant.components.media_player.const import (
+    MEDIA_TYPE_APP,
     MEDIA_TYPE_CHANNEL,
     SUPPORT_NEXT_TRACK,
     SUPPORT_PAUSE,
@@ -89,6 +90,8 @@ async def async_setup_entry(
 class SamsungTVDevice(MediaPlayerEntity):
     """Representation of a Samsung TV."""
 
+    _attr_source_list: list[str]
+
     def __init__(
         self,
         bridge: SamsungTVLegacyBridge | SamsungTVWSBridge,
@@ -109,6 +112,7 @@ def __init__(
         self._attr_is_volume_muted: bool = False
         self._attr_device_class = MediaPlayerDeviceClass.TV
         self._attr_source_list = list(SOURCES)
+        self._app_list: dict[str, str] | None = None
 
         if self._on_script or self._mac:
             self._attr_supported_features = SUPPORT_SAMSUNGTV | SUPPORT_TURN_ON
@@ -158,12 +162,21 @@ def update(self) -> None:
         else:
             self._attr_state = STATE_ON if self._bridge.is_on() else STATE_OFF
 
-    def send_key(self, key: str) -> None:
+        if self._attr_state == STATE_ON and self._app_list is None:
+            self._app_list = {}  # Ensure that we don't update it twice in parallel
+            self.hass.async_add_job(self._update_app_list)
+
+    def _update_app_list(self) -> None:
+        self._app_list = self._bridge.get_app_list()
+        if self._app_list is not None:
+            self._attr_source_list.extend(self._app_list)
+
+    def send_key(self, key: str, key_type: str | None = None) -> None:
         """Send a key to the tv and handles exceptions."""
         if self._power_off_in_progress() and key != "KEY_POWEROFF":
             LOGGER.info("TV is powering off, not sending command: %s", key)
             return
-        self._bridge.send_key(key)
+        self._bridge.send_key(key, key_type)
 
     def _power_off_in_progress(self) -> bool:
         return (
@@ -232,6 +245,10 @@ async def async_play_media(
         self, media_type: str, media_id: str, **kwargs: Any
     ) -> None:
         """Support changing a channel."""
+        if media_type == MEDIA_TYPE_APP:
+            await self.hass.async_add_executor_job(self.send_key, media_id, "run_app")
+            return
+
         if media_type != MEDIA_TYPE_CHANNEL:
             LOGGER.error("Unsupported media type")
             return
@@ -264,8 +281,12 @@ async def async_turn_on(self) -> None:
 
     def select_source(self, source: str) -> None:
         """Select input source."""
-        if source not in SOURCES:
-            LOGGER.error("Unsupported source")
+        if self._app_list and source in self._app_list:
+            self.send_key(self._app_list[source], "run_app")
+            return
+
+        if source in SOURCES:
+            self.send_key(SOURCES[source])
             return
 
-        self.send_key(SOURCES[source])
+        LOGGER.error("Unsupported source")
@@ -8,6 +8,8 @@
 
 import homeassistant.util.dt as dt_util
 
+from .const import SAMPLE_APP_LIST
+
 
 @pytest.fixture(autouse=True)
 def fake_host_fixture() -> None:
@@ -49,6 +51,7 @@ def remotews_fixture() -> Mock:
                 "networkType": "wireless",
             },
         }
+        remotews.app_list.return_value = SAMPLE_APP_LIST
         remotews.token = "FAKE_TOKEN"
         remotews_class.return_value = remotews
         yield remotews

@@ -0,0 +1,24 @@
+"""Constants for the samsungtv tests."""
+SAMPLE_APP_LIST = [
+    {
+        "appId": "111299001912",
+        "app_type": 2,
+        "icon": "/opt/share/webappservice/apps_icon/FirstScreen/111299001912/250x250.png",
+        "is_lock": 0,
+        "name": "YouTube",
+    },
+    {
+        "appId": "3201608010191",
+        "app_type": 2,
+        "icon": "/opt/share/webappservice/apps_icon/FirstScreen/3201608010191/250x250.png",
+        "is_lock": 0,
+        "name": "Deezer",
+    },
+    {
+        "appId": "3201606009684",
+        "app_type": 2,
+        "icon": "/opt/share/webappservice/apps_icon/FirstScreen/3201606009684/250x250.png",
+        "is_lock": 0,
+        "name": "Spotify - Music and Podcasts",
+    },
+]
@@ -44,6 +44,8 @@
 from homeassistant.core import HomeAssistant
 from homeassistant.setup import async_setup_component
 
+from .const import SAMPLE_APP_LIST
+
 from tests.common import MockConfigEntry
 
 RESULT_ALREADY_CONFIGURED = "already_configured"
@@ -817,6 +819,7 @@ async def test_autodetect_websocket(hass: HomeAssistant) -> None:
         remote = Mock(SamsungTVWS)
         remote.__enter__ = Mock(return_value=remote)
         remote.__exit__ = Mock(return_value=False)
+        remote.app_list.return_value = SAMPLE_APP_LIST
         remote.rest_device_info.return_value = {
             "id": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4",
             "device": {
@@ -863,6 +866,7 @@ async def test_websocket_no_mac(hass: HomeAssistant) -> None:
         remote = Mock(SamsungTVWS)
         remote.__enter__ = Mock(return_value=remote)
         remote.__exit__ = Mock(return_value=False)
+        remote.app_list.return_value = SAMPLE_APP_LIST
         remote.rest_device_info.return_value = {
             "id": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4",
             "device": {

@@ -17,6 +17,7 @@
     ATTR_MEDIA_CONTENT_TYPE,
     ATTR_MEDIA_VOLUME_MUTED,
     DOMAIN,
+    MEDIA_TYPE_APP,
     MEDIA_TYPE_CHANNEL,
     MEDIA_TYPE_URL,
     SERVICE_PLAY_MEDIA,
@@ -61,6 +62,8 @@
 from homeassistant.setup import async_setup_component
 import homeassistant.util.dt as dt_util
 
+from .const import SAMPLE_APP_LIST
+
 from tests.common import MockConfigEntry, async_fire_time_changed
 
 ENTITY_ID = f"{DOMAIN}.fake"
@@ -160,6 +163,7 @@ async def test_setup_websocket(hass: HomeAssistant) -> None:
         remote = Mock(SamsungTVWS)
         remote.__enter__ = Mock(return_value=remote)
         remote.__exit__ = Mock()
+        remote.app_list.return_value = SAMPLE_APP_LIST
         remote.rest_device_info.return_value = {
             "id": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4",
             "device": {
@@ -208,6 +212,7 @@ async def test_setup_websocket_2(hass: HomeAssistant, mock_now: datetime) -> Non
         remote = Mock(SamsungTVWS)
         remote.__enter__ = Mock(return_value=remote)
         remote.__exit__ = Mock()
+        remote.app_list.return_value = SAMPLE_APP_LIST
         remote.rest_device_info.return_value = {
             "id": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4",
             "device": {
@@ -860,3 +865,34 @@ async def test_select_source_invalid_source(hass: HomeAssistant) -> None:
         assert remote.control.call_count == 0
         assert remote.close.call_count == 0
         assert remote.call_count == 1
+
+
+async def test_play_media_app(hass: HomeAssistant, remotews: Mock) -> None:
+    """Test for play_media."""
+    await setup_samsungtv(hass, MOCK_CONFIGWS)
+
+    assert await hass.services.async_call(
+        DOMAIN,
+        SERVICE_PLAY_MEDIA,
+        {
+            ATTR_ENTITY_ID: ENTITY_ID,
+            ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_APP,
+            ATTR_MEDIA_CONTENT_ID: "3201608010191",
+        },
+        True,
+    )
+    assert remotews.run_app.call_count == 1
+    assert remotews.run_app.call_args_list == [call("3201608010191")]
+
+
+async def test_select_source_app(hass: HomeAssistant, remotews: Mock) -> None:
+    """Test for select_source."""
+    await setup_samsungtv(hass, MOCK_CONFIGWS)
+    assert await hass.services.async_call(
+        DOMAIN,
+        SERVICE_SELECT_SOURCE,
+        {ATTR_ENTITY_ID: ENTITY_ID, ATTR_INPUT_SOURCE: "Deezer"},
+        True,
+    )
+    assert remotews.run_app.call_count == 1
+    assert remotews.run_app.call_args_list == [call("3201608010191")]