Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use callback to get app_list in SamsungTV (#68506) 

Co-authored-by: epenet <epenet@users.noreply.github.com>
pull/68651/head
epenet 2022-03-23 22:12:12 +01:00 committed by GitHub
parent 8293430e25
commit 9ba0475644
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 133 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

69
homeassistant/components/samsungtv/bridge.py
View File

@ -14,7 +14,11 @@ from samsungtvws.async_remote import SamsungTVWSAsyncRemote
from samsungtvws.async_rest import SamsungTVAsyncRest
from samsungtvws.command import SamsungTVCommand
from samsungtvws.encrypted.remote import SamsungTVEncryptedWSAsyncRemote
from samsungtvws.event import MS_ERROR_EVENT
from samsungtvws.event import (
ED_INSTALLED_APP_EVENT,
MS_ERROR_EVENT,
parse_installed_app,
)
from samsungtvws.exceptions import ConnectionFailure, HttpApiError
from samsungtvws.remote import ChannelEmitCommand, SendRemoteKey
from websockets.exceptions import ConnectionClosedError, WebSocketException
@ -128,6 +132,7 @@ class SamsungTVBridge(ABC):
self._reauth_callback: CALLBACK_TYPE | None = None
self._reload_callback: CALLBACK_TYPE | None = None
self._update_config_entry: Callable[[Mapping[str, Any]], None] | None = None
self._app_list_callback: Callable[[dict[str, str]], None] | None = None
def register_reauth_callback(self, func: CALLBACK_TYPE) -> None:
"""Register a callback function."""
@ -143,6 +148,12 @@ class SamsungTVBridge(ABC):
"""Register a callback function."""
self._update_config_entry = func
def register_app_list_callback(
self, func: Callable[[dict[str, str]], None]
) -> None:
"""Register app_list callback function."""
self._app_list_callback = func
@abstractmethod
async def async_try_connect(self) -> str:
"""Try to connect to the TV."""
@ -151,9 +162,15 @@ class SamsungTVBridge(ABC):
async def async_device_info(self) -> dict[str, Any] | None:
"""Try to gather infos of this TV."""
@abstractmethod
async def async_get_app_list(self) -> dict[str, str] | None:
"""Get installed app list."""
async def async_request_app_list(self) -> None:
"""Request app list."""
# Overridden in SamsungTVWSBridge
LOGGER.debug(
"App list request is not supported on %s TV: %s",
self.method,
self.host,
)
self._notify_app_list_callback({})
@abstractmethod
async def async_is_on(self) -> bool:
@ -186,6 +203,11 @@ class SamsungTVBridge(ABC):
if self._update_config_entry is not None:
self._update_config_entry(updates)
def _notify_app_list_callback(self, app_list: dict[str, str]) -> None:
"""Notify update config callback."""
if self._app_list_callback is not None:
self._app_list_callback(app_list)
class SamsungTVLegacyBridge(SamsungTVBridge):
"""The Bridge for Legacy TVs."""
@ -206,10 +228,6 @@ class SamsungTVLegacyBridge(SamsungTVBridge):
}
self._remote: Remote | None = None
async def async_get_app_list(self) -> dict[str, str]:
"""Get installed app list."""
return {}
async def async_is_on(self) -> bool:
"""Tells if the TV is on."""
return await self.hass.async_add_executor_job(self._is_on)
@ -345,26 +363,10 @@ class SamsungTVWSBridge(SamsungTVBridge):
if entry_data:
self.token = entry_data.get(CONF_TOKEN)
self._rest_api: SamsungTVAsyncRest | None = None
self._app_list: dict[str, str] | None = None
self._device_info: dict[str, Any] | None = None
self._remote: SamsungTVWSAsyncRemote | None = None
self._remote_lock = asyncio.Lock()
async def async_get_app_list(self) -> dict[str, str] | None:
"""Get installed app list."""
if self._app_list is None:
if remote := await self._async_get_remote():
raw_app_list = await remote.app_list()
self._app_list = {
app["name"]: app["appId"]
for app in sorted(
raw_app_list or [],
key=lambda app: cast(str, app["name"]),
)
}
LOGGER.debug("Generated app list: %s", self._app_list)
return self._app_list
def _get_device_spec(self, key: str) -> Any | None:
"""Check if a flag exists in latest device info."""
if not ((info := self._device_info) and (device := info.get("device"))):
@ -456,6 +458,10 @@ class SamsungTVWSBridge(SamsungTVBridge):
"""Send the launch_app command using websocket protocol."""
await self._async_send_commands([ChannelEmitCommand.launch_app(app_id)])
async def async_request_app_list(self) -> None:
"""Get installed app list."""
await self._async_send_commands([ChannelEmitCommand.get_installed_app()])
async def async_send_keys(self, keys: list[str]) -> None:
"""Send a list of keys using websocket protocol."""
await self._async_send_commands([SendRemoteKey.click(key) for key in keys])
@ -546,6 +552,17 @@ class SamsungTVWSBridge(SamsungTVBridge):
def _remote_event(self, event: str, response: Any) -> None:
"""Received event from remote websocket."""
if event == ED_INSTALLED_APP_EVENT:
self._notify_app_list_callback(
{
app["name"]: app["appId"]
for app in sorted(
parse_installed_app(response),
key=lambda app: cast(str, app["name"]),
)
}
)
return
if event == MS_ERROR_EVENT:
# { 'event': 'ms.error',
# 'data': {'message': 'unrecognized method value : ms.remote.control'}}
@ -608,10 +625,6 @@ class SamsungTVEncryptedBridge(SamsungTVBridge):
self._remote: SamsungTVEncryptedWSAsyncRemote | None = None
self._remote_lock = asyncio.Lock()
async def async_get_app_list(self) -> dict[str, str]:
"""Get installed app list."""
return {}
async def async_is_on(self) -> bool:
"""Tells if the TV is on."""
LOGGER.debug("Checking if TV %s is on using websocket", self.host)

