diff --git a/homeassistant/components/samsungtv/__init__.py b/homeassistant/components/samsungtv/__init__.py index a5deb3ca629..b7d400ce831 100644 --- a/homeassistant/components/samsungtv/__init__.py +++ b/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) diff --git a/homeassistant/components/samsungtv/bridge.py b/homeassistant/components/samsungtv/bridge.py index a2558367995..ba09cf9fe3b 100644 --- a/homeassistant/components/samsungtv/bridge.py +++ b/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() diff --git a/homeassistant/components/samsungtv/entity.py b/homeassistant/components/samsungtv/entity.py new file mode 100644 index 00000000000..418feecbf94 --- /dev/null +++ b/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) + } diff --git a/homeassistant/components/samsungtv/media_player.py b/homeassistant/components/samsungtv/media_player.py index d4c04942e63..2f82c979b94 100644 --- a/homeassistant/components/samsungtv/media_player.py +++ b/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: diff --git a/homeassistant/components/samsungtv/remote.py b/homeassistant/components/samsungtv/remote.py new file mode 100644 index 00000000000..dd9e46dedcb --- /dev/null +++ b/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) diff --git a/tests/components/samsungtv/test_remote.py b/tests/components/samsungtv/test_remote.py new file mode 100644 index 00000000000..d6c43060b85 --- /dev/null +++ b/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"