self-hosted/home-assistant-core
self-hosted/home-assistant-core
1
0
Fork 0
mirror of https://github.com/home-assistant/core synced 2024-10-04 20:42:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Import Generator from typing_extensions (1) (#118986)

Browse source
This commit is contained in:
Marc Mueller 2024-06-06 17:02:13 +02:00 committed by GitHub
parent 69708db8e0
commit fe21e2b8ba
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
34 changed files with 134 additions and 118 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
homeassistant/components/alexa/capabilities.py
View file

@ -2,10 +2,11 @@
from __future__ import annotations
from collections.abc import Generator
import logging
from typing import Any
from typing_extensions import Generator
from homeassistant.components import (
button,
climate,
@ -260,7 +261,7 @@ class AlexaCapability:
return result
def serialize_properties(self) -> Generator[dict[str, Any], None, None]:
def serialize_properties(self) -> Generator[dict[str, Any]]:
"""Return properties serialized for an API response."""
for prop in self.properties_supported():
prop_name = prop["name"]

50
homeassistant/components/alexa/entities.py
View file

@ -2,10 +2,12 @@
from __future__ import annotations
from collections.abc import Generator, Iterable
from collections.abc import Iterable
import logging
from typing import TYPE_CHECKING, Any
from typing_extensions import Generator
from homeassistant.components import (
alarm_control_panel,
alert,
@ -319,7 +321,7 @@ class AlexaEntity:
"""
raise NotImplementedError
def serialize_properties(self) -> Generator[dict[str, Any], None, None]:
def serialize_properties(self) -> Generator[dict[str, Any]]:
"""Yield each supported property in API format."""
for interface in self.interfaces():
if not interface.properties_proactively_reported():
@ -405,7 +407,7 @@ class GenericCapabilities(AlexaEntity):
return [DisplayCategory.OTHER]
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
yield AlexaPowerController(self.entity)
yield AlexaEndpointHealth(self.hass, self.entity)
@ -428,7 +430,7 @@ class SwitchCapabilities(AlexaEntity):
return [DisplayCategory.SWITCH]
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
yield AlexaPowerController(self.entity)
yield AlexaContactSensor(self.hass, self.entity)
@ -445,7 +447,7 @@ class ButtonCapabilities(AlexaEntity):
"""Return the display categories for this entity."""
return [DisplayCategory.ACTIVITY_TRIGGER]
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
yield AlexaSceneController(self.entity, supports_deactivation=False)
yield AlexaEventDetectionSensor(self.hass, self.entity)
@ -464,7 +466,7 @@ class ClimateCapabilities(AlexaEntity):
return [DisplayCategory.WATER_HEATER]
return [DisplayCategory.THERMOSTAT]
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
# If we support two modes, one being off, we allow turning on too.
supported_features = self.entity.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
@ -532,7 +534,7 @@ class CoverCapabilities(AlexaEntity):
return [DisplayCategory.OTHER]
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
device_class = self.entity.attributes.get(ATTR_DEVICE_CLASS)
if device_class not in (
@ -570,7 +572,7 @@ class EventCapabilities(AlexaEntity):
return [DisplayCategory.DOORBELL]
return None
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
if self.default_display_categories() is not None:
yield AlexaDoorbellEventSource(self.entity)
@ -586,7 +588,7 @@ class LightCapabilities(AlexaEntity):
"""Return the display categories for this entity."""
return [DisplayCategory.LIGHT]
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
yield AlexaPowerController(self.entity)
@ -610,7 +612,7 @@ class FanCapabilities(AlexaEntity):
"""Return the display categories for this entity."""
return [DisplayCategory.FAN]
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
yield AlexaPowerController(self.entity)
force_range_controller = True
@ -653,7 +655,7 @@ class HumidifierCapabilities(AlexaEntity):
"""Return the display categories for this entity."""
return [DisplayCategory.OTHER]
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
yield AlexaPowerController(self.entity)
supported = self.entity.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
@ -677,7 +679,7 @@ class LockCapabilities(AlexaEntity):
"""Return the display categories for this entity."""
return [DisplayCategory.SMARTLOCK]
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
yield AlexaLockController(self.entity)
yield AlexaEndpointHealth(self.hass, self.entity)
@ -696,7 +698,7 @@ class MediaPlayerCapabilities(AlexaEntity):
return [DisplayCategory.TV]
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
yield AlexaPowerController(self.entity)
@ -766,7 +768,7 @@ class SceneCapabilities(AlexaEntity):
"""Return the display categories for this entity."""
return [DisplayCategory.SCENE_TRIGGER]
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
yield AlexaSceneController(self.entity, supports_deactivation=False)
yield Alexa(self.entity)
@ -780,7 +782,7 @@ class ScriptCapabilities(AlexaEntity):
"""Return the display categories for this entity."""
return [DisplayCategory.ACTIVITY_TRIGGER]
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
yield AlexaSceneController(self.entity, supports_deactivation=True)
yield Alexa(self.entity)
@ -796,7 +798,7 @@ class SensorCapabilities(AlexaEntity):
# sensors are currently ignored.
return [DisplayCategory.TEMPERATURE_SENSOR]
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
attrs = self.entity.attributes
if attrs.get(ATTR_UNIT_OF_MEASUREMENT) in {
@ -827,7 +829,7 @@ class BinarySensorCapabilities(AlexaEntity):
return [DisplayCategory.CAMERA]
return None
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
sensor_type = self.get_type()
if sensor_type is self.TYPE_CONTACT:
@ -883,7 +885,7 @@ class AlarmControlPanelCapabilities(AlexaEntity):
"""Return the display categories for this entity."""
return [DisplayCategory.SECURITY_PANEL]
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
if not self.entity.attributes.get("code_arm_required"):
yield AlexaSecurityPanelController(self.hass, self.entity)
@ -899,7 +901,7 @@ class ImageProcessingCapabilities(AlexaEntity):
"""Return the display categories for this entity."""
return [DisplayCategory.CAMERA]
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
yield AlexaEventDetectionSensor(self.hass, self.entity)
yield AlexaEndpointHealth(self.hass, self.entity)
@ -915,7 +917,7 @@ class InputNumberCapabilities(AlexaEntity):
"""Return the display categories for this entity."""
return [DisplayCategory.OTHER]
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
domain = self.entity.domain
yield AlexaRangeController(self.entity, instance=f"{domain}.value")
@ -931,7 +933,7 @@ class TimerCapabilities(AlexaEntity):
"""Return the display categories for this entity."""
return [DisplayCategory.OTHER]
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
yield AlexaTimeHoldController(self.entity, allow_remote_resume=True)
yield AlexaPowerController(self.entity)
@ -946,7 +948,7 @@ class VacuumCapabilities(AlexaEntity):
"""Return the display categories for this entity."""
return [DisplayCategory.VACUUM_CLEANER]
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
supported = self.entity.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
if (
@ -981,7 +983,7 @@ class ValveCapabilities(AlexaEntity):
"""Return the display categories for this entity."""
return [DisplayCategory.OTHER]
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
supported = self.entity.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
if supported & valve.ValveEntityFeature.SET_POSITION:
@ -1006,7 +1008,7 @@ class CameraCapabilities(AlexaEntity):
"""Return the display categories for this entity."""
return [DisplayCategory.CAMERA]
def interfaces(self) -> Generator[AlexaCapability, None, None]:
def interfaces(self) -> Generator[AlexaCapability]:
"""Yield the supported interfaces."""
if self._check_requirements():
supported = self.entity.attributes.get(ATTR_SUPPORTED_FEATURES, 0)

11
homeassistant/components/assist_pipeline/pipeline.py
View file

@ -5,7 +5,7 @@ from __future__ import annotations
import array
import asyncio
from collections import defaultdict, deque
from collections.abc import AsyncGenerator, AsyncIterable, Callable, Iterable
from collections.abc import AsyncIterable, Callable, Iterable
from dataclasses import asdict, dataclass, field
from enum import StrEnum
import logging
@ -16,6 +16,7 @@ import time
from typing import TYPE_CHECKING, Any, Final, Literal, cast
import wave
from typing_extensions import AsyncGenerator
import voluptuous as vol
if TYPE_CHECKING:
@ -922,7 +923,7 @@ class PipelineRun:
stt_vad: VoiceCommandSegmenter | None,
sample_rate: int = 16000,
sample_width: int = 2,
) -> AsyncGenerator[bytes, None]:
) -> AsyncGenerator[bytes]:
"""Yield audio chunks until VAD detects silence or speech-to-text completes."""
chunk_seconds = AUDIO_PROCESSOR_SAMPLES / sample_rate
sent_vad_start = False
@ -1185,7 +1186,7 @@ class PipelineRun:
audio_stream: AsyncIterable[bytes],
sample_rate: int = 16000,
sample_width: int = 2,
) -> AsyncGenerator[ProcessedAudioChunk, None]:
) -> AsyncGenerator[ProcessedAudioChunk]:
"""Apply volume transformation only (no VAD/audio enhancements) with optional chunking."""
ms_per_sample = sample_rate // 1000
ms_per_chunk = (AUDIO_PROCESSOR_SAMPLES // sample_width) // ms_per_sample
@ -1220,7 +1221,7 @@ class PipelineRun:
audio_stream: AsyncIterable[bytes],
sample_rate: int = 16000,
sample_width: int = 2,
) -> AsyncGenerator[ProcessedAudioChunk, None]:
) -> AsyncGenerator[ProcessedAudioChunk]:
"""Split audio into 10 ms chunks and apply VAD/noise suppression/auto gain/volume transformation."""
assert self.audio_processor is not None
@ -1386,7 +1387,7 @@ class PipelineInput:
# Send audio in the buffer first to speech-to-text, then move on to stt_stream.
# This is basically an async itertools.chain.
async def buffer_then_audio_stream() -> (
AsyncGenerator[ProcessedAudioChunk, None]
AsyncGenerator[ProcessedAudioChunk]
):
# Buffered audio
for chunk in stt_audio_buffer:

5
homeassistant/components/assist_pipeline/websocket_api.py
View file

@ -5,12 +5,13 @@ import asyncio
# Suppressing disable=deprecated-module is needed for Python 3.11
import audioop # pylint: disable=deprecated-module
import base64
from collections.abc import AsyncGenerator, Callable
from collections.abc import Callable
import contextlib
import logging
import math
from typing import Any, Final
from typing_extensions import AsyncGenerator
import voluptuous as vol
from homeassistant.components import conversation, stt, tts, websocket_api
@ -165,7 +166,7 @@ async def websocket_run(
elif start_stage == PipelineStage.STT:
wake_word_phrase = msg["input"].get("wake_word_phrase")
async def stt_stream() -> AsyncGenerator[bytes, None]:
async def stt_stream() -> AsyncGenerator[bytes]:
state = None
# Yield until we receive an empty chunk

5
homeassistant/components/automation/trace.py
View file

@ -2,10 +2,11 @@
from __future__ import annotations
from collections.abc import Generator
from contextlib import contextmanager
from typing import Any
from typing_extensions import Generator
from homeassistant.components.trace import (
CONF_STORED_TRACES,
ActionTrace,
@ -55,7 +56,7 @@ def trace_automation(
blueprint_inputs: ConfigType | None,
context: Context,
trace_config: ConfigType,
) -> Generator[AutomationTrace, None, None]:
) -> Generator[AutomationTrace]:
"""Trace action execution of automation with automation_id."""
trace = AutomationTrace(automation_id, config, blueprint_inputs, context)
async_store_trace(hass, trace, trace_config[CONF_STORED_TRACES])

6
homeassistant/components/bluetooth/passive_update_coordinator.py
View file

@ -15,9 +15,11 @@ from homeassistant.helpers.update_coordinator import (
from .update_coordinator import BasePassiveBluetoothCoordinator
if TYPE_CHECKING:
from collections.abc import Callable, Generator
from collections.abc import Callable
import logging
from typing_extensions import Generator
from . import BluetoothChange, BluetoothScanningMode, BluetoothServiceInfoBleak
_PassiveBluetoothDataUpdateCoordinatorT = TypeVar(
@ -81,7 +83,7 @@ class PassiveBluetoothDataUpdateCoordinator(
self._listeners[remove_listener] = (update_callback, context)
return remove_listener
def async_contexts(self) -> Generator[Any, None, None]:
def async_contexts(self) -> Generator[Any]:
"""Return all registered contexts."""
yield from (
context for _, context in self._listeners.values() if context is not None

5
homeassistant/components/ecovacs/controller.py
View file

@ -2,7 +2,7 @@
from __future__ import annotations
from collections.abc import Generator, Mapping
from collections.abc import Mapping
import logging
import ssl
from typing import Any
@ -18,6 +18,7 @@ from deebot_client.mqtt_client import MqttClient, create_mqtt_config
from deebot_client.util import md5
from deebot_client.util.continents import get_continent
from sucks import EcoVacsAPI, VacBot
from typing_extensions import Generator
from homeassistant.const import CONF_COUNTRY, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant, callback
@ -119,7 +120,7 @@ class EcovacsController:
await self._authenticator.teardown()
@callback
def devices(self, capability: type[Capabilities]) -> Generator[Device, None, None]:
def devices(self, capability: type[Capabilities]) -> Generator[Device]:
"""Return generator for devices with a specific capability."""
for device in self._devices:
if isinstance(device.capabilities, capability):

4
homeassistant/components/homekit/aidmanager.py
View file

@ -11,10 +11,10 @@ This module generates and stores them in a HA storage.
from __future__ import annotations
from collections.abc import Generator
import random
from fnv_hash_fast import fnv1a_32
from typing_extensions import Generator
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import entity_registry as er
@ -39,7 +39,7 @@ def get_system_unique_id(entity: er.RegistryEntry, entity_unique_id: str) -> str
return f"{entity.platform}.{entity.domain}.{entity_unique_id}"
def _generate_aids(unique_id: str | None, entity_id: str) -> Generator[int, None, None]:
def _generate_aids(unique_id: str | None, entity_id: str) -> Generator[int]:
"""Generate accessory aid."""
if unique_id:

5
homeassistant/components/homekit_controller/device_trigger.py
View file

@ -2,13 +2,14 @@
from __future__ import annotations
from collections.abc import Callable, Generator
from collections.abc import Callable
from typing import TYPE_CHECKING, Any
from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.characteristics.const import InputEventValues
from aiohomekit.model.services import Service, ServicesTypes
from aiohomekit.utils import clamp_enum_to_char
from typing_extensions import Generator
import voluptuous as vol
from homeassistant.components.device_automation import DEVICE_TRIGGER_BASE_SCHEMA
@ -88,7 +89,7 @@ class TriggerSource:
for event_handler in self._callbacks.get(trigger_key, []):
event_handler(ev)
def async_get_triggers(self) -> Generator[tuple[str, str], None, None]:
def async_get_triggers(self) -> Generator[tuple[str, str]]:
"""List device triggers for HomeKit devices."""
yield from self._triggers

4
homeassistant/components/knx/config_flow.py
View file

@ -3,9 +3,9 @@
from __future__ import annotations
from abc import ABC, abstractmethod
from collections.abc import AsyncGenerator
from typing import Any, Final
from typing_extensions import AsyncGenerator
import voluptuous as vol
from xknx import XKNX
from xknx.exceptions.exception import (
@ -118,7 +118,7 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow):
self._tunnel_endpoints: list[XMLInterface] = []
self._gatewayscanner: GatewayScanner | None = None
self._async_scan_gen: AsyncGenerator[GatewayDescriptor, None] | None = None
self._async_scan_gen: AsyncGenerator[GatewayDescriptor] | None = None
@abstractmethod
def finish_flow(self) -> ConfigFlowResult:

9
homeassistant/components/logbook/processor.py
View file

@ -2,7 +2,7 @@
from __future__ import annotations
from collections.abc import Callable, Generator, Sequence
from collections.abc import Callable, Sequence
from contextlib import suppress
from dataclasses import dataclass
from datetime import datetime as dt
@ -11,6 +11,7 @@ from typing import Any
from sqlalchemy.engine import Result
from sqlalchemy.engine.row import Row
from typing_extensions import Generator
from homeassistant.components.recorder import get_instance
from homeassistant.components.recorder.filters import Filters
@ -173,7 +174,7 @@ class EventProcessor:
)
def humanify(
self, rows: Generator[EventAsRow, None, None] | Sequence[Row] | Result
self, rows: Generator[EventAsRow] | Sequence[Row] | Result
) -> list[dict[str, str]]:
"""Humanify rows."""
return list(
@ -189,11 +190,11 @@ class EventProcessor:
def _humanify(
hass: HomeAssistant,
rows: Generator[EventAsRow, None, None] | Sequence[Row] | Result,
rows: Generator[EventAsRow] | Sequence[Row] | Result,
ent_reg: er.EntityRegistry,
logbook_run: LogbookRun,
context_augmenter: ContextAugmenter,
) -> Generator[dict[str, Any], None, None]:
) -> Generator[dict[str, Any]]:
"""Generate a converted list of events into entries."""
# Continuous sensors, will be excluded from the logbook
continuous_sensors: dict[str, bool] = {}

7
homeassistant/components/matter/discovery.py
View file

@ -2,10 +2,9 @@
from __future__ import annotations
from collections.abc import Generator
from chip.clusters.Objects import ClusterAttributeDescriptor
from matter_server.client.models.node import MatterEndpoint
from typing_extensions import Generator
from homeassistant.const import Platform
from homeassistant.core import callback
@ -36,7 +35,7 @@ SUPPORTED_PLATFORMS = tuple(DISCOVERY_SCHEMAS)
@callback
def iter_schemas() -> Generator[MatterDiscoverySchema, None, None]:
def iter_schemas() -> Generator[MatterDiscoverySchema]:
"""Iterate over all available discovery schemas."""
for platform_schemas in DISCOVERY_SCHEMAS.values():
yield from platform_schemas
@ -45,7 +44,7 @@ def iter_schemas() -> Generator[MatterDiscoverySchema, None, None]:
@callback
def async_discover_entities(
endpoint: MatterEndpoint,
) -> Generator[MatterEntityInfo, None, None]:
) -> Generator[MatterEntityInfo]:
"""Run discovery on MatterEndpoint and return matching MatterEntityInfo(s)."""
discovered_attributes: set[type[ClusterAttributeDescriptor]] = set()
device_info = endpoint.device_info

5
homeassistant/components/mqtt/client.py
View file

@ -4,7 +4,7 @@ from __future__ import annotations
import asyncio
from collections import defaultdict
from collections.abc import AsyncGenerator, Callable, Coroutine, Iterable
from collections.abc import Callable, Coroutine, Iterable
import contextlib
from dataclasses import dataclass
from functools import lru_cache, partial
@ -18,6 +18,7 @@ from typing import TYPE_CHECKING, Any
import uuid
import certifi
from typing_extensions import AsyncGenerator
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
@ -521,7 +522,7 @@ class MQTT:
self._cleanup_on_unload.pop()()
@contextlib.asynccontextmanager
async def _async_connect_in_executor(self) -> AsyncGenerator[None, None]:
async def _async_connect_in_executor(self) -> AsyncGenerator[None]:
# While we are connecting in the executor we need to
# handle on_socket_open and on_socket_register_write
# in the executor as well.

4
homeassistant/components/profiler/__init__.py
View file

@ -1,7 +1,6 @@
"""The profiler integration."""
import asyncio
from collections.abc import Generator
import contextlib
from contextlib import suppress
from datetime import timedelta
@ -15,6 +14,7 @@ import traceback
from typing import Any, cast
from lru import LRU
from typing_extensions import Generator
import voluptuous as vol
from homeassistant.components import persistent_notification
@ -586,7 +586,7 @@ def _log_object_sources(
@contextlib.contextmanager
def _increase_repr_limit() -> Generator[None, None, None]:
def _increase_repr_limit() -> Generator[None]:
"""Increase the repr limit."""
arepr = reprlib.aRepr
original_maxstring = arepr.maxstring

7
homeassistant/components/recorder/util.py
View file

@ -2,7 +2,7 @@
from __future__ import annotations
from collections.abc import Callable, Generator, Sequence
from collections.abc import Callable, Sequence
import contextlib
from contextlib import contextmanager
from datetime import date, datetime, timedelta
@ -25,6 +25,7 @@ from sqlalchemy.exc import OperationalError, SQLAlchemyError, StatementError
from sqlalchemy.orm.query import Query
from sqlalchemy.orm.session import Session
from sqlalchemy.sql.lambdas import StatementLambdaElement
from typing_extensions import Generator
import voluptuous as vol
from homeassistant.core import HomeAssistant, callback
@ -118,7 +119,7 @@ def session_scope(
session: Session | None = None,
exception_filter: Callable[[Exception], bool] | None = None,
read_only: bool = False,
) -> Generator[Session, None, None]:
) -> Generator[Session]:
"""Provide a transactional scope around a series of operations.
read_only is used to indicate that the session is only used for reading
@ -714,7 +715,7 @@ def periodic_db_cleanups(instance: Recorder) -> None:
@contextmanager
def write_lock_db_sqlite(instance: Recorder) -> Generator[None, None, None]:
def write_lock_db_sqlite(instance: Recorder) -> Generator[None]:
"""Lock database for writes."""
assert instance.engine is not None
with instance.engine.connect() as connection:

5
homeassistant/components/stream/fmp4utils.py
View file

@ -2,9 +2,10 @@
from __future__ import annotations
from collections.abc import Generator
from typing import TYPE_CHECKING
from typing_extensions import Generator
from homeassistant.exceptions import HomeAssistantError
from .core import Orientation
@ -15,7 +16,7 @@ if TYPE_CHECKING:
def find_box(
mp4_bytes: bytes, target_type: bytes, box_start: int = 0
) -> Generator[int, None, None]:
) -> Generator[int]:
"""Find location of first box (or sub box if box_start provided) of given type."""
if box_start == 0:
index = 0

5
homeassistant/components/stream/worker.py
View file

@ -3,7 +3,7 @@
from __future__ import annotations
from collections import defaultdict, deque
from collections.abc import Callable, Generator, Iterator, Mapping
from collections.abc import Callable, Iterator, Mapping
import contextlib
from dataclasses import fields
import datetime
@ -13,6 +13,7 @@ from threading import Event
from typing import Any, Self, cast
import av
from typing_extensions import Generator
from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util
@ -415,7 +416,7 @@ class PeekIterator(Iterator):
self._next = self._iterator.__next__
return self._next()
def peek(self) -> Generator[av.Packet, None, None]:
def peek(self) -> Generator[av.Packet]:
"""Return items without consuming from the iterator."""
# Items consumed are added to a buffer for future calls to __next__
# or peek. First iterate over the buffer from previous calls to peek.

4
homeassistant/components/unifiprotect/camera.py
View file

@ -2,7 +2,6 @@
from __future__ import annotations
from collections.abc import Generator
import logging
from typing import Any, cast
@ -14,6 +13,7 @@ from pyunifiprotect.data import (
ProtectModelWithId,
StateType,
)
from typing_extensions import Generator
from homeassistant.components.camera import Camera, CameraEntityFeature
from homeassistant.config_entries import ConfigEntry
@ -71,7 +71,7 @@ def _get_camera_channels(
entry: ConfigEntry,
data: ProtectData,
ufp_device: UFPCamera | None = None,
) -> Generator[tuple[UFPCamera, CameraChannel, bool], None, None]:
) -> Generator[tuple[UFPCamera, CameraChannel, bool]]:
"""Get all the camera channels."""
devices = (

5
homeassistant/components/unifiprotect/data.py
View file

@ -2,7 +2,7 @@
from __future__ import annotations
from collections.abc import Callable, Generator, Iterable
from collections.abc import Callable, Iterable
from datetime import datetime, timedelta
from functools import partial
import logging
@ -22,6 +22,7 @@ from pyunifiprotect.data import (
)
from pyunifiprotect.exceptions import ClientError, NotAuthorized
from pyunifiprotect.utils import log_event
from typing_extensions import Generator
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
@ -94,7 +95,7 @@ class ProtectData:
def get_by_types(
self, device_types: Iterable[ModelType], ignore_unadopted: bool = True
) -> Generator[ProtectAdoptableDeviceModel, None, None]:
) -> Generator[ProtectAdoptableDeviceModel]:
"""Get all devices matching types."""
for device_type in device_types:
devices = async_get_devices_by_type(

5
homeassistant/components/unifiprotect/utils.py
View file

@ -2,7 +2,7 @@
from __future__ import annotations
from collections.abc import Generator, Iterable
from collections.abc import Iterable
import contextlib
from enum import Enum
from pathlib import Path
@ -19,6 +19,7 @@ from pyunifiprotect.data import (
LightModeType,
ProtectAdoptableDeviceModel,
)
from typing_extensions import Generator
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
@ -99,7 +100,7 @@ def async_get_devices_by_type(
@callback
def async_get_devices(
bootstrap: Bootstrap, model_type: Iterable[ModelType]
) -> Generator[ProtectAdoptableDeviceModel, None, None]:
) -> Generator[ProtectAdoptableDeviceModel]:
"""Return all device by type."""
return (
device

4
homeassistant/components/wemo/entity.py
View file

@ -2,11 +2,11 @@
from __future__ import annotations
from collections.abc import Generator
import contextlib
import logging
from pywemo.exceptions import ActionException
from typing_extensions import Generator
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
@ -65,7 +65,7 @@ class WemoEntity(CoordinatorEntity[DeviceCoordinator]):
return self._device_info
@contextlib.contextmanager
def _wemo_call_wrapper(self, message: str) -> Generator[None, None, None]:
def _wemo_call_wrapper(self, message: str) -> Generator[None]:
"""Wrap calls to the device that change its state.
1. Takes care of making available=False when communications with the

4
homeassistant/components/wyoming/satellite.py
View file

@ -1,12 +1,12 @@
"""Support for Wyoming satellite services."""
import asyncio
from collections.abc import AsyncGenerator
import io
import logging
from typing import Final
import wave
from typing_extensions import AsyncGenerator
from wyoming.asr import Transcribe, Transcript
from wyoming.audio import AudioChunk, AudioChunkConverter, AudioStart, AudioStop
from wyoming.client import AsyncTcpClient
@ -550,7 +550,7 @@ class WyomingSatellite:
await self._client.write_event(AudioStop(timestamp=timestamp).event())
_LOGGER.debug("TTS streaming complete")
async def _stt_stream(self) -> AsyncGenerator[bytes, None]:
async def _stt_stream(self) -> AsyncGenerator[bytes]:
"""Yield audio chunks from a queue."""
try:
is_first_chunk = True

8
homeassistant/components/zwave_js/discovery.py
View file

@ -2,12 +2,12 @@
from __future__ import annotations
from collections.abc import Generator
from dataclasses import asdict, dataclass, field
from enum import StrEnum
from typing import TYPE_CHECKING, Any, cast
from awesomeversion import AwesomeVersion
from typing_extensions import Generator
from zwave_js_server.const import (
CURRENT_STATE_PROPERTY,
CURRENT_VALUE_PROPERTY,
@ -1186,7 +1186,7 @@ DISCOVERY_SCHEMAS = [
@callback
def async_discover_node_values(
node: ZwaveNode, device: DeviceEntry, discovered_value_ids: dict[str, set[str]]
) -> Generator[ZwaveDiscoveryInfo, None, None]:
) -> Generator[ZwaveDiscoveryInfo]:
"""Run discovery on ZWave node and return matching (primary) values."""
for value in node.values.values():
# We don't need to rediscover an already processed value_id
@ -1197,7 +1197,7 @@ def async_discover_node_values(
@callback
def async_discover_single_value(
value: ZwaveValue, device: DeviceEntry, discovered_value_ids: dict[str, set[str]]
) -> Generator[ZwaveDiscoveryInfo, None, None]:
) -> Generator[ZwaveDiscoveryInfo]:
"""Run discovery on a single ZWave value and return matching schema info."""
discovered_value_ids[device.id].add(value.value_id)
for schema in DISCOVERY_SCHEMAS:
@ -1318,7 +1318,7 @@ def async_discover_single_value(
@callback
def async_discover_single_configuration_value(
value: ConfigurationValue,
) -> Generator[ZwaveDiscoveryInfo, None, None]:
) -> Generator[ZwaveDiscoveryInfo]:
"""Run discovery on single Z-Wave configuration value and return schema matches."""
if value.metadata.writeable and value.metadata.readable:
if value.configuration_value_type == ConfigurationValueType.ENUMERATED:

5
homeassistant/components/zwave_js/services.py
View file

@ -3,11 +3,12 @@
from __future__ import annotations
import asyncio
from collections.abc import Collection, Generator, Sequence
from collections.abc import Collection, Sequence
import logging
import math
from typing import Any
from typing_extensions import Generator
import voluptuous as vol
from zwave_js_server.client import Client as ZwaveClient
from zwave_js_server.const import SET_VALUE_SUCCESS, CommandClass, CommandStatus
@ -83,7 +84,7 @@ def broadcast_command(val: dict[str, Any]) -> dict[str, Any]:
def get_valid_responses_from_results[_T: ZwaveNode | Endpoint](
zwave_objects: Sequence[_T], results: Sequence[Any]
) -> Generator[tuple[_T, Any], None, None]:
) -> Generator[tuple[_T, Any]]:
"""Return valid responses from a list of results."""
for zwave_object, result in zip(zwave_objects, results, strict=False):
if not isinstance(result, Exception):

14
homeassistant/config_entries.py
View file

@ -4,15 +4,7 @@ from __future__ import annotations
import asyncio
from collections import UserDict
from collections.abc import (
Callable,
Coroutine,
Generator,
Hashable,
Iterable,
Mapping,
ValuesView,
)
from collections.abc import Callable, Coroutine, Hashable, Iterable, Mapping, ValuesView
from contextvars import ContextVar
from copy import deepcopy
from enum import Enum, StrEnum
@ -24,7 +16,7 @@ from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Generic, Self, cast
from async_interrupt import interrupt
from typing_extensions import TypeVar
from typing_extensions import Generator, TypeVar
from . import data_entry_flow, loader
from .components import persistent_notification
@ -1105,7 +1097,7 @@ class ConfigEntry(Generic[_DataT]):
@callback
def async_get_active_flows(
self, hass: HomeAssistant, sources: set[str]
) -> Generator[ConfigFlowResult, None, None]:
) -> Generator[ConfigFlowResult]:
"""Get any active flows of certain sources for this entry."""
return (
flow

12
homeassistant/exceptions.py
View file

@ -2,10 +2,12 @@
from __future__ import annotations
from collections.abc import Callable, Generator, Sequence
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
from typing_extensions import Generator
from .util.event_type import EventType
if TYPE_CHECKING:
@ -138,7 +140,7 @@ class ConditionError(HomeAssistantError):
"""Return indentation."""
return " " * indent + message
def output(self, indent: int) -> Generator[str, None, None]:
def output(self, indent: int) -> Generator[str]:
"""Yield an indented representation."""
raise NotImplementedError
@ -154,7 +156,7 @@ class ConditionErrorMessage(ConditionError):
# A message describing this error
message: str
def output(self, indent: int) -> Generator[str, None, None]:
def output(self, indent: int) -> Generator[str]:
"""Yield an indented representation."""
yield self._indent(indent, f"In '{self.type}' condition: {self.message}")
@ -170,7 +172,7 @@ class ConditionErrorIndex(ConditionError):
# The error that this error wraps
error: ConditionError
def output(self, indent: int) -> Generator[str, None, None]:
def output(self, indent: int) -> Generator[str]:
"""Yield an indented representation."""
if self.total > 1:
yield self._indent(
@ -189,7 +191,7 @@ class ConditionErrorContainer(ConditionError):
# List of ConditionErrors that this error wraps
errors: Sequence[ConditionError]
def output(self, indent: int) -> Generator[str, None, None]:
def output(self, indent: int) -> Generator[str]:
"""Yield an indented representation."""
for item in self.errors:
yield from item.output(indent)

5
homeassistant/helpers/condition.py
View file

@ -4,7 +4,7 @@ from __future__ import annotations
import asyncio
from collections import deque
from collections.abc import Callable, Container, Generator
from collections.abc import Callable, Container
from contextlib import contextmanager
from datetime import datetime, time as dt_time, timedelta
import functools as ft
@ -12,6 +12,7 @@ import re
import sys
from typing import Any, Protocol, cast
from typing_extensions import Generator
import voluptuous as vol
from homeassistant.components import zone as zone_cmp
@ -150,7 +151,7 @@ def condition_trace_update_result(**kwargs: Any) -> None:
@contextmanager
def trace_condition(variables: TemplateVarsType) -> Generator[TraceElement, None, None]:
def trace_condition(variables: TemplateVarsType) -> Generator[TraceElement]:
"""Trace condition evaluation."""
should_pop = True
trace_element = trace_stack_top(trace_stack_cv)

5
homeassistant/helpers/script.py
View file

@ -3,7 +3,7 @@
from __future__ import annotations
import asyncio
from collections.abc import AsyncGenerator, Callable, Mapping, Sequence
from collections.abc import Callable, Mapping, Sequence
from contextlib import asynccontextmanager
from contextvars import ContextVar
from copy import copy
@ -16,6 +16,7 @@ from types import MappingProxyType
from typing import Any, Literal, TypedDict, cast
import async_interrupt
from typing_extensions import AsyncGenerator
import voluptuous as vol
from homeassistant import exceptions
@ -190,7 +191,7 @@ async def trace_action(
script_run: _ScriptRun,
stop: asyncio.Future[None],
variables: dict[str, Any],
) -> AsyncGenerator[TraceElement, None]:
) -> AsyncGenerator[TraceElement]:
"""Trace action execution."""
path = trace_path_get()
trace_element = action_trace_append(variables, path)

9
homeassistant/helpers/template.py
View file

@ -6,7 +6,7 @@ from ast import literal_eval
import asyncio
import base64
import collections.abc
from collections.abc import Callable, Generator, Iterable
from collections.abc import Callable, Iterable
from contextlib import AbstractContextManager
from contextvars import ContextVar
from datetime import date, datetime, time, timedelta
@ -34,6 +34,7 @@ from jinja2.sandbox import ImmutableSandboxedEnvironment
from jinja2.utils import Namespace
from lru import LRU
import orjson
from typing_extensions import Generator
import voluptuous as vol
from homeassistant.const import (
@ -882,7 +883,7 @@ class AllStates:
if (render_info := _render_info.get()) is not None:
render_info.all_states_lifecycle = True
def __iter__(self) -> Generator[TemplateState, None, None]:
def __iter__(self) -> Generator[TemplateState]:
"""Return all states."""
self._collect_all()
return _state_generator(self._hass, None)
@ -972,7 +973,7 @@ class DomainStates:
if (entity_collect := _render_info.get()) is not None:
entity_collect.domains_lifecycle.add(self._domain) # type: ignore[attr-defined]
def __iter__(self) -> Generator[TemplateState, None, None]:
def __iter__(self) -> Generator[TemplateState]:
"""Return the iteration over all the states."""
self._collect_domain()
return _state_generator(self._hass, self._domain)
@ -1160,7 +1161,7 @@ def _collect_state(hass: HomeAssistant, entity_id: str) -> None:
def _state_generator(
hass: HomeAssistant, domain: str | None
) -> Generator[TemplateState, None, None]:
) -> Generator[TemplateState]:
"""State generator for a domain or all states."""
states = hass.states
# If domain is None, we want to iterate over all states, but making

6
homeassistant/helpers/trace.py
View file

@ -3,12 +3,14 @@
from __future__ import annotations
from collections import deque
from collections.abc import Callable, Coroutine, Generator
from collections.abc import Callable, Coroutine
from contextlib import contextmanager
from contextvars import ContextVar
from functools import wraps
from typing import Any
from typing_extensions import Generator
from homeassistant.core import ServiceResponse
import homeassistant.util.dt as dt_util
@ -248,7 +250,7 @@ def script_execution_get() -> str | None:
@contextmanager
def trace_path(suffix: str | list[str]) -> Generator[None, None, None]:
def trace_path(suffix: str | list[str]) -> Generator[None]:
"""Go deeper in the config tree.
Can not be used as a decorator on couroutine functions.

6
homeassistant/helpers/update_coordinator.py
View file

@ -4,7 +4,7 @@ from __future__ import annotations
from abc import abstractmethod
import asyncio
from collections.abc import Awaitable, Callable, Coroutine, Generator
from collections.abc import Awaitable, Callable, Coroutine
from datetime import datetime, timedelta
import logging
from random import randint
@ -14,7 +14,7 @@ import urllib.error
import aiohttp
import requests
from typing_extensions import TypeVar
from typing_extensions import Generator, TypeVar
from homeassistant import config_entries
from homeassistant.const import EVENT_HOMEASSISTANT_STOP
@ -177,7 +177,7 @@ class DataUpdateCoordinator(BaseDataUpdateCoordinatorProtocol, Generic[_DataT]):
self._async_unsub_refresh()
self._debounced_refresh.async_cancel()
def async_contexts(self) -> Generator[Any, None, None]:
def async_contexts(self) -> Generator[Any]:
"""Return all registered contexts."""
yield from (
context for _, context in self._listeners.values() if context is not None

10
homeassistant/setup.py
View file

@ -4,7 +4,7 @@ from __future__ import annotations
import asyncio
from collections import defaultdict
from collections.abc import Awaitable, Callable, Generator, Mapping
from collections.abc import Awaitable, Callable, Mapping
import contextlib
import contextvars
from enum import StrEnum
@ -14,6 +14,8 @@ import time
from types import ModuleType
from typing import Any, Final, TypedDict
from typing_extensions import Generator
from . import config as conf_util, core, loader, requirements
from .const import (
BASE_PLATFORMS, # noqa: F401
@ -674,9 +676,7 @@ def _setup_started(
@contextlib.contextmanager
def async_pause_setup(
hass: core.HomeAssistant, phase: SetupPhases
) -> Generator[None, None, None]:
def async_pause_setup(hass: core.HomeAssistant, phase: SetupPhases) -> Generator[None]:
"""Keep track of time we are blocked waiting for other operations.
We want to count the time we wait for importing and
@ -724,7 +724,7 @@ def async_start_setup(
integration: str,
phase: SetupPhases,
group: str | None = None,
) -> Generator[None, None, None]:
) -> Generator[None]:
"""Keep track of when setup starts and finishes.
:param hass: Home Assistant instance

4
script/scaffold/templates/config_flow/tests/conftest.py
View file

@ -1,13 +1,13 @@
"""Common fixtures for the NEW_NAME tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
import pytest
from typing_extensions import Generator
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock, None, None]:
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.NEW_DOMAIN.async_setup_entry", return_value=True

4
script/scaffold/templates/config_flow_helper/tests/conftest.py
View file

@ -1,13 +1,13 @@
"""Common fixtures for the NEW_NAME tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
import pytest
from typing_extensions import Generator
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock, None, None]:
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.NEW_DOMAIN.async_setup_entry", return_value=True