39
homeassistant/components/samsungtv/media_player.py
View File

@ -1,6 +1,7 @@
"""Support for interface with an Samsung TV."""
from __future__ import annotations
import asyncio
from datetime import datetime, timedelta
from typing import Any
@ -75,6 +76,9 @@ SCAN_INTERVAL_PLUS_OFF_TIME = entity_component.DEFAULT_SCAN_INTERVAL + timedelta
seconds=5
)
# Max delay waiting for app_list to return, as some TVs simply ignore the request
APP_LIST_DELAY = 3
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
@ -119,6 +123,7 @@ class SamsungTVDevice(MediaPlayerEntity):
self._attr_device_class = MediaPlayerDeviceClass.TV
self._attr_source_list = list(SOURCES)
self._app_list: dict[str, str] | None = None
self._app_list_event: asyncio.Event = asyncio.Event()
if config_entry.data.get(CONF_METHOD) != METHOD_ENCRYPTED_WEBSOCKET:
# Encrypted websockets currently only support ON/OFF status
@ -146,6 +151,18 @@ class SamsungTVDevice(MediaPlayerEntity):
self._bridge = bridge
self._auth_failed = False
self._bridge.register_reauth_callback(self.access_denied)
self._bridge.register_app_list_callback(self._app_list_callback)
def _update_sources(self) -> None:
self._attr_source_list = list(SOURCES)
if app_list := self._app_list:
self._attr_source_list.extend(app_list)
def _app_list_callback(self, app_list: dict[str, str]) -> None:
"""App list callback."""
self._app_list = app_list
self._update_sources()
self._app_list_event.set()
def access_denied(self) -> None:
"""Access denied callback."""
@ -173,14 +190,20 @@ class SamsungTVDevice(MediaPlayerEntity):
STATE_ON if await self._bridge.async_is_on() else STATE_OFF
)
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
await self._async_update_app_list()
async def _async_update_app_list(self) -> None:
self._app_list = await self._bridge.async_get_app_list()
if self._app_list is not None:
self._attr_source_list.extend(self._app_list)
if self._attr_state == STATE_ON and not self._app_list_event.is_set():
await self._bridge.async_request_app_list()
if self._app_list_event.is_set():
# The try+wait_for is a bit expensive so we should try not to
# enter it unless we have to (Python 3.11 will have zero cost try)
return
try:
await asyncio.wait_for(self._app_list_event.wait(), APP_LIST_DELAY)
except asyncio.TimeoutError as err:
# No need to try again
self._app_list_event.set()
LOGGER.debug(
"Failed to load app list from %s: %s", self._host, err.__repr__()
)
async def _async_launch_app(self, app_id: str) -> None:
"""Send launch_app to the tv."""

