Improve CoordinatorEntity typing (#68441)

This commit is contained in:
Marc Mueller 2022-03-21 10:22:30 +01:00 committed by GitHub
parent 3320606a1b
commit 830cc278d3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
23 changed files with 75 additions and 60 deletions

View File

@ -9,13 +9,13 @@ from homeassistant.components.sensor import SensorEntity, SensorStateClass
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_ATTRIBUTION, PERCENTAGE from homeassistant.const import ATTR_ATTRIBUTION, PERCENTAGE
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers import update_coordinator
from homeassistant.helpers.device_registry import DeviceEntryType from homeassistant.helpers.device_registry import DeviceEntryType
from homeassistant.helpers.entity import DeviceInfo from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import CO2SignalCoordinator, CO2SignalResponse from . import CO2SignalCoordinator
from .const import ATTRIBUTION, DOMAIN from .const import ATTRIBUTION, DOMAIN
SCAN_INTERVAL = timedelta(minutes=3) SCAN_INTERVAL = timedelta(minutes=3)
@ -55,7 +55,7 @@ async def async_setup_entry(
async_add_entities(CO2Sensor(coordinator, description) for description in SENSORS) async_add_entities(CO2Sensor(coordinator, description) for description in SENSORS)
class CO2Sensor(update_coordinator.CoordinatorEntity[CO2SignalResponse], SensorEntity): class CO2Sensor(CoordinatorEntity[CO2SignalCoordinator], SensorEntity):
"""Implementation of the CO2Signal sensor.""" """Implementation of the CO2Signal sensor."""
_attr_state_class = SensorStateClass.MEASUREMENT _attr_state_class = SensorStateClass.MEASUREMENT

View File

@ -67,7 +67,7 @@ async def async_setup_entry(
async_setup_entry_platform(hass, config_entry, async_add_entities, _constructor) async_setup_entry_platform(hass, config_entry, async_add_entities, _constructor)
class BinarySensor(CoordinatorEntity[State], BinarySensorEntity): class BinarySensor(CoordinatorEntity[DataUpdateCoordinator[State]], BinarySensorEntity):
"""Grease filter sensor.""" """Grease filter sensor."""
entity_description: EntityDescription entity_description: EntityDescription

View File

@ -70,7 +70,7 @@ async def async_setup_entry(
async_setup_entry_platform(hass, config_entry, async_add_entities, _constructor) async_setup_entry_platform(hass, config_entry, async_add_entities, _constructor)
class Fan(CoordinatorEntity[State], FanEntity): class Fan(CoordinatorEntity[DataUpdateCoordinator[State]], FanEntity):
"""Fan entity.""" """Fan entity."""
def __init__( def __init__(

View File

@ -37,7 +37,7 @@ async def async_setup_entry(
async_setup_entry_platform(hass, config_entry, async_add_entities, _constructor) async_setup_entry_platform(hass, config_entry, async_add_entities, _constructor)
class Light(CoordinatorEntity[State], LightEntity): class Light(CoordinatorEntity[DataUpdateCoordinator[State]], LightEntity):
"""Light device.""" """Light device."""
def __init__( def __init__(

View File

@ -34,7 +34,9 @@ async def async_setup_entry(
async_setup_entry_platform(hass, config_entry, async_add_entities, _constructor) async_setup_entry_platform(hass, config_entry, async_add_entities, _constructor)
class PeriodicVentingTime(CoordinatorEntity[State], NumberEntity): class PeriodicVentingTime(
CoordinatorEntity[DataUpdateCoordinator[State]], NumberEntity
):
"""Periodic Venting.""" """Periodic Venting."""
_attr_max_value: float = 59 _attr_max_value: float = 59

View File

@ -39,7 +39,7 @@ async def async_setup_entry(
async_setup_entry_platform(hass, config_entry, async_add_entities, _constructor) async_setup_entry_platform(hass, config_entry, async_add_entities, _constructor)
class RssiSensor(CoordinatorEntity[State], SensorEntity): class RssiSensor(CoordinatorEntity[DataUpdateCoordinator[State]], SensorEntity):
"""Sensor device.""" """Sensor device."""
def __init__( def __init__(

View File

@ -172,12 +172,11 @@ async def async_setup_entry(
) )
class GitHubSensorEntity(CoordinatorEntity[dict[str, Any]], SensorEntity): class GitHubSensorEntity(CoordinatorEntity[GitHubDataUpdateCoordinator], SensorEntity):
"""Defines a GitHub sensor entity.""" """Defines a GitHub sensor entity."""
_attr_attribution = "Data provided by the GitHub API" _attr_attribution = "Data provided by the GitHub API"
coordinator: GitHubDataUpdateCoordinator
entity_description: GitHubSensorEntityDescription entity_description: GitHubSensorEntityDescription
def __init__( def __init__(

View File

@ -11,7 +11,7 @@ from . import DOMAIN, HassioDataUpdateCoordinator
from .const import ATTR_SLUG, DATA_KEY_ADDONS, DATA_KEY_OS from .const import ATTR_SLUG, DATA_KEY_ADDONS, DATA_KEY_OS
class HassioAddonEntity(CoordinatorEntity): class HassioAddonEntity(CoordinatorEntity[HassioDataUpdateCoordinator]):
"""Base entity for a Hass.io add-on.""" """Base entity for a Hass.io add-on."""
def __init__( def __init__(
@ -38,7 +38,7 @@ class HassioAddonEntity(CoordinatorEntity):
) )
class HassioOSEntity(CoordinatorEntity): class HassioOSEntity(CoordinatorEntity[HassioDataUpdateCoordinator]):
"""Base Entity for Hass.io OS.""" """Base Entity for Hass.io OS."""
def __init__( def __init__(

View File

@ -142,7 +142,7 @@ async def async_setup_entry(
async_add_entities(entities) async_add_entities(entities)
class HWEnergySensor(CoordinatorEntity[DeviceResponseEntry], SensorEntity): class HWEnergySensor(CoordinatorEntity[HWEnergyDeviceUpdateCoordinator], SensorEntity):
"""Representation of a HomeWizard Sensor.""" """Representation of a HomeWizard Sensor."""
def __init__( def __init__(

View File

@ -31,11 +31,11 @@ async def async_setup_entry(
) )
class HWEnergySwitchEntity(CoordinatorEntity, SwitchEntity): class HWEnergySwitchEntity(
CoordinatorEntity[HWEnergyDeviceUpdateCoordinator], SwitchEntity
):
"""Representation switchable entity.""" """Representation switchable entity."""
coordinator: HWEnergyDeviceUpdateCoordinator
def __init__( def __init__(
self, self,
coordinator: HWEnergyDeviceUpdateCoordinator, coordinator: HWEnergyDeviceUpdateCoordinator,

View File

@ -11,7 +11,10 @@ from homeassistant.helpers import device_registry
from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import DeviceInfo from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from . import KrakenData from . import KrakenData
from .const import ( from .const import (
@ -86,7 +89,9 @@ async def async_setup_entry(
) )
class KrakenSensor(CoordinatorEntity[Optional[KrakenResponse]], SensorEntity): class KrakenSensor(
CoordinatorEntity[DataUpdateCoordinator[Optional[KrakenResponse]]], SensorEntity
):
"""Define a Kraken sensor.""" """Define a Kraken sensor."""
entity_description: KrakenSensorEntityDescription entity_description: KrakenSensorEntityDescription

View File

@ -45,7 +45,7 @@ from homeassistant.util.distance import convert as convert_distance
from homeassistant.util.pressure import convert as convert_pressure from homeassistant.util.pressure import convert as convert_pressure
from homeassistant.util.speed import convert as convert_speed from homeassistant.util.speed import convert as convert_speed
from . import MetDataUpdateCoordinator, MetWeatherData from . import MetDataUpdateCoordinator
from .const import ( from .const import (
ATTR_FORECAST_PRECIPITATION, ATTR_FORECAST_PRECIPITATION,
ATTR_MAP, ATTR_MAP,
@ -127,11 +127,9 @@ def format_condition(condition: str) -> str:
return condition return condition
class MetWeather(CoordinatorEntity[MetWeatherData], WeatherEntity): class MetWeather(CoordinatorEntity[MetDataUpdateCoordinator], WeatherEntity):
"""Implementation of a Met.no weather condition.""" """Implementation of a Met.no weather condition."""
coordinator: MetDataUpdateCoordinator
def __init__( def __init__(
self, self,
coordinator: MetDataUpdateCoordinator, coordinator: MetDataUpdateCoordinator,

View File

@ -126,8 +126,6 @@ class ModernFormsDataUpdateCoordinator(DataUpdateCoordinator[ModernFormsDeviceSt
class ModernFormsDeviceEntity(CoordinatorEntity[ModernFormsDataUpdateCoordinator]): class ModernFormsDeviceEntity(CoordinatorEntity[ModernFormsDataUpdateCoordinator]):
"""Defines a Modern Forms device entity.""" """Defines a Modern Forms device entity."""
coordinator: ModernFormsDataUpdateCoordinator
def __init__( def __init__(
self, self,
*, *,

View File

@ -12,14 +12,12 @@ from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN from .const import DOMAIN
from .coordinator import PlugwiseData, PlugwiseDataUpdateCoordinator from .coordinator import PlugwiseDataUpdateCoordinator
class PlugwiseEntity(CoordinatorEntity[PlugwiseData]): class PlugwiseEntity(CoordinatorEntity[PlugwiseDataUpdateCoordinator]):
"""Represent a PlugWise Entity.""" """Represent a PlugWise Entity."""
coordinator: PlugwiseDataUpdateCoordinator
def __init__( def __init__(
self, self,
coordinator: PlugwiseDataUpdateCoordinator, coordinator: PlugwiseDataUpdateCoordinator,

View File

@ -1,7 +1,10 @@
"""The Tesla Powerwall integration base entity.""" """The Tesla Powerwall integration base entity."""
from homeassistant.helpers.entity import DeviceInfo from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from .const import ( from .const import (
DOMAIN, DOMAIN,
@ -13,7 +16,7 @@ from .const import (
from .models import PowerwallData, PowerwallRuntimeData from .models import PowerwallData, PowerwallRuntimeData
class PowerWallEntity(CoordinatorEntity[PowerwallData]): class PowerWallEntity(CoordinatorEntity[DataUpdateCoordinator[PowerwallData]]):
"""Base class for powerwall entities.""" """Base class for powerwall entities."""
def __init__(self, powerwall_data: PowerwallRuntimeData) -> None: def __init__(self, powerwall_data: PowerwallRuntimeData) -> None:

View File

@ -78,10 +78,11 @@ async def async_setup_entry(
) )
class PureEnergieSensorEntity(CoordinatorEntity[PureEnergieData], SensorEntity): class PureEnergieSensorEntity(
CoordinatorEntity[PureEnergieDataUpdateCoordinator], SensorEntity
):
"""Defines an Pure Energie sensor.""" """Defines an Pure Energie sensor."""
coordinator: PureEnergieDataUpdateCoordinator
entity_description: PureEnergieSensorEntityDescription entity_description: PureEnergieSensorEntityDescription
def __init__( def __init__(

View File

@ -4,7 +4,7 @@ from __future__ import annotations
from collections.abc import Awaitable, Callable from collections.abc import Awaitable, Callable
from datetime import timedelta from datetime import timedelta
import logging import logging
from typing import TypeVar from typing import Optional, TypeVar
from renault_api.kamereon.exceptions import ( from renault_api.kamereon.exceptions import (
AccessDeniedException, AccessDeniedException,
@ -16,7 +16,7 @@ from renault_api.kamereon.models import KamereonVehicleDataAttributes
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
T = TypeVar("T", bound=KamereonVehicleDataAttributes) T = TypeVar("T", bound=Optional[KamereonVehicleDataAttributes])
class RenaultDataUpdateCoordinator(DataUpdateCoordinator[T]): class RenaultDataUpdateCoordinator(DataUpdateCoordinator[T]):

View File

@ -2,14 +2,14 @@
from __future__ import annotations from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
from typing import Optional, cast from typing import cast
from homeassistant.const import ATTR_NAME from homeassistant.const import ATTR_NAME
from homeassistant.helpers.entity import Entity, EntityDescription from homeassistant.helpers.entity import Entity, EntityDescription
from homeassistant.helpers.typing import StateType from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .renault_coordinator import T from .renault_coordinator import RenaultDataUpdateCoordinator, T
from .renault_vehicle import RenaultVehicleProxy from .renault_vehicle import RenaultVehicleProxy
@ -50,7 +50,9 @@ class RenaultEntity(Entity):
return f"{self.vehicle.device_info[ATTR_NAME]} {self.entity_description.name}" return f"{self.vehicle.device_info[ATTR_NAME]} {self.entity_description.name}"
class RenaultDataEntity(CoordinatorEntity[Optional[T]], RenaultEntity): class RenaultDataEntity(
CoordinatorEntity[RenaultDataUpdateCoordinator[T]], RenaultEntity
):
"""Implementation of a Renault entity with a data coordinator.""" """Implementation of a Renault entity with a data coordinator."""
def __init__( def __init__(
@ -65,5 +67,5 @@ class RenaultDataEntity(CoordinatorEntity[Optional[T]], RenaultEntity):
def _get_data_attr(self, key: str) -> StateType: def _get_data_attr(self, key: str) -> StateType:
"""Return the attribute value from the coordinator data.""" """Return the attribute value from the coordinator data."""
if self.coordinator.data is None: if self.coordinator.data is None:
return None return None # type: ignore[unreachable]
return cast(StateType, getattr(self.coordinator.data, key)) return cast(StateType, getattr(self.coordinator.data, key))

View File

@ -4,7 +4,7 @@ from __future__ import annotations
from collections.abc import Callable from collections.abc import Callable
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime from datetime import datetime
from typing import TYPE_CHECKING, cast from typing import TYPE_CHECKING, Any, Generic, cast
from renault_api.kamereon.enums import ChargeState, PlugState from renault_api.kamereon.enums import ChargeState, PlugState
from renault_api.kamereon.models import ( from renault_api.kamereon.models import (
@ -45,18 +45,18 @@ from .renault_vehicle import RenaultVehicleProxy
@dataclass @dataclass
class RenaultSensorRequiredKeysMixin: class RenaultSensorRequiredKeysMixin(Generic[T]):
"""Mixin for required keys.""" """Mixin for required keys."""
data_key: str data_key: str
entity_class: type[RenaultSensor] entity_class: type[RenaultSensor[T]]
@dataclass @dataclass
class RenaultSensorEntityDescription( class RenaultSensorEntityDescription(
SensorEntityDescription, SensorEntityDescription,
RenaultDataEntityDescription, RenaultDataEntityDescription,
RenaultSensorRequiredKeysMixin, RenaultSensorRequiredKeysMixin[T],
): ):
"""Class describing Renault sensor entities.""" """Class describing Renault sensor entities."""
@ -73,7 +73,7 @@ async def async_setup_entry(
) -> None: ) -> None:
"""Set up the Renault entities from config entry.""" """Set up the Renault entities from config entry."""
proxy: RenaultHub = hass.data[DOMAIN][config_entry.entry_id] proxy: RenaultHub = hass.data[DOMAIN][config_entry.entry_id]
entities: list[RenaultSensor] = [ entities: list[RenaultSensor[Any]] = [
description.entity_class(vehicle, description) description.entity_class(vehicle, description)
for vehicle in proxy.vehicles.values() for vehicle in proxy.vehicles.values()
for description in SENSOR_TYPES for description in SENSOR_TYPES
@ -87,7 +87,7 @@ async def async_setup_entry(
class RenaultSensor(RenaultDataEntity[T], SensorEntity): class RenaultSensor(RenaultDataEntity[T], SensorEntity):
"""Mixin for sensor specific attributes.""" """Mixin for sensor specific attributes."""
entity_description: RenaultSensorEntityDescription entity_description: RenaultSensorEntityDescription[T]
@property @property
def data(self) -> StateType: def data(self) -> StateType:
@ -157,7 +157,7 @@ def _get_utc_value(entity: RenaultSensor[T]) -> datetime:
return as_utc(original_dt) return as_utc(original_dt)
SENSOR_TYPES: tuple[RenaultSensorEntityDescription, ...] = ( SENSOR_TYPES: tuple[RenaultSensorEntityDescription[Any], ...] = (
RenaultSensorEntityDescription( RenaultSensorEntityDescription(
key="battery_level", key="battery_level",
coordinator="battery", coordinator="battery",

View File

@ -13,13 +13,14 @@ from homeassistant.const import (
) )
from homeassistant.core import HomeAssistant, callback from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import TemplateError from homeassistant.exceptions import TemplateError
from homeassistant.helpers import template, update_coordinator from homeassistant.helpers import template
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import TriggerUpdateCoordinator from . import TriggerUpdateCoordinator
from .const import CONF_ATTRIBUTES, CONF_AVAILABILITY, CONF_PICTURE from .const import CONF_ATTRIBUTES, CONF_AVAILABILITY, CONF_PICTURE
class TriggerEntity(update_coordinator.CoordinatorEntity): class TriggerEntity(CoordinatorEntity[TriggerUpdateCoordinator]):
"""Template entity based on trigger data.""" """Template entity based on trigger data."""
domain: str domain: str

View File

@ -32,11 +32,14 @@ from homeassistant.const import (
) )
from homeassistant.core import HomeAssistant, callback from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import PlatformNotReady from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers import update_coordinator
from homeassistant.helpers.device_registry import async_get as async_get_dev_reg from homeassistant.helpers.device_registry import async_get as async_get_dev_reg
from homeassistant.helpers.entity import DeviceInfo, EntityCategory from homeassistant.helpers.entity import DeviceInfo, EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.entity_registry import async_get as async_get_entity_reg from homeassistant.helpers.entity_registry import async_get as async_get_entity_reg
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from homeassistant.util import Throttle, dt as dt_util from homeassistant.util import Throttle, dt as dt_util
from .const import DOMAIN as TIBBER_DOMAIN, MANUFACTURER from .const import DOMAIN as TIBBER_DOMAIN, MANUFACTURER
@ -239,7 +242,7 @@ async def async_setup_entry(
entity_registry = async_get_entity_reg(hass) entity_registry = async_get_entity_reg(hass)
device_registry = async_get_dev_reg(hass) device_registry = async_get_dev_reg(hass)
coordinator: update_coordinator.DataUpdateCoordinator | None = None coordinator: TibberDataCoordinator | None = None
entities: list[TibberSensor] = [] entities: list[TibberSensor] = []
for home in tibber_connection.get_homes(only_active=False): for home in tibber_connection.get_homes(only_active=False):
try: try:
@ -392,13 +395,13 @@ class TibberSensorElPrice(TibberSensor):
]["estimatedAnnualConsumption"] ]["estimatedAnnualConsumption"]
class TibberDataSensor(TibberSensor, update_coordinator.CoordinatorEntity): class TibberDataSensor(TibberSensor, CoordinatorEntity["TibberDataCoordinator"]):
"""Representation of a Tibber sensor.""" """Representation of a Tibber sensor."""
def __init__( def __init__(
self, self,
tibber_home, tibber_home,
coordinator: update_coordinator.DataUpdateCoordinator, coordinator: TibberDataCoordinator,
entity_description: SensorEntityDescription, entity_description: SensorEntityDescription,
): ):
"""Initialize the sensor.""" """Initialize the sensor."""
@ -420,7 +423,7 @@ class TibberDataSensor(TibberSensor, update_coordinator.CoordinatorEntity):
return getattr(self._tibber_home, self.entity_description.key) return getattr(self._tibber_home, self.entity_description.key)
class TibberSensorRT(TibberSensor, update_coordinator.CoordinatorEntity): class TibberSensorRT(TibberSensor, CoordinatorEntity["TibberRtDataCoordinator"]):
"""Representation of a Tibber sensor for real time consumption.""" """Representation of a Tibber sensor for real time consumption."""
def __init__( def __init__(
@ -450,7 +453,7 @@ class TibberSensorRT(TibberSensor, update_coordinator.CoordinatorEntity):
@callback @callback
def _handle_coordinator_update(self) -> None: def _handle_coordinator_update(self) -> None:
if not (live_measurement := self.coordinator.get_live_measurement()): # type: ignore[attr-defined] if not (live_measurement := self.coordinator.get_live_measurement()):
return return
state = live_measurement.get(self.entity_description.key) state = live_measurement.get(self.entity_description.key)
if state is None: if state is None:
@ -479,7 +482,7 @@ class TibberSensorRT(TibberSensor, update_coordinator.CoordinatorEntity):
self.async_write_ha_state() self.async_write_ha_state()
class TibberRtDataCoordinator(update_coordinator.DataUpdateCoordinator): class TibberRtDataCoordinator(DataUpdateCoordinator):
"""Handle Tibber realtime data.""" """Handle Tibber realtime data."""
def __init__(self, async_add_entities, tibber_home, hass): def __init__(self, async_add_entities, tibber_home, hass):
@ -538,7 +541,7 @@ class TibberRtDataCoordinator(update_coordinator.DataUpdateCoordinator):
return self.data.get("data", {}).get("liveMeasurement") return self.data.get("data", {}).get("liveMeasurement")
class TibberDataCoordinator(update_coordinator.DataUpdateCoordinator): class TibberDataCoordinator(DataUpdateCoordinator):
"""Handle Tibber data and insert statistics.""" """Handle Tibber data and insert statistics."""
def __init__(self, hass, tibber_connection): def __init__(self, hass, tibber_connection):

View File

@ -17,12 +17,14 @@ from homeassistant.helpers.update_coordinator import (
from .const import DOMAIN from .const import DOMAIN
class TwenteMilieuEntity(CoordinatorEntity[dict[WasteType, list[date]]], Entity): class TwenteMilieuEntity(
CoordinatorEntity[DataUpdateCoordinator[dict[WasteType, list[date]]]], Entity
):
"""Defines a Twente Milieu entity.""" """Defines a Twente Milieu entity."""
def __init__( def __init__(
self, self,
coordinator: DataUpdateCoordinator, coordinator: DataUpdateCoordinator[dict[WasteType, list[date]]],
entry: ConfigEntry, entry: ConfigEntry,
) -> None: ) -> None:
"""Initialize the Twente Milieu entity.""" """Initialize the Twente Milieu entity."""

View File

@ -6,7 +6,7 @@ from collections.abc import Awaitable, Callable
from datetime import datetime, timedelta from datetime import datetime, timedelta
import logging import logging
from time import monotonic from time import monotonic
from typing import Generic, TypeVar from typing import Any, Generic, TypeVar # pylint: disable=unused-import
import urllib.error import urllib.error
import aiohttp import aiohttp
@ -24,6 +24,9 @@ REQUEST_REFRESH_DEFAULT_COOLDOWN = 10
REQUEST_REFRESH_DEFAULT_IMMEDIATE = True REQUEST_REFRESH_DEFAULT_IMMEDIATE = True
_T = TypeVar("_T") _T = TypeVar("_T")
_DataUpdateCoordinatorT = TypeVar(
"_DataUpdateCoordinatorT", bound="DataUpdateCoordinator[Any]"
)
class UpdateFailed(Exception): class UpdateFailed(Exception):
@ -295,10 +298,10 @@ class DataUpdateCoordinator(Generic[_T]):
self._unsub_refresh = None self._unsub_refresh = None
class CoordinatorEntity(Generic[_T], entity.Entity): class CoordinatorEntity(entity.Entity, Generic[_DataUpdateCoordinatorT]):
"""A class for entities using DataUpdateCoordinator.""" """A class for entities using DataUpdateCoordinator."""
def __init__(self, coordinator: DataUpdateCoordinator[_T]) -> None: def __init__(self, coordinator: _DataUpdateCoordinatorT) -> None:
"""Create the entity with a DataUpdateCoordinator.""" """Create the entity with a DataUpdateCoordinator."""
self.coordinator = coordinator self.coordinator = coordinator