mirror of
https://github.com/home-assistant/core
synced 2024-09-09 12:51:22 +02:00
Add command support to SamsungTV H/J models (#68301)
Co-authored-by: epenet <epenet@users.noreply.github.com> Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
parent
46072d2997
commit
b13e14b80c
@ -13,7 +13,11 @@ from samsungctl.exceptions import AccessDenied, ConnectionClosed, UnhandledRespo
|
||||
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.encrypted.command import SamsungTVEncryptedCommand
|
||||
from samsungtvws.encrypted.remote import (
|
||||
SamsungTVEncryptedWSAsyncRemote,
|
||||
SendRemoteKey as SendEncryptedRemoteKey,
|
||||
)
|
||||
from samsungtvws.event import (
|
||||
ED_INSTALLED_APP_EVENT,
|
||||
MS_ERROR_EVENT,
|
||||
@ -33,12 +37,12 @@ from homeassistant.const import (
|
||||
CONF_TOKEN,
|
||||
)
|
||||
from homeassistant.core import CALLBACK_TYPE, HomeAssistant
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.helpers.device_registry import format_mac
|
||||
|
||||
from .const import (
|
||||
CONF_DESCRIPTION,
|
||||
CONF_MODEL,
|
||||
CONF_SESSION_ID,
|
||||
ENCRYPTED_WEBSOCKET_PORT,
|
||||
LEGACY_PORT,
|
||||
@ -59,6 +63,9 @@ from .const import (
|
||||
|
||||
KEY_PRESS_TIMEOUT = 1.2
|
||||
|
||||
ENCRYPTED_MODEL_USES_POWER_OFF = {"H6400"}
|
||||
ENCRYPTED_MODEL_USES_POWER = {"JU6400", "JU641D"}
|
||||
|
||||
|
||||
def mac_from_device_info(info: dict[str, Any]) -> str | None:
|
||||
"""Extract the mac address from the device info."""
|
||||
@ -617,9 +624,16 @@ class SamsungTVEncryptedBridge(SamsungTVBridge):
|
||||
) -> None:
|
||||
"""Initialize Bridge."""
|
||||
super().__init__(hass, method, host, port)
|
||||
self._power_off_warning_logged: bool = False
|
||||
self._model: str | None = None
|
||||
self._short_model: str | None = None
|
||||
if entry_data:
|
||||
self.token = entry_data.get(CONF_TOKEN)
|
||||
self.session_id = entry_data.get(CONF_SESSION_ID)
|
||||
self._model = entry_data.get(CONF_MODEL)
|
||||
if self._model and len(self._model) > 4:
|
||||
self._short_model = self._model[4:]
|
||||
|
||||
self._rest_api_port: int | None = None
|
||||
self._device_info: dict[str, Any] | None = None
|
||||
self._remote: SamsungTVEncryptedWSAsyncRemote | None = None
|
||||
@ -693,10 +707,33 @@ class SamsungTVEncryptedBridge(SamsungTVBridge):
|
||||
|
||||
async def async_send_keys(self, keys: list[str]) -> None:
|
||||
"""Send a list of keys using websocket protocol."""
|
||||
raise HomeAssistantError(
|
||||
"Sending commands to encrypted TVs is not yet supported"
|
||||
await self._async_send_commands(
|
||||
[SendEncryptedRemoteKey.click(key) for key in keys]
|
||||
)
|
||||
|
||||
async def _async_send_commands(
|
||||
self, commands: list[SamsungTVEncryptedCommand]
|
||||
) -> None:
|
||||
"""Send the commands using websocket protocol."""
|
||||
try:
|
||||
# recreate connection if connection was dead
|
||||
retry_count = 1
|
||||
for _ in range(retry_count + 1):
|
||||
try:
|
||||
if remote := await self._async_get_remote():
|
||||
await remote.send_commands(commands)
|
||||
break
|
||||
except (
|
||||
BrokenPipeError,
|
||||
WebSocketException,
|
||||
):
|
||||
# BrokenPipe can occur when the commands is sent to fast
|
||||
# WebSocketException can occur when timed out
|
||||
self._remote = None
|
||||
except OSError:
|
||||
# Different reasons, e.g. hostname not resolveable
|
||||
pass
|
||||
|
||||
async def _async_get_remote(self) -> SamsungTVEncryptedWSAsyncRemote | None:
|
||||
"""Create or return a remote control instance."""
|
||||
if (remote := self._remote) and remote.is_alive():
|
||||
@ -737,9 +774,24 @@ class SamsungTVEncryptedBridge(SamsungTVBridge):
|
||||
|
||||
async def async_power_off(self) -> None:
|
||||
"""Send power off command to remote."""
|
||||
raise HomeAssistantError(
|
||||
"Sending commands to encrypted TVs is not yet supported"
|
||||
power_off_commands: list[SamsungTVEncryptedCommand] = []
|
||||
if self._short_model in ENCRYPTED_MODEL_USES_POWER_OFF:
|
||||
power_off_commands.append(SendEncryptedRemoteKey.click("KEY_POWEROFF"))
|
||||
elif self._short_model in ENCRYPTED_MODEL_USES_POWER:
|
||||
power_off_commands.append(SendEncryptedRemoteKey.click("KEY_POWER"))
|
||||
else:
|
||||
if self._model and not self._power_off_warning_logged:
|
||||
LOGGER.warning(
|
||||
"Unknown power_off command for %s (%s): sending KEY_POWEROFF and KEY_POWER",
|
||||
self._model,
|
||||
self.host,
|
||||
)
|
||||
self._power_off_warning_logged = True
|
||||
power_off_commands.append(SendEncryptedRemoteKey.click("KEY_POWEROFF"))
|
||||
power_off_commands.append(SendEncryptedRemoteKey.click("KEY_POWER"))
|
||||
await self._async_send_commands(power_off_commands)
|
||||
# Force closing of remote session to provide instant UI feedback
|
||||
await self.async_close_remote()
|
||||
|
||||
async def async_close_remote(self) -> None:
|
||||
"""Close remote object."""
|
||||
|
@ -27,14 +27,7 @@ from homeassistant.components.media_player.const import (
|
||||
SUPPORT_VOLUME_STEP,
|
||||
)
|
||||
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntry
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_MAC,
|
||||
CONF_METHOD,
|
||||
CONF_NAME,
|
||||
STATE_OFF,
|
||||
STATE_ON,
|
||||
)
|
||||
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_NAME, STATE_OFF, STATE_ON
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import entity_component
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
@ -52,7 +45,6 @@ from .const import (
|
||||
DEFAULT_NAME,
|
||||
DOMAIN,
|
||||
LOGGER,
|
||||
METHOD_ENCRYPTED_WEBSOCKET,
|
||||
)
|
||||
|
||||
SOURCES = {"TV": "KEY_TV", "HDMI": "KEY_HDMI"}
|
||||
@ -125,10 +117,7 @@ class SamsungTVDevice(MediaPlayerEntity):
|
||||
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
|
||||
self._attr_supported_features = SUPPORT_SAMSUNGTV
|
||||
|
||||
if self._on_script or self._mac:
|
||||
# Add turn-on if on_script or mac is available
|
||||
self._attr_supported_features |= SUPPORT_TURN_ON
|
||||
|
@ -8,7 +8,10 @@ import pytest
|
||||
from samsungctl import exceptions
|
||||
from samsungtvws.async_remote import SamsungTVWSAsyncRemote
|
||||
from samsungtvws.command import SamsungTVSleepCommand
|
||||
from samsungtvws.encrypted.remote import SamsungTVEncryptedWSAsyncRemote
|
||||
from samsungtvws.encrypted.remote import (
|
||||
SamsungTVEncryptedCommand,
|
||||
SamsungTVEncryptedWSAsyncRemote,
|
||||
)
|
||||
from samsungtvws.exceptions import ConnectionFailure, HttpApiError
|
||||
from samsungtvws.remote import ChannelEmitCommand, SendRemoteKey
|
||||
from websockets.exceptions import ConnectionClosedError, WebSocketException
|
||||
@ -28,6 +31,7 @@ from homeassistant.components.media_player.const import (
|
||||
SUPPORT_TURN_ON,
|
||||
)
|
||||
from homeassistant.components.samsungtv.const import (
|
||||
CONF_MODEL,
|
||||
CONF_ON_ACTION,
|
||||
DOMAIN as SAMSUNGTV_DOMAIN,
|
||||
ENCRYPTED_WEBSOCKET_PORT,
|
||||
@ -605,11 +609,9 @@ async def test_send_key_websocketexception_encrypted(
|
||||
"""Testing unhandled response exception."""
|
||||
await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS)
|
||||
remoteencws.send_commands = Mock(side_effect=WebSocketException("Boom"))
|
||||
with pytest.raises(HomeAssistantError) as exc_info:
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
assert exc_info.match("media_player.fake does not support this service.")
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == STATE_ON
|
||||
|
||||
@ -631,11 +633,9 @@ async def test_send_key_os_error_ws_encrypted(
|
||||
"""Testing unhandled response exception."""
|
||||
await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS)
|
||||
remoteencws.send_commands = Mock(side_effect=OSError("Boom"))
|
||||
with pytest.raises(HomeAssistantError) as exc_info:
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
assert exc_info.match("media_player.fake does not support this service.")
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == STATE_ON
|
||||
|
||||
@ -805,22 +805,64 @@ async def test_turn_off_encrypted_websocket(
|
||||
hass: HomeAssistant, remoteencws: Mock, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Test for turn_off."""
|
||||
await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS)
|
||||
entry_data = deepcopy(MOCK_ENTRYDATA_ENCRYPTED_WS)
|
||||
entry_data[CONF_MODEL] = "UE48UNKNOWN"
|
||||
await setup_samsungtv_entry(hass, entry_data)
|
||||
|
||||
remoteencws.send_commands.reset_mock()
|
||||
|
||||
with pytest.raises(HomeAssistantError) as exc_info:
|
||||
caplog.clear()
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
assert exc_info.match("media_player.fake does not support this service.")
|
||||
# key called
|
||||
assert remoteencws.send_commands.call_count == 1
|
||||
commands = remoteencws.send_commands.call_args_list[0].args[0]
|
||||
assert len(commands) == 2
|
||||
assert isinstance(command := commands[0], SamsungTVEncryptedCommand)
|
||||
assert command.body["param3"] == "KEY_POWEROFF"
|
||||
assert isinstance(command := commands[1], SamsungTVEncryptedCommand)
|
||||
assert command.body["param3"] == "KEY_POWER"
|
||||
assert "Unknown power_off command for UE48UNKNOWN (fake_host)" in caplog.text
|
||||
|
||||
# commands not sent : power off in progress
|
||||
with pytest.raises(HomeAssistantError) as exc_info:
|
||||
remoteencws.send_commands.reset_mock()
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
assert exc_info.match("media_player.fake does not support this service.")
|
||||
assert "TV is powering off, not sending keys: ['KEY_VOLUP']" in caplog.text
|
||||
remoteencws.send_commands.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("model", "expected_key_type"),
|
||||
[("UE50H6400", "KEY_POWEROFF"), ("UN75JU641D", "KEY_POWER")],
|
||||
)
|
||||
async def test_turn_off_encrypted_websocket_key_type(
|
||||
hass: HomeAssistant,
|
||||
remoteencws: Mock,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
model: str,
|
||||
expected_key_type: str,
|
||||
) -> None:
|
||||
"""Test for turn_off."""
|
||||
entry_data = deepcopy(MOCK_ENTRYDATA_ENCRYPTED_WS)
|
||||
entry_data[CONF_MODEL] = model
|
||||
await setup_samsungtv_entry(hass, entry_data)
|
||||
|
||||
remoteencws.send_commands.reset_mock()
|
||||
|
||||
caplog.clear()
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
# key called
|
||||
assert remoteencws.send_commands.call_count == 1
|
||||
commands = remoteencws.send_commands.call_args_list[0].args[0]
|
||||
assert len(commands) == 1
|
||||
assert isinstance(command := commands[0], SamsungTVEncryptedCommand)
|
||||
assert command.body["param3"] == expected_key_type
|
||||
assert "Unknown power_off command for" not in caplog.text
|
||||
|
||||
|
||||
async def test_turn_off_legacy(hass: HomeAssistant, remote: Mock) -> None:
|
||||
@ -867,11 +909,10 @@ async def test_turn_off_encryptedws_os_error(
|
||||
caplog.set_level(logging.DEBUG)
|
||||
await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS)
|
||||
remoteencws.close = Mock(side_effect=OSError("BOOM"))
|
||||
with pytest.raises(HomeAssistantError) as exc_info:
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
assert exc_info.match("media_player.fake does not support this service.")
|
||||
assert "Error closing connection" in caplog.text
|
||||
|
||||
|
||||
async def test_volume_up(hass: HomeAssistant, remote: Mock) -> None:
|
||||
|
Loading…
Reference in New Issue
Block a user