29
tests/components/samsungtv/conftest.py
View File

@ -9,11 +9,14 @@ from unittest.mock import AsyncMock, Mock, patch
import pytest
from samsungctl import Remote
from samsungtvws.async_remote import SamsungTVWSAsyncRemote
from samsungtvws.command import SamsungTVCommand
from samsungtvws.encrypted.remote import SamsungTVEncryptedWSAsyncRemote
from samsungtvws.event import ED_INSTALLED_APP_EVENT
from samsungtvws.remote import ChannelEmitCommand
import homeassistant.util.dt as dt_util
from .const import SAMPLE_APP_LIST, SAMPLE_DEVICE_INFO_WIFI
from .const import SAMPLE_DEVICE_INFO_WIFI
@pytest.fixture(autouse=True)
@ -26,6 +29,13 @@ def fake_host_fixture() -> None:
yield
@pytest.fixture(autouse=True)
def app_list_delay_fixture() -> None:
"""Patch APP_LIST_DELAY."""
with patch("homeassistant.components.samsungtv.media_player.APP_LIST_DELAY", 0):
yield
@pytest.fixture(name="remote")
def remote_fixture() -> Mock:
"""Patch the samsungctl Remote."""
@ -56,19 +66,32 @@ def remotews_fixture() -> Mock:
remotews = Mock(SamsungTVWSAsyncRemote)
remotews.__aenter__ = AsyncMock(return_value=remotews)
remotews.__aexit__ = AsyncMock()
remotews.app_list.return_value = SAMPLE_APP_LIST
remotews.token = "FAKE_TOKEN"
remotews.app_list_data = None
def _start_listening(
async def _start_listening(
ws_event_callback: Callable[[str, Any], Awaitable[None] | None] | None = None
):
remotews.ws_event_callback = ws_event_callback
async def _send_commands(commands: list[SamsungTVCommand]):
if (
len(commands) == 1
and isinstance(commands[0], ChannelEmitCommand)
and commands[0].params["event"] == "ed.installedApp.get"
and remotews.app_list_data is not None
):
remotews.raise_mock_ws_event_callback(
ED_INSTALLED_APP_EVENT,
remotews.app_list_data,
)
def _mock_ws_event_callback(event: str, response: Any):
if remotews.ws_event_callback:
remotews.ws_event_callback(event, response)
remotews.start_listening.side_effect = _start_listening
remotews.send_commands.side_effect = _send_commands
remotews.raise_mock_ws_event_callback = Mock(side_effect=_mock_ws_event_callback)
with patch(

56
tests/components/samsungtv/const.py
View File

@ -1,4 +1,6 @@
"""Constants for the samsungtv tests."""
from samsungtvws.event import ED_INSTALLED_APP_EVENT
from homeassistant.components.samsungtv.const import CONF_SESSION_ID
from homeassistant.const import (
CONF_HOST,
@ -24,30 +26,6 @@ MOCK_ENTRYDATA_ENCRYPTED_WS = {
CONF_SESSION_ID: "2",
}
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",
},
]
SAMPLE_DEVICE_INFO_WIFI = {
"id": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4",
"device": {
@ -127,3 +105,33 @@ SAMPLE_DEVICE_INFO_UE48JU6400 = {
"type": "Samsung SmartTV",
"uri": "https://1.2.3.4:8002/api/v2/",
}
SAMPLE_EVENT_ED_INSTALLED_APP = {
"event": ED_INSTALLED_APP_EVENT,
"from": "host",
"data": {
"data": [
{
"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",
},
]
},
}

3
tests/components/samsungtv/test_config_flow.py
View File

@ -55,7 +55,6 @@ from . import setup_samsungtv_entry
from .const import (
MOCK_CONFIG_ENCRYPTED_WS,
MOCK_ENTRYDATA_ENCRYPTED_WS,
SAMPLE_APP_LIST,
SAMPLE_DEVICE_INFO_FRAME,
)
@ -910,7 +909,6 @@ async def test_autodetect_websocket(hass: HomeAssistant) -> None:
remote = Mock(SamsungTVWSAsyncRemote)
remote.__aenter__ = AsyncMock(return_value=remote)
remote.__aexit__ = AsyncMock(return_value=False)
remote.app_list.return_value = SAMPLE_APP_LIST
rest_api_class.return_value.rest_device_info = AsyncMock(
return_value={
"id": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4",
@ -957,7 +955,6 @@ async def test_websocket_no_mac(hass: HomeAssistant, mac_address: Mock) -> None:
remote = Mock(SamsungTVWSAsyncRemote)
remote.__aenter__ = AsyncMock(return_value=remote)
remote.__aexit__ = AsyncMock(return_value=False)
remote.app_list.return_value = SAMPLE_APP_LIST
rest_api_class.return_value.rest_device_info = AsyncMock(
return_value={
"id": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4",

6
tests/components/samsungtv/test_media_player.py
View File

@ -72,8 +72,8 @@ import homeassistant.util.dt as dt_util
from . import setup_samsungtv_entry
from .const import (
MOCK_ENTRYDATA_ENCRYPTED_WS,
SAMPLE_APP_LIST,
SAMPLE_DEVICE_INFO_FRAME,
SAMPLE_EVENT_ED_INSTALLED_APP,
)
from tests.common import MockConfigEntry, async_fire_time_changed
@ -175,7 +175,6 @@ async def test_setup_websocket(hass: HomeAssistant) -> None:
remote = Mock(SamsungTVWSAsyncRemote)
remote.__aenter__ = AsyncMock(return_value=remote)
remote.__aexit__ = AsyncMock()
remote.app_list.return_value = SAMPLE_APP_LIST
remote.token = "123456789"
remote_class.return_value = remote
@ -213,7 +212,6 @@ async def test_setup_websocket_2(hass: HomeAssistant, mock_now: datetime) -> Non
remote = Mock(SamsungTVWSAsyncRemote)
remote.__aenter__ = AsyncMock(return_value=remote)
remote.__aexit__ = AsyncMock()
remote.app_list.return_value = SAMPLE_APP_LIST
remote.token = "987654321"
remote_class.return_value = remote
assert await async_setup_component(hass, SAMSUNGTV_DOMAIN, {})
@ -738,6 +736,7 @@ async def test_turn_off_websocket(
hass: HomeAssistant, remotews: Mock, caplog: pytest.LogCaptureFixture
) -> None:
"""Test for turn_off."""
remotews.app_list_data = SAMPLE_EVENT_ED_INSTALLED_APP
with patch(
"homeassistant.components.samsungtv.bridge.Remote",
side_effect=[OSError("Boom"), DEFAULT_MOCK],
@ -1178,6 +1177,7 @@ async def test_play_media_app(hass: HomeAssistant, remotews: Mock) -> None:
async def test_select_source_app(hass: HomeAssistant, remotews: Mock) -> None:
"""Test for select_source."""
remotews.app_list_data = SAMPLE_EVENT_ED_INSTALLED_APP
await setup_samsungtv(hass, MOCK_CONFIGWS)
remotews.send_commands.reset_mock()