Use PEP 695 misc (2) (#117814)

This commit is contained in:
Marc Mueller 2024-05-21 09:45:57 +02:00 committed by GitHub
parent c1b4c977e9
commit d44f949b19
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 52 additions and 65 deletions

View File

@ -2,8 +2,6 @@
from __future__ import annotations
from typing import Generic, TypeVar
from pydeconz.models.deconz_device import DeconzDevice as PydeconzDevice
from pydeconz.models.group import Group as PydeconzGroup
from pydeconz.models.light import LightBase as PydeconzLightBase
@ -19,13 +17,12 @@ from .const import DOMAIN as DECONZ_DOMAIN
from .hub import DeconzHub
from .util import serial_from_unique_id
_DeviceT = TypeVar(
"_DeviceT",
bound=PydeconzGroup | PydeconzLightBase | PydeconzSensorBase | PydeconzScene,
type _DeviceType = (
PydeconzGroup | PydeconzLightBase | PydeconzSensorBase | PydeconzScene
)
class DeconzBase(Generic[_DeviceT]):
class DeconzBase[_DeviceT: _DeviceType]:
"""Common base for deconz entities and events."""
unique_id_suffix: str | None = None
@ -71,7 +68,7 @@ class DeconzBase(Generic[_DeviceT]):
)
class DeconzDevice(DeconzBase[_DeviceT], Entity):
class DeconzDevice[_DeviceT: _DeviceType](DeconzBase[_DeviceT], Entity):
"""Representation of a deCONZ device."""
_attr_should_poll = False

View File

@ -2,8 +2,6 @@
from __future__ import annotations
from typing import TypeVar
from devolo_plc_api.device_api import (
ConnectedStationInfo,
NeighborAPInfo,
@ -21,16 +19,13 @@ from homeassistant.helpers.update_coordinator import (
from . import DevoloHomeNetworkConfigEntry
from .const import DOMAIN
_DataT = TypeVar(
"_DataT",
bound=(
LogicalNetwork
| DataRate
| list[ConnectedStationInfo]
| list[NeighborAPInfo]
| WifiGuestAccessGet
| bool
),
type _DataType = (
LogicalNetwork
| DataRate
| list[ConnectedStationInfo]
| list[NeighborAPInfo]
| WifiGuestAccessGet
| bool
)
@ -62,7 +57,7 @@ class DevoloEntity(Entity):
)
class DevoloCoordinatorEntity(
class DevoloCoordinatorEntity[_DataT: _DataType](
CoordinatorEntity[DataUpdateCoordinator[_DataT]], DevoloEntity
):
"""Representation of a coordinated devolo home network device."""

View File

@ -1,7 +1,6 @@
"""Entity for the SleepIQ integration."""
from abc import abstractmethod
from typing import TypeVar
from asyncsleepiq import SleepIQBed, SleepIQSleeper
@ -14,10 +13,7 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import ENTITY_TYPES, ICON_OCCUPIED
from .coordinator import SleepIQDataUpdateCoordinator, SleepIQPauseUpdateCoordinator
_SleepIQCoordinatorT = TypeVar(
"_SleepIQCoordinatorT",
bound=SleepIQDataUpdateCoordinator | SleepIQPauseUpdateCoordinator,
)
type _DataCoordinatorType = SleepIQDataUpdateCoordinator | SleepIQPauseUpdateCoordinator
def device_from_bed(bed: SleepIQBed) -> DeviceInfo:
@ -47,7 +43,9 @@ class SleepIQEntity(Entity):
self._attr_device_info = device_from_bed(bed)
class SleepIQBedEntity(CoordinatorEntity[_SleepIQCoordinatorT]):
class SleepIQBedEntity[_SleepIQCoordinatorT: _DataCoordinatorType](
CoordinatorEntity[_SleepIQCoordinatorT]
):
"""Implementation of a SleepIQ sensor."""
_attr_icon = ICON_OCCUPIED
@ -75,7 +73,9 @@ class SleepIQBedEntity(CoordinatorEntity[_SleepIQCoordinatorT]):
"""Update sensor attributes."""
class SleepIQSleeperEntity(SleepIQBedEntity[_SleepIQCoordinatorT]):
class SleepIQSleeperEntity[_SleepIQCoordinatorT: _DataCoordinatorType](
SleepIQBedEntity[_SleepIQCoordinatorT]
):
"""Implementation of a SleepIQ sensor."""
_attr_icon = ICON_OCCUPIED

View File

@ -1,7 +1,7 @@
"""Support for SwitchBee entity."""
import logging
from typing import Generic, TypeVar, cast
from typing import cast
from switchbee import SWITCHBEE_BRAND
from switchbee.device import DeviceType, SwitchBeeBaseDevice
@ -12,13 +12,12 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import SwitchBeeCoordinator
_DeviceTypeT = TypeVar("_DeviceTypeT", bound=SwitchBeeBaseDevice)
_LOGGER = logging.getLogger(__name__)
class SwitchBeeEntity(CoordinatorEntity[SwitchBeeCoordinator], Generic[_DeviceTypeT]):
class SwitchBeeEntity[_DeviceTypeT: SwitchBeeBaseDevice](
CoordinatorEntity[SwitchBeeCoordinator]
):
"""Representation of a Switchbee entity."""
_attr_has_entity_name = True
@ -35,7 +34,9 @@ class SwitchBeeEntity(CoordinatorEntity[SwitchBeeCoordinator], Generic[_DeviceTy
self._attr_unique_id = f"{coordinator.unique_id}-{device.id}"
class SwitchBeeDeviceEntity(SwitchBeeEntity[_DeviceTypeT]):
class SwitchBeeDeviceEntity[_DeviceTypeT: SwitchBeeBaseDevice](
SwitchBeeEntity[_DeviceTypeT]
):
"""Representation of a Switchbee device entity."""
def __init__(

View File

@ -3,12 +3,10 @@
from __future__ import annotations
from collections.abc import Callable
from typing import Any, TypeVar
_TypeT = TypeVar("_TypeT", bound=type[Any])
from typing import Any
class DictRegistry(dict[int | str, _TypeT]):
class DictRegistry[_TypeT: type[Any]](dict[int | str, _TypeT]):
"""Dict Registry of items."""
def register(self, name: int | str) -> Callable[[_TypeT], _TypeT]:
@ -22,7 +20,9 @@ class DictRegistry(dict[int | str, _TypeT]):
return decorator
class NestedDictRegistry(dict[int | str, dict[int | str | None, _TypeT]]):
class NestedDictRegistry[_TypeT: type[Any]](
dict[int | str, dict[int | str | None, _TypeT]]
):
"""Dict Registry of multiple items per key."""
def register(
@ -43,7 +43,9 @@ class NestedDictRegistry(dict[int | str, dict[int | str | None, _TypeT]]):
class SetRegistry(set[int | str]):
"""Set Registry of items."""
def register(self, name: int | str) -> Callable[[_TypeT], _TypeT]:
def register[_TypeT: type[Any]](
self, name: int | str
) -> Callable[[_TypeT], _TypeT]:
"""Return decorator to register item with a specific name."""
def decorator(cluster_handler: _TypeT) -> _TypeT:

View File

@ -6,7 +6,7 @@ import collections
from collections.abc import Callable
import dataclasses
from operator import attrgetter
from typing import TYPE_CHECKING, TypeVar
from typing import TYPE_CHECKING
import attr
from zigpy import zcl
@ -23,9 +23,6 @@ if TYPE_CHECKING:
from .cluster_handlers import ClientClusterHandler, ClusterHandler
_ZhaEntityT = TypeVar("_ZhaEntityT", bound=type["ZhaEntity"])
_ZhaGroupEntityT = TypeVar("_ZhaGroupEntityT", bound=type["ZhaGroupEntity"])
GROUP_ENTITY_DOMAINS = [Platform.LIGHT, Platform.SWITCH, Platform.FAN]
IKEA_AIR_PURIFIER_CLUSTER = 0xFC7D
@ -387,7 +384,7 @@ class ZHAEntityRegistry:
"""Match a ZHA group to a ZHA Entity class."""
return self._group_registry.get(component)
def strict_match(
def strict_match[_ZhaEntityT: type[ZhaEntity]](
self,
component: Platform,
cluster_handler_names: set[str] | str | None = None,
@ -418,7 +415,7 @@ class ZHAEntityRegistry:
return decorator
def multipass_match(
def multipass_match[_ZhaEntityT: type[ZhaEntity]](
self,
component: Platform,
cluster_handler_names: set[str] | str | None = None,
@ -453,7 +450,7 @@ class ZHAEntityRegistry:
return decorator
def config_diagnostic_match(
def config_diagnostic_match[_ZhaEntityT: type[ZhaEntity]](
self,
component: Platform,
cluster_handler_names: set[str] | str | None = None,
@ -488,7 +485,7 @@ class ZHAEntityRegistry:
return decorator
def group_match(
def group_match[_ZhaGroupEntityT: type[ZhaGroupEntity]](
self, component: Platform
) -> Callable[[_ZhaGroupEntityT], _ZhaGroupEntityT]:
"""Decorate a group match rule."""

View File

@ -35,9 +35,6 @@ CHANGE_ADDED = "added"
CHANGE_UPDATED = "updated"
CHANGE_REMOVED = "removed"
_ItemT = TypeVar("_ItemT")
_StoreT = TypeVar("_StoreT", bound="SerializedStorageCollection")
_StorageCollectionT = TypeVar("_StorageCollectionT", bound="StorageCollection")
_EntityT = TypeVar("_EntityT", bound=Entity, default=Entity)
@ -129,7 +126,7 @@ class CollectionEntity(Entity):
"""Handle updated configuration."""
class ObservableCollection(ABC, Generic[_ItemT]):
class ObservableCollection[_ItemT](ABC):
"""Base collection type that can be observed."""
def __init__(self, id_manager: IDManager | None) -> None:
@ -236,7 +233,9 @@ class SerializedStorageCollection(TypedDict):
items: list[dict[str, Any]]
class StorageCollection(ObservableCollection[_ItemT], Generic[_ItemT, _StoreT]):
class StorageCollection[_ItemT, _StoreT: SerializedStorageCollection](
ObservableCollection[_ItemT]
):
"""Offer a CRUD interface on top of JSON storage."""
def __init__(
@ -512,7 +511,7 @@ def sync_entity_lifecycle(
).async_setup()
class StorageCollectionWebsocket(Generic[_StorageCollectionT]):
class StorageCollectionWebsocket[_StorageCollectionT: StorageCollection]:
"""Class to expose storage collection management over websocket."""
def __init__(

View File

@ -12,7 +12,7 @@ from json import JSONDecodeError, JSONEncoder
import logging
import os
from pathlib import Path
from typing import Any, Generic, TypeVar
from typing import Any
from homeassistant.const import (
EVENT_HOMEASSISTANT_FINAL_WRITE,
@ -48,11 +48,9 @@ STORAGE_MANAGER: HassKey[_StoreManager] = HassKey("storage_manager")
MANAGER_CLEANUP_DELAY = 60
_T = TypeVar("_T", bound=Mapping[str, Any] | Sequence[Any])
@bind_hass
async def async_migrator(
async def async_migrator[_T: Mapping[str, Any] | Sequence[Any]](
hass: HomeAssistant,
old_path: str,
store: Store[_T],
@ -229,7 +227,7 @@ class _StoreManager:
@bind_hass
class Store(Generic[_T]):
class Store[_T: Mapping[str, Any] | Sequence[Any]]:
"""Class to help storing data."""
def __init__(

View File

@ -33,9 +33,6 @@ REQUEST_REFRESH_DEFAULT_COOLDOWN = 10
REQUEST_REFRESH_DEFAULT_IMMEDIATE = True
_DataT = TypeVar("_DataT", default=dict[str, Any])
_BaseDataUpdateCoordinatorT = TypeVar(
"_BaseDataUpdateCoordinatorT", bound="BaseDataUpdateCoordinatorProtocol"
)
_DataUpdateCoordinatorT = TypeVar(
"_DataUpdateCoordinatorT",
bound="DataUpdateCoordinator[Any]",
@ -462,7 +459,9 @@ class TimestampDataUpdateCoordinator(DataUpdateCoordinator[_DataT]):
self.last_update_success_time = utcnow()
class BaseCoordinatorEntity(entity.Entity, Generic[_BaseDataUpdateCoordinatorT]):
class BaseCoordinatorEntity[
_BaseDataUpdateCoordinatorT: BaseDataUpdateCoordinatorProtocol
](entity.Entity):
"""Base class for all Coordinator entities."""
def __init__(

View File

@ -3,13 +3,12 @@
from collections.abc import Callable
import contextlib
from enum import Enum
from typing import TYPE_CHECKING, Any, TypeVar
from typing import TYPE_CHECKING, Any
# https://github.com/python/mypy/issues/5107
if TYPE_CHECKING:
_LruCacheT = TypeVar("_LruCacheT", bound=Callable)
def lru_cache(func: _LruCacheT) -> _LruCacheT:
def lru_cache[_T: Callable[..., Any]](func: _T) -> _T:
"""Stub for lru_cache."""
else: