Add LG webOS TV screen power switch entity (#177938)

Co-authored-by: Shay Levy <levyshay1@gmail.com>
This commit is contained in:
darkrain-nl
2026-08-03 23:28:30 +03:00
committed by GitHub
co-authored by Shay Levy
parent f8fcf39bd4
commit 206227238c
8 changed files with 261 additions and 69 deletions
+1 -1
View File
@@ -10,7 +10,7 @@ from homeassistant.const import Platform
DOMAIN = "webostv"
LOGGER = logging.getLogger(__package__)
PLATFORMS = [Platform.MEDIA_PLAYER]
PLATFORMS = [Platform.MEDIA_PLAYER, Platform.SWITCH]
DEFAULT_NAME = "LG webOS TV"
ATTR_PAYLOAD = "payload"
@@ -0,0 +1,62 @@
"""Base entity for the LG webOS TV integration."""
from collections.abc import Callable, Coroutine
from functools import wraps
from typing import Any, Concatenate, cast
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, WEBOSTV_EXCEPTIONS
from .coordinator import WebOsTvConfigEntry, WebOsTvDataUpdateCoordinator
class WebOsTvEntity(CoordinatorEntity[WebOsTvDataUpdateCoordinator]):
"""Base entity for the LG webOS TV integration."""
_attr_has_entity_name = True
_attr_device_info: DeviceInfo
def __init__(self, entry: WebOsTvConfigEntry) -> None:
"""Initialize the entity."""
super().__init__(entry.runtime_data)
self._client = entry.runtime_data.client
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, cast(str, entry.unique_id))},
manufacturer="LG",
name=entry.title,
)
def cmd[_EntityT: WebOsTvEntity, _R, **_P](
func: Callable[Concatenate[_EntityT, _P], Coroutine[Any, Any, _R]],
) -> Callable[Concatenate[_EntityT, _P], Coroutine[Any, Any, _R]]:
"""Catch command exceptions."""
@wraps(func)
async def cmd_wrapper(self: _EntityT, *args: _P.args, **kwargs: _P.kwargs) -> _R:
"""Wrap all command methods."""
if not self._client.tv_state.is_on and func.__name__ != "async_turn_off":
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="device_off",
translation_placeholders={
"name": self.coordinator.name,
"func": func.__name__,
},
)
try:
return await func(self, *args, **kwargs)
except WEBOSTV_EXCEPTIONS as error:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="communication_error",
translation_placeholders={
"name": self.coordinator.name,
"func": func.__name__,
"error": str(error),
},
) from error
return cmd_wrapper
@@ -1,4 +1,11 @@
{
"entity": {
"switch": {
"screen": {
"default": "mdi:television"
}
}
},
"services": {
"button": {
"service": "mdi:button-pointer"
@@ -1,11 +1,9 @@
"""Support for interface with an LG webOS TV."""
import asyncio
from collections.abc import Callable, Coroutine
from contextlib import suppress
from functools import wraps
from http import HTTPStatus
from typing import Any, Concatenate, cast, override
from typing import Any, cast, override
from homeassistant.components.media_player import (
MediaPlayerDeviceClass,
@@ -18,10 +16,8 @@ from homeassistant.const import EntityStateAttribute
from homeassistant.core import HomeAssistant, ServiceResponse, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import (
ATTR_PAYLOAD,
@@ -30,9 +26,9 @@ from .const import (
DOMAIN,
LIVE_TV_APP_ID,
LOGGER,
WEBOSTV_EXCEPTIONS,
)
from .coordinator import WebOsTvConfigEntry, WebOsTvDataUpdateCoordinator
from .coordinator import WebOsTvConfigEntry
from .entity import WebOsTvEntity, cmd
from .triggers.turn_on import async_get_turn_on_trigger
SUPPORT_WEBOSTV = (
@@ -63,57 +59,16 @@ async def async_setup_entry(
async_add_entities([LgWebOSMediaPlayerEntity(entry)])
def cmd[_R, **_P](
func: Callable[Concatenate[LgWebOSMediaPlayerEntity, _P], Coroutine[Any, Any, _R]],
) -> Callable[Concatenate[LgWebOSMediaPlayerEntity, _P], Coroutine[Any, Any, _R]]:
"""Catch command exceptions."""
@wraps(func)
async def cmd_wrapper(
self: LgWebOSMediaPlayerEntity, *args: _P.args, **kwargs: _P.kwargs
) -> _R:
"""Wrap all command methods."""
if self.state is MediaPlayerState.OFF and func.__name__ != "async_turn_off":
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="device_off",
translation_placeholders={
"name": str(self._entry.title),
"func": func.__name__,
},
)
try:
return await func(self, *args, **kwargs)
except WEBOSTV_EXCEPTIONS as error:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="communication_error",
translation_placeholders={
"name": str(self._entry.title),
"func": func.__name__,
"error": str(error),
},
) from error
return cmd_wrapper
class LgWebOSMediaPlayerEntity(
CoordinatorEntity[WebOsTvDataUpdateCoordinator], RestoreEntity, MediaPlayerEntity
):
class LgWebOSMediaPlayerEntity(WebOsTvEntity, RestoreEntity, MediaPlayerEntity):
"""Representation of a LG webOS TV."""
_attr_device_class = MediaPlayerDeviceClass.TV
_attr_has_entity_name = True
_attr_name = None
def __init__(self, entry: WebOsTvConfigEntry) -> None:
"""Initialize the webos device."""
super().__init__(entry.runtime_data)
self._entry = entry
self._client = entry.runtime_data.client
super().__init__(entry)
self._attr_assumed_state = True
self._device_name = entry.title
self._attr_unique_id = entry.unique_id
self._sources = entry.options.get(CONF_SOURCES)
@@ -163,12 +118,6 @@ class LgWebOSMediaPlayerEntity(
self._attr_extra_state_attributes = {}
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, cast(str, self.unique_id))},
manufacturer="LG",
name=self._device_name,
)
if tv_state.is_on or not self._supported_features:
supported = SUPPORT_WEBOSTV
if tv_state.sound_output == "external_speaker":
@@ -52,20 +52,12 @@ rules:
dynamic-devices:
status: exempt
comment: The integration connects to a single device.
entity-category:
status: exempt
comment: The integration only registers one entity.
entity-category: done
entity-device-class: done
entity-disabled-by-default:
status: exempt
comment: The integration only registers one entity.
entity-translations:
status: exempt
comment: There are no entities to translate.
entity-disabled-by-default: done
entity-translations: done
exception-translations: done
icon-translations:
status: exempt
comment: The only entity can use the device class.
icon-translations: done
reconfiguration-flow: done
repair-issues:
status: exempt
@@ -45,6 +45,13 @@
"webostv.turn_on": "Device is requested to turn on"
}
},
"entity": {
"switch": {
"screen": {
"name": "Screen"
}
}
},
"exceptions": {
"auth_failed": {
"message": "Pairing for {device} failed, make sure to accept the pairing request on your TV."
@@ -0,0 +1,56 @@
"""Support for LG webOS TV switch."""
from typing import Any, override
from homeassistant.components.switch import SwitchEntity
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import WebOsTvConfigEntry
from .entity import WebOsTvEntity, cmd
PARALLEL_UPDATES = 0
async def async_setup_entry(
hass: HomeAssistant,
entry: WebOsTvConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the LG webOS TV switch platform."""
async_add_entities([LgWebOSScreenSwitchEntity(entry)])
class LgWebOSScreenSwitchEntity(WebOsTvEntity, SwitchEntity):
"""Representation of a LG webOS TV Screen Switch."""
_attr_translation_key = "screen"
def __init__(self, entry: WebOsTvConfigEntry) -> None:
"""Initialize the screen switch entity."""
super().__init__(entry)
self._attr_unique_id = f"{entry.unique_id}_screen"
@property
@override
def available(self) -> bool:
"""Return true if the entity is available."""
return super().available and self._client.tv_state.is_on
@property
@override
def is_on(self) -> bool:
"""Return true if screen is on."""
return self._client.tv_state.is_screen_on
@cmd
@override
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the screen on."""
await self._client.set_screen_state(True)
@cmd
@override
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the screen off."""
await self._client.set_screen_state(False)
+119
View File
@@ -0,0 +1,119 @@
"""Tests for LG webOS TV switch platform."""
from unittest.mock import AsyncMock
from aiowebostv import WebOsTvCommandError
import pytest
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.components.webostv.const import DOMAIN
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from . import setup_webostv
from .const import FAKE_UUID
SWITCH_ENTITY_ID = f"{SWITCH_DOMAIN}.lg_webos_tv_model_screen"
async def test_screen_switch_setup(
hass: HomeAssistant,
client: AsyncMock,
entity_registry: er.EntityRegistry,
) -> None:
"""Test setup of LG webOS TV screen switch."""
await setup_webostv(hass)
entry = entity_registry.async_get(SWITCH_ENTITY_ID)
assert entry is not None
assert entry.unique_id == f"{FAKE_UUID}_screen"
state = hass.states.get(SWITCH_ENTITY_ID)
assert state is not None
assert state.state == STATE_OFF
async def test_screen_switch_state_updates(
hass: HomeAssistant,
client: AsyncMock,
) -> None:
"""Test screen switch state updates from client."""
await setup_webostv(hass)
state = hass.states.get(SWITCH_ENTITY_ID)
assert state is not None
assert state.state == STATE_OFF
client.tv_state.is_screen_on = True
await client.mock_state_update()
await hass.async_block_till_done()
state = hass.states.get(SWITCH_ENTITY_ID)
assert state is not None
assert state.state == STATE_ON
client.tv_state.is_on = False
client.tv_state.is_screen_on = False
await client.mock_state_update()
await hass.async_block_till_done()
state = hass.states.get(SWITCH_ENTITY_ID)
assert state is not None
assert state.state == STATE_UNAVAILABLE
@pytest.mark.parametrize(
("service", "screen_state"),
[
(SERVICE_TURN_ON, True),
(SERVICE_TURN_OFF, False),
],
)
async def test_screen_switch_commands(
hass: HomeAssistant,
client: AsyncMock,
service: str,
screen_state: bool,
) -> None:
"""Test the screen switch sets the screen state."""
await setup_webostv(hass)
await hass.services.async_call(
SWITCH_DOMAIN,
service,
{ATTR_ENTITY_ID: SWITCH_ENTITY_ID},
blocking=True,
)
client.set_screen_state.assert_called_once_with(screen_state)
@pytest.mark.parametrize("service", [SERVICE_TURN_ON, SERVICE_TURN_OFF])
async def test_screen_switch_command_error(
hass: HomeAssistant,
client: AsyncMock,
service: str,
) -> None:
"""Test a failing screen command raises a translated error."""
await setup_webostv(hass)
client.set_screen_state.side_effect = WebOsTvCommandError("Communication error")
with pytest.raises(HomeAssistantError) as err:
await hass.services.async_call(
SWITCH_DOMAIN,
service,
{ATTR_ENTITY_ID: SWITCH_ENTITY_ID},
blocking=True,
)
assert err.value.translation_domain == DOMAIN
assert err.value.translation_key == "communication_error"