From b13e14b80cd0667b62228f3121cfe28ab5dfcd92 Mon Sep 17 00:00:00 2001 From: epenet <6771947+epenet@users.noreply.github.com> Date: Thu, 24 Mar 2022 18:58:58 +0100 Subject: [PATCH] Add command support to SamsungTV H/J models (#68301) Co-authored-by: epenet Co-authored-by: J. Nick Koston --- homeassistant/components/samsungtv/bridge.py | 66 +++++++++++-- .../components/samsungtv/media_player.py | 15 +-- .../components/samsungtv/test_media_player.py | 95 +++++++++++++------ 3 files changed, 129 insertions(+), 47 deletions(-) diff --git a/homeassistant/components/samsungtv/bridge.py b/homeassistant/components/samsungtv/bridge.py index 73764d51dc0..b89c76f028e 100644 --- a/homeassistant/components/samsungtv/bridge.py +++ b/homeassistant/components/samsungtv/bridge.py @@ -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.""" diff --git a/homeassistant/components/samsungtv/media_player.py b/homeassistant/components/samsungtv/media_player.py index 677dbfab66a..edd929273b1 100644 --- a/homeassistant/components/samsungtv/media_player.py +++ b/homeassistant/components/samsungtv/media_player.py @@ -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 - + 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 diff --git a/tests/components/samsungtv/test_media_player.py b/tests/components/samsungtv/test_media_player.py index 96b939d08b9..a0b712e575b 100644 --- a/tests/components/samsungtv/test_media_player.py +++ b/tests/components/samsungtv/test_media_player.py @@ -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.") + assert await hass.services.async_call( + DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) 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.") + assert await hass.services.async_call( + DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) 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: - 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.") + 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) == 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: - 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.") + remoteencws.send_commands.reset_mock() + assert await hass.services.async_call( + DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + 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 await hass.services.async_call( + DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + assert "Error closing connection" in caplog.text async def test_volume_up(hass: HomeAssistant, remote: Mock) -> None: