Improve `lookin` generic typing (#84636)

pull/84639/head
Marc Mueller 2022-12-27 20:37:47 +01:00 committed by GitHub
parent acd31d4ae3
commit 653805584b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 38 additions and 25 deletions

View File

@ -102,7 +102,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
push_coordinator = LookinPushCoordinator(entry.title) push_coordinator = LookinPushCoordinator(entry.title)
if lookin_device.model >= 2: if lookin_device.model >= 2:
meteo_coordinator: LookinDataUpdateCoordinator = LookinDataUpdateCoordinator( meteo_coordinator = LookinDataUpdateCoordinator[MeteoSensor](
hass, hass,
push_coordinator, push_coordinator,
name=entry.title, name=entry.title,
@ -113,7 +113,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
) )
await meteo_coordinator.async_config_entry_first_refresh() await meteo_coordinator.async_config_entry_first_refresh()
device_coordinators: dict[str, LookinDataUpdateCoordinator] = {} device_coordinators: dict[str, LookinDataUpdateCoordinator[Remote]] = {}
for remote in devices: for remote in devices:
if (platform := TYPE_TO_PLATFORM.get(remote["Type"])) is None: if (platform := TYPE_TO_PLATFORM.get(remote["Type"])) is None:
continue continue

View File

@ -4,7 +4,7 @@ from __future__ import annotations
import logging import logging
from typing import Any, Final, cast from typing import Any, Final, cast
from aiolookin import Climate, MeteoSensor from aiolookin import Climate, MeteoSensor, Remote
from aiolookin.models import UDPCommandType, UDPEvent from aiolookin.models import UDPCommandType, UDPEvent
from homeassistant.components.climate import ( from homeassistant.components.climate import (
@ -75,7 +75,7 @@ async def async_setup_entry(
continue continue
uuid = remote["UUID"] uuid = remote["UUID"]
coordinator = lookin_data.device_coordinators[uuid] coordinator = lookin_data.device_coordinators[uuid]
device: Climate = coordinator.data device = cast(Climate, coordinator.data)
entities.append( entities.append(
ConditionerEntity( ConditionerEntity(
uuid=uuid, uuid=uuid,
@ -110,7 +110,7 @@ class ConditionerEntity(LookinCoordinatorEntity, ClimateEntity):
uuid: str, uuid: str,
device: Climate, device: Climate,
lookin_data: LookinData, lookin_data: LookinData,
coordinator: LookinDataUpdateCoordinator, coordinator: LookinDataUpdateCoordinator[Remote],
) -> None: ) -> None:
"""Init the ConditionerEntity.""" """Init the ConditionerEntity."""
super().__init__(coordinator, uuid, device, lookin_data) super().__init__(coordinator, uuid, device, lookin_data)

View File

@ -5,12 +5,13 @@ from collections.abc import Awaitable, Callable
from datetime import timedelta from datetime import timedelta
import logging import logging
import time import time
from typing import cast from typing import TypeVar
from homeassistant.core import HomeAssistant, callback from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
_DataT = TypeVar("_DataT")
NEVER_TIME = -120.0 # Time that will never match time.monotonic() NEVER_TIME = -120.0 # Time that will never match time.monotonic()
ACTIVE_UPDATES_INTERVAL = 3 # Consider active for 3x the update interval ACTIVE_UPDATES_INTERVAL = 3 # Consider active for 3x the update interval
@ -43,7 +44,7 @@ class LookinPushCoordinator:
return is_active return is_active
class LookinDataUpdateCoordinator(DataUpdateCoordinator): class LookinDataUpdateCoordinator(DataUpdateCoordinator[_DataT]):
"""DataUpdateCoordinator to gather data for a specific lookin devices.""" """DataUpdateCoordinator to gather data for a specific lookin devices."""
def __init__( def __init__(
@ -52,7 +53,7 @@ class LookinDataUpdateCoordinator(DataUpdateCoordinator):
push_coordinator: LookinPushCoordinator, push_coordinator: LookinPushCoordinator,
name: str, name: str,
update_interval: timedelta | None = None, update_interval: timedelta | None = None,
update_method: Callable[[], Awaitable[dict]] | None = None, update_method: Callable[[], Awaitable[_DataT]] | None = None,
) -> None: ) -> None:
"""Initialize DataUpdateCoordinator to gather data for specific device.""" """Initialize DataUpdateCoordinator to gather data for specific device."""
self.push_coordinator = push_coordinator self.push_coordinator = push_coordinator
@ -65,12 +66,12 @@ class LookinDataUpdateCoordinator(DataUpdateCoordinator):
) )
@callback @callback
def async_set_updated_data(self, data: dict) -> None: def async_set_updated_data(self, data: _DataT) -> None:
"""Manually update data, notify listeners and reset refresh interval, and remember.""" """Manually update data, notify listeners and reset refresh interval, and remember."""
self.push_coordinator.update() self.push_coordinator.update()
super().async_set_updated_data(data) super().async_set_updated_data(data)
async def _async_update_data(self) -> dict: async def _async_update_data(self) -> _DataT:
"""Fetch data only if we have not received a push inside the interval.""" """Fetch data only if we have not received a push inside the interval."""
interval = self.update_interval interval = self.update_interval
if ( if (
@ -82,4 +83,4 @@ class LookinDataUpdateCoordinator(DataUpdateCoordinator):
data = self.data data = self.data
else: else:
data = await super()._async_update_data() data = await super()._async_update_data()
return cast(dict, data) return data

View File

@ -3,9 +3,15 @@ from __future__ import annotations
from abc import abstractmethod from abc import abstractmethod
import logging import logging
from typing import cast
from aiolookin import POWER_CMD, POWER_OFF_CMD, POWER_ON_CMD, Climate, Remote from aiolookin import (
POWER_CMD,
POWER_OFF_CMD,
POWER_ON_CMD,
Climate,
MeteoSensor,
Remote,
)
from aiolookin.models import Device, UDPCommandType, UDPEvent from aiolookin.models import Device, UDPCommandType, UDPEvent
from homeassistant.helpers.entity import DeviceInfo from homeassistant.helpers.entity import DeviceInfo
@ -53,7 +59,7 @@ class LookinDeviceMixIn:
class LookinDeviceCoordinatorEntity( class LookinDeviceCoordinatorEntity(
LookinDeviceMixIn, CoordinatorEntity[LookinDataUpdateCoordinator] LookinDeviceMixIn, CoordinatorEntity[LookinDataUpdateCoordinator[MeteoSensor]]
): ):
"""A lookin device entity on the device itself that uses the coordinator.""" """A lookin device entity on the device itself that uses the coordinator."""
@ -86,7 +92,9 @@ class LookinEntityMixIn:
class LookinCoordinatorEntity( class LookinCoordinatorEntity(
LookinDeviceMixIn, LookinEntityMixIn, CoordinatorEntity[LookinDataUpdateCoordinator] LookinDeviceMixIn,
LookinEntityMixIn,
CoordinatorEntity[LookinDataUpdateCoordinator[Remote]],
): ):
"""A lookin device entity for an external device that uses the coordinator.""" """A lookin device entity for an external device that uses the coordinator."""
@ -95,7 +103,7 @@ class LookinCoordinatorEntity(
def __init__( def __init__(
self, self,
coordinator: LookinDataUpdateCoordinator, coordinator: LookinDataUpdateCoordinator[Remote],
uuid: str, uuid: str,
device: Remote | Climate, device: Remote | Climate,
lookin_data: LookinData, lookin_data: LookinData,
@ -122,7 +130,7 @@ class LookinPowerEntity(LookinCoordinatorEntity):
def __init__( def __init__(
self, self,
coordinator: LookinDataUpdateCoordinator, coordinator: LookinDataUpdateCoordinator[Remote],
uuid: str, uuid: str,
device: Remote | Climate, device: Remote | Climate,
lookin_data: LookinData, lookin_data: LookinData,
@ -142,7 +150,7 @@ class LookinPowerPushRemoteEntity(LookinPowerEntity):
def __init__( def __init__(
self, self,
coordinator: LookinDataUpdateCoordinator, coordinator: LookinDataUpdateCoordinator[Remote],
uuid: str, uuid: str,
device: Remote, device: Remote,
lookin_data: LookinData, lookin_data: LookinData,
@ -154,7 +162,7 @@ class LookinPowerPushRemoteEntity(LookinPowerEntity):
@property @property
def _remote(self) -> Remote: def _remote(self) -> Remote:
return cast(Remote, self.coordinator.data) return self.coordinator.data
@abstractmethod @abstractmethod
def _update_from_status(self, status: str) -> None: def _update_from_status(self, status: str) -> None:

View File

@ -4,8 +4,6 @@ from __future__ import annotations
import logging import logging
from typing import Any from typing import Any
from aiolookin import Remote
from homeassistant.components.light import ColorMode, LightEntity from homeassistant.components.light import ColorMode, LightEntity
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform from homeassistant.const import Platform
@ -33,7 +31,7 @@ async def async_setup_entry(
continue continue
uuid = remote["UUID"] uuid = remote["UUID"]
coordinator = lookin_data.device_coordinators[uuid] coordinator = lookin_data.device_coordinators[uuid]
device: Remote = coordinator.data device = coordinator.data
entities.append( entities.append(
LookinLightEntity( LookinLightEntity(
coordinator=coordinator, coordinator=coordinator,

View File

@ -4,7 +4,13 @@ from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any from typing import Any
from aiolookin import Device, LookInHttpProtocol, LookinUDPSubscriptions from aiolookin import (
Device,
LookInHttpProtocol,
LookinUDPSubscriptions,
MeteoSensor,
Remote,
)
from .coordinator import LookinDataUpdateCoordinator from .coordinator import LookinDataUpdateCoordinator
@ -16,7 +22,7 @@ class LookinData:
host: str host: str
lookin_udp_subs: LookinUDPSubscriptions lookin_udp_subs: LookinUDPSubscriptions
lookin_device: Device lookin_device: Device
meteo_coordinator: LookinDataUpdateCoordinator | None meteo_coordinator: LookinDataUpdateCoordinator[MeteoSensor] | None
devices: list[dict[str, Any]] devices: list[dict[str, Any]]
lookin_protocol: LookInHttpProtocol lookin_protocol: LookInHttpProtocol
device_coordinators: dict[str, LookinDataUpdateCoordinator] device_coordinators: dict[str, LookinDataUpdateCoordinator[Remote]]