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

Improve typing in DuneHD integration (#51025)

Browse source 
* Improve typing

* One more typing fix

* Run hassfest

* Fix test

* Fix return from constructor

* Add missing Final

* Improve long string format

* Use bool for mute

* Remove unnecessary str type

* Fix host type

* Add missing Final

* Increase test coverage

* Suggested change

Co-authored-by: Ruslan Sayfutdinov <ruslan@sayfutdinov.com>

Co-authored-by: Ruslan Sayfutdinov <ruslan@sayfutdinov.com>
This commit is contained in:
Maciej Bieniek 2021-05-24 21:09:57 +02:00 • committed by GitHub
parent 394e044c66
commit 12e2c59a4c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 116 additions and 68 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.strict-typing
View file

@ -23,6 +23,7 @@ homeassistant.components.canary.*
homeassistant.components.cover.*
homeassistant.components.device_automation.*
homeassistant.components.device_tracker.*
homeassistant.components.dunehd.*
homeassistant.components.elgato.*
homeassistant.components.fitbit.*
homeassistant.components.fritzbox.*

14
homeassistant/components/dunehd/__init__.py
View file

@ -1,16 +1,22 @@
"""The Dune HD component."""
from __future__ import annotations
from typing import Final
from pdunehd import DuneHDPlayer
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from .const import DOMAIN
PLATFORMS = ["media_player"]
PLATFORMS: Final[list[str]] = ["media_player"]
async def async_setup_entry(hass, entry):
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a config entry."""
host = entry.data[CONF_HOST]
host: str = entry.data[CONF_HOST]
player = DuneHDPlayer(host)
@ -22,7 +28,7 @@ async def async_setup_entry(hass, entry):
return True
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)
if unload_ok:

52
homeassistant/components/dunehd/config_flow.py
View file

@ -1,29 +1,34 @@
"""Adds config flow for Dune HD integration."""
from __future__ import annotations
import ipaddress
import logging
import re
from typing import Any, Final
from pdunehd import DuneHDPlayer
import voluptuous as vol
from homeassistant import config_entries, exceptions
from homeassistant.const import CONF_HOST
from homeassistant.data_entry_flow import FlowResult
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
_LOGGER: Final = logging.getLogger(__name__)
def host_valid(host):
def host_valid(host: str) -> bool:
"""Return True if hostname or IP address is valid."""
try:
if ipaddress.ip_address(host).version in [4, 6]:
return True
except ValueError:
if len(host) > 253:
return False
allowed = re.compile(r"(?!-)[A-Z\d\-\_]{1,63}(?<!-)$", re.IGNORECASE)
return all(allowed.match(x) for x in host.split("."))
pass
if len(host) > 253:
return False
allowed = re.compile(r"(?!-)[A-Z\d\-\_]{1,63}(?<!-)$", re.IGNORECASE)
return all(allowed.match(x) for x in host.split("."))
class DuneHDConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
@ -31,35 +36,33 @@ class DuneHDConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1
def __init__(self):
"""Initialize."""
self.host = None
async def init_device(self, host):
async def init_device(self, host: str) -> None:
"""Initialize Dune HD player."""
player = DuneHDPlayer(host)
state = await self.hass.async_add_executor_job(player.update_state)
if not state:
raise CannotConnect()
async def async_step_user(self, user_input=None):
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
errors = {}
if user_input is not None:
if host_valid(user_input[CONF_HOST]):
self.host = user_input[CONF_HOST]
host: str = user_input[CONF_HOST]
try:
if self.host_already_configured(self.host):
if self.host_already_configured(host):
raise AlreadyConfigured()
await self.init_device(self.host)
await self.init_device(host)
except CannotConnect:
errors[CONF_HOST] = "cannot_connect"
except AlreadyConfigured:
errors[CONF_HOST] = "already_configured"
else:
return self.async_create_entry(title=self.host, data=user_input)
return self.async_create_entry(title=host, data=user_input)
else:
errors[CONF_HOST] = "invalid_host"
@ -69,21 +72,24 @@ class DuneHDConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def async_step_import(self, user_input=None):
async def async_step_import(
self, user_input: dict[str, str] | None = None
) -> FlowResult:
"""Handle configuration by yaml file."""
self.host = user_input[CONF_HOST]
assert user_input is not None
host: str = user_input[CONF_HOST]
self._async_abort_entries_match({CONF_HOST: self.host})
self._async_abort_entries_match({CONF_HOST: host})
try:
await self.init_device(self.host)
await self.init_device(host)
except CannotConnect:
_LOGGER.error("Import aborted, cannot connect to %s", self.host)
_LOGGER.error("Import aborted, cannot connect to %s", host)
return self.async_abort(reason="cannot_connect")
else:
return self.async_create_entry(title=self.host, data=user_input)
return self.async_create_entry(title=host, data=user_input)
def host_already_configured(self, host):
def host_already_configured(self, host: str) -> bool:
"""See if we already have a dunehd entry matching user input configured."""
existing_hosts = {
entry.data[CONF_HOST] for entry in self._async_current_entries()

10
homeassistant/components/dunehd/const.py
View file

@ -1,4 +1,8 @@
"""Constants for Dune HD integration."""
ATTR_MANUFACTURER = "Dune"
DOMAIN = "dunehd"
DEFAULT_NAME = "Dune HD"
from __future__ import annotations
from typing import Final
ATTR_MANUFACTURER: Final = "Dune"
DOMAIN: Final = "dunehd"
DEFAULT_NAME: Final = "Dune HD"

88
homeassistant/components/dunehd/media_player.py
View file

@ -1,7 +1,15 @@
"""Dune HD implementation of the media player."""
from __future__ import annotations
from typing import Any, Final
from pdunehd import DuneHDPlayer
import voluptuous as vol
from homeassistant.components.media_player import PLATFORM_SCHEMA, MediaPlayerEntity
from homeassistant.components.media_player import (
PLATFORM_SCHEMA as PARENT_PLATFORM_SCHEMA,
MediaPlayerEntity,
)
from homeassistant.components.media_player.const import (
SUPPORT_NEXT_TRACK,
SUPPORT_PAUSE,
@ -10,7 +18,7 @@ from homeassistant.components.media_player.const import (
SUPPORT_TURN_OFF,
SUPPORT_TURN_ON,
)
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
CONF_HOST,
CONF_NAME,
@ -19,13 +27,17 @@ from homeassistant.const import (
STATE_PAUSED,
STATE_PLAYING,
)
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType, StateType
from .const import ATTR_MANUFACTURER, DEFAULT_NAME, DOMAIN
CONF_SOURCES = "sources"
CONF_SOURCES: Final = "sources"
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
PLATFORM_SCHEMA: Final = PARENT_PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_SOURCES): vol.Schema({cv.string: cv.string}),
@ -33,7 +45,7 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
}
)
DUNEHD_PLAYER_SUPPORT = (
DUNEHD_PLAYER_SUPPORT: Final[int] = (
SUPPORT_PAUSE
| SUPPORT_TURN_ON
| SUPPORT_TURN_OFF
@ -43,9 +55,14 @@ DUNEHD_PLAYER_SUPPORT = (
)
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the Dune HD media player platform."""
host = config.get(CONF_HOST)
host: str = config[CONF_HOST]
hass.async_create_task(
hass.config_entries.flow.async_init(
@ -54,11 +71,13 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
)
async def async_setup_entry(hass, config_entry, async_add_entities):
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Add Dune HD entities from a config_entry."""
unique_id = config_entry.entry_id
unique_id = entry.entry_id
player = hass.data[DOMAIN][config_entry.entry_id]
player: str = hass.data[DOMAIN][entry.entry_id]
async_add_entities([DuneHDPlayerEntity(player, DEFAULT_NAME, unique_id)], True)
@ -66,22 +85,22 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
class DuneHDPlayerEntity(MediaPlayerEntity):
"""Implementation of the Dune HD player."""
def __init__(self, player, name, unique_id):
def __init__(self, player: DuneHDPlayer, name: str, unique_id: str) -> None:
"""Initialize entity to control Dune HD."""
self._player = player
self._name = name
self._media_title = None
self._state = None
self._media_title: str | None = None
self._state: dict[str, Any] = {}
self._unique_id = unique_id
def update(self):
def update(self) -> bool:
"""Update internal status of the entity."""
self._state = self._player.update_state()
self.__update_title()
return True
@property
def state(self):
def state(self) -> StateType:
"""Return player state."""
state = STATE_OFF
if "playback_position" in self._state:
@ -95,22 +114,22 @@ class DuneHDPlayerEntity(MediaPlayerEntity):
return state
@property
def name(self):
def name(self) -> str:
"""Return the name of the device."""
return self._name
@property
def available(self):
def available(self) -> bool:
"""Return True if entity is available."""
return bool(self._state)
return len(self._state) > 0
@property
def unique_id(self):
def unique_id(self) -> str:
"""Return a unique_id for this entity."""
return self._unique_id
@property
def device_info(self):
def device_info(self) -> DeviceInfo:
"""Return the device info."""
return {
"identifiers": {(DOMAIN, self._unique_id)},
@ -119,57 +138,58 @@ class DuneHDPlayerEntity(MediaPlayerEntity):
}
@property
def volume_level(self):
def volume_level(self) -> float:
"""Return the volume level of the media player (0..1)."""
return int(self._state.get("playback_volume", 0)) / 100
@property
def is_volume_muted(self):
def is_volume_muted(self) -> bool:
"""Return a boolean if volume is currently muted."""
return int(self._state.get("playback_mute", 0)) == 1
@property
def supported_features(self):
def supported_features(self) -> int:
"""Flag media player features that are supported."""
return DUNEHD_PLAYER_SUPPORT
def volume_up(self):
def volume_up(self) -> None:
"""Volume up media player."""
self._state = self._player.volume_up()
def volume_down(self):
def volume_down(self) -> None:
"""Volume down media player."""
self._state = self._player.volume_down()
def mute_volume(self, mute):
def mute_volume(self, mute: bool) -> None:
"""Mute/unmute player volume."""
self._state = self._player.mute(mute)
def turn_off(self):
def turn_off(self) -> None:
"""Turn off media player."""
self._media_title = None
self._state = self._player.turn_off()
def turn_on(self):
def turn_on(self) -> None:
"""Turn off media player."""
self._state = self._player.turn_on()
def media_play(self):
def media_play(self) -> None:
"""Play media player."""
self._state = self._player.play()
def media_pause(self):
def media_pause(self) -> None:
"""Pause media player."""
self._state = self._player.pause()
@property
def media_title(self):
def media_title(self) -> str | None:
"""Return the current media source."""
self.__update_title()
if self._media_title:
return self._media_title
return None
def __update_title(self):
def __update_title(self) -> None:
if self._state.get("player_state") == "bluray_playback":
self._media_title = "Blu-Ray"
elif self._state.get("player_state") == "photo_viewer":
@ -179,10 +199,10 @@ class DuneHDPlayerEntity(MediaPlayerEntity):
else:
self._media_title = None
def media_previous_track(self):
def media_previous_track(self) -> None:
"""Send previous track command."""
self._state = self._player.previous_track()
def media_next_track(self):
def media_next_track(self) -> None:
"""Send next track command."""
self._state = self._player.next_track()

11
mypy.ini
View file

@ -264,6 +264,17 @@ no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.dunehd.*]
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.elgato.*]
check_untyped_defs = true
disallow_incomplete_defs = true

8
tests/components/dunehd/test_config_flow.py
View file

@ -67,10 +67,10 @@ async def test_user_invalid_host(hass):
async def test_user_very_long_host(hass):
"""Test that errors are shown when the host is longer than 253 chars."""
long_host = (
"very_long_host_very_long_host_very_long_host_very_long_host_very_long"
"host_very_long_host_very_long_host_very_long_host_very_long_host_very_long_ho"
"st_very_long_host_very_long_host_very_long_host_very_long_host_very_long_host_"
"very_long_host_very_long_host"
"very_long_host_very_long_host_very_long_host_very_long_host_very_long_"
"host_very_long_host_very_long_host_very_long_host_very_long_host_very_long_"
"host_very_long_host_very_long_host_very_long_host_very_long_host_very_long_"
"host_very_long_host_very_long_host"
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: long_host}