Add first batch of Ruff PYI rules (#115100)

Co-authored-by: Jan Bouwhuis <jbouwh@users.noreply.github.com>

homeassistant/auth/permissions/__init__.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 from collections.abc import Callable
-from typing import Any
 
 import voluptuous as vol
 
@@ -64,7 +63,7 @@ class PolicyPermissions(AbstractPermissions):
         """Return a function that can test entity access."""
         return compile_entities(self._policy.get(CAT_ENTITIES), self._perm_lookup)
 
-    def __eq__(self, other: Any) -> bool:
+    def __eq__(self, other: object) -> bool:
         """Equals check."""
         return isinstance(other, PolicyPermissions) and other._policy == self._policy
 

homeassistant/components/assist_pipeline/pipeline.py

@@ -587,7 +587,7 @@ class PipelineRun:
                 self.audio_settings.noise_suppression_level,
             )
 
-    def __eq__(self, other: Any) -> bool:
+    def __eq__(self, other: object) -> bool:
         """Compare pipeline runs by id."""
         if isinstance(other, PipelineRun):
             return self.id == other.id

homeassistant/components/control4/director_utils.py

@@ -1,7 +1,6 @@
 """Provides data updates from the Control4 controller for platforms."""
 
 from collections import defaultdict
-from collections.abc import Set
 import logging
 from typing import Any
 
@@ -20,7 +19,7 @@ _LOGGER = logging.getLogger(__name__)
 
 
 async def _update_variables_for_config_entry(
-    hass: HomeAssistant, entry: ConfigEntry, variable_names: Set[str]
+    hass: HomeAssistant, entry: ConfigEntry, variable_names: set[str]
 ) -> dict[int, dict[str, Any]]:
     """Retrieve data from the Control4 director."""
     director: C4Director = hass.data[DOMAIN][entry.entry_id][CONF_DIRECTOR]
@@ -32,7 +31,7 @@ async def _update_variables_for_config_entry(
 
 
 async def update_variables_for_config_entry(
-    hass: HomeAssistant, entry: ConfigEntry, variable_names: Set[str]
+    hass: HomeAssistant, entry: ConfigEntry, variable_names: set[str]
 ) -> dict[int, dict[str, Any]]:
     """Try to Retrieve data from the Control4 director for update_coordinator."""
     try:

homeassistant/components/deconz/binary_sensor.py

@@ -33,8 +33,6 @@ from .const import ATTR_DARK, ATTR_ON
 from .deconz_device import DeconzDevice
 from .hub import DeconzHub
 
-_SensorDeviceT = TypeVar("_SensorDeviceT", bound=PydeconzSensorBase)
-
 ATTR_ORIENTATION = "orientation"
 ATTR_TILTANGLE = "tiltangle"
 ATTR_VIBRATIONSTRENGTH = "vibrationstrength"

homeassistant/components/here_travel_time/__init__.py

@@ -50,7 +50,7 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
         departure=departure,
     )
 
-    cls: type[HERETransitDataUpdateCoordinator] | type[HERERoutingDataUpdateCoordinator]
+    cls: type[HERETransitDataUpdateCoordinator | HERERoutingDataUpdateCoordinator]
     if config_entry.data[CONF_MODE] in {TRAVEL_MODE_PUBLIC, "publicTransportTimeTable"}:
         cls = HERETransitDataUpdateCoordinator
     else:

homeassistant/components/homekit_controller/switch.py

@@ -192,7 +192,7 @@ class DeclarativeCharacteristicSwitch(CharacteristicEntity, SwitchEntity):
         )
 
 
