Cool_Tools/hOn
1
0
Fork 0
mirror of https://github.com/Andre0512/hon.git synced 2024-11-23 23:01:00 +01:00
Code Issues Packages Projects Releases Wiki Activity

Rebuild to single data coordinator

Browse Source
This commit is contained in:
Andre Basche 2024-03-29 01:22:44 +01:00
parent 8f1fc627e6
commit a6c2c3e992
16 changed files with 148 additions and 203 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
custom_components/hon/__init__.py
View File

@ -1,11 +1,13 @@
import logging
from pathlib import Path
from typing import Any
import voluptuous as vol # type: ignore[import-untyped]
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.helpers import config_validation as cv, aiohttp_client
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from pyhon import Hon
from .const import DOMAIN, PLATFORMS, MOBILE_ID, CONF_REFRESH_TOKEN
@ -29,23 +31,27 @@ async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool
session = aiohttp_client.async_get_clientsession(hass)
if (config_dir := hass.config.config_dir) is None:
raise ValueError("Missing Config Dir")
kwargs = {
"email": entry.data[CONF_EMAIL],
"password": entry.data[CONF_PASSWORD],
"mobile_id": MOBILE_ID,
"session": session,
"test_data_path": Path(config_dir),
}
if refresh_token := entry.data.get(CONF_REFRESH_TOKEN):
kwargs["refresh_token"] = refresh_token
hon = await Hon(**kwargs).create()
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.unique_id] = hon
hon = await Hon(
email=entry.data[CONF_EMAIL],
password=entry.data[CONF_PASSWORD],
mobile_id=MOBILE_ID,
session=session,
# test_data_path=Path(config_dir),
refresh_token=entry.data.get(CONF_REFRESH_TOKEN, ""),
).create()
# Save the new refresh token
hass.config_entries.async_update_entry(
entry, data={**entry.data, CONF_REFRESH_TOKEN: hon.api.auth.refresh_token}
)
hass.data[DOMAIN]["coordinators"] = {}
coordinator: DataUpdateCoordinator[dict[str, Any]] = DataUpdateCoordinator(
hass, _LOGGER, name=DOMAIN
)
hon.subscribe_updates(coordinator.async_set_updated_data)
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.unique_id] = {"hon": hon, "coordinator": coordinator}
for platform in PLATFORMS:
hass.async_create_task(
@ -55,7 +61,7 @@ async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool
async def async_unload_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool:
refresh_token = hass.data[DOMAIN][entry.unique_id].api.auth.refresh_token
refresh_token = hass.data[DOMAIN][entry.unique_id]["hon"].api.auth.refresh_token
hass.config_entries.async_update_entry(
entry, data={**entry.data, CONF_REFRESH_TOKEN: refresh_token}

5
custom_components/hon/binary_sensor.py
View File

@ -12,7 +12,8 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import HomeAssistantType
from .const import DOMAIN
from .hon import HonEntity, unique_entities
from .entity import HonEntity
from .util import unique_entities
_LOGGER = logging.getLogger(__name__)
@ -319,7 +320,7 @@ 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 device in hass.data[DOMAIN][entry.unique_id]["hon"].appliances:
for description in BINARY_SENSORS.get(device.appliance_type, []):
if device.get(description.key) is None:
continue

18
custom_components/hon/button.py
View File

@ -10,7 +10,7 @@ from homeassistant.helpers.typing import HomeAssistantType
from pyhon.appliance import HonAppliance
from .const import DOMAIN
from .hon import HonEntity
from .entity import HonEntity
from .typedefs import HonButtonType
_LOGGER = logging.getLogger(__name__)
@ -59,7 +59,7 @@ 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 device in hass.data[DOMAIN][entry.unique_id]["hon"].appliances:
for description in BUTTONS.get(device.appliance_type, []):
if not device.commands.get(description.key):
continue
@ -96,19 +96,14 @@ class HonDeviceInfo(HonEntity, ButtonEntity):
self._attr_icon = "mdi:information"
self._attr_name = "Show Device Info"
self._attr_entity_category = EntityCategory.DIAGNOSTIC
if "beta" not in self.coordinator.info.hon_version:
self._attr_entity_registry_enabled_default = False
self._attr_entity_registry_enabled_default = False
async def async_press(self) -> None:
versions = "versions:\n"
versions += f" hon: {self.coordinator.info.hon_version}\n"
versions += f" pyhOn: {self.coordinator.info.pyhon_version}\n"
info = f"{self._device.diagnose}{versions}"
title = f"{self._device.nick_name} Device Info"
persistent_notification.create(
self._hass, f"````\n```\n{info}\n```\n````", title
self._hass, f"````\n```\n{self._device.diagnose}\n```\n````", title
)
_LOGGER.info(info.replace(" ", "\u200B "))
_LOGGER.info(self._device.diagnose.replace(" ", "\u200B "))
class HonDataArchive(HonEntity, ButtonEntity):
@ -121,8 +116,7 @@ class HonDataArchive(HonEntity, ButtonEntity):
self._attr_icon = "mdi:archive-arrow-down"
self._attr_name = "Create Data Archive"
self._attr_entity_category = EntityCategory.DIAGNOSTIC
if "beta" not in self.coordinator.info.hon_version:
self._attr_entity_registry_enabled_default = False
self._attr_entity_registry_enabled_default = False
async def async_press(self) -> None:
if (config_dir := self._hass.config.config_dir) is None:

8
custom_components/hon/climate.py
View File

@ -26,7 +26,7 @@ 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
from .entity import HonEntity
_LOGGER = logging.getLogger(__name__)
@ -108,7 +108,7 @@ async def async_setup_entry(
) -> None:
entities = []
entity: HonClimateEntity | HonACClimateEntity
for device in hass.data[DOMAIN][entry.unique_id].appliances:
for device in hass.data[DOMAIN][entry.unique_id]["hon"].appliances:
for description in CLIMATES.get(device.appliance_type, []):
if isinstance(description, HonACClimateEntityDescription):
if description.key not in list(device.commands):
@ -223,7 +223,7 @@ class HonACClimateEntity(HonEntity, ClimateEntity):
self._device.sync_command("startProgram", "settings")
self._set_temperature_bound()
self._handle_coordinator_update(update=False)
await self.coordinator.async_refresh()
self.async_write_ha_state()
self._attr_preset_mode = preset_mode
await self._device.commands["startProgram"].send()
self.async_write_ha_state()
@ -390,7 +390,7 @@ class HonClimateEntity(HonEntity, ClimateEntity):
self._device.sync_command(command, "settings")
self._set_temperature_bound()
self._attr_preset_mode = preset_mode
await self.coordinator.async_refresh()
self.async_write_ha_state()
await self._device.commands[command].send()
self.async_write_ha_state()

54
custom_components/hon/entity.py Normal file
View File

@ -0,0 +1,54 @@
from typing import Optional, Any
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
from .typedefs import HonEntityDescription
class HonEntity(CoordinatorEntity[DataUpdateCoordinator[dict[str, Any]]]):
_attr_has_entity_name = True
_attr_should_poll = False
def __init__(
self,
hass: HomeAssistantType,
entry: ConfigEntry,
device: HonAppliance,
description: Optional[HonEntityDescription] = None,
) -> None:
self.coordinator = hass.data[DOMAIN][entry.unique_id]["coordinator"]
super().__init__(self.coordinator)
self._hon = hass.data[DOMAIN][entry.unique_id]["hon"]
self._hass = hass
self._device: HonAppliance = device
if description is not None:
self.entity_description = description
self._attr_unique_id = f"{self._device.unique_id}{description.key}"
else:
self._attr_unique_id = self._device.unique_id
self._handle_coordinator_update(update=False)
@property
def device_info(self) -> DeviceInfo:
return DeviceInfo(
identifiers={(DOMAIN, self._device.unique_id)},
manufacturer=self._device.get("brand", ""),
name=self._device.nick_name,
model=self._device.model_name,
sw_version=self._device.get("fwVersion", ""),
)
@callback
def _handle_coordinator_update(self, update: bool = True) -> None:
if update:
self.async_write_ha_state()

4
custom_components/hon/fan.py
View File

@ -19,7 +19,7 @@ from pyhon.appliance import HonAppliance
from pyhon.parameter.range import HonParameterRange
from .const import DOMAIN
from .hon import HonEntity
from .entity import HonEntity
_LOGGER = logging.getLogger(__name__)
@ -39,7 +39,7 @@ 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 device in hass.data[DOMAIN][entry.unique_id]["hon"].appliances:
for description in FANS.get(device.appliance_type, []):
if (
description.key not in device.available_settings

142
custom_components/hon/hon.py
View File

@ -1,142 +0,0 @@
import json
import logging
from contextlib import suppress
from datetime import timedelta
from pathlib import Path
from typing import Optional, Any
import pkg_resources # type: ignore[import, unused-ignore]
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
from .typedefs import HonEntityDescription, HonOptionEntityDescription, T
_LOGGER = logging.getLogger(__name__)
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,
)
self._device = device
self._info = HonInfo()
self._device.subscribe(self.async_set_updated_data)
async def _async_update_data(self) -> None:
return
@property
def info(self) -> HonInfo:
return self._info
class HonEntity(CoordinatorEntity[HonCoordinator]):
_attr_has_entity_name = True
_attr_should_poll = False
def __init__(
self,
hass: HomeAssistantType,
entry: ConfigEntry,
device: HonAppliance,
description: Optional[HonEntityDescription] = None,
) -> None:
coordinator = get_coordinator(hass, device)
super().__init__(coordinator)
self._hon = hass.data[DOMAIN][entry.unique_id]
self._hass = hass
self._coordinator = coordinator
self._device: HonAppliance = device
if description is not None:
self.entity_description = description
self._attr_unique_id = f"{self._device.unique_id}{description.key}"
else:
self._attr_unique_id = self._device.unique_id
self._handle_coordinator_update(update=False)
@property
def device_info(self) -> DeviceInfo:
return DeviceInfo(
identifiers={(DOMAIN, self._device.unique_id)},
manufacturer=self._device.get("brand", ""),
name=self._device.nick_name,
model=self._device.model_name,
sw_version=self._device.get("fwVersion", ""),
)
@callback
def _handle_coordinator_update(self, update: bool = True) -> None:
if update:
self.async_write_ha_state()
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: HomeAssistantType, appliance: HonAppliance) -> HonCoordinator:
coordinators = hass.data[DOMAIN]["coordinators"]
if appliance.unique_id in coordinators:
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: HonOptionEntityDescription, value: float | str
) -> float | str:
if description.option_list is not None:
with suppress(ValueError):
return description.option_list.get(int(value), value)
return value

