jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 09:17:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add installed apps to samsungtv sources (#66752)

Browse Source 
Co-authored-by: epenet <epenet@users.noreply.github.com>
This commit is contained in:
epenet 2022-02-18 22:33:49 +01:00 committed by GitHub
parent cfd908218d
commit 3aa18ea5d8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 123 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -121,6 +121,10 @@ class SamsungTVBridge(ABC):
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 @@ class SamsungTVBridge(ABC):
# 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 @@ class SamsungTVBridge(ABC):
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 @@ class SamsungTVLegacyBridge(SamsungTVBridge):
"""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 @@ class SamsungTVLegacyBridge(SamsungTVBridge):
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 @@ class SamsungTVWSBridge(SamsungTVBridge):
"""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 @@ class SamsungTVWSBridge(SamsungTVBridge):
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":
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."""

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

@ -13,6 +13,7 @@ from homeassistant.components.media_player import (
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 @@ class SamsungTVDevice(MediaPlayerEntity):
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 @@ class SamsungTVDevice(MediaPlayerEntity):
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 @@ class SamsungTVDevice(MediaPlayerEntity):
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 @@ class SamsungTVDevice(MediaPlayerEntity):
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
self.send_key(SOURCES[source])
if source in SOURCES:
self.send_key(SOURCES[source])
return
LOGGER.error("Unsupported source")

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

@ -8,6 +8,8 @@ from samsungtvws import SamsungTVWS
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

24
tests/components/samsungtv/const.py Normal file
View File

@ -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",
},
]

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

@ -44,6 +44,8 @@ from homeassistant.const import (
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": {

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

@ -17,6 +17,7 @@ from homeassistant.components.media_player.const import (
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.helpers.typing import ConfigType
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")]