-ENTITY_TYPES: dict[str, type[HomeKitSwitch] | type[HomeKitValve]] = {
+ENTITY_TYPES: dict[str, type[HomeKitSwitch | HomeKitValve]] = {
     ServicesTypes.SWITCH: HomeKitSwitch,
     ServicesTypes.OUTLET: HomeKitSwitch,
     ServicesTypes.VALVE: HomeKitValve,

homeassistant/components/mobile_app/helpers.py

@@ -38,7 +38,7 @@ _LOGGER = logging.getLogger(__name__)
 
 
 def setup_decrypt(
-    key_encoder: type[RawEncoder] | type[HexEncoder],
+    key_encoder: type[RawEncoder | HexEncoder],
 ) -> Callable[[bytes, bytes], bytes]:
     """Return decryption function and length of key.
 
@@ -55,7 +55,7 @@ def setup_decrypt(
 
 
 def setup_encrypt(
-    key_encoder: type[RawEncoder] | type[HexEncoder],
+    key_encoder: type[RawEncoder | HexEncoder],
 ) -> Callable[[bytes, bytes], bytes]:
     """Return encryption function and length of key.
 
@@ -75,7 +75,7 @@ def _decrypt_payload_helper(
     key: str | bytes,
     ciphertext: bytes,
     key_bytes: bytes,
-    key_encoder: type[RawEncoder] | type[HexEncoder],
+    key_encoder: type[RawEncoder | HexEncoder],
 ) -> JsonValueType | None:
     """Decrypt encrypted payload."""
     try:

homeassistant/components/mqtt/__init__.py

@@ -6,7 +6,7 @@ import asyncio
 from collections.abc import Callable
 from datetime import datetime
 import logging
-from typing import TYPE_CHECKING, Any, TypeVar, cast
+from typing import TYPE_CHECKING, Any, cast
 
 import voluptuous as vol
 
@@ -143,8 +143,6 @@ CONFIG_ENTRY_CONFIG_KEYS = [
     CONF_WILL_MESSAGE,
 ]
 
-_T = TypeVar("_T")
-
 REMOVED_OPTIONS = vol.All(
     cv.removed(CONF_BIRTH_MESSAGE),  # Removed in HA Core 2023.4
     cv.removed(CONF_BROKER),  # Removed in HA Core 2023.4

homeassistant/components/noaa_tides/sensor.py

@@ -85,7 +85,7 @@ class NOAATidesData(TypedDict):
     """Representation of a single tide."""
 
     time_stamp: list[Timestamp]
-    hi_lo: list[Literal["L"] | Literal["H"]]
+    hi_lo: list[Literal["L", "H"]]
     predicted_wl: list[float]
 
 

homeassistant/components/rainbird/coordinator.py

@@ -7,7 +7,6 @@ from dataclasses import dataclass
 import datetime
 from functools import cached_property
 import logging
-from typing import TypeVar
 
 import aiohttp
 from pyrainbird.async_client import (
@@ -39,8 +38,6 @@ CONECTION_LIMIT = 1
 
 _LOGGER = logging.getLogger(__name__)
 
-_T = TypeVar("_T")
-
 
 @dataclass
 class RainbirdDeviceState:

homeassistant/components/recorder/statistics.py

@@ -684,7 +684,7 @@ def get_metadata_with_session(
     session: Session,
     *,
     statistic_ids: set[str] | None = None,
-    statistic_type: Literal["mean"] | Literal["sum"] | None = None,
+    statistic_type: Literal["mean", "sum"] | None = None,
     statistic_source: str | None = None,
 ) -> dict[str, tuple[int, StatisticMetaData]]:
     """Fetch meta data.
@@ -705,7 +705,7 @@ def get_metadata(
     hass: HomeAssistant,
     *,
     statistic_ids: set[str] | None = None,
-    statistic_type: Literal["mean"] | Literal["sum"] | None = None,
+    statistic_type: Literal["mean", "sum"] | None = None,
     statistic_source: str | None = None,
 ) -> dict[str, tuple[int, StatisticMetaData]]:
     """Return metadata for statistic_ids."""
@@ -753,7 +753,7 @@ def update_statistics_metadata(
 async def async_list_statistic_ids(
     hass: HomeAssistant,
     statistic_ids: set[str] | None = None,
-    statistic_type: Literal["mean"] | Literal["sum"] | None = None,
+    statistic_type: Literal["mean", "sum"] | None = None,
 ) -> list[dict]:
     """Return all statistic_ids (or filtered one) and unit of measurement.
 
@@ -823,7 +823,7 @@ def _flatten_list_statistic_ids_metadata_result(
 def list_statistic_ids(
     hass: HomeAssistant,
     statistic_ids: set[str] | None = None,
-    statistic_type: Literal["mean"] | Literal["sum"] | None = None,
+    statistic_type: Literal["mean", "sum"] | None = None,
 ) -> list[dict]:
     """Return all statistic_ids (or filtered one) and unit of measurement.
 

homeassistant/components/recorder/table_managers/statistics_meta.py

@@ -36,7 +36,7 @@ QUERY_STATISTIC_META = (
 
 def _generate_get_metadata_stmt(
     statistic_ids: set[str] | None = None,
-    statistic_type: Literal["mean"] | Literal["sum"] | None = None,
+    statistic_type: Literal["mean", "sum"] | None = None,
     statistic_source: str | None = None,
 ) -> StatementLambdaElement:
     """Generate a statement to fetch metadata."""
@@ -88,7 +88,7 @@ class StatisticsMetaManager:
         self,
         session: Session,
         statistic_ids: set[str] | None = None,
-        statistic_type: Literal["mean"] | Literal["sum"] | None = None,
+        statistic_type: Literal["mean", "sum"] | None = None,
         statistic_source: str | None = None,
     ) -> dict[str, tuple[int, StatisticMetaData]]:
         """Fetch meta data and process it into results and/or cache."""
@@ -202,7 +202,7 @@ class StatisticsMetaManager:
         self,
         session: Session,
         statistic_ids: set[str] | None = None,
-        statistic_type: Literal["mean"] | Literal["sum"] | None = None,
+        statistic_type: Literal["mean", "sum"] | None = None,
         statistic_source: str | None = None,
     ) -> dict[str, tuple[int, StatisticMetaData]]:
         """Fetch meta data.

homeassistant/components/recorder/websocket_api.py

@@ -235,7 +235,7 @@ async def ws_get_statistics_during_period(
 def _ws_get_list_statistic_ids(
     hass: HomeAssistant,
     msg_id: int,
-    statistic_type: Literal["mean"] | Literal["sum"] | None = None,
+    statistic_type: Literal["mean", "sum"] | None = None,
 ) -> bytes:
     """Fetch a list of available statistic_id and convert them to JSON.
 

homeassistant/components/zha/core/helpers.py

@@ -14,7 +14,7 @@ from dataclasses import dataclass
 import enum
 import logging
 import re
-from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar, overload
+from typing import TYPE_CHECKING, Any, TypeVar, overload
 
 import voluptuous as vol
 import zigpy.exceptions
@@ -59,14 +59,10 @@ from .const import CLUSTER_TYPE_IN, CLUSTER_TYPE_OUT, CUSTOM_CONFIGURATION, DATA
 from .registries import BINDABLE_CLUSTERS
 
 if TYPE_CHECKING:
-    from .cluster_handlers import ClusterHandler
     from .device import ZHADevice
     from .gateway import ZHAGateway
 
-_ClusterHandlerT = TypeVar("_ClusterHandlerT", bound="ClusterHandler")
 _T = TypeVar("_T")
-_R = TypeVar("_R")
-_P = ParamSpec("_P")
 _LOGGER = logging.getLogger(__name__)
 
 
@@ -508,9 +504,9 @@ def validate_device_class(
 
 
 def validate_device_class(
-    device_class_enum: type[BinarySensorDeviceClass]
+    device_class_enum: type[
-    | type[SensorDeviceClass]
+        BinarySensorDeviceClass | SensorDeviceClass | NumberDeviceClass
-    | type[NumberDeviceClass],
+    ],
     metadata_value: enum.Enum,
     platform: str,
     logger: logging.Logger,

homeassistant/core.py

@@ -381,7 +381,7 @@ class HomeAssistant:
     http: HomeAssistantHTTP = None  # type: ignore[assignment]
     config_entries: ConfigEntries = None  # type: ignore[assignment]
 
-    def __new__(cls, config_dir: str) -> HomeAssistant:
+    def __new__(cls, config_dir: str) -> Self:
         """Set the _hass thread local data."""
         hass = super().__new__(cls)
         _hass.hass = hass
@@ -1168,9 +1168,9 @@ class Context:
         self.parent_id = parent_id
         self.origin_event: Event[Any] | None = None
 
-    def __eq__(self, other: Any) -> bool:
+    def __eq__(self, other: object) -> bool:
         """Compare contexts."""
-        return bool(self.__class__ == other.__class__ and self.id == other.id)
+        return isinstance(other, Context) and self.id == other.id
 
     @cached_property
     def _as_dict(self) -> dict[str, str | None]:

homeassistant/helpers/httpx_client.py

@@ -57,7 +57,7 @@ class HassHttpXAsyncClient(httpx.AsyncClient):
         """Prevent an integration from reopen of the client via context manager."""
         return self
 
-    async def __aexit__(self, *args: Any) -> None:
+    async def __aexit__(self, *args: object) -> None:
         """Prevent an integration from close of the client via context manager."""
 
 

homeassistant/helpers/intent.py

@@ -10,7 +10,7 @@ from dataclasses import dataclass
 from enum import Enum
 from functools import cached_property
 import logging
-from typing import Any, TypeVar
+from typing import Any
 
 import voluptuous as vol
 
@@ -34,7 +34,6 @@ from . import (
 
 _LOGGER = logging.getLogger(__name__)
 _SlotsType = dict[str, Any]
-_T = TypeVar("_T")
 
 INTENT_TURN_OFF = "HassTurnOff"
 INTENT_TURN_ON = "HassTurnOn"

homeassistant/helpers/reload.py

@@ -156,7 +156,7 @@ async def async_integration_yaml_config(
     hass: HomeAssistant,
     integration_name: str,
     *,
-    raise_on_failure: Literal[False] | bool,
+    raise_on_failure: Literal[False],
 ) -> ConfigType | None: ...
 
 

homeassistant/helpers/template.py

@@ -28,6 +28,7 @@ from typing import (
     Literal,
     NoReturn,
     ParamSpec,
+    Self,
     TypeVar,
     cast,
     overload,
@@ -310,7 +311,7 @@ class TupleWrapper(tuple, ResultWrapper):
 
     # This is all magic to be allowed to subclass a tuple.
 
-    def __new__(cls, value: tuple, *, render_result: str | None = None) -> TupleWrapper:
+    def __new__(cls, value: tuple, *, render_result: str | None = None) -> Self:
         """Create a new tuple class."""
         return super().__new__(cls, tuple(value))
 
@@ -1102,7 +1103,7 @@ class TemplateStateBase(State):
             return f"{state} {unit}"
         return state
 
-    def __eq__(self, other: Any) -> bool:
+    def __eq__(self, other: object) -> bool:
         """Ensure we collect on equality check."""
         self._collect_state()
         return self._state.__eq__(other)

homeassistant/helpers/typing.py

@@ -3,7 +3,7 @@
 from collections.abc import Mapping
 from enum import Enum
 from functools import partial
-from typing import Any, TypeVar
+from typing import Any
 
 import homeassistant.core
 
@@ -14,8 +14,6 @@ from .deprecation import (
     dir_with_deprecated_constants,
 )
 
-_DataT = TypeVar("_DataT")
-
 GPSType = tuple[float, float]
 ConfigType = dict[str, Any]
 DiscoveryInfoType = dict[str, Any]

homeassistant/util/dt.py

@@ -188,7 +188,7 @@ def parse_datetime(dt_str: str, *, raise_on_error: Literal[True]) -> dt.datetime
 
 @overload
 def parse_datetime(
-    dt_str: str, *, raise_on_error: Literal[False] | bool
+    dt_str: str, *, raise_on_error: Literal[False]
 ) -> dt.datetime | None: ...
 
 

homeassistant/util/signal_type.py

@@ -19,7 +19,7 @@ class _SignalTypeBase(Generic[*_Ts]):
 
         return hash(self.name)
 
-    def __eq__(self, other: Any) -> bool:
+    def __eq__(self, other: object) -> bool:
         """Check equality for dict keys to be compatible with str."""
 
         if isinstance(other, str):

homeassistant/util/yaml/loader.py

@@ -276,7 +276,7 @@ def _parse_yaml_python(
 
 
 def _parse_yaml(
-    loader: type[FastSafeLoader] | type[PythonSafeLoader],
+    loader: type[FastSafeLoader | PythonSafeLoader],
     content: str | TextIO,
     secrets: Secrets | None = None,
 ) -> JSON_TYPE:

pyproject.toml

@@ -638,6 +638,7 @@ select = [
     "PIE", # flake8-pie
     "PL", # pylint
     "PT", # flake8-pytest-style
+    "PYI", # flake8-pyi
     "RET", # flake8-return
     "RSE", # flake8-raise
     "RUF005", # Consider iterable unpacking instead of concatenation
@@ -719,6 +720,9 @@ ignore = [
 
     # temporarily disabled
     "PT019",
+    "PYI024", # Use typing.NamedTuple instead of collections.namedtuple
+    "PYI036",
+    "PYI041",
     "RET503",
     "RET502",
     "RET501",

tests/common.py

@@ -1527,12 +1527,12 @@ class _HA_ANY:
 
     _other = _SENTINEL
 
-    def __eq__(self, other: Any) -> bool:
+    def __eq__(self, other: object) -> bool:
         """Test equal."""
         self._other = other
         return True
 
-    def __ne__(self, other: Any) -> bool:
+    def __ne__(self, other: object) -> bool:
         """Test not equal."""
         self._other = other
         return False

tests/components/recorder/db_schema_25.py

@@ -653,7 +653,7 @@ class LazyState(State):
             "last_updated": last_updated_isoformat,
         }
 
-    def __eq__(self, other: Any) -> bool:
+    def __eq__(self, other: object) -> bool:
         """Return the comparison."""
         return (
             other.__class__ in [self.__class__, State]

tests/components/recorder/db_schema_28.py

@@ -818,7 +818,7 @@ class LazyState(State):
             "last_updated": last_updated_isoformat,
         }
 
-    def __eq__(self, other: Any) -> bool:
+    def __eq__(self, other: object) -> bool:
         """Return the comparison."""
         return (
             other.__class__ in [self.__class__, State]