Improve Hive typing (#122314)

This commit is contained in:
Joost Lekkerkerker 2024-07-21 16:21:45 +02:00 committed by GitHub
parent b0a4140b4d
commit e8796cd725
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 68 additions and 26 deletions

View file

@ -1,6 +1,9 @@
"""Support for the Hive binary sensors."""
from datetime import timedelta
from typing import Any
from apyhiveapi import Hive
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
@ -68,7 +71,12 @@ async def async_setup_entry(
class HiveBinarySensorEntity(HiveEntity, BinarySensorEntity):
"""Representation of a Hive binary sensor."""
def __init__(self, hive, hive_device, entity_description):
def __init__(
self,
hive: Hive,
hive_device: dict[str, Any],
entity_description: BinarySensorEntityDescription,
) -> None:
"""Initialise hive binary sensor."""
super().__init__(hive, hive_device)
self.entity_description = entity_description

View file

@ -4,6 +4,7 @@ from datetime import timedelta
import logging
from typing import Any
from apyhiveapi import Hive
import voluptuous as vol
from homeassistant.components.climate import (
@ -46,7 +47,10 @@ HIVE_TO_HASS_HVAC_ACTION = {
True: HVACAction.HEATING,
}
TEMP_UNIT = {"C": UnitOfTemperature.CELSIUS, "F": UnitOfTemperature.FAHRENHEIT}
TEMP_UNIT = {
"C": UnitOfTemperature.CELSIUS,
"F": UnitOfTemperature.FAHRENHEIT,
}
PARALLEL_UPDATES = 0
SCAN_INTERVAL = timedelta(seconds=15)
_LOGGER = logging.getLogger()
@ -97,11 +101,11 @@ class HiveClimateEntity(HiveEntity, ClimateEntity):
)
_enable_turn_on_off_backwards_compatibility = False
def __init__(self, hive_session, hive_device):
def __init__(self, hive: Hive, hive_device: dict[str, Any]) -> None:
"""Initialize the Climate device."""
super().__init__(hive_session, hive_device)
super().__init__(hive, hive_device)
self.thermostat_node_id = hive_device["device_id"]
self._attr_temperature_unit = TEMP_UNIT.get(hive_device["temperatureunit"])
self._attr_temperature_unit = TEMP_UNIT[hive_device["temperatureunit"]]
@refresh_system
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
@ -132,7 +136,7 @@ class HiveClimateEntity(HiveEntity, ClimateEntity):
await self.hive.heating.setBoostOn(self.device, time_period, temperature)
@refresh_system
async def async_heating_boost_off(self):
async def async_heating_boost_off(self) -> None:
"""Handle boost heating service call."""
await self.hive.heating.setBoostOff(self.device)

View file

@ -31,19 +31,21 @@ class HiveFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle a Hive config flow."""
VERSION = CONFIG_ENTRY_VERSION
hive_auth: Auth
def __init__(self):
def __init__(self) -> None:
"""Initialize the config flow."""
self.hive_auth = None
self.data = {}
self.tokens = {}
self.entry = None
self.device_registration = False
self.data: dict[str, Any] = {}
self.tokens: dict[str, str] = {}
self.entry: ConfigEntry | None = None
self.device_registration: bool = False
self.device_name = "Home Assistant"
async def async_step_user(self, user_input=None):
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Prompt user input. Create or edit entry."""
errors = {}
errors: dict[str, str] = {}
# Login to Hive with user data.
if user_input is not None:
self.data.update(user_input)
@ -83,7 +85,9 @@ class HiveFlowHandler(ConfigFlow, domain=DOMAIN):
)
return self.async_show_form(step_id="user", data_schema=schema, errors=errors)
async def async_step_2fa(self, user_input=None):
async def async_step_2fa(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle 2fa step."""
errors = {}
@ -108,7 +112,9 @@ class HiveFlowHandler(ConfigFlow, domain=DOMAIN):
schema = vol.Schema({vol.Required(CONF_CODE): str})
return self.async_show_form(step_id="2fa", data_schema=schema, errors=errors)
async def async_step_configuration(self, user_input=None):
async def async_step_configuration(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle hive configuration step."""
errors = {}
@ -130,7 +136,7 @@ class HiveFlowHandler(ConfigFlow, domain=DOMAIN):
step_id="configuration", data_schema=schema, errors=errors
)
async def async_setup_hive_entry(self):
async def async_setup_hive_entry(self) -> ConfigFlowResult:
"""Finish setup and create the config entry."""
if "AuthenticationResult" not in self.tokens:
@ -139,6 +145,7 @@ class HiveFlowHandler(ConfigFlow, domain=DOMAIN):
# Setup the config entry
self.data["tokens"] = self.tokens
if self.context["source"] == SOURCE_REAUTH:
assert self.entry
self.hass.config_entries.async_update_entry(
self.entry, title=self.data["username"], data=self.data
)
@ -156,7 +163,9 @@ class HiveFlowHandler(ConfigFlow, domain=DOMAIN):
}
return await self.async_step_user(data)
async def async_step_import(self, user_input=None):
async def async_step_import(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Import user."""
return await self.async_step_user(user_input)
@ -178,16 +187,21 @@ class HiveOptionsFlowHandler(OptionsFlow):
self.config_entry = config_entry
self.interval = config_entry.options.get(CONF_SCAN_INTERVAL, 120)
async def async_step_init(self, user_input=None):
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Manage the options."""
return await self.async_step_user()
async def async_step_user(self, user_input=None):
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle a flow initialized by the user."""
self.hive = self.hass.data["hive"][self.config_entry.entry_id]
errors = {}
errors: dict[str, str] = {}
if user_input is not None:
new_interval = user_input.get(CONF_SCAN_INTERVAL)
assert self.hive
await self.hive.updateInterval(new_interval)
return self.async_create_entry(title="", data=user_input)

View file

@ -1,6 +1,9 @@
"""Support for the Hive sensors."""
from datetime import timedelta
from typing import Any
from apyhiveapi import Hive
from homeassistant.components.sensor import (
SensorDeviceClass,
@ -70,7 +73,12 @@ async def async_setup_entry(
class HiveSensorEntity(HiveEntity, SensorEntity):
"""Hive Sensor Entity."""
def __init__(self, hive, hive_device, entity_description):
def __init__(
self,
hive: Hive,
hive_device: dict[str, Any],
entity_description: SensorEntityDescription,
) -> None:
"""Initialise hive sensor."""
super().__init__(hive, hive_device)
self.entity_description = entity_description

View file

@ -5,6 +5,8 @@ from __future__ import annotations
from datetime import timedelta
from typing import Any
from apyhiveapi import Hive
from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory
@ -52,7 +54,12 @@ async def async_setup_entry(
class HiveSwitch(HiveEntity, SwitchEntity):
"""Hive Active Plug."""
def __init__(self, hive, hive_device, entity_description):
def __init__(
self,
hive: Hive,
hive_device: dict[str, Any],
entity_description: SwitchEntityDescription,
) -> None:
"""Initialise hive switch."""
super().__init__(hive, hive_device)
self.entity_description = entity_description

View file

@ -1,6 +1,7 @@
"""Support for hive water heaters."""
from datetime import timedelta
from typing import Any
import voluptuous as vol
@ -76,12 +77,12 @@ class HiveWaterHeater(HiveEntity, WaterHeaterEntity):
_attr_operation_list = SUPPORT_WATER_HEATER
@refresh_system
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on hotwater."""
await self.hive.hotwater.setMode(self.device, "MANUAL")
@refresh_system
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn on hotwater."""
await self.hive.hotwater.setMode(self.device, "OFF")
@ -92,7 +93,7 @@ class HiveWaterHeater(HiveEntity, WaterHeaterEntity):
await self.hive.hotwater.setMode(self.device, new_mode)
@refresh_system
async def async_hot_water_boost(self, time_period, on_off):
async def async_hot_water_boost(self, time_period: int, on_off: str) -> None:
"""Handle the service call."""
if on_off == "on":
await self.hive.hotwater.setBoostOn(self.device, time_period)