Add mypy check, add missing types and fix type issues

This commit is contained in:
Andre Basche 2023-07-23 21:52:42 +02:00
parent f0fb5742a4
commit 9d6b8297b2
19 changed files with 542 additions and 239 deletions

View File

@ -24,12 +24,17 @@ jobs:
- name: Install dependencies
run: |
python -m pip install --upgrade pip
python -m pip install flake8 pylint black
python -m pip install -r requirements.txt
python -m pip install -r requirements_dev.txt
- name: Lint with flake8
run: |
# stop the build if there are Python syntax errors or undefined names
flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
flake8 . --count --exit-zero --max-complexity=10 --max-line-length=88 --statistics
- name: Type check with mypy
run: |
touch "$(python -c 'import inspect, homeassistant, os; print(os.path.dirname(inspect.getfile(homeassistant)))')"/py.typed
mypy -p custom_components.hon
# - name: Analysing the code with pylint
# run: |
# pylint --max-line-length 88 $(git ls-files '*.py')

View File

@ -1,7 +1,7 @@
import logging
from pathlib import Path
import voluptuous as vol
import voluptuous as vol # type: ignore[import]
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.helpers import config_validation as cv, aiohttp_client
@ -25,13 +25,15 @@ CONFIG_SCHEMA = vol.Schema(
)
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry):
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry) -> None:
session = aiohttp_client.async_get_clientsession(hass)
if (config_dir := hass.config.config_dir) is None:
raise ValueError("Missing Config Dir")
hon = await Hon(
entry.data["email"],
entry.data["password"],
session=session,
test_data_path=Path(hass.config.config_dir),
test_data_path=Path(config_dir),
).create()
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.unique_id] = hon
@ -41,10 +43,10 @@ async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry):
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, platform)
)
return True
return
async def async_unload_entry(hass, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool:
unload = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload:
if not hass.data[DOMAIN]:

View File

@ -8,6 +8,8 @@ from homeassistant.components.binary_sensor import (
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import HomeAssistantType
from .const import DOMAIN
from .hon import HonEntity, unique_entities
@ -287,7 +289,9 @@ BINARY_SENSORS: dict[str, tuple[HonBinarySensorEntityDescription, ...]] = {
BINARY_SENSORS["WD"] = unique_entities(BINARY_SENSORS["WM"], BINARY_SENSORS["TD"])
async def async_setup_entry(hass, entry: ConfigEntry, async_add_entities) -> None:
async def async_setup_entry(
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
entities = []
for device in hass.data[DOMAIN][entry.unique_id].appliances:
for description in BINARY_SENSORS.get(device.appliance_type, []):
@ -304,13 +308,13 @@ class HonBinarySensorEntity(HonEntity, BinarySensorEntity):
@property
def is_on(self) -> bool:
return (
return bool(
self._device.get(self.entity_description.key, "")
== self.entity_description.on_value
)
@callback
def _handle_coordinator_update(self, update=True) -> None:
def _handle_coordinator_update(self, update: bool = True) -> None:
self._attr_native_value = (
self._device.get(self.entity_description.key, "")
== self.entity_description.on_value

View File

@ -5,10 +5,13 @@ from homeassistant.components import persistent_notification
from homeassistant.components.button import ButtonEntityDescription, ButtonEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import HomeAssistantType
from pyhon.appliance import HonAppliance
from .const import DOMAIN
from .hon import HonEntity
from .typedefs import HonButtonType
_LOGGER = logging.getLogger(__name__)
@ -38,8 +41,10 @@ BUTTONS: dict[str, tuple[ButtonEntityDescription, ...]] = {
}
async def async_setup_entry(hass, entry: ConfigEntry, async_add_entities) -> None:
entities = []
async def async_setup_entry(
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
entities: list[HonButtonType] = []
for device in hass.data[DOMAIN][entry.unique_id].appliances:
for description in BUTTONS.get(device.appliance_type, []):
if not device.commands.get(description.key):
@ -70,7 +75,9 @@ class HonButtonEntity(HonEntity, ButtonEntity):
class HonDeviceInfo(HonEntity, ButtonEntity):
def __init__(self, hass, entry, device: HonAppliance) -> None:
def __init__(
self, hass: HomeAssistantType, entry: ConfigEntry, device: HonAppliance
) -> None:
super().__init__(hass, entry, device)
self._attr_unique_id = f"{super().unique_id}_show_device_info"
@ -93,7 +100,9 @@ class HonDeviceInfo(HonEntity, ButtonEntity):
class HonDataArchive(HonEntity, ButtonEntity):
def __init__(self, hass, entry, device: HonAppliance) -> None:
def __init__(
self, hass: HomeAssistantType, entry: ConfigEntry, device: HonAppliance
) -> None:
super().__init__(hass, entry, device)
self._attr_unique_id = f"{super().unique_id}_create_data_archive"
@ -104,7 +113,9 @@ class HonDataArchive(HonEntity, ButtonEntity):
self._attr_entity_registry_enabled_default = False
async def async_press(self) -> None:
path = Path(self._hass.config.config_dir) / "www"
if (config_dir := self._hass.config.config_dir) is None:
raise ValueError("Missing Config Dir")
path = Path(config_dir) / "www"
data = await self._device.data_archive(path)
title = f"{self._device.nick_name} Data Archive"
text = (

View File

@ -1,5 +1,6 @@
import logging
from dataclasses import dataclass
from typing import Any
from homeassistant.components.climate import (
ClimateEntity,
@ -19,7 +20,10 @@ from homeassistant.const import (
TEMP_CELSIUS,
)
from homeassistant.core import callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import HomeAssistantType
from pyhon.appliance import HonAppliance
from pyhon.parameter.range import HonParameterRange
from .const import HON_HVAC_MODE, HON_FAN, DOMAIN, HON_HVAC_PROGRAM
from .hon import HonEntity
@ -34,10 +38,12 @@ class HonACClimateEntityDescription(ClimateEntityDescription):
@dataclass
class HonClimateEntityDescription(ClimateEntityDescription):
mode: HVACMode = "auto"
mode: HVACMode = HVACMode.AUTO
CLIMATES = {
CLIMATES: dict[
str, tuple[HonACClimateEntityDescription | HonClimateEntityDescription, ...]
] = {
"AC": (
HonACClimateEntityDescription(
key="settings",
@ -90,8 +96,11 @@ CLIMATES = {
}
async def async_setup_entry(hass, entry: ConfigEntry, async_add_entities) -> None:
async def async_setup_entry(
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
entities = []
entity: HonClimateEntity | HonACClimateEntity
for device in hass.data[DOMAIN][entry.unique_id].appliances:
for description in CLIMATES.get(device.appliance_type, []):
if isinstance(description, HonACClimateEntityDescription):
@ -103,14 +112,22 @@ async def async_setup_entry(hass, entry: ConfigEntry, async_add_entities) -> Non
continue
entity = HonClimateEntity(hass, entry, device, description)
else:
continue
continue # type: ignore[unreachable]
await entity.coordinator.async_config_entry_first_refresh()
entities.append(entity)
async_add_entities(entities)
class HonACClimateEntity(HonEntity, ClimateEntity):
def __init__(self, hass, entry, device: HonAppliance, description) -> None:
entity_description: HonACClimateEntityDescription
def __init__(
self,
hass: HomeAssistantType,
entry: ConfigEntry,
device: HonAppliance,
description: HonACClimateEntityDescription,
) -> None:
super().__init__(hass, entry, device, description)
self._attr_temperature_unit = TEMP_CELSIUS
@ -138,37 +155,38 @@ class HonACClimateEntity(HonEntity, ClimateEntity):
self._handle_coordinator_update(update=False)
def _set_temperature_bound(self) -> None:
self._attr_target_temperature_step = self._device.settings[
"settings.tempSel"
].step
self._attr_max_temp = self._device.settings["settings.tempSel"].max
self._attr_min_temp = self._device.settings["settings.tempSel"].min
temperature = self._device.settings[self.entity_description.key]
if not isinstance(temperature, HonParameterRange):
raise ValueError
self._attr_max_temp = temperature.max
self._attr_target_temperature_step = temperature.step
self._attr_min_temp = temperature.min
@property
def target_temperature(self) -> int | None:
def target_temperature(self) -> float | None:
"""Return the temperature we try to reach."""
return self._device.get("tempSel")
return self._device.get("tempSel", 0.0)
@property
def current_temperature(self) -> float | None:
"""Return the current temperature."""
return self._device.get("tempIndoor")
return self._device.get("tempIndoor", 0.0)
async def async_set_temperature(self, **kwargs):
async def async_set_temperature(self, **kwargs: Any) -> None:
if (temperature := kwargs.get(ATTR_TEMPERATURE)) is None:
return False
return
self._device.settings["settings.tempSel"].value = str(int(temperature))
await self._device.commands["settings"].send()
self.async_write_ha_state()
@property
def hvac_mode(self) -> HVACMode | str | None:
def hvac_mode(self) -> HVACMode:
if self._device.get("onOffStatus") == 0:
return HVACMode.OFF
else:
return HON_HVAC_MODE[self._device.get("machMode")]
async def async_set_hvac_mode(self, hvac_mode):
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
self._attr_hvac_mode = hvac_mode
if hvac_mode == HVACMode.OFF:
await self._device.commands["stopProgram"].send()
@ -215,7 +233,7 @@ class HonACClimateEntity(HonEntity, ClimateEntity):
"""Return the fan setting."""
return HON_FAN[self._device.get("windSpeed")]
async def async_set_fan_mode(self, fan_mode):
async def async_set_fan_mode(self, fan_mode: str) -> None:
fan_modes = {}
for mode in reversed(self._device.settings["settings.windSpeed"].values):
fan_modes[HON_FAN[int(mode)]] = mode
@ -231,14 +249,13 @@ class HonACClimateEntity(HonEntity, ClimateEntity):
vertical = self._device.get("windDirectionVertical")
if horizontal == 7 and vertical == 8:
return SWING_BOTH
elif horizontal == 7:
if horizontal == 7:
return SWING_HORIZONTAL
elif vertical == 8:
if vertical == 8:
return SWING_VERTICAL
else:
return SWING_OFF
return SWING_OFF
async def async_set_swing_mode(self, swing_mode):
async def async_set_swing_mode(self, swing_mode: str) -> None:
horizontal = self._device.settings["settings.windDirectionHorizontal"]
vertical = self._device.settings["settings.windDirectionVertical"]
if swing_mode in [SWING_BOTH, SWING_HORIZONTAL]:
@ -254,13 +271,7 @@ class HonACClimateEntity(HonEntity, ClimateEntity):
self.async_write_ha_state()
@callback
def _handle_coordinator_update(self, update=True) -> None:
self._attr_target_temperature = self.target_temperature
self._attr_current_temperature = self.current_temperature
self._attr_hvac_mode = self.hvac_mode
self._attr_fan_modes = self.fan_modes
self._attr_fan_mode = self.fan_mode
self._attr_swing_mode = self.swing_mode
def _handle_coordinator_update(self, update: bool = True) -> None:
if update:
self.async_write_ha_state()
@ -268,7 +279,13 @@ class HonACClimateEntity(HonEntity, ClimateEntity):
class HonClimateEntity(HonEntity, ClimateEntity):
entity_description: HonClimateEntityDescription
def __init__(self, hass, entry, device: HonAppliance, description) -> None:
def __init__(
self,
hass: HomeAssistantType,
entry: ConfigEntry,
device: HonAppliance,
description: HonClimateEntityDescription,
) -> None:
super().__init__(hass, entry, device, description)
self._attr_temperature_unit = TEMP_CELSIUS
@ -288,7 +305,9 @@ class HonClimateEntity(HonEntity, ClimateEntity):
for mode, data in device.commands["startProgram"].categories.items():
if mode not in data.parameters["program"].values:
continue
if zone := data.parameters.get("zone"):
if (zone := data.parameters.get("zone")) and isinstance(
self.entity_description.name, str
):
if self.entity_description.name.lower() in zone.values:
modes.append(mode)
else:
@ -300,29 +319,29 @@ class HonClimateEntity(HonEntity, ClimateEntity):
@property
def target_temperature(self) -> float | None:
"""Return the temperature we try to reach."""
return self._device.get(self.entity_description.key)
return self._device.get(self.entity_description.key, 0.0)
@property
def current_temperature(self) -> float | None:
"""Return the current temperature."""
temp_key = self.entity_description.key.split(".")[-1].replace("Sel", "")
return self._device.get(temp_key)
return self._device.get(temp_key, 0.0)
async def async_set_temperature(self, **kwargs):
async def async_set_temperature(self, **kwargs: Any) -> None:
if (temperature := kwargs.get(ATTR_TEMPERATURE)) is None:
return False
return
self._device.settings[self.entity_description.key].value = str(int(temperature))
await self._device.commands["settings"].send()
self.async_write_ha_state()
@property
def hvac_mode(self) -> HVACMode | str | None:
def hvac_mode(self) -> HVACMode:
if self._device.get("onOffStatus") == 0:
return HVACMode.OFF
else:
return self.entity_description.mode
async def async_set_hvac_mode(self, hvac_mode):
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
if len(self.hvac_modes) <= 1:
return
if hvac_mode == HVACMode.OFF:
@ -347,7 +366,8 @@ class HonClimateEntity(HonEntity, ClimateEntity):
command = "stopProgram" if preset_mode == "no_mode" else "startProgram"
if program := self._device.settings.get(f"{command}.program"):
program.value = preset_mode
if zone := self._device.settings.get(f"{command}.zone"):
zone = self._device.settings.get(f"{command}.zone")
if zone and isinstance(self.entity_description.name, str):
zone.value = self.entity_description.name.lower()
self._device.sync_command(command, "settings")
self._set_temperature_bound()
@ -356,18 +376,15 @@ class HonClimateEntity(HonEntity, ClimateEntity):
self._attr_preset_mode = preset_mode
self.async_write_ha_state()
def _set_temperature_bound(self):
self._attr_target_temperature_step = self._device.settings[
self.entity_description.key
].step
self._attr_max_temp = self._device.settings[self.entity_description.key].max
self._attr_min_temp = self._device.settings[self.entity_description.key].min
def _set_temperature_bound(self) -> None:
temperature = self._device.settings[self.entity_description.key]
if not isinstance(temperature, HonParameterRange):
raise ValueError
self._attr_max_temp = temperature.max
self._attr_target_temperature_step = temperature.step
self._attr_min_temp = temperature.min
@callback
def _handle_coordinator_update(self, update=True) -> None:
self._attr_target_temperature = self.target_temperature
self._attr_current_temperature = self.current_temperature
self._attr_hvac_mode = self.hvac_mode
self._attr_preset_mode = self.preset_mode
def _handle_coordinator_update(self, update: bool = True) -> None:
if update:
self.async_write_ha_state()

View File

@ -1,8 +1,10 @@
import logging
from typing import Any
import voluptuous as vol
import voluptuous as vol # type: ignore[import]
from homeassistant import config_entries
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.data_entry_flow import FlowResult
from .const import DOMAIN
@ -13,11 +15,13 @@ class HonFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
def __init__(self):
self._email = None
self._password = None
def __init__(self) -> None:
self._email: str | None = None
self._password: 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:
if user_input is None:
return self.async_show_form(
step_id="user",
@ -29,6 +33,14 @@ class HonFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
self._email = user_input[CONF_EMAIL]
self._password = user_input[CONF_PASSWORD]
if self._email is None or self._password is None:
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{vol.Required(CONF_EMAIL): str, vol.Required(CONF_PASSWORD): str}
),
)
# Check if already configured
await self.async_set_unique_id(self._email)
self._abort_if_unique_id_configured()
@ -41,5 +53,5 @@ class HonFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
},
)
async def async_step_import(self, user_input=None):
async def async_step_import(self, user_input: dict[str, str]) -> FlowResult:
return await self.async_step_user(user_input)

View File

@ -6,10 +6,10 @@ from homeassistant.components.climate import (
FAN_AUTO,
)
DOMAIN = "hon"
UPDATE_INTERVAL = 10
DOMAIN: str = "hon"
UPDATE_INTERVAL: int = 10
PLATFORMS = [
PLATFORMS: list[str] = [
"sensor",
"select",
"number",
@ -22,7 +22,7 @@ PLATFORMS = [
"lock",
]
APPLIANCES = {
APPLIANCES: dict[str, str] = {
"AC": "Air Conditioner",
"AP": "Air Purifier",
"AS": "Air Scanner",
@ -40,7 +40,7 @@ APPLIANCES = {
"WM": "Washing Machine",
}
HON_HVAC_MODE = {
HON_HVAC_MODE: dict[int, HVACMode] = {
0: HVACMode.AUTO,
1: HVACMode.COOL,
2: HVACMode.DRY,
@ -50,7 +50,7 @@ HON_HVAC_MODE = {
6: HVACMode.FAN_ONLY,
}
HON_HVAC_PROGRAM = {
HON_HVAC_PROGRAM: dict[str, str] = {
HVACMode.AUTO: "iot_auto",
HVACMode.COOL: "iot_cool",
HVACMode.DRY: "iot_dry",
@ -58,7 +58,7 @@ HON_HVAC_PROGRAM = {
HVACMode.FAN_ONLY: "iot_fan",
}
HON_FAN = {
HON_FAN: dict[int, str] = {
1: FAN_HIGH,
2: FAN_MEDIUM,
3: FAN_LOW,
@ -67,7 +67,7 @@ HON_FAN = {
}
# These languages are official supported by hOn
LANGUAGES = [
LANGUAGES: list[str] = [
"cs", # Czech
"de", # German
"el", # Greek
@ -89,7 +89,7 @@ LANGUAGES = [
"zh", # Chinese
]
WASHING_PR_PHASE = {
WASHING_PR_PHASE: dict[int, str] = {
0: "ready",
1: "washing",
2: "washing",
@ -116,7 +116,7 @@ WASHING_PR_PHASE = {
27: "washing",
}
MACH_MODE = {
MACH_MODE: dict[int, str] = {
0: "ready", # NO_STATE
1: "ready", # SELECTION_MODE
2: "running", # EXECUTION_MODE
@ -129,7 +129,7 @@ MACH_MODE = {
9: "ending", # STOP_MODE
}
TUMBLE_DRYER_PR_PHASE = {
TUMBLE_DRYER_PR_PHASE: dict[int, str] = {
0: "ready",
1: "heat_stroke",
2: "drying",
@ -147,21 +147,21 @@ TUMBLE_DRYER_PR_PHASE = {
20: "drying",
}
DIRTY_LEVEL = {
DIRTY_LEVEL: dict[int, str] = {
0: "unknown",
1: "little",
2: "normal",
3: "very",
}
STEAM_LEVEL = {
STEAM_LEVEL: dict[int, str] = {
0: "no_steam",
1: "cotton",
2: "delicate",
3: "synthetic",
}
DISHWASHER_PR_PHASE = {
DISHWASHER_PR_PHASE: dict[int, str] = {
0: "ready",
1: "prewash",
2: "washing",
@ -171,7 +171,7 @@ DISHWASHER_PR_PHASE = {
6: "hot_rinse",
}
TUMBLE_DRYER_DRY_LEVEL = {
TUMBLE_DRYER_DRY_LEVEL: dict[int, str] = {
0: "no_dry",
1: "iron_dry",
2: "no_dry_iron",
@ -184,7 +184,7 @@ TUMBLE_DRYER_DRY_LEVEL = {
15: "extra_dry",
}
AC_MACH_MODE = {
AC_MACH_MODE: dict[int, str] = {
0: "auto",
1: "cool",
2: "cool",
@ -194,7 +194,7 @@ AC_MACH_MODE = {
6: "fan",
}
AC_FAN_MODE = {
AC_FAN_MODE: dict[int, str] = {
1: "high",
2: "mid",
3: "low",
@ -202,14 +202,14 @@ AC_FAN_MODE = {
5: "auto",
}
AC_HUMAN_SENSE = {
AC_HUMAN_SENSE: dict[int, str] = {
0: "touch_off",
1: "avoid_touch",
2: "follow_touch",
3: "unknown",
}
AP_MACH_MODE = {
AP_MACH_MODE: dict[int, str] = {
0: "standby",
1: "sleep",
2: "auto",
@ -217,7 +217,7 @@ AP_MACH_MODE = {
4: "max",
}
AP_DIFFUSER_LEVEL = {
AP_DIFFUSER_LEVEL: dict[int, str] = {
0: "off",
1: "soft",
2: "mid",
@ -225,4 +225,4 @@ AP_DIFFUSER_LEVEL = {
4: "custom",
}
REF_HUMIDITY_LEVELS = {1: "low", 2: "mid", 3: "high"}
REF_HUMIDITY_LEVELS: dict[int, str] = {1: "low", 2: "mid", 3: "high"}

View File

@ -1,6 +1,5 @@
import logging
import math
from dataclasses import dataclass
from typing import Any
from homeassistant.components.fan import (
@ -10,6 +9,8 @@ from homeassistant.components.fan import (
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.util.percentage import (
percentage_to_ranged_value,
ranged_value_to_percentage,
@ -19,18 +20,14 @@ from pyhon.parameter.range import HonParameterRange
from .const import DOMAIN
from .hon import HonEntity
from .typedefs import HonEntityDescription
_LOGGER = logging.getLogger(__name__)
@dataclass
class HonFanEntityDescription(FanEntityDescription):
pass
FANS = {
FANS: dict[str, tuple[FanEntityDescription, ...]] = {
"HO": (
HonFanEntityDescription(
FanEntityDescription(
key="settings.windSpeed",
name="Wind Speed",
translation_key="air_extraction",
@ -39,30 +36,36 @@ FANS = {
}
async def async_setup_entry(hass, entry: ConfigEntry, async_add_entities) -> None:
async def async_setup_entry(
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
entities = []
for device in hass.data[DOMAIN][entry.unique_id].appliances:
for description in FANS.get(device.appliance_type, []):
if isinstance(description, HonFanEntityDescription):
if (
description.key not in device.available_settings
or device.get(description.key.split(".")[-1]) is None
):
continue
entity = HonFanEntity(hass, entry, device, description)
else:
if (
description.key not in device.available_settings
or device.get(description.key.split(".")[-1]) is None
):
continue
entity = HonFanEntity(hass, entry, device, description)
await entity.coordinator.async_config_entry_first_refresh()
entities.append(entity)
async_add_entities(entities)
class HonFanEntity(HonEntity, FanEntity):
entity_description: HonFanEntityDescription
entity_description: FanEntityDescription
def __init__(self, hass, entry, device: HonAppliance, description) -> None:
def __init__(
self,
hass: HomeAssistantType,
entry: ConfigEntry,
device: HonAppliance,
description: FanEntityDescription,
) -> None:
self._attr_supported_features = FanEntityFeature.SET_SPEED
self._wind_speed: HonParameterRange = device.settings.get(description.key)
self._wind_speed: HonParameterRange
self._speed_range: tuple[int, int]
self._command, self._parameter = description.key.split(".")
super().__init__(hass, entry, device, description)
@ -89,8 +92,10 @@ class HonFanEntity(HonEntity, FanEntity):
@property
def is_on(self) -> bool | None:
"""Return true if device is on."""
if self.percentage is None:
return False
mode = math.ceil(percentage_to_ranged_value(self._speed_range, self.percentage))
return mode > self._wind_speed.min
return bool(mode > self._wind_speed.min)
async def async_turn_on(
self,
@ -112,9 +117,10 @@ class HonFanEntity(HonEntity, FanEntity):
self.async_write_ha_state()
@callback
def _handle_coordinator_update(self, update=True) -> None:
self._wind_speed = self._device.settings.get(self.entity_description.key)
if len(self._wind_speed.values) > 1:
def _handle_coordinator_update(self, update: bool = True) -> None:
wind_speed = self._device.settings.get(self.entity_description.key)
if isinstance(wind_speed, HonParameterRange) and len(wind_speed.values) > 1:
self._wind_speed = wind_speed
self._speed_range = (
int(self._wind_speed.values[1]),
int(self._wind_speed.values[-1]),

View File

@ -3,23 +3,79 @@ import logging
from contextlib import suppress
from datetime import timedelta
from pathlib import Path
from typing import Optional, Any, TypeVar
import pkg_resources
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import callback
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from pyhon.appliance import HonAppliance
from .const import DOMAIN, UPDATE_INTERVAL
from .typedefs import HonEntityDescription, HonOptionEntityDescription, T
_LOGGER = logging.getLogger(__name__)
class HonEntity(CoordinatorEntity):
class HonInfo:
def __init__(self) -> None:
self._manifest: dict[str, Any] = self._get_manifest()
self._hon_version: str = self._manifest.get("version", "")
self._pyhon_version: str = pkg_resources.get_distribution("pyhon").version
@staticmethod
def _get_manifest() -> dict[str, Any]:
manifest = Path(__file__).parent / "manifest.json"
with open(manifest, "r", encoding="utf-8") as file:
result: dict[str, Any] = json.loads(file.read())
return result
@property
def manifest(self) -> dict[str, Any]:
return self._manifest
@property
def hon_version(self) -> str:
return self._hon_version
@property
def pyhon_version(self) -> str:
return self._pyhon_version
class HonCoordinator(DataUpdateCoordinator[None]):
def __init__(self, hass: HomeAssistantType, device: HonAppliance):
"""Initialize my coordinator."""
super().__init__(
hass,
_LOGGER,
name=device.unique_id,
update_interval=timedelta(seconds=UPDATE_INTERVAL),
)
self._device = device
self._info = HonInfo()
async def _async_update_data(self) -> None:
return await self._device.update()
@property
def info(self) -> HonInfo:
return self._info
class HonEntity(CoordinatorEntity[HonCoordinator]):
_attr_has_entity_name = True
def __init__(self, hass, entry, device: HonAppliance, description=None) -> None:
def __init__(
self,
hass: HomeAssistantType,
entry: ConfigEntry,
device: HonAppliance,
description: Optional[HonEntityDescription] = None,
) -> None:
coordinator = get_coordinator(hass, device)
super().__init__(coordinator)
@ -36,7 +92,7 @@ class HonEntity(CoordinatorEntity):
self._handle_coordinator_update(update=False)
@property
def device_info(self):
def device_info(self) -> DeviceInfo:
return DeviceInfo(
identifiers={(DOMAIN, self._device.unique_id)},
manufacturer=self._device.get("brand", ""),
@ -51,71 +107,34 @@ class HonEntity(CoordinatorEntity):
self.async_write_ha_state()
class HonInfo:
def __init__(self):
self._manifest = self._get_manifest()
self._hon_version = self._manifest.get("version", "")
self._pyhon_version = pkg_resources.get_distribution("pyhon").version
@staticmethod
def _get_manifest():
manifest = Path(__file__).parent / "manifest.json"
with open(manifest, "r", encoding="utf-8") as file:
return json.loads(file.read())
@property
def manifest(self):
return self._manifest
@property
def hon_version(self):
return self._hon_version
@property
def pyhon_version(self):
return self._pyhon_version
class HonCoordinator(DataUpdateCoordinator):
def __init__(self, hass, device: HonAppliance):
"""Initialize my coordinator."""
super().__init__(
hass,
_LOGGER,
name=device.unique_id,
update_interval=timedelta(seconds=UPDATE_INTERVAL),
)
self._device = device
self._info = HonInfo()
async def _async_update_data(self):
await self._device.update()
@property
def info(self) -> HonInfo:
return self._info
def unique_entities(base_entities, new_entities):
def unique_entities(
base_entities: tuple[T, ...],
new_entities: tuple[T, ...],
) -> tuple[T, ...]:
result = list(base_entities)
existing_entities = [entity.key for entity in base_entities]
entity: HonEntityDescription
for entity in new_entities:
if entity.key not in existing_entities:
result.append(entity)
return tuple(result)
def get_coordinator(hass, appliance):
def get_coordinator(hass: HomeAssistantType, appliance: HonAppliance) -> HonCoordinator:
coordinators = hass.data[DOMAIN]["coordinators"]
if appliance.unique_id in coordinators:
coordinator = hass.data[DOMAIN]["coordinators"][appliance.unique_id]
coordinator: HonCoordinator = hass.data[DOMAIN]["coordinators"][
appliance.unique_id
]
else:
coordinator = HonCoordinator(hass, appliance)
hass.data[DOMAIN]["coordinators"][appliance.unique_id] = coordinator
return coordinator
def get_readable(description, value):
def get_readable(
description: HonOptionEntityDescription, value: float | str
) -> float | str:
if description.option_list is not None:
with suppress(ValueError):
return description.option_list.get(int(value), value)

View File

@ -9,6 +9,8 @@ from homeassistant.components.light import (
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import HomeAssistantType
from pyhon.appliance import HonAppliance
from pyhon.parameter.range import HonParameterRange
@ -18,7 +20,7 @@ from .hon import HonEntity
_LOGGER = logging.getLogger(__name__)
LIGHTS = {
LIGHTS: dict[str, tuple[LightEntityDescription, ...]] = {
"WC": (
LightEntityDescription(
key="settings.lightStatus",
@ -43,7 +45,9 @@ LIGHTS = {
}
async def async_setup_entry(hass, entry: ConfigEntry, async_add_entities) -> None:
async def async_setup_entry(
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
entities = []
for device in hass.data[DOMAIN][entry.unique_id].appliances:
for description in LIGHTS.get(device.appliance_type, []):
@ -61,8 +65,16 @@ async def async_setup_entry(hass, entry: ConfigEntry, async_add_entities) -> Non
class HonLightEntity(HonEntity, LightEntity):
entity_description: LightEntityDescription
def __init__(self, hass, entry, device: HonAppliance, description) -> None:
light: HonParameterRange = device.settings.get(description.key)
def __init__(
self,
hass: HomeAssistantType,
entry: ConfigEntry,
device: HonAppliance,
description: LightEntityDescription,
) -> None:
light = self._device.settings.get(self.entity_description.key)
if not isinstance(light, HonParameterRange):
raise ValueError()
self._light_range = (light.min, light.max)
self._attr_supported_color_modes: set[ColorMode] = set()
if len(light.values) == 2:
@ -76,13 +88,13 @@ class HonLightEntity(HonEntity, LightEntity):
@property
def is_on(self) -> bool:
"""Return true if light is on."""
return self._device.get(self.entity_description.key.split(".")[-1]) > 0
return bool(self._device.get(self.entity_description.key.split(".")[-1]) > 0)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on or control the light."""
light: HonParameterRange = self._device.settings.get(
self.entity_description.key
)
light = self._device.settings.get(self.entity_description.key)
if not isinstance(light, HonParameterRange):
raise ValueError()
if ColorMode.BRIGHTNESS in self._attr_supported_color_modes:
percent = int(100 / 255 * kwargs.get(ATTR_BRIGHTNESS, 128))
light.value = round(light.max / 100 * percent)
@ -96,9 +108,9 @@ class HonLightEntity(HonEntity, LightEntity):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Instruct the light to turn off."""
light: HonParameterRange = self._device.settings.get(
self.entity_description.key
)
light = self._device.settings.get(self.entity_description.key)
if not isinstance(light, HonParameterRange):
raise ValueError()
light.value = light.min
await self._device.commands[self._command].send()
self.async_write_ha_state()
@ -106,15 +118,15 @@ class HonLightEntity(HonEntity, LightEntity):
@property
def brightness(self) -> int | None:
"""Return the brightness of the light."""
light: HonParameterRange = self._device.settings.get(
self.entity_description.key
)
light = self._device.settings.get(self.entity_description.key)
if not isinstance(light, HonParameterRange):
raise ValueError()
if light.value == light.min:
return None
return int(255 / light.max * light.value)
return int(255 / light.max * float(light.value))
@callback
def _handle_coordinator_update(self, update=True) -> None:
def _handle_coordinator_update(self, update: bool = True) -> None:
self._attr_is_on = self.is_on
self._attr_brightness = self.brightness
if update:
@ -122,7 +134,6 @@ class HonLightEntity(HonEntity, LightEntity):
@property
def available(self) -> bool:
return (
super().available
and len(self._device.settings.get(self.entity_description.key).values) > 1
)
if (entity := self._device.settings.get(self.entity_description.key)) is None:
return False
return super().available and len(entity.values) > 1

View File

@ -4,6 +4,8 @@ from typing import Any
from homeassistant.components.lock import LockEntity, LockEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import HomeAssistantType
from pyhon.parameter.base import HonParameter
from pyhon.parameter.range import HonParameterRange
@ -23,7 +25,9 @@ LOCKS: dict[str, tuple[LockEntityDescription, ...]] = {
}
async def async_setup_entry(hass, entry: ConfigEntry, async_add_entities) -> None:
async def async_setup_entry(
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
entities = []
for device in hass.data[DOMAIN][entry.unique_id].appliances:
for description in LOCKS.get(device.appliance_type, []):
@ -45,13 +49,12 @@ class HonLockEntity(HonEntity, LockEntity):
@property
def is_locked(self) -> bool | None:
"""Return a boolean for the state of the lock."""
"""Return True if entity is on."""
return self._device.get(self.entity_description.key, 0) == 1
return bool(self._device.get(self.entity_description.key, 0) == 1)
async def async_lock(self, **kwargs: Any) -> None:
"""Lock method."""
setting = self._device.settings[f"settings.{self.entity_description.key}"]
if type(setting) == HonParameter:
setting = self._device.settings.get(f"settings.{self.entity_description.key}")
if type(setting) == HonParameter or setting is None:
return
setting.value = setting.max if isinstance(setting, HonParameterRange) else 1
self.async_write_ha_state()
@ -78,8 +81,7 @@ class HonLockEntity(HonEntity, LockEntity):
)
@callback
def _handle_coordinator_update(self, update=True) -> None:
value = self._device.get(self.entity_description.key, 0)
def _handle_coordinator_update(self, update: bool = True) -> None:
self._attr_is_locked = self.is_locked
if update:
self.async_write_ha_state()

View File

@ -10,6 +10,9 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import UnitOfTime, UnitOfTemperature
from homeassistant.core import callback
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import HomeAssistantType
from pyhon.appliance import HonAppliance
from pyhon.parameter.range import HonParameterRange
from .const import DOMAIN
@ -183,8 +186,11 @@ NUMBERS: dict[str, tuple[NumberEntityDescription, ...]] = {
NUMBERS["WD"] = unique_entities(NUMBERS["WM"], NUMBERS["TD"])
async def async_setup_entry(hass, entry: ConfigEntry, async_add_entities) -> None:
async def async_setup_entry(
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
entities = []
entity: HonNumberEntity | HonConfigNumberEntity
for device in hass.data[DOMAIN][entry.unique_id].appliances:
for description in NUMBERS.get(device.appliance_type, []):
if description.key not in device.available_settings:
@ -203,7 +209,13 @@ async def async_setup_entry(hass, entry: ConfigEntry, async_add_entities) -> Non
class HonNumberEntity(HonEntity, NumberEntity):
entity_description: HonNumberEntityDescription
def __init__(self, hass, entry, device, description) -> None:
def __init__(
self,
hass: HomeAssistantType,
entry: ConfigEntry,
device: HonAppliance,
description: HonNumberEntityDescription,
) -> None:
super().__init__(hass, entry, device, description)
self._data = device.settings[description.key]
@ -214,7 +226,9 @@ class HonNumberEntity(HonEntity, NumberEntity):
@property
def native_value(self) -> float | None:
return self._device.get(self.entity_description.key.split(".")[-1])
if value := self._device.get(self.entity_description.key.split(".")[-1]):
return float(value)
return None
async def async_set_native_value(self, value: float) -> None:
setting = self._device.settings[self.entity_description.key]
@ -227,7 +241,7 @@ class HonNumberEntity(HonEntity, NumberEntity):
await self.coordinator.async_refresh()
@callback
def _handle_coordinator_update(self, update=True) -> None:
def _handle_coordinator_update(self, update: bool = True) -> None:
setting = self._device.settings[self.entity_description.key]
if isinstance(setting, HonParameterRange):
self._attr_native_max_value = setting.max
@ -247,14 +261,31 @@ class HonNumberEntity(HonEntity, NumberEntity):
)
class HonConfigNumberEntity(HonNumberEntity):
class HonConfigNumberEntity(HonEntity, NumberEntity):
entity_description: HonConfigNumberEntityDescription
def __init__(
self,
hass: HomeAssistantType,
entry: ConfigEntry,
device: HonAppliance,
description: HonConfigNumberEntityDescription,
) -> None:
super().__init__(hass, entry, device, description)
self._data = device.settings[description.key]
if isinstance(self._data, HonParameterRange):
self._attr_native_max_value = self._data.max
self._attr_native_min_value = self._data.min
self._attr_native_step = self._data.step
@property
def native_value(self) -> float | None:
return self._device.settings[self.entity_description.key].value
if value := self._device.settings[self.entity_description.key].value:
return float(value)
return None
async def async_set_native_value(self, value: str) -> None:
async def async_set_native_value(self, value: float) -> None:
setting = self._device.settings[self.entity_description.key]
if isinstance(setting, HonParameterRange):
setting.value = value
@ -264,3 +295,14 @@ class HonConfigNumberEntity(HonNumberEntity):
def available(self) -> bool:
"""Return True if entity is available."""
return super(NumberEntity, self).available
@callback
def _handle_coordinator_update(self, update: bool = True) -> None:
setting = self._device.settings[self.entity_description.key]
if isinstance(setting, HonParameterRange):
self._attr_native_max_value = setting.max
self._attr_native_min_value = setting.min
self._attr_native_step = setting.step
self._attr_native_value = self.native_value
if update:
self.async_write_ha_state()

View File

@ -2,13 +2,14 @@ from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import Dict, List
from homeassistant.components.select import SelectEntity, SelectEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import UnitOfTemperature, UnitOfTime, REVOLUTIONS_PER_MINUTE
from homeassistant.core import callback
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import HomeAssistantType
from . import const
from .const import DOMAIN
@ -19,16 +20,16 @@ _LOGGER = logging.getLogger(__name__)
@dataclass
class HonSelectEntityDescription(SelectEntityDescription):
option_list: Dict[int, str] = None
option_list: dict[int, str] | None = None
@dataclass
class HonConfigSelectEntityDescription(SelectEntityDescription):
entity_category: EntityCategory = EntityCategory.CONFIG
option_list: Dict[int, str] = None
option_list: dict[int, str] | None = None
SELECTS = {
SELECTS: dict[str, tuple[SelectEntityDescription, ...]] = {
"WM": (
HonConfigSelectEntityDescription(
key="startProgram.spinSpeed",
@ -168,8 +169,11 @@ SELECTS = {
SELECTS["WD"] = unique_entities(SELECTS["WM"], SELECTS["TD"])
async def async_setup_entry(hass, entry: ConfigEntry, async_add_entities) -> None:
async def async_setup_entry(
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
entities = []
entity: HonSelectEntity | HonConfigSelectEntity
for device in hass.data[DOMAIN][entry.unique_id].appliances:
for description in SELECTS.get(device.appliance_type, []):
if description.key not in device.available_settings:
@ -195,16 +199,18 @@ class HonConfigSelectEntity(HonEntity, SelectEntity):
value = get_readable(self.entity_description, setting.value)
<