Improve daikin typing (#108039)

This commit is contained in:
Marc Mueller 2024-01-18 09:24:48 +01:00 committed by GitHub
parent afcb7a26cd
commit 3761d13915
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 46 additions and 21 deletions

View File

@ -1,7 +1,10 @@
"""Platform for the Daikin AC.""" """Platform for the Daikin AC."""
from __future__ import annotations
import asyncio import asyncio
from datetime import timedelta from datetime import timedelta
import logging import logging
from typing import Any
from aiohttp import ClientConnectionError from aiohttp import ClientConnectionError
from pydaikin.daikin_base import Appliance from pydaikin.daikin_base import Appliance
@ -68,7 +71,13 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload_ok return unload_ok
async def daikin_api_setup(hass: HomeAssistant, host, key, uuid, password): async def daikin_api_setup(
hass: HomeAssistant,
host: str,
key: str | None,
uuid: str | None,
password: str | None,
) -> DaikinApi | None:
"""Create a Daikin instance only once.""" """Create a Daikin instance only once."""
session = async_get_clientsession(hass) session = async_get_clientsession(hass)
@ -103,7 +112,7 @@ class DaikinApi:
self._available = True self._available = True
@Throttle(MIN_TIME_BETWEEN_UPDATES) @Throttle(MIN_TIME_BETWEEN_UPDATES)
async def async_update(self, **kwargs): async def async_update(self, **kwargs: Any) -> None:
"""Pull the latest data from Daikin.""" """Pull the latest data from Daikin."""
try: try:
await self.device.update_status() await self.device.update_status()

View File

@ -113,7 +113,7 @@ async def async_setup_entry(
async_add_entities([DaikinClimate(daikin_api)], update_before_add=True) async_add_entities([DaikinClimate(daikin_api)], update_before_add=True)
def format_target_temperature(target_temperature): def format_target_temperature(target_temperature: float) -> str:
"""Format target temperature to be sent to the Daikin unit, rounding to nearest half degree.""" """Format target temperature to be sent to the Daikin unit, rounding to nearest half degree."""
return str(round(float(target_temperature) * 2, 0) / 2).rstrip("0").rstrip(".") return str(round(float(target_temperature) * 2, 0) / 2).rstrip("0").rstrip(".")
@ -126,6 +126,8 @@ class DaikinClimate(ClimateEntity):
_attr_temperature_unit = UnitOfTemperature.CELSIUS _attr_temperature_unit = UnitOfTemperature.CELSIUS
_attr_hvac_modes = list(HA_STATE_TO_DAIKIN) _attr_hvac_modes = list(HA_STATE_TO_DAIKIN)
_attr_target_temperature_step = 1 _attr_target_temperature_step = 1
_attr_fan_modes: list[str]
_attr_swing_modes: list[str]
def __init__(self, api: DaikinApi) -> None: def __init__(self, api: DaikinApi) -> None:
"""Initialize the climate device.""" """Initialize the climate device."""
@ -134,7 +136,7 @@ class DaikinClimate(ClimateEntity):
self._attr_fan_modes = api.device.fan_rate self._attr_fan_modes = api.device.fan_rate
self._attr_swing_modes = api.device.swing_modes self._attr_swing_modes = api.device.swing_modes
self._attr_device_info = api.device_info self._attr_device_info = api.device_info
self._list = { self._list: dict[str, list[Any]] = {
ATTR_HVAC_MODE: self._attr_hvac_modes, ATTR_HVAC_MODE: self._attr_hvac_modes,
ATTR_FAN_MODE: self._attr_fan_modes, ATTR_FAN_MODE: self._attr_fan_modes,
ATTR_SWING_MODE: self._attr_swing_modes, ATTR_SWING_MODE: self._attr_swing_modes,
@ -151,9 +153,9 @@ class DaikinClimate(ClimateEntity):
if api.device.support_swing_mode: if api.device.support_swing_mode:
self._attr_supported_features |= ClimateEntityFeature.SWING_MODE self._attr_supported_features |= ClimateEntityFeature.SWING_MODE
async def _set(self, settings): async def _set(self, settings: dict[str, Any]) -> None:
"""Set device settings using API.""" """Set device settings using API."""
values = {} values: dict[str, Any] = {}
for attr in (ATTR_TEMPERATURE, ATTR_FAN_MODE, ATTR_SWING_MODE, ATTR_HVAC_MODE): for attr in (ATTR_TEMPERATURE, ATTR_FAN_MODE, ATTR_SWING_MODE, ATTR_HVAC_MODE):
if (value := settings.get(attr)) is None: if (value := settings.get(attr)) is None:
@ -180,17 +182,17 @@ class DaikinClimate(ClimateEntity):
await self._api.device.set(values) await self._api.device.set(values)
@property @property
def unique_id(self): def unique_id(self) -> str:
"""Return a unique ID.""" """Return a unique ID."""
return self._api.device.mac return self._api.device.mac
@property @property
def current_temperature(self): def current_temperature(self) -> float | None:
"""Return the current temperature.""" """Return the current temperature."""
return self._api.device.inside_temperature return self._api.device.inside_temperature
@property @property
def target_temperature(self): def target_temperature(self) -> float | None:
"""Return the temperature we try to reach.""" """Return the temperature we try to reach."""
return self._api.device.target_temperature return self._api.device.target_temperature
@ -221,7 +223,7 @@ class DaikinClimate(ClimateEntity):
await self._set({ATTR_HVAC_MODE: hvac_mode}) await self._set({ATTR_HVAC_MODE: hvac_mode})
@property @property
def fan_mode(self): def fan_mode(self) -> str:
"""Return the fan setting.""" """Return the fan setting."""
return self._api.device.represent(HA_ATTR_TO_DAIKIN[ATTR_FAN_MODE])[1].title() return self._api.device.represent(HA_ATTR_TO_DAIKIN[ATTR_FAN_MODE])[1].title()
@ -230,7 +232,7 @@ class DaikinClimate(ClimateEntity):
await self._set({ATTR_FAN_MODE: fan_mode}) await self._set({ATTR_FAN_MODE: fan_mode})
@property @property
def swing_mode(self): def swing_mode(self) -> str:
"""Return the fan setting.""" """Return the fan setting."""
return self._api.device.represent(HA_ATTR_TO_DAIKIN[ATTR_SWING_MODE])[1].title() return self._api.device.represent(HA_ATTR_TO_DAIKIN[ATTR_SWING_MODE])[1].title()
@ -239,7 +241,7 @@ class DaikinClimate(ClimateEntity):
await self._set({ATTR_SWING_MODE: swing_mode}) await self._set({ATTR_SWING_MODE: swing_mode})
@property @property
def preset_mode(self): def preset_mode(self) -> str:
"""Return the preset_mode.""" """Return the preset_mode."""
if ( if (
self._api.device.represent(HA_ATTR_TO_DAIKIN[ATTR_PRESET_MODE])[1] self._api.device.represent(HA_ATTR_TO_DAIKIN[ATTR_PRESET_MODE])[1]
@ -282,7 +284,7 @@ class DaikinClimate(ClimateEntity):
) )
@property @property
def preset_modes(self): def preset_modes(self) -> list[str]:
"""List of available preset modes.""" """List of available preset modes."""
ret = [PRESET_NONE] ret = [PRESET_NONE]
if self._api.device.support_away_mode: if self._api.device.support_away_mode:

View File

@ -1,6 +1,9 @@
"""Config flow for the Daikin platform.""" """Config flow for the Daikin platform."""
from __future__ import annotations
import asyncio import asyncio
import logging import logging
from typing import Any
from uuid import uuid4 from uuid import uuid4
from aiohttp import ClientError, web_exceptions from aiohttp import ClientError, web_exceptions
@ -24,12 +27,12 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
def __init__(self): def __init__(self) -> None:
"""Initialize the Daikin config flow.""" """Initialize the Daikin config flow."""
self.host = None self.host: str | None = None
@property @property
def schema(self): def schema(self) -> vol.Schema:
"""Return current schema.""" """Return current schema."""
return vol.Schema( return vol.Schema(
{ {
@ -39,7 +42,14 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
} }
) )
async def _create_entry(self, host, mac, key=None, uuid=None, password=None): async def _create_entry(
self,
host: str,
mac: str,
key: str | None = None,
uuid: str | None = None,
password: str | None = None,
) -> FlowResult:
"""Register new entry.""" """Register new entry."""
if not self.unique_id: if not self.unique_id:
await self.async_set_unique_id(mac) await self.async_set_unique_id(mac)
@ -56,7 +66,9 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
}, },
) )
async def _create_device(self, host, key=None, password=None): async def _create_device(
self, host: str, key: str | None = None, password: str | None = None
) -> FlowResult:
"""Create device.""" """Create device."""
# BRP07Cxx devices needs uuid together with key # BRP07Cxx devices needs uuid together with key
if key: if key:
@ -108,12 +120,14 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
mac = device.mac mac = device.mac
return await self._create_entry(host, mac, key, uuid, password) return await self._create_entry(host, mac, key, uuid, password)
async def async_step_user(self, user_input=None): async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""User initiated config flow.""" """User initiated config flow."""
if user_input is None: if user_input is None:
return self.async_show_form(step_id="user", data_schema=self.schema) return self.async_show_form(step_id="user", data_schema=self.schema)
if user_input.get(CONF_API_KEY) and user_input.get(CONF_PASSWORD): if user_input.get(CONF_API_KEY) and user_input.get(CONF_PASSWORD):
self.host = user_input.get(CONF_HOST) self.host = user_input[CONF_HOST]
return self.async_show_form( return self.async_show_form(
step_id="user", step_id="user",
data_schema=self.schema, data_schema=self.schema,

View File

@ -61,7 +61,7 @@ class DaikinZoneSwitch(SwitchEntity):
_attr_icon = ZONE_ICON _attr_icon = ZONE_ICON
_attr_has_entity_name = True _attr_has_entity_name = True
def __init__(self, api: DaikinApi, zone_id) -> None: def __init__(self, api: DaikinApi, zone_id: int) -> None:
"""Initialize the zone.""" """Initialize the zone."""
self._api = api self._api = api
self._zone_id = zone_id self._zone_id = zone_id