Improve hunterdouglas_powerview typing (#107445)

This commit is contained in:
Marc Mueller 2024-01-08 10:08:52 +01:00 committed by GitHub
parent a6fc4c2bd5
commit 5ef04fcc7b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 34 additions and 19 deletions

View File

@ -3,6 +3,7 @@ from __future__ import annotations
import asyncio import asyncio
import logging import logging
from typing import Any
from aiopvapi.helpers.aiorequest import AioRequest from aiopvapi.helpers.aiorequest import AioRequest
import voluptuous as vol import voluptuous as vol
@ -51,18 +52,20 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
def __init__(self): def __init__(self) -> None:
"""Initialize the powerview config flow.""" """Initialize the powerview config flow."""
self.powerview_config = {} self.powerview_config: dict[str, str] = {}
self.discovered_ip = None self.discovered_ip: str | None = None
self.discovered_name = None self.discovered_name: str | None = None
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.""" """Handle the initial step."""
errors = {} errors: dict[str, Any] = {}
if user_input is not None: if user_input is not None:
info, error = await self._async_validate_or_error(user_input[CONF_HOST]) info, error = await self._async_validate_or_error(user_input[CONF_HOST])
if not error: if info and not error:
await self.async_set_unique_id(info["unique_id"]) await self.async_set_unique_id(info["unique_id"])
return self.async_create_entry( return self.async_create_entry(
title=info["title"], data={CONF_HOST: user_input[CONF_HOST]} title=info["title"], data={CONF_HOST: user_input[CONF_HOST]}
@ -73,7 +76,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
step_id="user", data_schema=DATA_SCHEMA, errors=errors step_id="user", data_schema=DATA_SCHEMA, errors=errors
) )
async def _async_validate_or_error(self, host): async def _async_validate_or_error(
self, host: str
) -> tuple[dict[str, str], None] | tuple[None, str]:
self._async_abort_entries_match({CONF_HOST: host}) self._async_abort_entries_match({CONF_HOST: host})
try: try:
@ -110,21 +115,22 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self.discovered_name = name self.discovered_name = name
return await self.async_step_discovery_confirm() return await self.async_step_discovery_confirm()
async def async_step_discovery_confirm(self): async def async_step_discovery_confirm(self) -> FlowResult:
"""Confirm dhcp or homekit discovery.""" """Confirm dhcp or homekit discovery."""
# If we already have the host configured do # If we already have the host configured do
# not open connections to it if we can avoid it. # not open connections to it if we can avoid it.
assert self.discovered_ip and self.discovered_name
self.context[CONF_HOST] = self.discovered_ip self.context[CONF_HOST] = self.discovered_ip
for progress in self._async_in_progress(): for progress in self._async_in_progress():
if progress.get("context", {}).get(CONF_HOST) == self.discovered_ip: if progress.get("context", {}).get(CONF_HOST) == self.discovered_ip:
return self.async_abort(reason="already_in_progress") return self.async_abort(reason="already_in_progress")
self._async_abort_entries_match({CONF_HOST: self.discovered_ip}) self._async_abort_entries_match({CONF_HOST: self.discovered_ip})
info, error = await self._async_validate_or_error(self.discovered_ip) info, error = await self._async_validate_or_error(self.discovered_ip)
if error: if error:
return self.async_abort(reason=error) return self.async_abort(reason=error)
assert info is not None
await self.async_set_unique_id(info["unique_id"], raise_on_progress=False) await self.async_set_unique_id(info["unique_id"], raise_on_progress=False)
self._abort_if_unique_id_configured({CONF_HOST: self.discovered_ip}) self._abort_if_unique_id_configured({CONF_HOST: self.discovered_ip})
@ -134,7 +140,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
} }
return await self.async_step_link() return await self.async_step_link()
async def async_step_link(self, user_input=None): async def async_step_link(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Attempt to link with Powerview.""" """Attempt to link with Powerview."""
if user_input is not None: if user_input is not None:
return self.async_create_entry( return self.async_create_entry(

View File

@ -2,9 +2,9 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
from collections.abc import Iterable from collections.abc import Callable, Iterable
from contextlib import suppress from contextlib import suppress
from datetime import timedelta from datetime import datetime, timedelta
import logging import logging
from math import ceil from math import ceil
from typing import Any from typing import Any
@ -137,7 +137,7 @@ class PowerViewShadeBase(ShadeEntity, CoverEntity):
self._scheduled_transition_update: CALLBACK_TYPE | None = None self._scheduled_transition_update: CALLBACK_TYPE | None = None
if self._device_info.model != LEGACY_DEVICE_MODEL: if self._device_info.model != LEGACY_DEVICE_MODEL:
self._attr_supported_features |= CoverEntityFeature.STOP self._attr_supported_features |= CoverEntityFeature.STOP
self._forced_resync = None self._forced_resync: Callable[[], None] | None = None
@property @property
def assumed_state(self) -> bool: def assumed_state(self) -> bool:
@ -291,7 +291,7 @@ class PowerViewShadeBase(ShadeEntity, CoverEntity):
self._async_complete_schedule_update, self._async_complete_schedule_update,
) )
async def _async_complete_schedule_update(self, _): async def _async_complete_schedule_update(self, _: datetime) -> None:
"""Update status of the cover.""" """Update status of the cover."""
_LOGGER.debug("Processing scheduled update for %s", self.name) _LOGGER.debug("Processing scheduled update for %s", self.name)
self._scheduled_transition_update = None self._scheduled_transition_update = None
@ -382,7 +382,7 @@ class PowerViewShadeWithTiltBase(PowerViewShadeBase):
return hd_position_to_hass(self.positions.vane, self._max_tilt) return hd_position_to_hass(self.positions.vane, self._max_tilt)
@property @property
def transition_steps(self): def transition_steps(self) -> int:
"""Return the steps to make a move.""" """Return the steps to make a move."""
return hd_position_to_hass( return hd_position_to_hass(
self.positions.primary, MAX_POSITION self.positions.primary, MAX_POSITION

View File

@ -11,8 +11,9 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN, ROOM_NAME_UNICODE, STATE_ATTRIBUTE_ROOM_NAME from .const import DOMAIN, ROOM_NAME_UNICODE, STATE_ATTRIBUTE_ROOM_NAME
from .coordinator import PowerviewShadeUpdateCoordinator
from .entity import HDEntity from .entity import HDEntity
from .model import PowerviewEntryData from .model import PowerviewDeviceInfo, PowerviewEntryData
async def async_setup_entry( async def async_setup_entry(
@ -22,7 +23,7 @@ async def async_setup_entry(
pv_entry: PowerviewEntryData = hass.data[DOMAIN][entry.entry_id] pv_entry: PowerviewEntryData = hass.data[DOMAIN][entry.entry_id]
pvscenes = [] pvscenes: list[PowerViewScene] = []
for raw_scene in pv_entry.scene_data.values(): for raw_scene in pv_entry.scene_data.values():
scene = PvScene(raw_scene, pv_entry.api) scene = PvScene(raw_scene, pv_entry.api)
room_name = pv_entry.room_data.get(scene.room_id, {}).get(ROOM_NAME_UNICODE, "") room_name = pv_entry.room_data.get(scene.room_id, {}).get(ROOM_NAME_UNICODE, "")
@ -37,7 +38,13 @@ class PowerViewScene(HDEntity, Scene):
_attr_icon = "mdi:blinds" _attr_icon = "mdi:blinds"
def __init__(self, coordinator, device_info, room_name, scene): def __init__(
self,
coordinator: PowerviewShadeUpdateCoordinator,
device_info: PowerviewDeviceInfo,
room_name: str,
scene: PvScene,
) -> None:
"""Initialize the scene.""" """Initialize the scene."""
super().__init__(coordinator, device_info, room_name, scene.id) super().__init__(coordinator, device_info, room_name, scene.id)
self._scene = scene self._scene = scene