4
custom_components/hon/light.py
View File

@ -15,7 +15,7 @@ from pyhon.appliance import HonAppliance
from pyhon.parameter.range import HonParameterRange
from .const import DOMAIN
from .hon import HonEntity
from .entity import HonEntity
_LOGGER = logging.getLogger(__name__)
@ -56,7 +56,7 @@ 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 device in hass.data[DOMAIN][entry.unique_id]["hon"].appliances:
for description in LIGHTS.get(device.appliance_type, []):
if (
description.key not in device.available_settings

8
custom_components/hon/lock.py
View File

@ -10,7 +10,7 @@ from pyhon.parameter.base import HonParameter
from pyhon.parameter.range import HonParameterRange
from .const import DOMAIN
from .hon import HonEntity
from .entity import HonEntity
_LOGGER = logging.getLogger(__name__)
@ -29,7 +29,7 @@ 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 device in hass.data[DOMAIN][entry.unique_id]["hon"].appliances:
for description in LOCKS.get(device.appliance_type, []):
if (
f"settings.{description.key}" not in device.available_settings
@ -58,7 +58,7 @@ class HonLockEntity(HonEntity, LockEntity):
setting.value = setting.max if isinstance(setting, HonParameterRange) else 1
self.async_write_ha_state()
await self._device.commands["settings"].send()
await self.coordinator.async_refresh()
self.async_write_ha_state()
async def async_unlock(self, **kwargs: Any) -> None:
"""Unlock method."""
@ -68,7 +68,7 @@ class HonLockEntity(HonEntity, LockEntity):
setting.value = setting.min if isinstance(setting, HonParameterRange) else 0
self.async_write_ha_state()
await self._device.commands["settings"].send()
await self.coordinator.async_refresh()
self.async_write_ha_state()
@property
def available(self) -> bool:

4
custom_components/hon/manifest.json
View File

@ -9,7 +9,7 @@
"iot_class": "cloud_push",
"issue_tracker": "https://github.com/Andre0512/hon/issues",
"requirements": [
"pyhOn==0.17.1"
"pyhOn==0.17.2"
],
"version": "0.14.0-beta.3"
"version": "0.14.0-beta.4"
}

9
custom_components/hon/number.py
View File

@ -16,7 +16,8 @@ from pyhon.appliance import HonAppliance
from pyhon.parameter.range import HonParameterRange
from .const import DOMAIN
from .hon import HonEntity, unique_entities
from .entity import HonEntity
from .util import unique_entities
@dataclass(frozen=True)
@ -210,7 +211,7 @@ async def async_setup_entry(
) -> None:
entities = []
entity: HonNumberEntity | HonConfigNumberEntity
for device in hass.data[DOMAIN][entry.unique_id].appliances:
for device in hass.data[DOMAIN][entry.unique_id]["hon"].appliances:
for description in NUMBERS.get(device.appliance_type, []):
if description.key not in device.available_settings:
continue
@ -256,7 +257,7 @@ class HonNumberEntity(HonEntity, NumberEntity):
await self._device.commands[command].send()
if command != "settings":
self._device.sync_command(command, "settings")
await self.coordinator.async_refresh()
self.async_write_ha_state()
@callback
def _handle_coordinator_update(self, update: bool = True) -> None:
@ -307,7 +308,7 @@ class HonConfigNumberEntity(HonEntity, NumberEntity):
setting = self._device.settings[self.entity_description.key]
if isinstance(setting, HonParameterRange):
setting.value = value
await self.coordinator.async_refresh()
self.async_write_ha_state()
@property
def available(self) -> bool:

9
custom_components/hon/select.py
View File

@ -13,7 +13,8 @@ from homeassistant.helpers.typing import HomeAssistantType
from . import const
from .const import DOMAIN
from .hon import HonEntity, unique_entities, get_readable
from .entity import HonEntity
from .util import unique_entities, get_readable
_LOGGER = logging.getLogger(__name__)
@ -214,7 +215,7 @@ async def async_setup_entry(
) -> None:
entities = []
entity: HonSelectEntity | HonConfigSelectEntity
for device in hass.data[DOMAIN][entry.unique_id].appliances:
for device in hass.data[DOMAIN][entry.unique_id]["hon"].appliances:
for description in SELECTS.get(device.appliance_type, []):
if description.key not in device.available_settings:
continue
@ -262,7 +263,7 @@ class HonConfigSelectEntity(HonEntity, SelectEntity):
async def async_select_option(self, option: str) -> None:
setting = self._device.settings[self.entity_description.key]
setting.value = self._option_to_number(option, setting.values)
await self.coordinator.async_refresh()
self.async_write_ha_state()
@callback
def _handle_coordinator_update(self, update: bool = True) -> None:
@ -316,7 +317,7 @@ class HonSelectEntity(HonEntity, SelectEntity):
await self._device.commands[command].send()
if command != "settings":
self._device.sync_command(command, "settings")
await self.coordinator.async_refresh()
self.async_write_ha_state()
@property
def available(self) -> bool:

5
custom_components/hon/sensor.py
View File

@ -29,7 +29,8 @@ from homeassistant.helpers.typing import HomeAssistantType
from . import const
from .const import DOMAIN
from .hon import HonEntity, unique_entities, get_readable
from .entity import HonEntity
from .util import unique_entities, get_readable
_LOGGER = logging.getLogger(__name__)
@ -812,7 +813,7 @@ async def async_setup_entry(
) -> None:
entities = []
entity: HonSensorEntity | HonConfigSensorEntity
for device in hass.data[DOMAIN][entry.unique_id].appliances:
for device in hass.data[DOMAIN][entry.unique_id]["hon"].appliances:
for description in SENSORS.get(device.appliance_type, []):
if isinstance(description, HonSensorEntityDescription):
if device.get(description.key) is None:

17
custom_components/hon/switch.py
View File

@ -13,7 +13,8 @@ from pyhon.parameter.base import HonParameter
from pyhon.parameter.range import HonParameterRange
from .const import DOMAIN
from .hon import HonEntity, unique_entities
from .entity import HonEntity
from .util import unique_entities
_LOGGER = logging.getLogger(__name__)
@ -406,7 +407,7 @@ async def async_setup_entry(
) -> None:
entities = []
entity: HonConfigSwitchEntity | HonControlSwitchEntity | HonSwitchEntity
for device in hass.data[DOMAIN][entry.unique_id].appliances:
for device in hass.data[DOMAIN][entry.unique_id]["hon"].appliances:
for description in SWITCHES.get(device.appliance_type, []):
if isinstance(description, HonConfigSwitchEntityDescription):
if description.key not in device.available_settings:
@ -446,7 +447,7 @@ class HonSwitchEntity(HonEntity, SwitchEntity):
setting.value = setting.max if isinstance(setting, HonParameterRange) else 1
self.async_write_ha_state()
await self._device.commands["settings"].send()
await self.coordinator.async_refresh()
self.async_write_ha_state()
async def async_turn_off(self, **kwargs: Any) -> None:
setting = self._device.settings[f"settings.{self.entity_description.key}"]
@ -455,7 +456,7 @@ class HonSwitchEntity(HonEntity, SwitchEntity):
setting.value = setting.min if isinstance(setting, HonParameterRange) else 0
self.async_write_ha_state()
await self._device.commands["settings"].send()
await self.coordinator.async_refresh()
self.async_write_ha_state()
@property
def available(self) -> bool:
@ -488,14 +489,14 @@ class HonControlSwitchEntity(HonEntity, SwitchEntity):
async def async_turn_on(self, **kwargs: Any) -> None:
self._device.sync_command(self.entity_description.turn_on_key, "settings")
await self.coordinator.async_refresh()
self.async_write_ha_state()
await self._device.commands[self.entity_description.turn_on_key].send()
self._device.attributes[self.entity_description.key] = True
self.async_write_ha_state()
async def async_turn_off(self, **kwargs: Any) -> None:
self._device.sync_command(self.entity_description.turn_off_key, "settings")
await self.coordinator.async_refresh()
self.async_write_ha_state()
await self._device.commands[self.entity_description.turn_off_key].send()
self._device.attributes[self.entity_description.key] = False
self.async_write_ha_state()
@ -541,7 +542,7 @@ class HonConfigSwitchEntity(HonEntity, SwitchEntity):
return
setting.value = setting.max if isinstance(setting, HonParameterRange) else "1"
self.async_write_ha_state()
await self.coordinator.async_refresh()
self.async_write_ha_state()
async def async_turn_off(self, **kwargs: Any) -> None:
setting = self._device.settings[self.entity_description.key]
@ -549,7 +550,7 @@ class HonConfigSwitchEntity(HonEntity, SwitchEntity):
return
setting.value = setting.min if isinstance(setting, HonParameterRange) else "0"
self.async_write_ha_state()
await self.coordinator.async_refresh()
self.async_write_ha_state()
@callback
def _handle_coordinator_update(self, update: bool = True) -> None:

28
custom_components/hon/util.py Normal file
View File

@ -0,0 +1,28 @@
import logging
from contextlib import suppress
from .typedefs import HonEntityDescription, HonOptionEntityDescription, T
_LOGGER = logging.getLogger(__name__)
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_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)
return value

2
requirements.txt
View File

@ -1 +1 @@
pyhOn==0.17.0
pyhOn==0.17.2