Use PEP 695 misc (#117788)
This commit is contained in:
parent
649981e503
commit
f50973c76c
6 changed files with 40 additions and 47 deletions
|
@ -4,7 +4,6 @@ from __future__ import annotations
|
||||||
|
|
||||||
from collections.abc import Callable
|
from collections.abc import Callable
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from typing import Generic, TypeVar
|
|
||||||
|
|
||||||
from pydeconz.interfaces.sensors import SensorResources
|
from pydeconz.interfaces.sensors import SensorResources
|
||||||
from pydeconz.models.event import EventType
|
from pydeconz.models.event import EventType
|
||||||
|
@ -48,29 +47,28 @@ PROVIDES_EXTRA_ATTRIBUTES = (
|
||||||
"water",
|
"water",
|
||||||
)
|
)
|
||||||
|
|
||||||
T = TypeVar(
|
|
||||||
"T",
|
|
||||||
Alarm,
|
|
||||||
CarbonMonoxide,
|
|
||||||
Fire,
|
|
||||||
GenericFlag,
|
|
||||||
OpenClose,
|
|
||||||
Presence,
|
|
||||||
Vibration,
|
|
||||||
Water,
|
|
||||||
PydeconzSensorBase,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True, kw_only=True)
|
@dataclass(frozen=True, kw_only=True)
|
||||||
class DeconzBinarySensorDescription(Generic[T], BinarySensorEntityDescription):
|
class DeconzBinarySensorDescription[
|
||||||
|
_T: (
|
||||||
|
Alarm,
|
||||||
|
CarbonMonoxide,
|
||||||
|
Fire,
|
||||||
|
GenericFlag,
|
||||||
|
OpenClose,
|
||||||
|
Presence,
|
||||||
|
Vibration,
|
||||||
|
Water,
|
||||||
|
PydeconzSensorBase,
|
||||||
|
)
|
||||||
|
](BinarySensorEntityDescription):
|
||||||
"""Class describing deCONZ binary sensor entities."""
|
"""Class describing deCONZ binary sensor entities."""
|
||||||
|
|
||||||
instance_check: type[T] | None = None
|
instance_check: type[_T] | None = None
|
||||||
name_suffix: str = ""
|
name_suffix: str = ""
|
||||||
old_unique_id_suffix: str = ""
|
old_unique_id_suffix: str = ""
|
||||||
update_key: str
|
update_key: str
|
||||||
value_fn: Callable[[T], bool | None]
|
value_fn: Callable[[_T], bool | None]
|
||||||
|
|
||||||
|
|
||||||
ENTITY_DESCRIPTIONS: tuple[DeconzBinarySensorDescription, ...] = (
|
ENTITY_DESCRIPTIONS: tuple[DeconzBinarySensorDescription, ...] = (
|
||||||
|
|
|
@ -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 typing import Generic, Literal, TypeVar, cast
|
from typing import Any, Literal
|
||||||
|
|
||||||
from pytraccar import DeviceModel
|
from pytraccar import DeviceModel
|
||||||
|
|
||||||
|
@ -22,13 +22,9 @@ from .const import DOMAIN
|
||||||
from .coordinator import TraccarServerCoordinator
|
from .coordinator import TraccarServerCoordinator
|
||||||
from .entity import TraccarServerEntity
|
from .entity import TraccarServerEntity
|
||||||
|
|
||||||
_T = TypeVar("_T")
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True, kw_only=True)
|
@dataclass(frozen=True, kw_only=True)
|
||||||
class TraccarServerBinarySensorEntityDescription(
|
class TraccarServerBinarySensorEntityDescription[_T](BinarySensorEntityDescription):
|
||||||
Generic[_T], BinarySensorEntityDescription
|
|
||||||
):
|
|
||||||
"""Describe Traccar Server sensor entity."""
|
"""Describe Traccar Server sensor entity."""
|
||||||
|
|
||||||
data_key: Literal["position", "device", "geofence", "attributes"]
|
data_key: Literal["position", "device", "geofence", "attributes"]
|
||||||
|
@ -37,7 +33,9 @@ class TraccarServerBinarySensorEntityDescription(
|
||||||
value_fn: Callable[[_T], bool | None]
|
value_fn: Callable[[_T], bool | None]
|
||||||
|
|
||||||
|
|
||||||
TRACCAR_SERVER_BINARY_SENSOR_ENTITY_DESCRIPTIONS = (
|
TRACCAR_SERVER_BINARY_SENSOR_ENTITY_DESCRIPTIONS: tuple[
|
||||||
|
TraccarServerBinarySensorEntityDescription[Any], ...
|
||||||
|
] = (
|
||||||
TraccarServerBinarySensorEntityDescription[DeviceModel](
|
TraccarServerBinarySensorEntityDescription[DeviceModel](
|
||||||
key="attributes.motion",
|
key="attributes.motion",
|
||||||
data_key="position",
|
data_key="position",
|
||||||
|
@ -65,18 +63,18 @@ async def async_setup_entry(
|
||||||
TraccarServerBinarySensor(
|
TraccarServerBinarySensor(
|
||||||
coordinator=coordinator,
|
coordinator=coordinator,
|
||||||
device=entry["device"],
|
device=entry["device"],
|
||||||
description=cast(TraccarServerBinarySensorEntityDescription, description),
|
description=description,
|
||||||
)
|
)
|
||||||
for entry in coordinator.data.values()
|
for entry in coordinator.data.values()
|
||||||
for description in TRACCAR_SERVER_BINARY_SENSOR_ENTITY_DESCRIPTIONS
|
for description in TRACCAR_SERVER_BINARY_SENSOR_ENTITY_DESCRIPTIONS
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
class TraccarServerBinarySensor(TraccarServerEntity, BinarySensorEntity):
|
class TraccarServerBinarySensor[_T](TraccarServerEntity, BinarySensorEntity):
|
||||||
"""Represent a traccar server binary sensor."""
|
"""Represent a traccar server binary sensor."""
|
||||||
|
|
||||||
_attr_has_entity_name = True
|
_attr_has_entity_name = True
|
||||||
entity_description: TraccarServerBinarySensorEntityDescription
|
entity_description: TraccarServerBinarySensorEntityDescription[_T]
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
|
|
|
@ -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 typing import Generic, Literal, TypeVar, cast
|
from typing import Any, Literal
|
||||||
|
|
||||||
from pytraccar import DeviceModel, GeofenceModel, PositionModel
|
from pytraccar import DeviceModel, GeofenceModel, PositionModel
|
||||||
|
|
||||||
|
@ -24,11 +24,9 @@ from .const import DOMAIN
|
||||||
from .coordinator import TraccarServerCoordinator
|
from .coordinator import TraccarServerCoordinator
|
||||||
from .entity import TraccarServerEntity
|
from .entity import TraccarServerEntity
|
||||||
|
|
||||||
_T = TypeVar("_T")
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True, kw_only=True)
|
@dataclass(frozen=True, kw_only=True)
|
||||||
class TraccarServerSensorEntityDescription(Generic[_T], SensorEntityDescription):
|
class TraccarServerSensorEntityDescription[_T](SensorEntityDescription):
|
||||||
"""Describe Traccar Server sensor entity."""
|
"""Describe Traccar Server sensor entity."""
|
||||||
|
|
||||||
data_key: Literal["position", "device", "geofence", "attributes"]
|
data_key: Literal["position", "device", "geofence", "attributes"]
|
||||||
|
@ -37,7 +35,9 @@ class TraccarServerSensorEntityDescription(Generic[_T], SensorEntityDescription)
|
||||||
value_fn: Callable[[_T], StateType]
|
value_fn: Callable[[_T], StateType]
|
||||||
|
|
||||||
|
|
||||||
TRACCAR_SERVER_SENSOR_ENTITY_DESCRIPTIONS = (
|
TRACCAR_SERVER_SENSOR_ENTITY_DESCRIPTIONS: tuple[
|
||||||
|
TraccarServerSensorEntityDescription[Any], ...
|
||||||
|
] = (
|
||||||
TraccarServerSensorEntityDescription[PositionModel](
|
TraccarServerSensorEntityDescription[PositionModel](
|
||||||
key="attributes.batteryLevel",
|
key="attributes.batteryLevel",
|
||||||
data_key="position",
|
data_key="position",
|
||||||
|
@ -91,18 +91,18 @@ async def async_setup_entry(
|
||||||
TraccarServerSensor(
|
TraccarServerSensor(
|
||||||
coordinator=coordinator,
|
coordinator=coordinator,
|
||||||
device=entry["device"],
|
device=entry["device"],
|
||||||
description=cast(TraccarServerSensorEntityDescription, description),
|
description=description,
|
||||||
)
|
)
|
||||||
for entry in coordinator.data.values()
|
for entry in coordinator.data.values()
|
||||||
for description in TRACCAR_SERVER_SENSOR_ENTITY_DESCRIPTIONS
|
for description in TRACCAR_SERVER_SENSOR_ENTITY_DESCRIPTIONS
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
class TraccarServerSensor(TraccarServerEntity, SensorEntity):
|
class TraccarServerSensor[_T](TraccarServerEntity, SensorEntity):
|
||||||
"""Represent a tracked device."""
|
"""Represent a tracked device."""
|
||||||
|
|
||||||
_attr_has_entity_name = True
|
_attr_has_entity_name = True
|
||||||
entity_description: TraccarServerSensorEntityDescription
|
entity_description: TraccarServerSensorEntityDescription[_T]
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
|
|
|
@ -857,7 +857,7 @@ class HomeAssistant:
|
||||||
return task
|
return task
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
def async_add_executor_job[_T, *_Ts](
|
def async_add_executor_job[*_Ts, _T](
|
||||||
self, target: Callable[[*_Ts], _T], *args: *_Ts
|
self, target: Callable[[*_Ts], _T], *args: *_Ts
|
||||||
) -> asyncio.Future[_T]:
|
) -> asyncio.Future[_T]:
|
||||||
"""Add an executor job from within the event loop."""
|
"""Add an executor job from within the event loop."""
|
||||||
|
@ -871,7 +871,7 @@ class HomeAssistant:
|
||||||
return task
|
return task
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
def async_add_import_executor_job[_T, *_Ts](
|
def async_add_import_executor_job[*_Ts, _T](
|
||||||
self, target: Callable[[*_Ts], _T], *args: *_Ts
|
self, target: Callable[[*_Ts], _T], *args: *_Ts
|
||||||
) -> asyncio.Future[_T]:
|
) -> asyncio.Future[_T]:
|
||||||
"""Add an import executor job from within the event loop.
|
"""Add an import executor job from within the event loop.
|
||||||
|
|
|
@ -18,7 +18,7 @@ import re
|
||||||
from socket import ( # type: ignore[attr-defined] # private, not in typeshed
|
from socket import ( # type: ignore[attr-defined] # private, not in typeshed
|
||||||
_GLOBAL_DEFAULT_TIMEOUT,
|
_GLOBAL_DEFAULT_TIMEOUT,
|
||||||
)
|
)
|
||||||
from typing import Any, TypeVar, cast, overload
|
from typing import Any, cast, overload
|
||||||
from urllib.parse import urlparse
|
from urllib.parse import urlparse
|
||||||
from uuid import UUID
|
from uuid import UUID
|
||||||
|
|
||||||
|
@ -140,9 +140,6 @@ gps = vol.ExactSequence([latitude, longitude])
|
||||||
sun_event = vol.All(vol.Lower, vol.Any(SUN_EVENT_SUNSET, SUN_EVENT_SUNRISE))
|
sun_event = vol.All(vol.Lower, vol.Any(SUN_EVENT_SUNSET, SUN_EVENT_SUNRISE))
|
||||||
port = vol.All(vol.Coerce(int), vol.Range(min=1, max=65535))
|
port = vol.All(vol.Coerce(int), vol.Range(min=1, max=65535))
|
||||||
|
|
||||||
# typing typevar
|
|
||||||
_T = TypeVar("_T")
|
|
||||||
|
|
||||||
|
|
||||||
def path(value: Any) -> str:
|
def path(value: Any) -> str:
|
||||||
"""Validate it's a safe path."""
|
"""Validate it's a safe path."""
|
||||||
|
@ -288,14 +285,14 @@ def ensure_list(value: None) -> list[Any]: ...
|
||||||
|
|
||||||
|
|
||||||
@overload
|
@overload
|
||||||
def ensure_list(value: list[_T]) -> list[_T]: ...
|
def ensure_list[_T](value: list[_T]) -> list[_T]: ...
|
||||||
|
|
||||||
|
|
||||||
@overload
|
@overload
|
||||||
def ensure_list(value: list[_T] | _T) -> list[_T]: ...
|
def ensure_list[_T](value: list[_T] | _T) -> list[_T]: ...
|
||||||
|
|
||||||
|
|
||||||
def ensure_list(value: _T | None) -> list[_T] | list[Any]:
|
def ensure_list[_T](value: _T | None) -> list[_T] | list[Any]:
|
||||||
"""Wrap value in list if it is not one."""
|
"""Wrap value in list if it is not one."""
|
||||||
if value is None:
|
if value is None:
|
||||||
return []
|
return []
|
||||||
|
@ -540,7 +537,7 @@ def time_period_seconds(value: float | str) -> timedelta:
|
||||||
time_period = vol.Any(time_period_str, time_period_seconds, timedelta, time_period_dict)
|
time_period = vol.Any(time_period_str, time_period_seconds, timedelta, time_period_dict)
|
||||||
|
|
||||||
|
|
||||||
def match_all(value: _T) -> _T:
|
def match_all[_T](value: _T) -> _T:
|
||||||
"""Validate that matches all values."""
|
"""Validate that matches all values."""
|
||||||
return value
|
return value
|
||||||
|
|
||||||
|
@ -556,7 +553,7 @@ positive_time_period_dict = vol.All(time_period_dict, positive_timedelta)
|
||||||
positive_time_period = vol.All(time_period, positive_timedelta)
|
positive_time_period = vol.All(time_period, positive_timedelta)
|
||||||
|
|
||||||
|
|
||||||
def remove_falsy(value: list[_T]) -> list[_T]:
|
def remove_falsy[_T](value: list[_T]) -> list[_T]:
|
||||||
"""Remove falsy values from a list."""
|
"""Remove falsy values from a list."""
|
||||||
return [v for v in value if v]
|
return [v for v in value if v]
|
||||||
|
|
||||||
|
|
|
@ -30,6 +30,6 @@ MqttMockHAClient = MagicMock
|
||||||
"""MagicMock for `homeassistant.components.mqtt.MQTT`."""
|
"""MagicMock for `homeassistant.components.mqtt.MQTT`."""
|
||||||
MqttMockHAClientGenerator = Callable[..., Coroutine[Any, Any, MqttMockHAClient]]
|
MqttMockHAClientGenerator = Callable[..., Coroutine[Any, Any, MqttMockHAClient]]
|
||||||
"""MagicMock generator for `homeassistant.components.mqtt.MQTT`."""
|
"""MagicMock generator for `homeassistant.components.mqtt.MQTT`."""
|
||||||
type RecorderInstanceGenerator = Callable[..., Coroutine[Any, Any, "Recorder"]]
|
type RecorderInstanceGenerator = Callable[..., Coroutine[Any, Any, Recorder]]
|
||||||
"""Instance generator for `homeassistant.components.recorder.Recorder`."""
|
"""Instance generator for `homeassistant.components.recorder.Recorder`."""
|
||||||
WebSocketGenerator = Callable[..., Coroutine[Any, Any, MockHAClientWebSocket]]
|
WebSocketGenerator = Callable[..., Coroutine[Any, Any, MockHAClientWebSocket]]
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue