diff --git a/homeassistant/components/panasonic_viera/__init__.py b/homeassistant/components/panasonic_viera/__init__.py index 2cf91792800..69800d2ef1e 100644 --- a/homeassistant/components/panasonic_viera/__init__.py +++ b/homeassistant/components/panasonic_viera/__init__.py @@ -1,22 +1,18 @@ """The Panasonic Viera integration.""" +from collections.abc import Callable from functools import partial import logging +from typing import Any from urllib.error import HTTPError, URLError from panasonic_viera import EncryptionRequired, Keys, RemoteControl, SOAPError import voluptuous as vol +from homeassistant.components.media_player import MediaPlayerState, MediaType from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry -from homeassistant.const import ( - CONF_HOST, - CONF_NAME, - CONF_PORT, - STATE_OFF, - STATE_ON, - Platform, -) -from homeassistant.core import HomeAssistant +from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT, Platform +from homeassistant.core import Context, HomeAssistant import homeassistant.helpers.config_validation as cv from homeassistant.helpers.script import Script from homeassistant.helpers.typing import ConfigType @@ -132,13 +128,13 @@ class Remote: def __init__( self, - hass, - host, - port, - on_action=None, - app_id=None, - encryption_key=None, - ): + hass: HomeAssistant, + host: str, + port: int, + on_action: Script | None = None, + app_id: str | None = None, + encryption_key: str | None = None, + ) -> None: """Initialize the Remote class.""" self._hass = hass @@ -150,15 +146,14 @@ class Remote: self._app_id = app_id self._encryption_key = encryption_key - self.state = None - self.available = False - self.volume = 0 - self.muted = False - self.playing = True + self._control: RemoteControl | None = None + self.state: MediaPlayerState | None = None + self.available: bool = False + self.volume: float = 0 + self.muted: bool = False + self.playing: bool = True - self._control = None - - async def async_create_remote_control(self, during_setup=False): + async def async_create_remote_control(self, during_setup: bool = False) -> None: """Create remote control.""" try: params = {} @@ -175,15 +170,15 @@ class Remote: except (URLError, SOAPError, OSError) as err: _LOGGER.debug("Could not establish remote connection: %s", err) self._control = None - self.state = STATE_OFF + self.state = MediaPlayerState.OFF self.available = self._on_action is not None except Exception: _LOGGER.exception("An unknown error occurred") self._control = None - self.state = STATE_OFF + self.state = MediaPlayerState.OFF self.available = self._on_action is not None - async def async_update(self): + async def async_update(self) -> None: """Update device data.""" if self._control is None: await self.async_create_remote_control() @@ -191,8 +186,9 @@ class Remote: await self._handle_errors(self._update) - def _update(self): + def _update(self) -> None: """Retrieve the latest data.""" + assert self._control is not None self.muted = self._control.get_mute() self.volume = self._control.get_volume() / 100 @@ -203,39 +199,43 @@ class Remote: except (AttributeError, TypeError): key = getattr(key, "value", key) + assert self._control is not None await self._handle_errors(self._control.send_key, key) - async def async_turn_on(self, context): + async def async_turn_on(self, context: Context | None) -> None: """Turn on the TV.""" if self._on_action is not None: await self._on_action.async_run(context=context) await self.async_update() - elif self.state != STATE_ON: + elif self.state is not MediaPlayerState.ON: await self.async_send_key(Keys.POWER) await self.async_update() - async def async_turn_off(self): + async def async_turn_off(self) -> None: """Turn off the TV.""" - if self.state != STATE_OFF: + if self.state is not MediaPlayerState.OFF: await self.async_send_key(Keys.POWER) - self.state = STATE_OFF + self.state = MediaPlayerState.OFF await self.async_update() - async def async_set_mute(self, enable): + async def async_set_mute(self, enable: bool) -> None: """Set mute based on 'enable'.""" + assert self._control is not None await self._handle_errors(self._control.set_mute, enable) - async def async_set_volume(self, volume): + async def async_set_volume(self, volume: float) -> None: """Set volume level, range 0..1.""" + assert self._control is not None volume = int(volume * 100) await self._handle_errors(self._control.set_volume, volume) - async def async_play_media(self, media_type, media_id): + async def async_play_media(self, media_type: MediaType, media_id: str) -> None: """Play media.""" + assert self._control is not None _LOGGER.debug("Play media: %s (%s)", media_id, media_type) await self._handle_errors(self._control.open_webpage, media_id) - async def async_get_device_info(self): + async def async_get_device_info(self) -> dict[str, Any] | None: """Return device info.""" if self._control is None: return None @@ -243,7 +243,9 @@ class Remote: _LOGGER.debug("Fetched device info: %s", str(device_info)) return device_info - async def _handle_errors(self, func, *args): + async def _handle_errors[_R, *_Ts]( + self, func: Callable[[*_Ts], _R], *args: *_Ts + ) -> _R | None: """Handle errors from func, set available and reconnect if needed.""" try: result = await self._hass.async_add_executor_job(func, *args) @@ -252,23 +254,24 @@ class Remote: "The connection couldn't be encrypted. Please reconfigure your TV" ) self.available = False + return None except (SOAPError, HTTPError) as err: _LOGGER.debug("An error occurred: %s", err) - self.state = STATE_OFF + self.state = MediaPlayerState.OFF self.available = True await self.async_create_remote_control() return None except (URLError, OSError) as err: _LOGGER.debug("An error occurred: %s", err) - self.state = STATE_OFF + self.state = MediaPlayerState.OFF self.available = self._on_action is not None await self.async_create_remote_control() return None except Exception: _LOGGER.exception("An unknown error occurred") - self.state = STATE_OFF + self.state = MediaPlayerState.OFF self.available = self._on_action is not None return None - self.state = STATE_ON + self.state = MediaPlayerState.ON self.available = True return result diff --git a/homeassistant/components/panasonic_viera/config_flow.py b/homeassistant/components/panasonic_viera/config_flow.py index 9cb8fb5da83..0226fb33c9e 100644 --- a/homeassistant/components/panasonic_viera/config_flow.py +++ b/homeassistant/components/panasonic_viera/config_flow.py @@ -2,12 +2,13 @@ from functools import partial import logging +from typing import Any from urllib.error import URLError from panasonic_viera import TV_TYPE_ENCRYPTED, RemoteControl, SOAPError import voluptuous as vol -from homeassistant.config_entries import ConfigFlow +from homeassistant.config_entries import ConfigFlow, ConfigFlowResult from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PIN, CONF_PORT from .const import ( @@ -33,7 +34,7 @@ class PanasonicVieraConfigFlow(ConfigFlow, domain=DOMAIN): def __init__(self) -> None: """Initialize the Panasonic Viera config flow.""" - self._data = { + self._data: dict[str, Any] = { CONF_HOST: None, CONF_NAME: None, CONF_PORT: None, @@ -41,11 +42,13 @@ class PanasonicVieraConfigFlow(ConfigFlow, domain=DOMAIN): ATTR_DEVICE_INFO: None, } - self._remote = None + self._remote: RemoteControl | None = None - async def async_step_user(self, user_input=None): + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: """Handle the initial step.""" - errors = {} + errors: dict[str, str] = {} if user_input is not None: await self.async_load_data(user_input) @@ -53,7 +56,7 @@ class PanasonicVieraConfigFlow(ConfigFlow, domain=DOMAIN): self._remote = await self.hass.async_add_executor_job( partial(RemoteControl, self._data[CONF_HOST], self._data[CONF_PORT]) ) - + assert self._remote is not None self._data[ATTR_DEVICE_INFO] = await self.hass.async_add_executor_job( self._remote.get_device_info ) @@ -63,8 +66,7 @@ class PanasonicVieraConfigFlow(ConfigFlow, domain=DOMAIN): except Exception: _LOGGER.exception("An unknown error occurred") return self.async_abort(reason="unknown") - - if "base" not in errors: + else: await self.async_set_unique_id(self._data[ATTR_DEVICE_INFO][ATTR_UDN]) self._abort_if_unique_id_configured() @@ -102,9 +104,12 @@ class PanasonicVieraConfigFlow(ConfigFlow, domain=DOMAIN): errors=errors, ) - async def async_step_pairing(self, user_input=None): + async def async_step_pairing( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: """Handle the pairing step.""" - errors = {} + errors: dict[str, str] = {} + assert self._remote is not None if user_input is not None: pin = user_input[CONF_PIN] @@ -152,11 +157,13 @@ class PanasonicVieraConfigFlow(ConfigFlow, domain=DOMAIN): errors=errors, ) - async def async_step_import(self, import_config): + async def async_step_import( + self, import_config: dict[str, Any] + ) -> ConfigFlowResult: """Import a config entry from configuration.yaml.""" return await self.async_step_user(user_input=import_config) - async def async_load_data(self, config): + async def async_load_data(self, config: dict[str, Any]) -> None: """Load the data.""" self._data = config diff --git a/homeassistant/components/panasonic_viera/media_player.py b/homeassistant/components/panasonic_viera/media_player.py index 76ca76c1ca6..8738b897d29 100644 --- a/homeassistant/components/panasonic_viera/media_player.py +++ b/homeassistant/components/panasonic_viera/media_player.py @@ -13,6 +13,7 @@ from homeassistant.components.media_player import ( MediaPlayerDeviceClass, MediaPlayerEntity, MediaPlayerEntityFeature, + MediaPlayerState, MediaType, async_process_play_media_url, ) @@ -72,6 +73,7 @@ class PanasonicVieraTVEntity(MediaPlayerEntity): ) _attr_has_entity_name = True _attr_name = None + _attr_device_class = MediaPlayerDeviceClass.TV def __init__(self, remote, name, device_info): """Initialize the entity.""" @@ -88,12 +90,7 @@ class PanasonicVieraTVEntity(MediaPlayerEntity): self._attr_name = name @property - def device_class(self): - """Return the device class of the device.""" - return MediaPlayerDeviceClass.TV - - @property - def state(self): + def state(self) -> MediaPlayerState | None: """Return the state of the device.""" return self._remote.state @@ -103,12 +100,12 @@ class PanasonicVieraTVEntity(MediaPlayerEntity): return self._remote.available @property - def volume_level(self): + def volume_level(self) -> float | None: """Volume level of the media player (0..1).""" return self._remote.volume @property - def is_volume_muted(self): + def is_volume_muted(self) -> bool | None: """Boolean if volume is currently muted.""" return self._remote.muted diff --git a/homeassistant/components/panasonic_viera/remote.py b/homeassistant/components/panasonic_viera/remote.py index c47dce36306..ad40a97f700 100644 --- a/homeassistant/components/panasonic_viera/remote.py +++ b/homeassistant/components/panasonic_viera/remote.py @@ -12,6 +12,7 @@ from homeassistant.core import HomeAssistant from homeassistant.helpers.device_registry import DeviceInfo from homeassistant.helpers.entity_platform import AddEntitiesCallback +from . import Remote from .const import ( ATTR_DEVICE_INFO, ATTR_MANUFACTURER, @@ -43,7 +44,9 @@ async def async_setup_entry( class PanasonicVieraRemoteEntity(RemoteEntity): """Representation of a Panasonic Viera TV Remote.""" - def __init__(self, remote, name, device_info): + def __init__( + self, remote: Remote, name: str, device_info: dict[str, Any] | None = None + ) -> None: """Initialize the entity.""" # Save a reference to the imported class self._remote = remote @@ -51,7 +54,7 @@ class PanasonicVieraRemoteEntity(RemoteEntity): self._device_info = device_info @property - def unique_id(self): + def unique_id(self) -> str | None: """Return the unique ID of the device.""" if self._device_info is None: return None @@ -70,7 +73,7 @@ class PanasonicVieraRemoteEntity(RemoteEntity): ) @property - def name(self): + def name(self) -> str: """Return the name of the device.""" return self._name @@ -80,7 +83,7 @@ class PanasonicVieraRemoteEntity(RemoteEntity): return self._remote.available @property - def is_on(self): + def is_on(self) -> bool: """Return true if device is on.""" return self._remote.state == STATE_ON