jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Enable strict typing in webostv (#64193)

Browse Source 
* Enable strict typing in webostv

Enable strict typing in webostv

* Apply review comments
This commit is contained in:
Shay Levy 2022-01-16 23:07:29 +02:00 committed by GitHub
parent 3d7572843e
commit 32d4f104ff
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 172 additions and 116 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.strict-typing
View File

@ -190,6 +190,7 @@ homeassistant.components.wallbox.*
homeassistant.components.water_heater.*
homeassistant.components.watttime.*
homeassistant.components.weather.*
homeassistant.components.webostv.*
homeassistant.components.websocket_api.*
homeassistant.components.wemo.*
homeassistant.components.whois.*

69
homeassistant/components/webostv/__init__.py
View File

@ -16,7 +16,7 @@ import voluptuous as vol
from homeassistant.components import notify as hass_notify
from homeassistant.components.automation import AutomationActionType
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
ATTR_COMMAND,
ATTR_ENTITY_ID,
@ -29,7 +29,14 @@ from homeassistant.const import (
EVENT_HOMEASSISTANT_STOP,
Platform,
)
from homeassistant.core import Context, HassJob, HomeAssistant, ServiceCall, callback
from homeassistant.core import (
Context,
Event,
HassJob,
HomeAssistant,
ServiceCall,
callback,
)
from homeassistant.helpers import config_validation as cv, discovery, entity_registry
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.typing import ConfigType
@ -102,7 +109,7 @@ SERVICE_TO_METHOD = {
_LOGGER = logging.getLogger(__name__)
def read_client_keys(config_file):
def read_client_keys(config_file: str) -> dict[str, str]:
"""Read legacy client keys from file."""
if not os.path.isfile(config_file):
return {}
@ -111,7 +118,9 @@ def read_client_keys(config_file):
with open(config_file, encoding="utf8") as json_file:
try:
client_keys = json.load(json_file)
return client_keys
if isinstance(client_keys, dict):
return client_keys
return {}
except (json.JSONDecodeError, UnicodeDecodeError):
pass
@ -119,8 +128,8 @@ def read_client_keys(config_file):
engine = db.create_engine(f"sqlite:///{config_file}")
table = db.Table("unnamed", db.MetaData(), autoload=True, autoload_with=engine)
results = engine.connect().execute(db.select([table])).fetchall()
client_keys = {k: loads(v) for k, v in results}
return client_keys
db_client_keys = {k: loads(v) for k, v in results}
return db_client_keys
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
@ -139,7 +148,9 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
_LOGGER.debug("No pairing keys, Not importing webOS Smart TV YAML config")
return True
async def async_migrate_task(entity_id, conf, key):
async def async_migrate_task(
entity_id: str, conf: dict[str, str], key: str
) -> None:
_LOGGER.debug("Migrating webOS Smart TV entity %s unique_id", entity_id)
client = WebOsClient(conf[CONF_HOST], key)
tries = 0
@ -181,7 +192,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
if entity_id := ent_reg.async_get_entity_id(Platform.MEDIA_PLAYER, DOMAIN, key):
tasks.append(asyncio.create_task(async_migrate_task(entity_id, conf, key)))
async def async_tasks_cancel(_event):
async def async_tasks_cancel(_event: Event) -> None:
"""Cancel config flow import tasks."""
for task in tasks:
if not task.done():
@ -192,12 +203,12 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
return True
def _async_migrate_options_from_data(hass, config_entry):
def _async_migrate_options_from_data(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Migrate options from data."""
if config_entry.options:
if entry.options:
return
config = config_entry.data
config = entry.data
options = {}
# Get Preferred Sources
@ -206,15 +217,15 @@ def _async_migrate_options_from_data(hass, config_entry):
if not isinstance(sources, list):
options[CONF_SOURCES] = sources.split(",")
hass.config_entries.async_update_entry(config_entry, options=options)
hass.config_entries.async_update_entry(entry, options=options)
async def async_setup_entry(hass, config_entry):
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set the config entry up."""
_async_migrate_options_from_data(hass, config_entry)
_async_migrate_options_from_data(hass, entry)
host = config_entry.data[CONF_HOST]
key = config_entry.data[CONF_CLIENT_SECRET]
host = entry.data[CONF_HOST]
key = entry.data[CONF_CLIENT_SECRET]
wrapper = WebOsClientWrapper(host, client_key=key)
await wrapper.connect()
@ -231,8 +242,8 @@ async def async_setup_entry(hass, config_entry):
DOMAIN, service, async_service_handler, schema=schema
)
hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id] = wrapper
hass.config_entries.async_setup_platforms(config_entry, PLATFORMS)
hass.data[DOMAIN][DATA_CONFIG_ENTRY][entry.entry_id] = wrapper
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
# set up notify platform, no entry support for notify component yet,
# have to use discovery to load platform.
@ -242,31 +253,29 @@ async def async_setup_entry(hass, config_entry):
"notify",
DOMAIN,
{
CONF_NAME: config_entry.title,
ATTR_CONFIG_ENTRY_ID: config_entry.entry_id,
CONF_NAME: entry.title,
ATTR_CONFIG_ENTRY_ID: entry.entry_id,
},
hass.data[DOMAIN][DATA_HASS_CONFIG],
)
)
if not config_entry.update_listeners:
config_entry.async_on_unload(
config_entry.add_update_listener(async_update_options)
)
if not entry.update_listeners:
entry.async_on_unload(entry.add_update_listener(async_update_options))
async def async_on_stop(_event):
async def async_on_stop(_event: Event) -> None:
"""Unregister callbacks and disconnect."""
await wrapper.shutdown()
config_entry.async_on_unload(
entry.async_on_unload(
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, async_on_stop)
)
return True
async def async_update_options(hass, config_entry):
async def async_update_options(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Update options."""
await hass.config_entries.async_reload(config_entry.entry_id)
await hass.config_entries.async_reload(entry.entry_id)
async def async_control_connect(host: str, key: str | None) -> WebOsClient:
@ -281,7 +290,7 @@ async def async_control_connect(host: str, key: str | None) -> WebOsClient:
return client
async def async_unload_entry(hass, entry):
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
@ -305,7 +314,7 @@ class PluggableAction:
"""Initialize."""
self._actions: dict[Callable[[], None], tuple[HassJob, dict[str, Any]]] = {}
def __bool__(self):
def __bool__(self) -> bool:
"""Return if we have something attached."""
return bool(self._actions)

46
homeassistant/components/webostv/config_flow.py
View File

@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from typing import Any
from urllib.parse import urlparse
from aiowebostv import WebOsTvPairError
@ -13,6 +14,7 @@ from homeassistant.const import CONF_CLIENT_SECRET, CONF_HOST, CONF_NAME, CONF_U
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import ConfigType
from . import async_control_connect
from .const import CONF_SOURCES, DEFAULT_NAME, DOMAIN, WEBOSTV_EXCEPTIONS
@ -33,19 +35,21 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1
def __init__(self):
def __init__(self) -> None:
"""Initialize workflow."""
self._host = None
self._name = None
self._uuid = None
self._host: str = ""
self._name: str = ""
self._uuid: str | None = None
@staticmethod
@callback
def async_get_options_flow(config_entry):
def async_get_options_flow(
config_entry: config_entries.ConfigEntry,
) -> OptionsFlowHandler:
"""Get the options flow for this handler."""
return OptionsFlowHandler(config_entry)
async def async_step_import(self, import_info):
async def async_step_import(self, import_info: dict[str, Any]) -> FlowResult:
"""Set the config entry up from yaml."""
self._host = import_info[CONF_HOST]
self._name = import_info.get(CONF_NAME) or import_info[CONF_HOST]
@ -60,9 +64,11 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
_LOGGER.debug("WebOS Smart TV host %s imported from YAML config", self._host)
return self.async_create_entry(title=self._name, data=data)
async def async_step_user(self, user_input=None):
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initialized by the user."""
errors = {}
errors: dict[str, str] = {}
if user_input is not None:
self._host = user_input[CONF_HOST]
self._name = user_input[CONF_NAME]
@ -73,7 +79,7 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
)
@callback
def _async_check_configured_entry(self):
def _async_check_configured_entry(self) -> None:
"""Check if entry is configured, update unique_id if needed."""
for entry in self._async_current_entries(include_ignore=False):
if entry.data[CONF_HOST] != self._host:
@ -89,7 +95,9 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
raise data_entry_flow.AbortFlow("already_configured")
async def async_step_pairing(self, user_input=None):
async def async_step_pairing(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Display pairing form."""
self._async_check_configured_entry()
@ -119,10 +127,14 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_ssdp(self, discovery_info: ssdp.SsdpServiceInfo) -> FlowResult:
"""Handle a flow initialized by discovery."""
self._host = urlparse(discovery_info.ssdp_location).hostname
assert discovery_info.ssdp_location
host = urlparse(discovery_info.ssdp_location).hostname
assert host
self._host = host
self._name = discovery_info.upnp[ssdp.ATTR_UPNP_FRIENDLY_NAME]
uuid = discovery_info.upnp[ssdp.ATTR_UPNP_UDN]
assert uuid
if uuid.startswith("uuid:"):
uuid = uuid[5:]
await self.async_set_unique_id(uuid)
@ -139,14 +151,14 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
class OptionsFlowHandler(config_entries.OptionsFlow):
"""Handle options."""
def __init__(self, config_entry):
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
"""Initialize options flow."""
self.config_entry = config_entry
self.options = config_entry.options
self.host = config_entry.data[CONF_HOST]
self.key = config_entry.data[CONF_CLIENT_SECRET]
async def async_step_init(self, user_input=None):
async def async_step_init(self, user_input: ConfigType | None = None) -> FlowResult:
"""Manage the options."""
errors = {}
if user_input is not None:
@ -155,7 +167,7 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
# Get sources
sources = self.options.get(CONF_SOURCES, "")
sources_list = await async_get_sources(self.host, self.key)
if sources_list is None:
if not sources_list:
errors["base"] = "cannot_retrieve"
options_schema = vol.Schema(
@ -163,7 +175,7 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
vol.Optional(
CONF_SOURCES,
description={"suggested_value": sources},
): cv.multi_select(sources_list),
): cv.multi_select({source: source for source in sources_list}),
}
)
@ -172,12 +184,12 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
)
async def async_get_sources(host, key):
async def async_get_sources(host: str, key: str) -> list[str]:
"""Construct sources list."""
try:
client = await async_control_connect(host, key)
except WEBOSTV_EXCEPTIONS:
return None
return []
return [
*(app["title"] for app in client.apps.values()),

1
homeassistant/components/webostv/helpers.py
View File

@ -71,6 +71,7 @@ def async_get_client_wrapper_by_device_entry(
Raises ValueError if client wrapper is not found.
"""
for config_entry_id in device.config_entries:
wrapper: WebOsClientWrapper | None
if wrapper := hass.data[DOMAIN][DATA_CONFIG_ENTRY].get(config_entry_id):
break

125
homeassistant/components/webostv/media_player.py
View File

@ -1,12 +1,15 @@
"""Support for interface with an LG webOS Smart TV."""
from __future__ import annotations
from collections.abc import Awaitable, Callable, Coroutine
from contextlib import suppress
from datetime import timedelta
from functools import wraps
import logging
from typing import Any, TypeVar, cast
from aiowebostv import WebOsClient, WebOsTvPairError
from typing_extensions import Concatenate, ParamSpec
from homeassistant import util
from homeassistant.components.media_player import (
@ -37,6 +40,7 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import WebOsClientWrapper
@ -76,6 +80,7 @@ async def async_setup_entry(
) -> None:
"""Set up the LG webOS Smart TV platform."""
unique_id = config_entry.unique_id
assert unique_id
name = config_entry.title
sources = config_entry.options.get(CONF_SOURCES)
wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id]
@ -83,17 +88,23 @@ async def async_setup_entry(
async_add_entities([LgWebOSMediaPlayerEntity(wrapper, name, sources, unique_id)])
def cmd(func):
_T = TypeVar("_T", bound="LgWebOSMediaPlayerEntity")
_P = ParamSpec("_P")
def cmd(
func: Callable[Concatenate[_T, _P], Awaitable[None]] # type: ignore[misc]
) -> Callable[Concatenate[_T, _P], Coroutine[Any, Any, None]]: # type: ignore[misc]
"""Catch command exceptions."""
@wraps(func)
async def cmd_wrapper(obj, *args, **kwargs):
async def cmd_wrapper(self: _T, *args: _P.args, **kwargs: _P.kwargs) -> None:
"""Wrap all command methods."""
try:
await func(obj, *args, **kwargs)
await func(self, *args, **kwargs)
except WEBOSTV_EXCEPTIONS as exc:
# If TV is off, we expect calls to fail.
if obj.state == STATE_OFF:
if self.state == STATE_OFF:
level = logging.INFO
else:
level = logging.ERROR
@ -101,7 +112,7 @@ def cmd(func):
level,
"Error calling %s on entity %s: %r",
func.__name__,
obj.entity_id,
self.entity_id,
exc,
)
@ -112,11 +123,15 @@ class LgWebOSMediaPlayerEntity(MediaPlayerEntity):
"""Representation of a LG webOS Smart TV."""
def __init__(
self, wrapper: WebOsClientWrapper, name: str, sources, unique_id
self,
wrapper: WebOsClientWrapper,
name: str,
sources: list[str] | None,
unique_id: str,
) -> None:
"""Initialize the webos device."""
self._wrapper = wrapper
self._client = wrapper.client
self._client: WebOsClient = wrapper.client
self._name = name
self._unique_id = unique_id
self._sources = sources
@ -127,7 +142,7 @@ class LgWebOSMediaPlayerEntity(MediaPlayerEntity):
self._current_source = None
self._source_list: dict = {}
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Connect and subscribe to dispatcher signals and state updates."""
self.async_on_remove(
async_dispatcher_connect(self.hass, DOMAIN, self.async_signal_handler)
@ -137,11 +152,11 @@ class LgWebOSMediaPlayerEntity(MediaPlayerEntity):
self.async_handle_state_update
)
async def async_will_remove_from_hass(self):
async def async_will_remove_from_hass(self) -> None:
"""Call disconnect on removal."""
self._client.unregister_state_update_callback(self.async_handle_state_update)
async def async_signal_handler(self, data):
async def async_signal_handler(self, data: dict[str, Any]) -> None:
"""Handle domain-specific signal by calling appropriate method."""
if (entity_ids := data[ATTR_ENTITY_ID]) == ENTITY_MATCH_NONE:
return
@ -154,12 +169,12 @@ class LgWebOSMediaPlayerEntity(MediaPlayerEntity):
}
await getattr(self, data["method"])(**params)
async def async_handle_state_update(self, _client: WebOsClient):
async def async_handle_state_update(self, _client: WebOsClient) -> None:
"""Update state from WebOsClient."""
self.update_sources()
self.async_write_ha_state()
def update_sources(self):
def update_sources(self) -> None:
"""Update list of sources from current source, apps, inputs and configured list."""
source_list = self._source_list
self._source_list = {}
@ -211,7 +226,7 @@ class LgWebOSMediaPlayerEntity(MediaPlayerEntity):
self._source_list = source_list
@util.Throttle(MIN_TIME_BETWEEN_SCANS, MIN_TIME_BETWEEN_FORCED_SCANS)
async def async_update(self):
async def async_update(self) -> None:
"""Connect."""
if self._client.is_connected():
return
@ -220,22 +235,22 @@ class LgWebOSMediaPlayerEntity(MediaPlayerEntity):
await self._client.connect()
@property
def unique_id(self):
def unique_id(self) -> str:
"""Return the unique id of the device."""
return self._unique_id
@property
def name(self):
def name(self) -> str:
"""Return the name of the device."""
return self._name
@property
def device_class(self):
def device_class(self) -> MediaPlayerDeviceClass:
"""Return the device class of the device."""
return MediaPlayerDeviceClass.TV
@property
def state(self):
def state(self) -> str:
"""Return the state of the device."""
if self._client.is_on:
return STATE_ON
@ -243,30 +258,30 @@ class LgWebOSMediaPlayerEntity(MediaPlayerEntity):
return STATE_OFF
@property
def is_volume_muted(self):
def is_volume_muted(self) -> bool:
"""Boolean if volume is currently muted."""
return self._client.muted
return cast(bool, self._client.muted)
@property
def volume_level(self):
def volume_level(self) -> float | None:
"""Volume level of the media player (0..1)."""
if self._client.volume is not None:
return self._client.volume / 100.0
return cast(float, self._client.volume / 100.0)
return None
@property
def source(self):
def source(self) -> str | None:
"""Return the current input source."""
return self._current_source
@property
def source_list(self):
def source_list(self) -> list[str]:
"""List of available input sources."""
return sorted(self._source_list)
@property
def media_content_type(self):
def media_content_type(self) -> str | None:
"""Content type of current playing media."""
if self._client.current_app_id == LIVE_TV_APP_ID:
return MEDIA_TYPE_CHANNEL
@ -274,26 +289,26 @@ class LgWebOSMediaPlayerEntity(MediaPlayerEntity):
return None
@property
def media_title(self):
def media_title(self) -> str | None:
"""Title of current playing media."""
if (self._client.current_app_id == LIVE_TV_APP_ID) and (
self._client.current_channel is not None
):
return self._client.current_channel.get("channelName")
return cast(str, self._client.current_channel.get("channelName"))
return None
@property
def media_image_url(self):
def media_image_url(self) -> str | None:
"""Image url of current playing media."""
if self._client.current_app_id in self._client.apps:
icon = self._client.apps[self._client.current_app_id]["largeIcon"]
icon: str = self._client.apps[self._client.current_app_id]["largeIcon"]
if not icon.startswith("http"):
icon = self._client.apps[self._client.current_app_id]["icon"]
return icon
return None
@property
def supported_features(self):
def supported_features(self) -> int:
"""Flag media player features that are supported."""
supported = SUPPORT_WEBOSTV
@ -308,13 +323,13 @@ class LgWebOSMediaPlayerEntity(MediaPlayerEntity):
return supported
@property
def device_info(self):
def device_info(self) -> DeviceInfo:
"""Return device information."""
device_info = {
"identifiers": {(DOMAIN, self._unique_id)},
"manufacturer": "LG",
"name": self._name,
}
device_info = DeviceInfo(
identifiers={(DOMAIN, self._unique_id)},
manufacturer="LG",
name=self._name,
)
if self._client.system_info is None and self.state == STATE_OFF:
return device_info
@ -331,49 +346,49 @@ class LgWebOSMediaPlayerEntity(MediaPlayerEntity):
return device_info
@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> dict[str, str] | None:
"""Return device specific state attributes."""
if self._client.sound_output is None and self.state == STATE_OFF:
return {}
return None
return {ATTR_SOUND_OUTPUT: self._client.sound_output}
@cmd
async def async_turn_off(self):
async def async_turn_off(self) -> None:
"""Turn off media player."""
await self._client.power_off()
async def async_turn_on(self):
async def async_turn_on(self) -> None:
"""Turn on media player."""
self._wrapper.turn_on.async_run(self.hass, self._context)
@cmd
async def async_volume_up(self):
async def async_volume_up(self) -> None:
"""Volume up the media player."""
await self._client.volume_up()
@cmd
async def async_volume_down(self):
async def async_volume_down(self) -> None:
"""Volume down media player."""
await self._client.volume_down()
@cmd
async def async_set_volume_level(self, volume):
async def async_set_volume_level(self, volume: int) -> None:
"""Set volume level, range 0..1."""
tv_volume = int(round(volume * 100))
await self._client.set_volume(tv_volume)
@cmd
async def async_mute_volume(self, mute):
async def async_mute_volume(self, mute: bool) -> None:
"""Send mute command."""
await self._client.set_mute(mute)
@cmd
async def async_select_sound_output(self, sound_output):
async def async_select_sound_output(self, sound_output: str) -> None:
"""Select the sound output."""
await self._client.change_sound_output(sound_output)
@cmd
async def async_media_play_pause(self):
async def async_media_play_pause(self) -> None:
"""Simulate play pause media player."""
if self._paused:
await self.async_media_play()
@ -381,7 +396,7 @@ class LgWebOSMediaPlayerEntity(MediaPlayerEntity):
await self.async_media_pause()
@cmd
async def async_select_source(self, source):
async def async_select_source(self, source: str) -> None:
"""Select input source."""
if (source_dict := self._source_list.get(source)) is None:
_LOGGER.warning("Source %s not found for %s", source, self.name)
@ -392,7 +407,9 @@ class LgWebOSMediaPlayerEntity(MediaPlayerEntity):
await self._client.set_input(source_dict["id"])
@cmd
async def async_play_media(self, media_type, media_id, **kwargs):
async def async_play_media(
self, media_type: str, media_id: str, **kwargs: Any
) -> None:
"""Play a piece of media."""
_LOGGER.debug("Call play media type <%s>, Id <%s>", media_type, media_id)
@ -427,24 +444,24 @@ class LgWebOSMediaPlayerEntity(MediaPlayerEntity):
await self._client.set_channel(partial_match_channel_id)
@cmd
async def async_media_play(self):
async def async_media_play(self) -> None:
"""Send play command."""
self._paused = False
await self._client.play()
@cmd
async def async_media_pause(self):
async def async_media_pause(self) -> None:
"""Send media pause command to media player."""
self._paused = True
await self._client.pause()
@cmd
async def async_media_stop(self):
async def async_media_stop(self) -> None:
"""Send stop command to media player."""
await self._client.stop()
@cmd
async def async_media_next_track(self):
async def async_media_next_track(self) -> None:
"""Send next track command."""
current_input = self._client.get_input()
if current_input == LIVE_TV_APP_ID:
@ -453,7 +470,7 @@ class LgWebOSMediaPlayerEntity(MediaPlayerEntity):
await self._client.fast_forward()
@cmd
async def async_media_previous_track(self):
async def async_media_previous_track(self) -> None:
"""Send the previous track command."""
current_input = self._client.get_input()
if current_input == LIVE_TV_APP_ID:
@ -462,11 +479,11 @@ class LgWebOSMediaPlayerEntity(MediaPlayerEntity):
await self._client.rewind()
@cmd
async def async_button(self, button):
async def async_button(self, button: str) -> None:
"""Send a button press."""
await self._client.button(button)
@cmd
async def async_command(self, command, **kwargs):
async def async_command(self, command: str, **kwargs: Any) -> None:
"""Send a command."""
await self._client.request(command, payload=kwargs.get(ATTR_PAYLOAD))

25
homeassistant/components/webostv/notify.py
View File

@ -1,39 +1,46 @@
"""Support for LG WebOS TV notification service."""
import logging
from __future__ import annotations
from aiowebostv import WebOsTvPairError
import logging
from typing import Any
from aiowebostv import WebOsClient, WebOsTvPairError
from homeassistant.components.notify import ATTR_DATA, BaseNotificationService
from homeassistant.const import CONF_ICON, CONF_NAME
from homeassistant.const import CONF_ICON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from .const import ATTR_CONFIG_ENTRY_ID, DATA_CONFIG_ENTRY, DOMAIN, WEBOSTV_EXCEPTIONS
_LOGGER = logging.getLogger(__name__)
async def async_get_service(hass, _config, discovery_info=None):
async def async_get_service(
hass: HomeAssistant,
config: ConfigType,
discovery_info: DiscoveryInfoType | None = None,
) -> BaseNotificationService | None:
"""Return the notify service."""
if discovery_info is None:
return None
name = discovery_info.get(CONF_NAME)
client = hass.data[DOMAIN][DATA_CONFIG_ENTRY][
discovery_info[ATTR_CONFIG_ENTRY_ID]
].client
return LgWebOSNotificationService(client, name)
return LgWebOSNotificationService(client)
class LgWebOSNotificationService(BaseNotificationService):
"""Implement the notification service for LG WebOS TV."""
def __init__(self, client, name):
def __init__(self, client: WebOsClient) -> None:
"""Initialize the service."""
self._name = name
self._client = client
async def async_send_message(self, message="", **kwargs):
async def async_send_message(self, message: str = "", **kwargs: Any) -> None:
"""Send a message to the tv."""
try:
if not self._client.is_connected():

2
homeassistant/components/webostv/trigger.py
View File

@ -33,7 +33,7 @@ async def async_validate_trigger_config(
) -> ConfigType:
"""Validate config."""
platform = _get_trigger_platform(config)
return platform.TRIGGER_SCHEMA(config)
return cast(ConfigType, platform.TRIGGER_SCHEMA(config))
async def async_attach_trigger(

11
mypy.ini
View File

@ -1902,6 +1902,17 @@ no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.webostv.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.websocket_api.*]
check_untyped_defs = true
disallow_incomplete_defs = true

2
tests/components/webostv/__init__.py
View File

@ -13,7 +13,7 @@ ENTITY_ID = f"{MP_DOMAIN}.{TV_NAME}"
MOCK_CLIENT_KEYS = {"1.2.3.4": "some-secret"}
async def setup_webostv(hass, unique_id=None):
async def setup_webostv(hass, unique_id="some-unique-id"):
"""Initialize webostv and media_player for tests."""
entry = MockConfigEntry(
domain=DOMAIN,

6
tests/components/webostv/test_config_flow.py
View File

@ -119,8 +119,6 @@ async def test_options_flow(hass, client):
"""Test options config flow."""
entry = await setup_webostv(hass)
hass.states.async_set("script.test", "off", {"domain": "script"})
result = await hass.config_entries.options.async_init(entry.entry_id)
await hass.async_block_till_done()
@ -235,7 +233,7 @@ async def test_ssdp_in_progress(hass, client):
async def test_ssdp_update_uuid(hass, client):
"""Test that ssdp updates existing host entry uuid."""
entry = await setup_webostv(hass)
entry = await setup_webostv(hass, None)
assert client
assert entry.unique_id is None
@ -251,7 +249,7 @@ async def test_ssdp_update_uuid(hass, client):
async def test_ssdp_not_update_uuid(hass, client):
"""Test that ssdp not updates different host."""
entry = await setup_webostv(hass)
entry = await setup_webostv(hass, None)
assert client
assert entry.unique_id is None