Add ability to send custom keys to Samsung TV (#83439)

* Add SamsungTV Remote entity with support for turn-off and send command

* Fix SamsungTV remote tests

homeassistant/components/samsungtv/__init__.py

@@ -49,7 +49,7 @@ from .const import (
     UPNP_SVC_RENDERING_CONTROL,
 )
 
-PLATFORMS = [Platform.MEDIA_PLAYER]
+PLATFORMS = [Platform.MEDIA_PLAYER, Platform.REMOTE]
 
 CONFIG_SCHEMA = cv.removed(DOMAIN, raise_if_present=False)
 

homeassistant/components/samsungtv/bridge.py

@@ -6,6 +6,7 @@ import asyncio
 from asyncio.exceptions import TimeoutError as AsyncioTimeoutError
 from collections.abc import Callable, Iterable, Mapping
 import contextlib
+from datetime import datetime, timedelta
 from typing import Any, Generic, TypeVar, cast
 
 from samsungctl import Remote
@@ -43,8 +44,10 @@ from homeassistant.const import (
     CONF_TOKEN,
 )
 from homeassistant.core import CALLBACK_TYPE, HomeAssistant
+from homeassistant.helpers import entity_component
 from homeassistant.helpers.aiohttp_client import async_get_clientsession
 from homeassistant.helpers.device_registry import format_mac
+from homeassistant.util import dt as dt_util
 
 from .const import (
     CONF_DESCRIPTION,
@@ -67,6 +70,13 @@ from .const import (
     WEBSOCKET_PORTS,
 )
 
+# Since the TV will take a few seconds to go to sleep
+# and actually be seen as off, we need to wait just a bit
+# more than the next scan interval
+SCAN_INTERVAL_PLUS_OFF_TIME = entity_component.DEFAULT_SCAN_INTERVAL + timedelta(
+    seconds=5
+)
+
 KEY_PRESS_TIMEOUT = 1.2
 
 ENCRYPTED_MODEL_USES_POWER_OFF = {"H6400", "H6410"}
@@ -161,6 +171,10 @@ class SamsungTVBridge(ABC):
         self._update_config_entry: Callable[[Mapping[str, Any]], None] | None = None
         self._app_list_callback: Callable[[dict[str, str]], None] | None = None
 
+        # Mark the end of a shutdown command (need to wait 15 seconds before
+        # sending the next command to avoid turning the TV back ON).
+        self._end_of_power_off: datetime | None = None
+
     def register_reauth_callback(self, func: CALLBACK_TYPE) -> None:
         """Register a callback function."""
         self._reauth_callback = func
@@ -203,8 +217,17 @@ class SamsungTVBridge(ABC):
     async def async_send_keys(self, keys: list[str]) -> None:
         """Send a list of keys to the tv."""
 
+    @property
+    def power_off_in_progress(self) -> bool:
+        """Return if power off has been recently requested."""
+        return (
+            self._end_of_power_off is not None
+            and self._end_of_power_off > dt_util.utcnow()
+        )
+
     async def async_power_off(self) -> None:
         """Send power off command to remote and close."""
+        self._end_of_power_off = dt_util.utcnow() + SCAN_INTERVAL_PLUS_OFF_TIME
         await self._async_send_power_off()
         # Force closing of remote session to provide instant UI feedback
         await self.async_close_remote()

homeassistant/components/samsungtv/entity.py

@@ -0,0 +1,32 @@
+"""Base SamsungTV Entity."""
+from __future__ import annotations
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_MAC, CONF_MODEL, CONF_NAME
+from homeassistant.helpers import device_registry as dr
+from homeassistant.helpers.entity import DeviceInfo, Entity
+
+from .bridge import SamsungTVBridge
+from .const import CONF_MANUFACTURER, DOMAIN
+
+
+class SamsungTVEntity(Entity):
+    """Defines a base SamsungTV entity."""
+
+    def __init__(self, *, bridge: SamsungTVBridge, config_entry: ConfigEntry) -> None:
+        """Initialize the SamsungTV entity."""
+        self._bridge = bridge
+        self._mac = config_entry.data.get(CONF_MAC)
+        self._attr_name = config_entry.data.get(CONF_NAME)
+        self._attr_unique_id = config_entry.unique_id
+        self._attr_device_info = DeviceInfo(
+            name=self.name,
+            manufacturer=config_entry.data.get(CONF_MANUFACTURER),
+            model=config_entry.data.get(CONF_MODEL),
+        )
+        if self.unique_id:
+            self._attr_device_info["identifiers"] = {(DOMAIN, self.unique_id)}
+        if self._mac:
+            self._attr_device_info["connections"] = {
+                (dr.CONNECTION_NETWORK_MAC, self._mac)
+            }

homeassistant/components/samsungtv/media_player.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 
 import asyncio
 from collections.abc import Coroutine, Sequence
-from datetime import datetime, timedelta
 from typing import Any
 
 import async_timeout
@@ -31,26 +30,16 @@ from homeassistant.components.media_player import (
     MediaType,
 )
 from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntry
-from homeassistant.const import CONF_HOST, CONF_MAC, CONF_MODEL, CONF_NAME
+from homeassistant.const import CONF_HOST
 from homeassistant.core import HomeAssistant, callback
-from homeassistant.helpers import (
-    config_validation as cv,
-    device_registry as dr,
-    entity_component,
-)
+from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.aiohttp_client import async_get_clientsession
-from homeassistant.helpers.entity import DeviceInfo
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.trigger import PluggableAction
-from homeassistant.util import dt as dt_util
 
 from .bridge import SamsungTVBridge, SamsungTVWSBridge
-from .const import (
-    CONF_MANUFACTURER,
-    CONF_SSDP_RENDERING_CONTROL_LOCATION,
-    DOMAIN,
-    LOGGER,
-)
+from .const import CONF_SSDP_RENDERING_CONTROL_LOCATION, DOMAIN, LOGGER
+from .entity import SamsungTVEntity
 from .triggers.turn_on import async_get_turn_on_trigger
 
 SOURCES = {"TV": "KEY_TV", "HDMI": "KEY_HDMI"}
@@ -67,12 +56,6 @@ SUPPORT_SAMSUNGTV = (
     | MediaPlayerEntityFeature.PLAY_MEDIA
 )
 
-# Since the TV will take a few seconds to go to sleep
-# and actually be seen as off, we need to wait just a bit
-# more than the next scan interval
-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
@@ -86,7 +69,7 @@ async def async_setup_entry(
     async_add_entities([SamsungTVDevice(bridge, entry)], True)
 
 
-class SamsungTVDevice(MediaPlayerEntity):
+class SamsungTVDevice(SamsungTVEntity, MediaPlayerEntity):
     """Representation of a Samsung TV."""
 
     _attr_source_list: list[str]
@@ -97,9 +80,9 @@ class SamsungTVDevice(MediaPlayerEntity):
         config_entry: ConfigEntry,
     ) -> None:
         """Initialize the Samsung device."""
+        super().__init__(bridge=bridge, config_entry=config_entry)
         self._config_entry = config_entry
         self._host: str | None = config_entry.data[CONF_HOST]
-        self._mac: str | None = config_entry.data.get(CONF_MAC)
         self._ssdp_rendering_control_location: str | None = config_entry.data.get(
             CONF_SSDP_RENDERING_CONTROL_LOCATION
         )
@@ -107,8 +90,6 @@ class SamsungTVDevice(MediaPlayerEntity):
         # Assume that the TV is in Play mode
         self._playing: bool = True
 
-        self._attr_name: str | None = config_entry.data.get(CONF_NAME)
-        self._attr_unique_id = config_entry.unique_id
         self._attr_is_volume_muted: bool = False
         self._attr_device_class = MediaPlayerDeviceClass.TV
         self._attr_source_list = list(SOURCES)
@@ -123,22 +104,6 @@ class SamsungTVDevice(MediaPlayerEntity):
         if self._ssdp_rendering_control_location:
             self._attr_supported_features |= MediaPlayerEntityFeature.VOLUME_SET
 
-        self._attr_device_info = DeviceInfo(
-            name=self.name,
-            manufacturer=config_entry.data.get(CONF_MANUFACTURER),
-            model=config_entry.data.get(CONF_MODEL),
-        )
-        if self.unique_id:
-            self._attr_device_info["identifiers"] = {(DOMAIN, self.unique_id)}
-        if self._mac:
-            self._attr_device_info["connections"] = {
-                (dr.CONNECTION_NETWORK_MAC, self._mac)
-            }
-
-        # Mark the end of a shutdown command (need to wait 15 seconds before
-        # sending the next command to avoid turning the TV back ON).
-        self._end_of_power_off: datetime | None = None
-        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)
@@ -190,7 +155,7 @@ class SamsungTVDevice(MediaPlayerEntity):
         if self._auth_failed or self.hass.is_stopping:
             return
         old_state = self._attr_state
-        if self._power_off_in_progress():
+        if self._bridge.power_off_in_progress:
             self._attr_state = MediaPlayerState.OFF
         else:
             self._attr_state = (
@@ -333,7 +298,7 @@ class SamsungTVDevice(MediaPlayerEntity):
 
     async def _async_launch_app(self, app_id: str) -> None:
         """Send launch_app to the tv."""
-        if self._power_off_in_progress():
+        if self._bridge.power_off_in_progress:
             LOGGER.info("TV is powering off, not sending launch_app command")
             return
         assert isinstance(self._bridge, SamsungTVWSBridge)
@@ -342,17 +307,11 @@ class SamsungTVDevice(MediaPlayerEntity):
     async def _async_send_keys(self, keys: list[str]) -> None:
         """Send a key to the tv and handles exceptions."""
         assert keys
-        if self._power_off_in_progress() and keys[0] != "KEY_POWEROFF":
+        if self._bridge.power_off_in_progress and keys[0] != "KEY_POWEROFF":
             LOGGER.info("TV is powering off, not sending keys: %s", keys)
             return
         await self._bridge.async_send_keys(keys)
 
-    def _power_off_in_progress(self) -> bool:
-        return (
-            self._end_of_power_off is not None
-            and self._end_of_power_off > dt_util.utcnow()
-        )
-
     @property
     def available(self) -> bool:
         """Return the availability of the device."""
@@ -362,7 +321,7 @@ class SamsungTVDevice(MediaPlayerEntity):
             self.state == MediaPlayerState.ON
             or bool(self._turn_on)
             or self._mac is not None
-            or self._power_off_in_progress()
+            or self._bridge.power_off_in_progress
         )
 
     async def async_added_to_hass(self) -> None:
@@ -378,7 +337,6 @@ class SamsungTVDevice(MediaPlayerEntity):
 
     async def async_turn_off(self) -> None:
         """Turn off media player."""
-        self._end_of_power_off = dt_util.utcnow() + SCAN_INTERVAL_PLUS_OFF_TIME
         await self._bridge.async_power_off()
 
     async def async_set_volume_level(self, volume: float) -> None:

homeassistant/components/samsungtv/remote.py

@@ -0,0 +1,45 @@
+"""Support for the SamsungTV remote."""
+from __future__ import annotations
+
+from collections.abc import Iterable
+from typing import Any
+
+from homeassistant.components.remote import ATTR_NUM_REPEATS, RemoteEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from .const import DOMAIN, LOGGER
+from .entity import SamsungTVEntity
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
+    """Set up the Samsung TV from a config entry."""
+    bridge = hass.data[DOMAIN][entry.entry_id]
+    async_add_entities([SamsungTVRemote(bridge=bridge, config_entry=entry)], True)
+
+
+class SamsungTVRemote(SamsungTVEntity, RemoteEntity):
+    """Device that sends commands to a SamsungTV."""
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Turn the device off."""
+        await self._bridge.async_power_off()
+
+    async def async_send_command(self, command: Iterable[str], **kwargs: Any) -> None:
+        """Send a command to a device.
+
+        Supported keys vary between models.
+        See https://github.com/jaruba/ha-samsungtv-tizen/blob/master/Key_codes.md
+        """
+        if self._bridge.power_off_in_progress:
+            LOGGER.info("TV is powering off, not sending keys: %s", command)
+            return
+
+        num_repeats = kwargs[ATTR_NUM_REPEATS]
+        command_list = list(command)
+
+        for _ in range(num_repeats):
+            await self._bridge.async_send_keys(command_list)

tests/components/samsungtv/test_remote.py

@@ -0,0 +1,93 @@
+"""The tests for the SamsungTV remote platform."""
+from unittest.mock import Mock
+
+import pytest
+from samsungtvws.encrypted.remote import SamsungTVEncryptedCommand
+
+from homeassistant.components.remote import (
+    ATTR_COMMAND,
+    DOMAIN as REMOTE_DOMAIN,
+    SERVICE_SEND_COMMAND,
+)
+from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TURN_OFF
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import entity_registry as er
+
+from . import setup_samsungtv_entry
+from .test_media_player import MOCK_ENTRYDATA_ENCRYPTED_WS
+
+ENTITY_ID = f"{REMOTE_DOMAIN}.fake"
+
+
+@pytest.mark.usefixtures("remoteencws", "rest_api")
+async def test_setup(hass: HomeAssistant) -> None:
+    """Test setup with basic config."""
+    await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS)
+    assert hass.states.get(ENTITY_ID)
+
+
+@pytest.mark.usefixtures("remoteencws", "rest_api")
+async def test_unique_id(hass: HomeAssistant) -> None:
+    """Test unique id."""
+    await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS)
+
+    entity_registry = er.async_get(hass)
+
+    main = entity_registry.async_get(ENTITY_ID)
+    assert main.unique_id == "any"
+
+
+@pytest.mark.usefixtures("remoteencws", "rest_api")
+async def test_main_services(
+    hass: HomeAssistant, remoteencws: Mock, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Test for turn_off."""
+    await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS)
+
+    remoteencws.send_commands.reset_mock()
+
+    assert await hass.services.async_call(
+        REMOTE_DOMAIN,
+        SERVICE_TURN_OFF,
+        {ATTR_ENTITY_ID: ENTITY_ID},
+        blocking=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"
+
+    # commands not sent : power off in progress
+    remoteencws.send_commands.reset_mock()
+    assert await hass.services.async_call(
+        REMOTE_DOMAIN,
+        SERVICE_SEND_COMMAND,
+        {ATTR_ENTITY_ID: ENTITY_ID, ATTR_COMMAND: ["dash"]},
+        blocking=True,
+    )
+    assert "TV is powering off, not sending keys: ['dash']" in caplog.text
+    remoteencws.send_commands.assert_not_called()
+
+
+@pytest.mark.usefixtures("remoteencws", "rest_api")
+async def test_send_command_service(hass: HomeAssistant, remoteencws: Mock) -> None:
+    """Test the send command."""
+    await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS)
+
+    assert await hass.services.async_call(
+        REMOTE_DOMAIN,
+        SERVICE_SEND_COMMAND,
+        {ATTR_ENTITY_ID: ENTITY_ID, ATTR_COMMAND: ["dash"]},
+        blocking=True,
+    )
+
+    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"] == "dash"