Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add turn_on to SamsungTV remote (#117403) 

Co-authored-by: J. Nick Koston <nick@koston.org>
pull/117766/head^2
epenet 2024-05-20 08:14:20 +02:00 committed by GitHub
parent 14f1e8c520
commit 570d5f2b55
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 141 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
homeassistant/components/samsungtv/entity.py
View File

@ -2,10 +2,13 @@
from __future__ import annotations
from wakeonlan import send_magic_packet
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_CONNECTIONS,
ATTR_IDENTIFIERS,
CONF_HOST,
CONF_MAC,
CONF_MODEL,
CONF_NAME,
@ -13,9 +16,11 @@ from homeassistant.const import (
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.trigger import PluggableAction
from .bridge import SamsungTVBridge
from .const import CONF_MANUFACTURER, DOMAIN
from .triggers.turn_on import async_get_turn_on_trigger
class SamsungTVEntity(Entity):
@ -26,7 +31,8 @@ class SamsungTVEntity(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._mac: str | None = config_entry.data.get(CONF_MAC)
self._host: str | None = config_entry.data.get(CONF_HOST)
# Fallback for legacy models that doesn't have a API to retrieve MAC or SerialNumber
self._attr_unique_id = config_entry.unique_id or config_entry.entry_id
self._attr_device_info = DeviceInfo(
@ -40,3 +46,22 @@ class SamsungTVEntity(Entity):
self._attr_device_info[ATTR_CONNECTIONS] = {
(dr.CONNECTION_NETWORK_MAC, self._mac)
}
self._turn_on_action = PluggableAction(self.async_write_ha_state)
async def async_added_to_hass(self) -> None:
"""Connect and subscribe to dispatcher signals and state updates."""
await super().async_added_to_hass()
if (entry := self.registry_entry) and entry.device_id:
self.async_on_remove(
self._turn_on_action.async_register(
self.hass, async_get_turn_on_trigger(entry.device_id)
)
)
def _wake_on_lan(self) -> None:
"""Wake the device via wake on lan."""
send_magic_packet(self._mac, ip_address=self._host)
# If the ip address changed since we last saw the device
# broadcast a packet as well
send_magic_packet(self._mac)

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

@ -20,7 +20,6 @@ from async_upnp_client.exceptions import (
from async_upnp_client.profiles.dlna import DmrDevice
from async_upnp_client.utils import async_get_local_ip
import voluptuous as vol
from wakeonlan import send_magic_packet
from homeassistant.components.media_player import (
MediaPlayerDeviceClass,
@ -30,19 +29,16 @@ from homeassistant.components.media_player import (
MediaType,
)
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntry
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.trigger import PluggableAction
from homeassistant.util.async_ import create_eager_task
from . import SamsungTVConfigEntry
from .bridge import SamsungTVBridge, SamsungTVWSBridge
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"}
@ -90,11 +86,9 @@ class SamsungTVDevice(SamsungTVEntity, MediaPlayerEntity):
"""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._ssdp_rendering_control_location: str | None = config_entry.data.get(
CONF_SSDP_RENDERING_CONTROL_LOCATION
)
self._turn_on = PluggableAction(self.async_write_ha_state)
# Assume that the TV is in Play mode
self._playing: bool = True
@ -123,7 +117,7 @@ class SamsungTVDevice(SamsungTVEntity, MediaPlayerEntity):
"""Flag media player features that are supported."""
# `turn_on` triggers are not yet registered during initialisation,
# so this property needs to be dynamic
if self._turn_on:
if self._turn_on_action:
return self._attr_supported_features | MediaPlayerEntityFeature.TURN_ON
return self._attr_supported_features
@ -326,22 +320,11 @@ class SamsungTVDevice(SamsungTVEntity, MediaPlayerEntity):
return False
return (
self.state == MediaPlayerState.ON
or bool(self._turn_on)
or bool(self._turn_on_action)
or self._mac is not None
or self._bridge.power_off_in_progress
)
async def async_added_to_hass(self) -> None:
"""Connect and subscribe to dispatcher signals and state updates."""
await super().async_added_to_hass()
if (entry := self.registry_entry) and entry.device_id:
self.async_on_remove(
self._turn_on.async_register(
self.hass, async_get_turn_on_trigger(entry.device_id)
)
)
async def async_turn_off(self) -> None:
"""Turn off media player."""
await self._bridge.async_power_off()
@ -416,17 +399,10 @@ class SamsungTVDevice(SamsungTVEntity, MediaPlayerEntity):
keys=[f"KEY_{digit}" for digit in media_id] + ["KEY_ENTER"]
)
def _wake_on_lan(self) -> None:
"""Wake the device via wake on lan."""
send_magic_packet(self._mac, ip_address=self._host)
# If the ip address changed since we last saw the device
# broadcast a packet as well
send_magic_packet(self._mac)
async def async_turn_on(self) -> None:
"""Turn the media player on."""
if self._turn_on:
await self._turn_on.async_run(self.hass, self._context)
if self._turn_on_action:
await self._turn_on_action.async_run(self.hass, self._context)
elif self._mac:
await self.hass.async_add_executor_job(self._wake_on_lan)

12
homeassistant/components/samsungtv/remote.py
View File

@ -7,6 +7,7 @@ from typing import Any
from homeassistant.components.remote import ATTR_NUM_REPEATS, RemoteEntity
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import SamsungTVConfigEntry
@ -49,3 +50,14 @@ class SamsungTVRemote(SamsungTVEntity, RemoteEntity):
for _ in range(num_repeats):
await self._bridge.async_send_keys(command_list)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the remote on."""
if self._turn_on_action:
await self._turn_on_action.async_run(self.hass, self._context)
elif self._mac:
await self.hass.async_add_executor_job(self._wake_on_lan)
else:
raise HomeAssistantError(
f"Entity {self.entity_id} does not support this service."
)

21
tests/components/samsungtv/const.py
View File

@ -3,7 +3,11 @@
from samsungtvws.event import ED_INSTALLED_APP_EVENT
from homeassistant.components import ssdp
from homeassistant.components.samsungtv.const import CONF_SESSION_ID, METHOD_WEBSOCKET
from homeassistant.components.samsungtv.const import (
CONF_SESSION_ID,
METHOD_LEGACY,
METHOD_WEBSOCKET,
)
from homeassistant.components.ssdp import (
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_MANUFACTURER,
@ -21,6 +25,12 @@ from homeassistant.const import (
CONF_TOKEN,
)
MOCK_CONFIG = {
CONF_HOST: "fake_host",
CONF_NAME: "fake",
CONF_PORT: 55000,
CONF_METHOD: METHOD_LEGACY,
}
MOCK_CONFIG_ENCRYPTED_WS = {
CONF_HOST: "fake_host",
CONF_NAME: "fake",
@ -41,6 +51,15 @@ MOCK_ENTRYDATA_WS = {
CONF_MODEL: "any",
CONF_NAME: "any",
}
MOCK_ENTRY_WS_WITH_MAC = {
CONF_IP_ADDRESS: "test",
CONF_HOST: "fake_host",
CONF_METHOD: "websocket",
CONF_MAC: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "fake",
CONF_PORT: 8002,
CONF_TOKEN: "123456789",
}
MOCK_SSDP_DATA_RENDERING_CONTROL_ST = ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",

11
tests/components/samsungtv/test_device_trigger.py
View File

@ -15,7 +15,7 @@ from homeassistant.helpers.device_registry import async_get as get_dev_reg
from homeassistant.setup import async_setup_component
from . import setup_samsungtv_entry
from .test_media_player import ENTITY_ID, MOCK_ENTRYDATA_ENCRYPTED_WS
from .const import MOCK_ENTRYDATA_ENCRYPTED_WS
from tests.common import MockConfigEntry, async_get_device_automations
@ -48,6 +48,7 @@ async def test_if_fires_on_turn_on_request(
) -> None:
"""Test for turn_on and turn_off triggers firing."""
await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS)
entity_id = "media_player.fake"
device_reg = get_dev_reg(hass)
device = device_reg.async_get_device(identifiers={(DOMAIN, "any")})
@ -75,12 +76,12 @@ async def test_if_fires_on_turn_on_request(
{
"trigger": {
"platform": "samsungtv.turn_on",
"entity_id": ENTITY_ID,
"entity_id": entity_id,
},
"action": {
"service": "test.automation",
"data_template": {
"some": ENTITY_ID,
"some": entity_id,
"id": "{{ trigger.id }}",
},
},
@ -90,14 +91,14 @@ async def test_if_fires_on_turn_on_request(
)
await hass.services.async_call(
"media_player", "turn_on", {"entity_id": ENTITY_ID}, blocking=True
"media_player", "turn_on", {"entity_id": entity_id}, blocking=True
)
await hass.async_block_till_done()
assert len(calls) == 2
assert calls[0].data["some"] == device.id
assert calls[0].data["id"] == 0
assert calls[1].data["some"] == ENTITY_ID
assert calls[1].data["some"] == entity_id
assert calls[1].data["id"] == 0

2
tests/components/samsungtv/test_diagnostics.py
View File

@ -10,11 +10,11 @@ from homeassistant.core import HomeAssistant
from . import setup_samsungtv_entry
from .const import (
MOCK_ENTRY_WS_WITH_MAC,
MOCK_ENTRYDATA_ENCRYPTED_WS,
SAMPLE_DEVICE_INFO_UE48JU6400,
SAMPLE_DEVICE_INFO_WIFI,
)
from .test_media_player import MOCK_ENTRY_WS_WITH_MAC
from tests.components.diagnostics import get_diagnostics_for_config_entry
from tests.typing import ClientSessionGenerator

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

@ -42,7 +42,6 @@ from homeassistant.components.samsungtv.const import (
DOMAIN as SAMSUNGTV_DOMAIN,
ENCRYPTED_WEBSOCKET_PORT,
METHOD_ENCRYPTED_WEBSOCKET,
METHOD_LEGACY,
METHOD_WEBSOCKET,
TIMEOUT_WEBSOCKET,
)
@ -82,6 +81,8 @@ import homeassistant.util.dt as dt_util
from . import async_wait_config_entry_reload, setup_samsungtv_entry
from .const import (
MOCK_CONFIG,
MOCK_ENTRY_WS_WITH_MAC,
MOCK_ENTRYDATA_ENCRYPTED_WS,
SAMPLE_DEVICE_INFO_FRAME,
SAMPLE_DEVICE_INFO_WIFI,
@ -91,12 +92,6 @@ from .const import (
from tests.common import MockConfigEntry, async_fire_time_changed
ENTITY_ID = f"{DOMAIN}.fake"
MOCK_CONFIG = {
CONF_HOST: "fake_host",
CONF_NAME: "fake",
CONF_PORT: 55000,
CONF_METHOD: METHOD_LEGACY,
}
MOCK_CONFIGWS = {
CONF_HOST: "fake_host",
CONF_NAME: "fake",
@ -123,17 +118,6 @@ MOCK_ENTRY_WS = {
}
MOCK_ENTRY_WS_WITH_MAC = {
CONF_IP_ADDRESS: "test",
CONF_HOST: "fake_host",
CONF_METHOD: "websocket",
CONF_MAC: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "fake",
CONF_PORT: 8002,
CONF_TOKEN: "123456789",
}
@pytest.mark.usefixtures("remote")
async def test_setup(hass: HomeAssistant) -> None:
"""Test setup of platform."""
@ -1048,7 +1032,7 @@ async def test_turn_on_wol(hass: HomeAssistant) -> None:
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
with patch(
"homeassistant.components.samsungtv.media_player.send_magic_packet"
"homeassistant.components.samsungtv.entity.send_magic_packet"
) as mock_send_magic_packet:
await hass.services.async_call(
DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID}, True
@ -1060,7 +1044,7 @@ async def test_turn_on_wol(hass: HomeAssistant) -> None:
async def test_turn_on_without_turnon(hass: HomeAssistant, remote: Mock) -> None:
"""Test turn on."""
await setup_samsungtv_entry(hass, MOCK_CONFIG)
with pytest.raises(HomeAssistantError):
with pytest.raises(HomeAssistantError, match="does not support this service"):
await hass.services.async_call(
DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID}, True
)

42
tests/components/samsungtv/test_remote.py
View File

@ -1,6 +1,6 @@
"""The tests for the SamsungTV remote platform."""
from unittest.mock import Mock
from unittest.mock import Mock, patch
import pytest
from samsungtvws.encrypted.remote import SamsungTVEncryptedCommand
@ -10,12 +10,16 @@ from homeassistant.components.remote import (
DOMAIN as REMOTE_DOMAIN,
SERVICE_SEND_COMMAND,
)
from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TURN_OFF
from homeassistant.components.samsungtv.const import DOMAIN as SAMSUNGTV_DOMAIN
from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TURN_OFF, SERVICE_TURN_ON
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from . import setup_samsungtv_entry
from .test_media_player import MOCK_ENTRYDATA_ENCRYPTED_WS
from .const import MOCK_CONFIG, MOCK_ENTRY_WS_WITH_MAC, MOCK_ENTRYDATA_ENCRYPTED_WS
from tests.common import MockConfigEntry
ENTITY_ID = f"{REMOTE_DOMAIN}.fake"
@ -92,3 +96,35 @@ async def test_send_command_service(hass: HomeAssistant, remoteencws: Mock) -> N
assert len(commands) == 1
assert isinstance(command := commands[0], SamsungTVEncryptedCommand)
assert command.body["param3"] == "dash"
@pytest.mark.usefixtures("remotews", "rest_api")
async def test_turn_on_wol(hass: HomeAssistant) -> None:
"""Test turn on."""
entry = MockConfigEntry(
domain=SAMSUNGTV_DOMAIN,
data=MOCK_ENTRY_WS_WITH_MAC,
unique_id="any",
)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
with patch(
"homeassistant.components.samsungtv.entity.send_magic_packet"
) as mock_send_magic_packet:
await hass.services.async_call(
REMOTE_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
await hass.async_block_till_done()
assert mock_send_magic_packet.called
async def test_turn_on_without_turnon(hass: HomeAssistant, remote: Mock) -> None:
"""Test turn on."""
await setup_samsungtv_entry(hass, MOCK_CONFIG)
with pytest.raises(HomeAssistantError, match="does not support this service"):
await hass.services.async_call(
REMOTE_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
# nothing called as not supported feature
assert remote.control.call_count == 0

45
tests/components/samsungtv/test_trigger.py
View File

@ -6,24 +6,30 @@ import pytest
from homeassistant.components import automation
from homeassistant.components.samsungtv import DOMAIN
from homeassistant.const import SERVICE_RELOAD
from homeassistant.const import SERVICE_RELOAD, SERVICE_TURN_ON
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.helpers import device_registry as dr
from homeassistant.setup import async_setup_component
from . import setup_samsungtv_entry
from .test_media_player import ENTITY_ID, MOCK_ENTRYDATA_ENCRYPTED_WS
from .const import MOCK_ENTRYDATA_ENCRYPTED_WS
from tests.common import MockEntity, MockEntityPlatform
@pytest.mark.usefixtures("remoteencws", "rest_api")
@pytest.mark.parametrize("entity_domain", ["media_player", "remote"])
async def test_turn_on_trigger_device_id(
hass: HomeAssistant, calls: list[ServiceCall], device_registry: dr.DeviceRegistry
hass: HomeAssistant,
calls: list[ServiceCall],
device_registry: dr.DeviceRegistry,
entity_domain: str,
) -> None:
"""Test for turn_on triggers by device_id firing."""
await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS)
entity_id = f"{entity_domain}.fake"
device = device_registry.async_get_device(identifiers={(DOMAIN, "any")})
assert device, repr(device_registry.devices)
@ -50,7 +56,7 @@ async def test_turn_on_trigger_device_id(
)
await hass.services.async_call(
"media_player", "turn_on", {"entity_id": ENTITY_ID}, blocking=True
entity_domain, SERVICE_TURN_ON, {"entity_id": entity_id}, blocking=True
)
await hass.async_block_till_done()
@ -65,10 +71,10 @@ async def test_turn_on_trigger_device_id(
# Ensure WOL backup is called when trigger not present
with patch(
"homeassistant.components.samsungtv.media_player.send_magic_packet"
"homeassistant.components.samsungtv.entity.send_magic_packet"
) as mock_send_magic_packet:
await hass.services.async_call(
"media_player", "turn_on", {"entity_id": ENTITY_ID}, blocking=True
entity_domain, SERVICE_TURN_ON, {"entity_id": entity_id}, blocking=True
)
await hass.async_block_till_done()
@ -77,12 +83,15 @@ async def test_turn_on_trigger_device_id(
@pytest.mark.usefixtures("remoteencws", "rest_api")
@pytest.mark.parametrize("entity_domain", ["media_player", "remote"])
async def test_turn_on_trigger_entity_id(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, calls: list[ServiceCall], entity_domain: str
) -> None:
"""Test for turn_on triggers by entity_id firing."""
await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS)
entity_id = f"{entity_domain}.fake"
assert await async_setup_component(
hass,
automation.DOMAIN,
@ -91,12 +100,12 @@ async def test_turn_on_trigger_entity_id(
{
"trigger": {
"platform": "samsungtv.turn_on",
"entity_id": ENTITY_ID,
"entity_id": entity_id,
},
"action": {
"service": "test.automation",
"data_template": {
"some": ENTITY_ID,
"some": entity_id,
"id": "{{ trigger.id }}",
},
},
@ -106,21 +115,23 @@ async def test_turn_on_trigger_entity_id(
)
await hass.services.async_call(
"media_player", "turn_on", {"entity_id": ENTITY_ID}, blocking=True
entity_domain, SERVICE_TURN_ON, {"entity_id": entity_id}, blocking=True
)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == ENTITY_ID
assert calls[0].data["some"] == entity_id
assert calls[0].data["id"] == 0
@pytest.mark.usefixtures("remoteencws", "rest_api")
@pytest.mark.parametrize("entity_domain", ["media_player", "remote"])
async def test_wrong_trigger_platform_type(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, entity_domain: str
) -> None:
"""Test wrong trigger platform type."""
await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS)
entity_id = f"{entity_domain}.fake"
await async_setup_component(
hass,
@ -130,12 +141,12 @@ async def test_wrong_trigger_platform_type(
{
"trigger": {
"platform": "samsungtv.wrong_type",
"entity_id": ENTITY_ID,
"entity_id": entity_id,
},
"action": {
"service": "test.automation",
"data_template": {
"some": ENTITY_ID,
"some": entity_id,
"id": "{{ trigger.id }}",
},
},
@ -151,11 +162,13 @@ async def test_wrong_trigger_platform_type(
@pytest.mark.usefixtures("remoteencws", "rest_api")
@pytest.mark.parametrize("entity_domain", ["media_player", "remote"])
async def test_trigger_invalid_entity_id(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, entity_domain: str
) -> None:
"""Test turn on trigger using invalid entity_id."""
await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS)
entity_id = f"{entity_domain}.fake"
platform = MockEntityPlatform(hass)
@ -175,7 +188,7 @@ async def test_trigger_invalid_entity_id(
"action": {
"service": "test.automation",
"data_template": {
"some": ENTITY_ID,
"some": entity_id,
"id": "{{ trigger.id }}",
},
},