evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add typing to Panasonic Viera (#120772)

Browse source 
Co-authored-by: Marc Mueller <30130371+cdce8p@users.noreply.github.com>
This commit is contained in:
Joost Lekkerkerker 2024-07-01 12:30:20 +02:00 • committed by GitHub
parent 921430d497
commit f08638eead
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 76 additions and 66 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

87
homeassistant/components/panasonic_viera/__init__.py
View file

@ -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

31
homeassistant/components/panasonic_viera/config_flow.py
View file

@ -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

13
homeassistant/components/panasonic_viera/media_player.py
View file

@ -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

11
homeassistant/components/panasonic_viera/remote.py
View file

@ -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