Use typed config entry in tplink coordinator (#135182)

This commit is contained in:
Steven B. 2025-01-09 16:43:38 +00:00 committed by GitHub
parent a5f70dec96
commit 1ca5f79708
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 42 additions and 31 deletions

View File

@ -22,7 +22,6 @@ from kasa.iot import IotStrip
from homeassistant import config_entries
from homeassistant.components import network
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_ALIAS,
CONF_AUTHENTICATION,
@ -59,10 +58,7 @@ from .const import (
DOMAIN,
PLATFORMS,
)
from .coordinator import TPLinkDataUpdateCoordinator
from .models import TPLinkData
type TPLinkConfigEntry = ConfigEntry[TPLinkData]
from .coordinator import TPLinkConfigEntry, TPLinkData, TPLinkDataUpdateCoordinator
DISCOVERY_INTERVAL = timedelta(minutes=15)
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
@ -236,7 +232,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: TPLinkConfigEntry) -> bo
},
)
parent_coordinator = TPLinkDataUpdateCoordinator(hass, device, timedelta(seconds=5))
parent_coordinator = TPLinkDataUpdateCoordinator(
hass, device, timedelta(seconds=5), entry
)
child_coordinators: list[TPLinkDataUpdateCoordinator] = []
# The iot HS300 allows a limited number of concurrent requests and fetching the
@ -245,7 +243,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: TPLinkConfigEntry) -> bo
child_coordinators = [
# The child coordinators only update energy data so we can
# set a longer update interval to avoid flooding the device
TPLinkDataUpdateCoordinator(hass, child, timedelta(seconds=60))
TPLinkDataUpdateCoordinator(hass, child, timedelta(seconds=60), entry)
for child in device.children
]

View File

@ -2,38 +2,56 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import timedelta
import logging
from kasa import AuthenticationError, Device, KasaException
from kasa import AuthenticationError, Credentials, Device, KasaException
from homeassistant import config_entries
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.debounce import Debouncer
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
@dataclass(slots=True)
class TPLinkData:
"""Data for the tplink integration."""
parent_coordinator: TPLinkDataUpdateCoordinator
children_coordinators: list[TPLinkDataUpdateCoordinator]
camera_credentials: Credentials | None
live_view: bool | None
type TPLinkConfigEntry = ConfigEntry[TPLinkData]
REQUEST_REFRESH_DELAY = 0.35
class TPLinkDataUpdateCoordinator(DataUpdateCoordinator[None]):
"""DataUpdateCoordinator to gather data for a specific TPLink device."""
config_entry: config_entries.ConfigEntry
config_entry: TPLinkConfigEntry
def __init__(
self,
hass: HomeAssistant,
device: Device,
update_interval: timedelta,
config_entry: TPLinkConfigEntry,
) -> None:
"""Initialize DataUpdateCoordinator to gather data for specific SmartPlug."""
self.device = device
super().__init__(
hass,
_LOGGER,
config_entry=config_entry,
name=device.host,
update_interval=update_interval,
# We don't want an immediate refresh since the device
@ -48,6 +66,20 @@ class TPLinkDataUpdateCoordinator(DataUpdateCoordinator[None]):
try:
await self.device.update(update_children=False)
except AuthenticationError as ex:
raise ConfigEntryAuthFailed from ex
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,
translation_key="device_authentication",
translation_placeholders={
"func": "update",
"exc": str(ex),
},
) from ex
except KasaException as ex:
raise UpdateFailed(ex) from ex
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="device_error",
translation_placeholders={
"func": "update",
"exc": str(ex),
},
) from ex

View File

@ -1,19 +0,0 @@
"""The tplink integration models."""
from __future__ import annotations
from dataclasses import dataclass
from kasa import Credentials
from .coordinator import TPLinkDataUpdateCoordinator
@dataclass(slots=True)
class TPLinkData:
"""Data for the tplink integration."""
parent_coordinator: TPLinkDataUpdateCoordinator
children_coordinators: list[TPLinkDataUpdateCoordinator]
camera_credentials: Credentials | None
live_view: bool | None