diff --git a/.strict-typing b/.strict-typing index 8de6e9871d3..d62c32615fd 100644 --- a/.strict-typing +++ b/.strict-typing @@ -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.* diff --git a/homeassistant/components/webostv/__init__.py b/homeassistant/components/webostv/__init__.py index cdb3927b4e0..d5335c4fdcd 100644 --- a/homeassistant/components/webostv/__init__.py +++ b/homeassistant/components/webostv/__init__.py @@ -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) diff --git a/homeassistant/components/webostv/config_flow.py b/homeassistant/components/webostv/config_flow.py index 10f7686d807..91574d768bb 100644 --- a/homeassistant/components/webostv/config_flow.py +++ b/homeassistant/components/webostv/config_flow.py @@ -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()), diff --git a/homeassistant/components/webostv/helpers.py b/homeassistant/components/webostv/helpers.py index 86117d12e71..d3f5fec6826 100644 --- a/homeassistant/components/webostv/helpers.py +++ b/homeassistant/components/webostv/helpers.py @@ -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 diff --git a/homeassistant/components/webostv/media_player.py b/homeassistant/components/webostv/media_player.py index 45cfef5e1b0..74e7e63b2d0 100644 --- a/homeassistant/components/webostv/media_player.py +++ b/homeassistant/components/webostv/media_player.py @@ -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)) diff --git a/homeassistant/components/webostv/notify.py b/homeassistant/components/webostv/notify.py index db330eb0227..df2ed7e5063 100644 --- a/homeassistant/components/webostv/notify.py +++ b/homeassistant/components/webostv/notify.py @@ -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(): diff --git a/homeassistant/components/webostv/trigger.py b/homeassistant/components/webostv/trigger.py index a364c162fd2..1ad7058e1de 100644 --- a/homeassistant/components/webostv/trigger.py +++ b/homeassistant/components/webostv/trigger.py @@ -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( diff --git a/mypy.ini b/mypy.ini index 29e87110bf0..9603da1ae04 100644 --- a/mypy.ini +++ b/mypy.ini @@ -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 diff --git a/tests/components/webostv/__init__.py b/tests/components/webostv/__init__.py index 9fa66cf0863..96c83b33c41 100644 --- a/tests/components/webostv/__init__.py +++ b/tests/components/webostv/__init__.py @@ -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, diff --git a/tests/components/webostv/test_config_flow.py b/tests/components/webostv/test_config_flow.py index da0dd7f6ba5..a46af19fc04 100644 --- a/tests/components/webostv/test_config_flow.py +++ b/tests/components/webostv/test_config_flow.py @@ -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