From dcca29ef68e74cfcbca1c7bf7309e1041a8b7f90 Mon Sep 17 00:00:00 2001 From: Marc Mueller <30130371+cdce8p@users.noreply.github.com> Date: Thu, 18 Mar 2021 15:08:35 +0100 Subject: [PATCH] Update typing 14 (#48078) --- .../components/water_heater/device_action.py | 6 +- .../water_heater/reproduce_state.py | 12 ++-- .../components/websocket_api/__init__.py | 10 +-- .../components/websocket_api/connection.py | 8 ++- .../components/websocket_api/http.py | 5 +- .../components/websocket_api/messages.py | 9 +-- homeassistant/components/wemo/entity.py | 12 ++-- homeassistant/components/wilight/fan.py | 5 +- homeassistant/components/withings/__init__.py | 5 +- .../components/withings/binary_sensor.py | 6 +- homeassistant/components/withings/common.py | 52 ++++++++-------- .../components/withings/config_flow.py | 5 +- homeassistant/components/withings/sensor.py | 8 ++- homeassistant/components/wled/__init__.py | 6 +- homeassistant/components/wled/config_flow.py | 22 ++++--- homeassistant/components/wled/light.py | 34 +++++----- homeassistant/components/wled/sensor.py | 16 ++--- homeassistant/components/wled/switch.py | 12 ++-- .../components/wunderground/sensor.py | 26 ++++---- homeassistant/components/xbox/__init__.py | 19 +++--- homeassistant/components/xbox/base_sensor.py | 4 +- .../components/xbox/binary_sensor.py | 7 ++- homeassistant/components/xbox/browse_media.py | 8 +-- homeassistant/components/xbox/media_player.py | 6 +- homeassistant/components/xbox/media_source.py | 11 ++-- homeassistant/components/xbox/sensor.py | 7 ++- homeassistant/components/yeelight/__init__.py | 7 ++- homeassistant/components/zerproc/light.py | 10 +-- homeassistant/components/zha/climate.py | 23 +++---- homeassistant/components/zha/config_flow.py | 6 +- .../components/zha/core/channels/__init__.py | 32 +++++----- .../components/zha/core/channels/base.py | 5 +- .../components/zha/core/channels/general.py | 28 +++++---- .../zha/core/channels/homeautomation.py | 8 ++- .../components/zha/core/channels/hvac.py | 28 +++++---- .../components/zha/core/channels/lighting.py | 12 ++-- .../zha/core/channels/smartenergy.py | 6 +- homeassistant/components/zha/core/const.py | 5 +- .../components/zha/core/decorators.py | 8 ++- homeassistant/components/zha/core/device.py | 6 +- .../components/zha/core/discovery.py | 13 ++-- homeassistant/components/zha/core/gateway.py | 8 +-- homeassistant/components/zha/core/group.py | 28 +++++---- homeassistant/components/zha/core/helpers.py | 13 ++-- .../components/zha/core/registries.py | 46 +++++++------- homeassistant/components/zha/core/store.py | 10 +-- homeassistant/components/zha/cover.py | 11 ++-- homeassistant/components/zha/device_action.py | 4 +- homeassistant/components/zha/entity.py | 21 ++++--- homeassistant/components/zha/fan.py | 23 +++---- homeassistant/components/zha/light.py | 32 +++++----- homeassistant/components/zha/sensor.py | 22 ++++--- homeassistant/components/zha/switch.py | 8 ++- homeassistant/components/zone/__init__.py | 28 ++++----- homeassistant/components/zwave/climate.py | 13 ++-- homeassistant/components/zwave_js/__init__.py | 6 +- homeassistant/components/zwave_js/addon.py | 8 +-- homeassistant/components/zwave_js/api.py | 5 +- .../components/zwave_js/binary_sensor.py | 25 ++++---- homeassistant/components/zwave_js/climate.py | 56 +++++++++-------- .../components/zwave_js/config_flow.py | 62 ++++++++++--------- homeassistant/components/zwave_js/cover.py | 20 +++--- .../components/zwave_js/discovery.py | 35 ++++++----- homeassistant/components/zwave_js/entity.py | 16 ++--- homeassistant/components/zwave_js/fan.py | 18 +++--- homeassistant/components/zwave_js/helpers.py | 8 ++- homeassistant/components/zwave_js/light.py | 24 +++---- homeassistant/components/zwave_js/lock.py | 16 ++--- homeassistant/components/zwave_js/migrate.py | 5 +- homeassistant/components/zwave_js/number.py | 10 +-- homeassistant/components/zwave_js/sensor.py | 17 ++--- homeassistant/components/zwave_js/services.py | 8 +-- homeassistant/components/zwave_js/switch.py | 11 ++-- 73 files changed, 614 insertions(+), 521 deletions(-) diff --git a/homeassistant/components/water_heater/device_action.py b/homeassistant/components/water_heater/device_action.py index 991929580d1..f138c777d44 100644 --- a/homeassistant/components/water_heater/device_action.py +++ b/homeassistant/components/water_heater/device_action.py @@ -1,5 +1,5 @@ """Provides device automations for Water Heater.""" -from typing import List, Optional +from __future__ import annotations import voluptuous as vol @@ -28,7 +28,7 @@ ACTION_SCHEMA = cv.DEVICE_ACTION_BASE_SCHEMA.extend( ) -async def async_get_actions(hass: HomeAssistant, device_id: str) -> List[dict]: +async def async_get_actions(hass: HomeAssistant, device_id: str) -> list[dict]: """List device actions for Water Heater devices.""" registry = await entity_registry.async_get_registry(hass) actions = [] @@ -58,7 +58,7 @@ async def async_get_actions(hass: HomeAssistant, device_id: str) -> List[dict]: async def async_call_action_from_config( - hass: HomeAssistant, config: dict, variables: dict, context: Optional[Context] + hass: HomeAssistant, config: dict, variables: dict, context: Context | None ) -> None: """Execute a device action.""" config = ACTION_SCHEMA(config) diff --git a/homeassistant/components/water_heater/reproduce_state.py b/homeassistant/components/water_heater/reproduce_state.py index 77cdba93f96..4675cdb8621 100644 --- a/homeassistant/components/water_heater/reproduce_state.py +++ b/homeassistant/components/water_heater/reproduce_state.py @@ -1,7 +1,9 @@ """Reproduce an Water heater state.""" +from __future__ import annotations + import asyncio import logging -from typing import Any, Dict, Iterable, Optional +from typing import Any, Iterable from homeassistant.const import ( ATTR_ENTITY_ID, @@ -47,8 +49,8 @@ async def _async_reproduce_state( hass: HomeAssistantType, state: State, *, - context: Optional[Context] = None, - reproduce_options: Optional[Dict[str, Any]] = None, + context: Context | None = None, + reproduce_options: dict[str, Any] | None = None, ) -> None: """Reproduce a single state.""" cur_state = hass.states.get(state.entity_id) @@ -124,8 +126,8 @@ async def async_reproduce_states( hass: HomeAssistantType, states: Iterable[State], *, - context: Optional[Context] = None, - reproduce_options: Optional[Dict[str, Any]] = None, + context: Context | None = None, + reproduce_options: dict[str, Any] | None = None, ) -> None: """Reproduce Water heater states.""" await asyncio.gather( diff --git a/homeassistant/components/websocket_api/__init__.py b/homeassistant/components/websocket_api/__init__.py index 43f5c807770..e7b10e18889 100644 --- a/homeassistant/components/websocket_api/__init__.py +++ b/homeassistant/components/websocket_api/__init__.py @@ -1,5 +1,7 @@ """WebSocket based API for Home Assistant.""" -from typing import Optional, Union, cast +from __future__ import annotations + +from typing import cast import voluptuous as vol @@ -43,9 +45,9 @@ DEPENDENCIES = ("http",) @callback def async_register_command( hass: HomeAssistant, - command_or_handler: Union[str, const.WebSocketCommandHandler], - handler: Optional[const.WebSocketCommandHandler] = None, - schema: Optional[vol.Schema] = None, + command_or_handler: str | const.WebSocketCommandHandler, + handler: const.WebSocketCommandHandler | None = None, + schema: vol.Schema | None = None, ) -> None: """Register a websocket command.""" # pylint: disable=protected-access diff --git a/homeassistant/components/websocket_api/connection.py b/homeassistant/components/websocket_api/connection.py index 108d4de5ada..dd1bb333693 100644 --- a/homeassistant/components/websocket_api/connection.py +++ b/homeassistant/components/websocket_api/connection.py @@ -1,6 +1,8 @@ """Connection session.""" +from __future__ import annotations + import asyncio -from typing import Any, Callable, Dict, Hashable, Optional +from typing import Any, Callable, Hashable import voluptuous as vol @@ -26,7 +28,7 @@ class ActiveConnection: else: self.refresh_token_id = None - self.subscriptions: Dict[Hashable, Callable[[], Any]] = {} + self.subscriptions: dict[Hashable, Callable[[], Any]] = {} self.last_id = 0 def context(self, msg): @@ -37,7 +39,7 @@ class ActiveConnection: return Context(user_id=user.id) @callback - def send_result(self, msg_id: int, result: Optional[Any] = None) -> None: + def send_result(self, msg_id: int, result: Any | None = None) -> None: """Send a result message.""" self.send_message(messages.result_message(msg_id, result)) diff --git a/homeassistant/components/websocket_api/http.py b/homeassistant/components/websocket_api/http.py index daa8529e8bd..982bead296a 100644 --- a/homeassistant/components/websocket_api/http.py +++ b/homeassistant/components/websocket_api/http.py @@ -1,8 +1,9 @@ """View to accept incoming websocket connection.""" +from __future__ import annotations + import asyncio from contextlib import suppress import logging -from typing import Optional from aiohttp import WSMsgType, web import async_timeout @@ -57,7 +58,7 @@ class WebSocketHandler: """Initialize an active connection.""" self.hass = hass self.request = request - self.wsock: Optional[web.WebSocketResponse] = None + self.wsock: web.WebSocketResponse | None = None self._to_write: asyncio.Queue = asyncio.Queue(maxsize=MAX_PENDING_MSG) self._handle_task = None self._writer_task = None diff --git a/homeassistant/components/websocket_api/messages.py b/homeassistant/components/websocket_api/messages.py index f68beff5924..736a7ad59f0 100644 --- a/homeassistant/components/websocket_api/messages.py +++ b/homeassistant/components/websocket_api/messages.py @@ -1,8 +1,9 @@ """Message templates for websocket commands.""" +from __future__ import annotations from functools import lru_cache import logging -from typing import Any, Dict +from typing import Any import voluptuous as vol @@ -32,12 +33,12 @@ IDEN_TEMPLATE = "__IDEN__" IDEN_JSON_TEMPLATE = '"__IDEN__"' -def result_message(iden: int, result: Any = None) -> Dict: +def result_message(iden: int, result: Any = None) -> dict: """Return a success result message.""" return {"id": iden, "type": const.TYPE_RESULT, "success": True, "result": result} -def error_message(iden: int, code: str, message: str) -> Dict: +def error_message(iden: int, code: str, message: str) -> dict: """Return an error result message.""" return { "id": iden, @@ -47,7 +48,7 @@ def error_message(iden: int, code: str, message: str) -> Dict: } -def event_message(iden: JSON_TYPE, event: Any) -> Dict: +def event_message(iden: JSON_TYPE, event: Any) -> dict: """Return an event message.""" return {"id": iden, "type": "event", "event": event} diff --git a/homeassistant/components/wemo/entity.py b/homeassistant/components/wemo/entity.py index d9b4b0548f8..d10707f4590 100644 --- a/homeassistant/components/wemo/entity.py +++ b/homeassistant/components/wemo/entity.py @@ -1,8 +1,10 @@ """Classes shared among Wemo entities.""" +from __future__ import annotations + import asyncio import contextlib import logging -from typing import Any, Dict, Generator, Optional +from typing import Any, Generator import async_timeout from pywemo import WeMoDevice @@ -19,7 +21,7 @@ class ExceptionHandlerStatus: """Exit status from the _wemo_exception_handler context manager.""" # An exception if one was raised in the _wemo_exception_handler. - exception: Optional[Exception] = None + exception: Exception | None = None @property def success(self) -> bool: @@ -68,7 +70,7 @@ class WemoEntity(Entity): _LOGGER.info("Reconnected to %s", self.name) self._available = True - def _update(self, force_update: Optional[bool] = True): + def _update(self, force_update: bool | None = True): """Update the device state.""" raise NotImplementedError() @@ -99,7 +101,7 @@ class WemoEntity(Entity): self._available = False async def _async_locked_update( - self, force_update: bool, timeout: Optional[async_timeout.timeout] = None + self, force_update: bool, timeout: async_timeout.timeout | None = None ) -> None: """Try updating within an async lock.""" async with self._update_lock: @@ -124,7 +126,7 @@ class WemoSubscriptionEntity(WemoEntity): return self.wemo.serialnumber @property - def device_info(self) -> Dict[str, Any]: + def device_info(self) -> dict[str, Any]: """Return the device info.""" return { "name": self.name, diff --git a/homeassistant/components/wilight/fan.py b/homeassistant/components/wilight/fan.py index 35727b19927..49402ecb911 100644 --- a/homeassistant/components/wilight/fan.py +++ b/homeassistant/components/wilight/fan.py @@ -1,6 +1,5 @@ """Support for WiLight Fan.""" - -from typing import Optional +from __future__ import annotations from pywilight.const import ( FAN_V1, @@ -79,7 +78,7 @@ class WiLightFan(WiLightDevice, FanEntity): return self._status.get("direction", WL_DIRECTION_OFF) != WL_DIRECTION_OFF @property - def percentage(self) -> Optional[int]: + def percentage(self) -> int | None: """Return the current speed percentage.""" if "direction" in self._status: if self._status["direction"] == WL_DIRECTION_OFF: diff --git a/homeassistant/components/withings/__init__.py b/homeassistant/components/withings/__init__.py index 5efa22e6a86..7ccff020c0d 100644 --- a/homeassistant/components/withings/__init__.py +++ b/homeassistant/components/withings/__init__.py @@ -3,8 +3,9 @@ Support for the Withings API. For more details about this platform, please refer to the documentation at """ +from __future__ import annotations + import asyncio -from typing import Optional from aiohttp.web import Request, Response import voluptuous as vol @@ -175,7 +176,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_webhook_handler( hass: HomeAssistant, webhook_id: str, request: Request -) -> Optional[Response]: +) -> Response | None: """Handle webhooks calls.""" # Handle http head calls to the path. # When creating a notify subscription, Withings will check that the endpoint is running by sending a HEAD request. diff --git a/homeassistant/components/withings/binary_sensor.py b/homeassistant/components/withings/binary_sensor.py index 136a12a0e1d..a7d0a80e8e3 100644 --- a/homeassistant/components/withings/binary_sensor.py +++ b/homeassistant/components/withings/binary_sensor.py @@ -1,5 +1,7 @@ """Sensors flow for Withings.""" -from typing import Callable, List +from __future__ import annotations + +from typing import Callable from homeassistant.components.binary_sensor import ( DEVICE_CLASS_OCCUPANCY, @@ -16,7 +18,7 @@ from .common import BaseWithingsSensor, async_create_entities async def async_setup_entry( hass: HomeAssistant, entry: ConfigEntry, - async_add_entities: Callable[[List[Entity], bool], None], + async_add_entities: Callable[[list[Entity], bool], None], ) -> None: """Set up the sensor config entry.""" entities = await async_create_entities( diff --git a/homeassistant/components/withings/common.py b/homeassistant/components/withings/common.py index 54a0a3c94ee..96be4fd3063 100644 --- a/homeassistant/components/withings/common.py +++ b/homeassistant/components/withings/common.py @@ -1,4 +1,6 @@ """Common code for Withings.""" +from __future__ import annotations + import asyncio from dataclasses import dataclass import datetime @@ -6,7 +8,7 @@ from datetime import timedelta from enum import Enum, IntEnum import logging import re -from typing import Any, Callable, Dict, List, Optional, Tuple, Union +from typing import Any, Callable, Dict from aiohttp.web import Response import requests @@ -86,7 +88,7 @@ class WithingsAttribute: measute_type: Enum friendly_name: str unit_of_measurement: str - icon: Optional[str] + icon: str | None platform: str enabled_by_default: bool update_type: UpdateType @@ -461,12 +463,12 @@ WITHINGS_ATTRIBUTES = [ ), ] -WITHINGS_MEASUREMENTS_MAP: Dict[Measurement, WithingsAttribute] = { +WITHINGS_MEASUREMENTS_MAP: dict[Measurement, WithingsAttribute] = { attr.measurement: attr for attr in WITHINGS_ATTRIBUTES } -WITHINGS_MEASURE_TYPE_MAP: Dict[ - Union[NotifyAppli, GetSleepSummaryField, MeasureType], WithingsAttribute +WITHINGS_MEASURE_TYPE_MAP: dict[ + NotifyAppli | GetSleepSummaryField | MeasureType, WithingsAttribute ] = {attr.measute_type: attr for attr in WITHINGS_ATTRIBUTES} @@ -486,8 +488,8 @@ class ConfigEntryWithingsApi(AbstractWithingsApi): self.session = OAuth2Session(hass, config_entry, implementation) def _request( - self, path: str, params: Dict[str, Any], method: str = "GET" - ) -> Dict[str, Any]: + self, path: str, params: dict[str, Any], method: str = "GET" + ) -> dict[str, Any]: """Perform an async request.""" asyncio.run_coroutine_threadsafe( self.session.async_ensure_token_valid(), self._hass.loop @@ -524,7 +526,7 @@ class WebhookUpdateCoordinator: """Initialize the object.""" self._hass = hass self._user_id = user_id - self._listeners: List[CALLBACK_TYPE] = [] + self._listeners: list[CALLBACK_TYPE] = [] self.data: MeasurementData = {} def async_add_listener(self, listener: CALLBACK_TYPE) -> Callable[[], None]: @@ -573,10 +575,8 @@ class DataManager: self._notify_unsubscribe_delay = datetime.timedelta(seconds=1) self._is_available = True - self._cancel_interval_update_interval: Optional[CALLBACK_TYPE] = None - self._cancel_configure_webhook_subscribe_interval: Optional[ - CALLBACK_TYPE - ] = None + self._cancel_interval_update_interval: CALLBACK_TYPE | None = None + self._cancel_configure_webhook_subscribe_interval: CALLBACK_TYPE | None = None self._api_notification_id = f"withings_{self._user_id}" self.subscription_update_coordinator = DataUpdateCoordinator( @@ -600,7 +600,7 @@ class DataManager: self.webhook_update_coordinator = WebhookUpdateCoordinator( self._hass, self._user_id ) - self._cancel_subscription_update: Optional[Callable[[], None]] = None + self._cancel_subscription_update: Callable[[], None] | None = None self._subscribe_webhook_run_count = 0 @property @@ -728,7 +728,7 @@ class DataManager: self._api.notify_revoke, profile.callbackurl, profile.appli ) - async def async_get_all_data(self) -> Optional[Dict[MeasureType, Any]]: + async def async_get_all_data(self) -> dict[MeasureType, Any] | None: """Update all withings data.""" try: return await self._do_retry(self._async_get_all_data) @@ -764,14 +764,14 @@ class DataManager: raise exception - async def _async_get_all_data(self) -> Optional[Dict[MeasureType, Any]]: + async def _async_get_all_data(self) -> dict[MeasureType, Any] | None: _LOGGER.info("Updating all withings data") return { **await self.async_get_measures(), **await self.async_get_sleep_summary(), } - async def async_get_measures(self) -> Dict[MeasureType, Any]: + async def async_get_measures(self) -> dict[MeasureType, Any]: """Get the measures data.""" _LOGGER.debug("Updating withings measures") @@ -794,7 +794,7 @@ class DataManager: for measure in group.measures } - async def async_get_sleep_summary(self) -> Dict[MeasureType, Any]: + async def async_get_sleep_summary(self) -> dict[MeasureType, Any]: """Get the sleep summary data.""" _LOGGER.debug("Updating withing sleep summary") now = dt.utcnow() @@ -837,7 +837,7 @@ class DataManager: response = await self._hass.async_add_executor_job(get_sleep_summary) # Set the default to empty lists. - raw_values: Dict[GetSleepSummaryField, List[int]] = { + raw_values: dict[GetSleepSummaryField, list[int]] = { field: [] for field in GetSleepSummaryField } @@ -848,9 +848,9 @@ class DataManager: for field in GetSleepSummaryField: raw_values[field].append(dict(data)[field.value]) - values: Dict[GetSleepSummaryField, float] = {} + values: dict[GetSleepSummaryField, float] = {} - def average(data: List[int]) -> float: + def average(data: list[int]) -> float: return sum(data) / len(data) def set_value(field: GetSleepSummaryField, func: Callable) -> None: @@ -907,7 +907,7 @@ def get_attribute_unique_id(attribute: WithingsAttribute, user_id: int) -> str: async def async_get_entity_id( hass: HomeAssistant, attribute: WithingsAttribute, user_id: int -) -> Optional[str]: +) -> str | None: """Get an entity id for a user's attribute.""" entity_registry: EntityRegistry = ( await hass.helpers.entity_registry.async_get_registry() @@ -936,7 +936,7 @@ class BaseWithingsSensor(Entity): self._user_id = self._data_manager.user_id self._name = f"Withings {self._attribute.measurement.value} {self._profile}" self._unique_id = get_attribute_unique_id(self._attribute, self._user_id) - self._state_data: Optional[Any] = None + self._state_data: Any | None = None @property def should_poll(self) -> bool: @@ -1053,7 +1053,7 @@ async def async_get_data_manager( def get_data_manager_by_webhook_id( hass: HomeAssistant, webhook_id: str -) -> Optional[DataManager]: +) -> DataManager | None: """Get a data manager by it's webhook id.""" return next( iter( @@ -1067,7 +1067,7 @@ def get_data_manager_by_webhook_id( ) -def get_all_data_managers(hass: HomeAssistant) -> Tuple[DataManager, ...]: +def get_all_data_managers(hass: HomeAssistant) -> tuple[DataManager, ...]: """Get all configured data managers.""" return tuple( config_entry_data[const.DATA_MANAGER] @@ -1086,7 +1086,7 @@ async def async_create_entities( entry: ConfigEntry, create_func: Callable[[DataManager, WithingsAttribute], Entity], platform: str, -) -> List[Entity]: +) -> list[Entity]: """Create withings entities from config entry.""" data_manager = await async_get_data_manager(hass, entry) @@ -1096,7 +1096,7 @@ async def async_create_entities( ] -def get_platform_attributes(platform: str) -> Tuple[WithingsAttribute, ...]: +def get_platform_attributes(platform: str) -> tuple[WithingsAttribute, ...]: """Get withings attributes used for a specific platform.""" return tuple( attribute for attribute in WITHINGS_ATTRIBUTES if attribute.platform == platform diff --git a/homeassistant/components/withings/config_flow.py b/homeassistant/components/withings/config_flow.py index ddf51741c62..0cdc523266c 100644 --- a/homeassistant/components/withings/config_flow.py +++ b/homeassistant/components/withings/config_flow.py @@ -1,6 +1,7 @@ """Config flow for Withings.""" +from __future__ import annotations + import logging -from typing import Dict, Union import voluptuous as vol from withings_api.common import AuthScope @@ -19,7 +20,7 @@ class WithingsFlowHandler( DOMAIN = const.DOMAIN CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL # Temporarily holds authorization data during the profile step. - _current_data: Dict[str, Union[None, str, int]] = {} + _current_data: dict[str, None | str | int] = {} @property def logger(self) -> logging.Logger: diff --git a/homeassistant/components/withings/sensor.py b/homeassistant/components/withings/sensor.py index a55d83d717b..a7a7947dec5 100644 --- a/homeassistant/components/withings/sensor.py +++ b/homeassistant/components/withings/sensor.py @@ -1,5 +1,7 @@ """Sensors flow for Withings.""" -from typing import Callable, List, Union +from __future__ import annotations + +from typing import Callable from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN from homeassistant.config_entries import ConfigEntry @@ -12,7 +14,7 @@ from .common import BaseWithingsSensor, async_create_entities async def async_setup_entry( hass: HomeAssistant, entry: ConfigEntry, - async_add_entities: Callable[[List[Entity], bool], None], + async_add_entities: Callable[[list[Entity], bool], None], ) -> None: """Set up the sensor config entry.""" @@ -30,6 +32,6 @@ class WithingsHealthSensor(BaseWithingsSensor): """Implementation of a Withings sensor.""" @property - def state(self) -> Union[None, str, int, float]: + def state(self) -> None | str | int | float: """Return the state of the entity.""" return self._state_data diff --git a/homeassistant/components/wled/__init__.py b/homeassistant/components/wled/__init__.py index 72fee5f078c..2ec02931415 100644 --- a/homeassistant/components/wled/__init__.py +++ b/homeassistant/components/wled/__init__.py @@ -1,8 +1,10 @@ """Support for WLED.""" +from __future__ import annotations + import asyncio from datetime import timedelta import logging -from typing import Any, Dict +from typing import Any from wled import WLED, Device as WLEDDevice, WLEDConnectionError, WLEDError @@ -185,7 +187,7 @@ class WLEDDeviceEntity(WLEDEntity): """Defines a WLED device entity.""" @property - def device_info(self) -> Dict[str, Any]: + def device_info(self) -> dict[str, Any]: """Return device information about this WLED device.""" return { ATTR_IDENTIFIERS: {(DOMAIN, self.coordinator.data.info.mac_address)}, diff --git a/homeassistant/components/wled/config_flow.py b/homeassistant/components/wled/config_flow.py index 5915447f2f5..75e59c88086 100644 --- a/homeassistant/components/wled/config_flow.py +++ b/homeassistant/components/wled/config_flow.py @@ -1,5 +1,7 @@ """Config flow to configure the WLED integration.""" -from typing import Any, Dict, Optional +from __future__ import annotations + +from typing import Any import voluptuous as vol from wled import WLED, WLEDConnectionError @@ -23,14 +25,14 @@ class WLEDFlowHandler(ConfigFlow, domain=DOMAIN): CONNECTION_CLASS = CONN_CLASS_LOCAL_POLL async def async_step_user( - self, user_input: Optional[ConfigType] = None - ) -> Dict[str, Any]: + self, user_input: ConfigType | None = None + ) -> dict[str, Any]: """Handle a flow initiated by the user.""" return await self._handle_config_flow(user_input) async def async_step_zeroconf( - self, user_input: Optional[ConfigType] = None - ) -> Dict[str, Any]: + self, user_input: ConfigType | None = None + ) -> dict[str, Any]: """Handle zeroconf discovery.""" if user_input is None: return self.async_abort(reason="cannot_connect") @@ -53,13 +55,13 @@ class WLEDFlowHandler(ConfigFlow, domain=DOMAIN): async def async_step_zeroconf_confirm( self, user_input: ConfigType = None - ) -> Dict[str, Any]: + ) -> dict[str, Any]: """Handle a flow initiated by zeroconf.""" return await self._handle_config_flow(user_input) async def _handle_config_flow( - self, user_input: Optional[ConfigType] = None, prepare: bool = False - ) -> Dict[str, Any]: + self, user_input: ConfigType | None = None, prepare: bool = False + ) -> dict[str, Any]: """Config flow handler for WLED.""" source = self.context.get("source") @@ -100,7 +102,7 @@ class WLEDFlowHandler(ConfigFlow, domain=DOMAIN): data={CONF_HOST: user_input[CONF_HOST], CONF_MAC: user_input[CONF_MAC]}, ) - def _show_setup_form(self, errors: Optional[Dict] = None) -> Dict[str, Any]: + def _show_setup_form(self, errors: dict | None = None) -> dict[str, Any]: """Show the setup form to the user.""" return self.async_show_form( step_id="user", @@ -108,7 +110,7 @@ class WLEDFlowHandler(ConfigFlow, domain=DOMAIN): errors=errors or {}, ) - def _show_confirm_dialog(self, errors: Optional[Dict] = None) -> Dict[str, Any]: + def _show_confirm_dialog(self, errors: dict | None = None) -> dict[str, Any]: """Show the confirm dialog to the user.""" name = self.context.get(CONF_NAME) return self.async_show_form( diff --git a/homeassistant/components/wled/light.py b/homeassistant/components/wled/light.py index d25c0fc9064..df412ba4c36 100644 --- a/homeassistant/components/wled/light.py +++ b/homeassistant/components/wled/light.py @@ -1,6 +1,8 @@ """Support for LED lights.""" +from __future__ import annotations + from functools import partial -from typing import Any, Callable, Dict, List, Optional, Tuple, Union +from typing import Any, Callable import voluptuous as vol @@ -51,7 +53,7 @@ PARALLEL_UPDATES = 1 async def async_setup_entry( hass: HomeAssistantType, entry: ConfigEntry, - async_add_entities: Callable[[List[Entity], bool], None], + async_add_entities: Callable[[list[Entity], bool], None], ) -> None: """Set up WLED light based on a config entry.""" coordinator: WLEDDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id] @@ -115,7 +117,7 @@ class WLEDMasterLight(LightEntity, WLEDDeviceEntity): return SUPPORT_BRIGHTNESS | SUPPORT_TRANSITION @property - def brightness(self) -> Optional[int]: + def brightness(self) -> int | None: """Return the brightness of this light between 1..255.""" return self.coordinator.data.state.brightness @@ -188,7 +190,7 @@ class WLEDSegmentLight(LightEntity, WLEDDeviceEntity): return super().available @property - def extra_state_attributes(self) -> Optional[Dict[str, Any]]: + def extra_state_attributes(self) -> dict[str, Any] | None: """Return the state attributes of the entity.""" playlist = self.coordinator.data.state.playlist if playlist == -1: @@ -209,18 +211,18 @@ class WLEDSegmentLight(LightEntity, WLEDDeviceEntity): } @property - def hs_color(self) -> Optional[Tuple[float, float]]: + def hs_color(self) -> tuple[float, float] | None: """Return the hue and saturation color value [float, float].""" color = self.coordinator.data.state.segments[self._segment].color_primary return color_util.color_RGB_to_hs(*color[:3]) @property - def effect(self) -> Optional[str]: + def effect(self) -> str | None: """Return the current effect of the light.""" return self.coordinator.data.state.segments[self._segment].effect.name @property - def brightness(self) -> Optional[int]: + def brightness(self) -> int | None: """Return the brightness of this light between 1..255.""" state = self.coordinator.data.state @@ -234,7 +236,7 @@ class WLEDSegmentLight(LightEntity, WLEDDeviceEntity): return state.segments[self._segment].brightness @property - def white_value(self) -> Optional[int]: + def white_value(self) -> int | None: """Return the white value of this light between 0..255.""" color = self.coordinator.data.state.segments[self._segment].color_primary return color[-1] if self._rgbw else None @@ -256,7 +258,7 @@ class WLEDSegmentLight(LightEntity, WLEDDeviceEntity): return flags @property - def effect_list(self) -> List[str]: + def effect_list(self) -> list[str]: """Return the list of supported effects.""" return [effect.name for effect in self.coordinator.data.effects] @@ -357,11 +359,11 @@ class WLEDSegmentLight(LightEntity, WLEDDeviceEntity): @wled_exception_handler async def async_effect( self, - effect: Optional[Union[int, str]] = None, - intensity: Optional[int] = None, - palette: Optional[Union[int, str]] = None, - reverse: Optional[bool] = None, - speed: Optional[int] = None, + effect: int | str | None = None, + intensity: int | None = None, + palette: int | str | None = None, + reverse: bool | None = None, + speed: int | None = None, ) -> None: """Set the effect of a WLED light.""" data = {ATTR_SEGMENT_ID: self._segment} @@ -398,7 +400,7 @@ class WLEDSegmentLight(LightEntity, WLEDDeviceEntity): def async_update_segments( entry: ConfigEntry, coordinator: WLEDDataUpdateCoordinator, - current: Dict[int, WLEDSegmentLight], + current: dict[int, WLEDSegmentLight], async_add_entities, ) -> None: """Update segments.""" @@ -438,7 +440,7 @@ def async_update_segments( async def async_remove_entity( index: int, coordinator: WLEDDataUpdateCoordinator, - current: Dict[int, WLEDSegmentLight], + current: dict[int, WLEDSegmentLight], ) -> None: """Remove WLED segment light from Home Assistant.""" entity = current[index] diff --git a/homeassistant/components/wled/sensor.py b/homeassistant/components/wled/sensor.py index da002e1e8f0..4bfca885613 100644 --- a/homeassistant/components/wled/sensor.py +++ b/homeassistant/components/wled/sensor.py @@ -1,6 +1,8 @@ """Support for WLED sensors.""" +from __future__ import annotations + from datetime import timedelta -from typing import Any, Callable, Dict, List, Optional +from typing import Any, Callable from homeassistant.components.sensor import DEVICE_CLASS_CURRENT from homeassistant.config_entries import ConfigEntry @@ -22,7 +24,7 @@ from .const import ATTR_LED_COUNT, ATTR_MAX_POWER, CURRENT_MA, DOMAIN async def async_setup_entry( hass: HomeAssistantType, entry: ConfigEntry, - async_add_entities: Callable[[List[Entity], bool], None], + async_add_entities: Callable[[list[Entity], bool], None], ) -> None: """Set up WLED sensor based on a config entry.""" coordinator: WLEDDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id] @@ -52,7 +54,7 @@ class WLEDSensor(WLEDDeviceEntity): icon: str, key: str, name: str, - unit_of_measurement: Optional[str] = None, + unit_of_measurement: str | None = None, ) -> None: """Initialize WLED sensor.""" self._unit_of_measurement = unit_of_measurement @@ -92,7 +94,7 @@ class WLEDEstimatedCurrentSensor(WLEDSensor): ) @property - def extra_state_attributes(self) -> Optional[Dict[str, Any]]: + def extra_state_attributes(self) -> dict[str, Any] | None: """Return the state attributes of the entity.""" return { ATTR_LED_COUNT: self.coordinator.data.info.leds.count, @@ -105,7 +107,7 @@ class WLEDEstimatedCurrentSensor(WLEDSensor): return self.coordinator.data.info.leds.power @property - def device_class(self) -> Optional[str]: + def device_class(self) -> str | None: """Return the class of this sensor.""" return DEVICE_CLASS_CURRENT @@ -131,7 +133,7 @@ class WLEDUptimeSensor(WLEDSensor): return uptime.replace(microsecond=0).isoformat() @property - def device_class(self) -> Optional[str]: + def device_class(self) -> str | None: """Return the class of this sensor.""" return DEVICE_CLASS_TIMESTAMP @@ -199,7 +201,7 @@ class WLEDWifiRSSISensor(WLEDSensor): return self.coordinator.data.info.wifi.rssi @property - def device_class(self) -> Optional[str]: + def device_class(self) -> str | None: """Return the class of this sensor.""" return DEVICE_CLASS_SIGNAL_STRENGTH diff --git a/homeassistant/components/wled/switch.py b/homeassistant/components/wled/switch.py index e58b32425cb..5902cd246a0 100644 --- a/homeassistant/components/wled/switch.py +++ b/homeassistant/components/wled/switch.py @@ -1,5 +1,7 @@ """Support for WLED switches.""" -from typing import Any, Callable, Dict, List, Optional +from __future__ import annotations + +from typing import Any, Callable from homeassistant.components.switch import SwitchEntity from homeassistant.config_entries import ConfigEntry @@ -21,7 +23,7 @@ PARALLEL_UPDATES = 1 async def async_setup_entry( hass: HomeAssistantType, entry: ConfigEntry, - async_add_entities: Callable[[List[Entity], bool], None], + async_add_entities: Callable[[list[Entity], bool], None], ) -> None: """Set up WLED switch based on a config entry.""" coordinator: WLEDDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id] @@ -72,7 +74,7 @@ class WLEDNightlightSwitch(WLEDSwitch): ) @property - def extra_state_attributes(self) -> Optional[Dict[str, Any]]: + def extra_state_attributes(self) -> dict[str, Any] | None: """Return the state attributes of the entity.""" return { ATTR_DURATION: self.coordinator.data.state.nightlight.duration, @@ -110,7 +112,7 @@ class WLEDSyncSendSwitch(WLEDSwitch): ) @property - def extra_state_attributes(self) -> Optional[Dict[str, Any]]: + def extra_state_attributes(self) -> dict[str, Any] | None: """Return the state attributes of the entity.""" return {ATTR_UDP_PORT: self.coordinator.data.info.udp_port} @@ -144,7 +146,7 @@ class WLEDSyncReceiveSwitch(WLEDSwitch): ) @property - def extra_state_attributes(self) -> Optional[Dict[str, Any]]: + def extra_state_attributes(self) -> dict[str, Any] | None: """Return the state attributes of the entity.""" return {ATTR_UDP_PORT: self.coordinator.data.info.udp_port} diff --git a/homeassistant/components/wunderground/sensor.py b/homeassistant/components/wunderground/sensor.py index 1f4332c35e7..9e4b764f34c 100644 --- a/homeassistant/components/wunderground/sensor.py +++ b/homeassistant/components/wunderground/sensor.py @@ -1,9 +1,11 @@ """Support for WUnderground weather service.""" +from __future__ import annotations + import asyncio from datetime import timedelta import logging import re -from typing import Any, Callable, Optional, Union +from typing import Any, Callable import aiohttp import async_timeout @@ -64,10 +66,10 @@ class WUSensorConfig: def __init__( self, - friendly_name: Union[str, Callable], + friendly_name: str | Callable, feature: str, value: Callable[["WUndergroundData"], Any], - unit_of_measurement: Optional[str] = None, + unit_of_measurement: str | None = None, entity_picture=None, icon: str = "mdi:gauge", extra_state_attributes=None, @@ -99,10 +101,10 @@ class WUCurrentConditionsSensorConfig(WUSensorConfig): def __init__( self, - friendly_name: Union[str, Callable], + friendly_name: str | Callable, field: str, - icon: Optional[str] = "mdi:gauge", - unit_of_measurement: Optional[str] = None, + icon: str | None = "mdi:gauge", + unit_of_measurement: str | None = None, device_class=None, ): """Initialize current conditions sensor configuration. @@ -131,9 +133,7 @@ class WUCurrentConditionsSensorConfig(WUSensorConfig): class WUDailyTextForecastSensorConfig(WUSensorConfig): """Helper for defining sensor configurations for daily text forecasts.""" - def __init__( - self, period: int, field: str, unit_of_measurement: Optional[str] = None - ): + def __init__(self, period: int, field: str, unit_of_measurement: str | None = None): """Initialize daily text forecast sensor configuration. :param period: forecast period number @@ -166,8 +166,8 @@ class WUDailySimpleForecastSensorConfig(WUSensorConfig): friendly_name: str, period: int, field: str, - wu_unit: Optional[str] = None, - ha_unit: Optional[str] = None, + wu_unit: str | None = None, + ha_unit: str | None = None, icon=None, device_class=None, ): @@ -273,7 +273,7 @@ class WUAlmanacSensorConfig(WUSensorConfig): def __init__( self, - friendly_name: Union[str, Callable], + friendly_name: str | Callable, field: str, value_type: str, wu_unit: str, @@ -303,7 +303,7 @@ class WUAlmanacSensorConfig(WUSensorConfig): class WUAlertsSensorConfig(WUSensorConfig): """Helper for defining field configuration for alerts.""" - def __init__(self, friendly_name: Union[str, Callable]): + def __init__(self, friendly_name: str | Callable): """Initialiize alerts sensor configuration. :param friendly_name: Friendly name diff --git a/homeassistant/components/xbox/__init__.py b/homeassistant/components/xbox/__init__.py index 1d921f5fd18..dd358286802 100644 --- a/homeassistant/components/xbox/__init__.py +++ b/homeassistant/components/xbox/__init__.py @@ -1,9 +1,10 @@ """The xbox integration.""" +from __future__ import annotations + import asyncio from dataclasses import dataclass from datetime import timedelta import logging -from typing import Dict, Optional import voluptuous as vol from xbox.webapi.api.client import XboxLiveClient @@ -133,7 +134,7 @@ class ConsoleData: """Xbox console status data.""" status: SmartglassConsoleStatus - app_details: Optional[Product] + app_details: Product | None @dataclass @@ -149,7 +150,7 @@ class PresenceData: in_game: bool in_multiplayer: bool gamer_score: str - gold_tenure: Optional[str] + gold_tenure: str | None account_tier: str @@ -157,8 +158,8 @@ class PresenceData: class XboxData: """Xbox dataclass for update coordinator.""" - consoles: Dict[str, ConsoleData] - presence: Dict[str, PresenceData] + consoles: dict[str, ConsoleData] + presence: dict[str, PresenceData] class XboxUpdateCoordinator(DataUpdateCoordinator): @@ -184,9 +185,9 @@ class XboxUpdateCoordinator(DataUpdateCoordinator): async def _async_update_data(self) -> XboxData: """Fetch the latest console status.""" # Update Console Status - new_console_data: Dict[str, ConsoleData] = {} + new_console_data: dict[str, ConsoleData] = {} for console in self.consoles.result: - current_state: Optional[ConsoleData] = self.data.consoles.get(console.id) + current_state: ConsoleData | None = self.data.consoles.get(console.id) status: SmartglassConsoleStatus = ( await self.client.smartglass.get_console_status(console.id) ) @@ -198,7 +199,7 @@ class XboxUpdateCoordinator(DataUpdateCoordinator): ) # Setup focus app - app_details: Optional[Product] = None + app_details: Product | None = None if current_state is not None: app_details = current_state.app_details @@ -246,7 +247,7 @@ class XboxUpdateCoordinator(DataUpdateCoordinator): def _build_presence_data(person: Person) -> PresenceData: """Build presence data from a person.""" - active_app: Optional[PresenceDetail] = None + active_app: PresenceDetail | None = None try: active_app = next( presence for presence in person.presence_details if presence.is_primary diff --git a/homeassistant/components/xbox/base_sensor.py b/homeassistant/components/xbox/base_sensor.py index d19fbfb918d..c149ce74c32 100644 --- a/homeassistant/components/xbox/base_sensor.py +++ b/homeassistant/components/xbox/base_sensor.py @@ -1,5 +1,5 @@ """Base Sensor for the Xbox Integration.""" -from typing import Optional +from __future__ import annotations from yarl import URL @@ -24,7 +24,7 @@ class XboxBaseSensorEntity(CoordinatorEntity): return f"{self.xuid}_{self.attribute}" @property - def data(self) -> Optional[PresenceData]: + def data(self) -> PresenceData | None: """Return coordinator data for this console.""" return self.coordinator.data.presence.get(self.xuid) diff --git a/homeassistant/components/xbox/binary_sensor.py b/homeassistant/components/xbox/binary_sensor.py index 109b2839b68..98e06257146 100644 --- a/homeassistant/components/xbox/binary_sensor.py +++ b/homeassistant/components/xbox/binary_sensor.py @@ -1,6 +1,7 @@ """Xbox friends binary sensors.""" +from __future__ import annotations + from functools import partial -from typing import Dict, List from homeassistant.components.binary_sensor import BinarySensorEntity from homeassistant.core import callback @@ -44,7 +45,7 @@ class XboxBinarySensorEntity(XboxBaseSensorEntity, BinarySensorEntity): @callback def async_update_friends( coordinator: XboxUpdateCoordinator, - current: Dict[str, List[XboxBinarySensorEntity]], + current: dict[str, list[XboxBinarySensorEntity]], async_add_entities, ) -> None: """Update friends.""" @@ -73,7 +74,7 @@ def async_update_friends( async def async_remove_entities( xuid: str, coordinator: XboxUpdateCoordinator, - current: Dict[str, XboxBinarySensorEntity], + current: dict[str, XboxBinarySensorEntity], ) -> None: """Remove friend sensors from Home Assistant.""" registry = await async_get_entity_registry(coordinator.hass) diff --git a/homeassistant/components/xbox/browse_media.py b/homeassistant/components/xbox/browse_media.py index 2aebb07df1b..0c3eec95c6f 100644 --- a/homeassistant/components/xbox/browse_media.py +++ b/homeassistant/components/xbox/browse_media.py @@ -1,5 +1,5 @@ """Support for media browsing.""" -from typing import Dict, List, Optional +from __future__ import annotations from xbox.webapi.api.client import XboxLiveClient from xbox.webapi.api.provider.catalog.const import HOME_APP_IDS, SYSTEM_PFN_ID_MAP @@ -41,7 +41,7 @@ async def build_item_response( tv_configured: bool, media_content_type: str, media_content_id: str, -) -> Optional[BrowseMedia]: +) -> BrowseMedia | None: """Create response payload for the provided media query.""" apps: InstalledPackagesList = await client.smartglass.get_installed_apps(device_id) @@ -149,7 +149,7 @@ async def build_item_response( ) -def item_payload(item: InstalledPackage, images: Dict[str, List[Image]]): +def item_payload(item: InstalledPackage, images: dict[str, list[Image]]): """Create response payload for a single media item.""" thumbnail = None image = _find_media_image(images.get(item.one_store_product_id, [])) @@ -169,7 +169,7 @@ def item_payload(item: InstalledPackage, images: Dict[str, List[Image]]): ) -def _find_media_image(images: List[Image]) -> Optional[Image]: +def _find_media_image(images: list[Image]) -> Image | None: purpose_order = ["Poster", "Tile", "Logo", "BoxArt"] for purpose in purpose_order: for image in images: diff --git a/homeassistant/components/xbox/media_player.py b/homeassistant/components/xbox/media_player.py index 19e8a90d48e..e57f3971042 100644 --- a/homeassistant/components/xbox/media_player.py +++ b/homeassistant/components/xbox/media_player.py @@ -1,6 +1,8 @@ """Xbox Media Player Support.""" +from __future__ import annotations + import re -from typing import List, Optional +from typing import List from xbox.webapi.api.client import XboxLiveClient from xbox.webapi.api.provider.catalog.models import Image @@ -231,7 +233,7 @@ class XboxMediaPlayer(CoordinatorEntity, MediaPlayerEntity): } -def _find_media_image(images=List[Image]) -> Optional[Image]: +def _find_media_image(images=List[Image]) -> Image | None: purpose_order = ["FeaturePromotionalSquareArt", "Tile", "Logo", "BoxArt"] for purpose in purpose_order: for image in images: diff --git a/homeassistant/components/xbox/media_source.py b/homeassistant/components/xbox/media_source.py index 53346f3750d..1a459580d66 100644 --- a/homeassistant/components/xbox/media_source.py +++ b/homeassistant/components/xbox/media_source.py @@ -1,6 +1,7 @@ """Xbox Media Source Implementation.""" +from __future__ import annotations + from dataclasses import dataclass -from typing import List, Tuple from pydantic.error_wrappers import ValidationError # pylint: disable=no-name-in-module from xbox.webapi.api.client import XboxLiveClient @@ -50,7 +51,7 @@ async def async_get_media_source(hass: HomeAssistantType): @callback def async_parse_identifier( item: MediaSourceItem, -) -> Tuple[str, str, str]: +) -> tuple[str, str, str]: """Parse identifier.""" identifier = item.identifier or "" start = ["", "", ""] @@ -87,7 +88,7 @@ class XboxSource(MediaSource): return PlayMedia(url, MIME_TYPE_MAP[kind]) async def async_browse_media( - self, item: MediaSourceItem, media_types: Tuple[str] = MEDIA_MIME_TYPES + self, item: MediaSourceItem, media_types: tuple[str] = MEDIA_MIME_TYPES ) -> BrowseMediaSource: """Return media.""" title, category, _ = async_parse_identifier(item) @@ -136,7 +137,7 @@ class XboxSource(MediaSource): title_id, _, thumbnail = title.split("#", 2) owner, kind = category.split("#", 1) - items: List[XboxMediaItem] = [] + items: list[XboxMediaItem] = [] try: if kind == "gameclips": if owner == "my": @@ -206,7 +207,7 @@ class XboxSource(MediaSource): ) -def _build_game_item(item: InstalledPackage, images: List[Image]): +def _build_game_item(item: InstalledPackage, images: list[Image]): """Build individual game.""" thumbnail = "" image = _find_media_image(images.get(item.one_store_product_id, [])) diff --git a/homeassistant/components/xbox/sensor.py b/homeassistant/components/xbox/sensor.py index e0ab661b2d9..88bf112b728 100644 --- a/homeassistant/components/xbox/sensor.py +++ b/homeassistant/components/xbox/sensor.py @@ -1,6 +1,7 @@ """Xbox friends binary sensors.""" +from __future__ import annotations + from functools import partial -from typing import Dict, List from homeassistant.core import callback from homeassistant.helpers.entity_registry import ( @@ -43,7 +44,7 @@ class XboxSensorEntity(XboxBaseSensorEntity): @callback def async_update_friends( coordinator: XboxUpdateCoordinator, - current: Dict[str, List[XboxSensorEntity]], + current: dict[str, list[XboxSensorEntity]], async_add_entities, ) -> None: """Update friends.""" @@ -72,7 +73,7 @@ def async_update_friends( async def async_remove_entities( xuid: str, coordinator: XboxUpdateCoordinator, - current: Dict[str, XboxSensorEntity], + current: dict[str, XboxSensorEntity], ) -> None: """Remove friend sensors from Home Assistant.""" registry = await async_get_entity_registry(coordinator.hass) diff --git a/homeassistant/components/yeelight/__init__.py b/homeassistant/components/yeelight/__init__.py index a86af10fe1c..c1e0c555e02 100644 --- a/homeassistant/components/yeelight/__init__.py +++ b/homeassistant/components/yeelight/__init__.py @@ -1,8 +1,9 @@ """Support for Xiaomi Yeelight WiFi color bulb.""" +from __future__ import annotations + import asyncio from datetime import timedelta import logging -from typing import Optional import voluptuous as vol from yeelight import Bulb, BulbException, discover_bulbs @@ -181,7 +182,7 @@ async def async_setup(hass: HomeAssistant, config: dict) -> bool: async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Set up Yeelight from a config entry.""" - async def _initialize(host: str, capabilities: Optional[dict] = None) -> None: + async def _initialize(host: str, capabilities: dict | None = None) -> None: remove_dispatcher = async_dispatcher_connect( hass, DEVICE_INITIALIZED.format(host), @@ -593,7 +594,7 @@ async def _async_get_device( hass: HomeAssistant, host: str, entry: ConfigEntry, - capabilities: Optional[dict], + capabilities: dict | None, ) -> YeelightDevice: # Get model from config and capabilities model = entry.options.get(CONF_MODEL) diff --git a/homeassistant/components/zerproc/light.py b/homeassistant/components/zerproc/light.py index bf9f917f230..16fe5607925 100644 --- a/homeassistant/components/zerproc/light.py +++ b/homeassistant/components/zerproc/light.py @@ -1,7 +1,9 @@ """Zerproc light platform.""" +from __future__ import annotations + from datetime import timedelta import logging -from typing import Callable, List, Optional +from typing import Callable import pyzerproc @@ -29,7 +31,7 @@ SUPPORT_ZERPROC = SUPPORT_BRIGHTNESS | SUPPORT_COLOR DISCOVERY_INTERVAL = timedelta(seconds=60) -async def discover_entities(hass: HomeAssistant) -> List[Entity]: +async def discover_entities(hass: HomeAssistant) -> list[Entity]: """Attempt to discover new lights.""" lights = await pyzerproc.discover() @@ -49,7 +51,7 @@ async def discover_entities(hass: HomeAssistant) -> List[Entity]: async def async_setup_entry( hass: HomeAssistantType, config_entry: ConfigEntry, - async_add_entities: Callable[[List[Entity], bool], None], + async_add_entities: Callable[[list[Entity], bool], None], ) -> None: """Set up Zerproc light devices.""" warned = False @@ -122,7 +124,7 @@ class ZerprocLight(LightEntity): } @property - def icon(self) -> Optional[str]: + def icon(self) -> str | None: """Return the icon to use in the frontend.""" return "mdi:string-lights" diff --git a/homeassistant/components/zha/climate.py b/homeassistant/components/zha/climate.py index f947012d3af..e2c4faa8539 100644 --- a/homeassistant/components/zha/climate.py +++ b/homeassistant/components/zha/climate.py @@ -4,11 +4,12 @@ Climate on Zigbee Home Automation networks. For more details on this platform, please refer to the documentation at https://home-assistant.io/components/zha.climate/ """ +from __future__ import annotations + from datetime import datetime, timedelta import enum import functools from random import randint -from typing import List, Optional, Tuple from homeassistant.components.climate import ClimateEntity from homeassistant.components.climate.const import ( @@ -212,7 +213,7 @@ class Thermostat(ZhaEntity, ClimateEntity): return data @property - def fan_mode(self) -> Optional[str]: + def fan_mode(self) -> str | None: """Return current FAN mode.""" if self._thrm.running_state is None: return FAN_AUTO @@ -224,14 +225,14 @@ class Thermostat(ZhaEntity, ClimateEntity): return FAN_AUTO @property - def fan_modes(self) -> Optional[List[str]]: + def fan_modes(self) -> list[str] | None: """Return supported FAN modes.""" if not self._fan: return None return [FAN_AUTO, FAN_ON] @property - def hvac_action(self) -> Optional[str]: + def hvac_action(self) -> str | None: """Return the current HVAC action.""" if ( self._thrm.pi_heating_demand is None @@ -241,7 +242,7 @@ class Thermostat(ZhaEntity, ClimateEntity): return self._pi_demand_action @property - def _rm_rs_action(self) -> Optional[str]: + def _rm_rs_action(self) -> str | None: """Return the current HVAC action based on running mode and running state.""" running_mode = self._thrm.running_mode @@ -260,7 +261,7 @@ class Thermostat(ZhaEntity, ClimateEntity): return CURRENT_HVAC_OFF @property - def _pi_demand_action(self) -> Optional[str]: + def _pi_demand_action(self) -> str | None: """Return the current HVAC action based on pi_demands.""" heating_demand = self._thrm.pi_heating_demand @@ -275,12 +276,12 @@ class Thermostat(ZhaEntity, ClimateEntity): return CURRENT_HVAC_OFF @property - def hvac_mode(self) -> Optional[str]: + def hvac_mode(self) -> str | None: """Return HVAC operation mode.""" return SYSTEM_MODE_2_HVAC.get(self._thrm.system_mode) @property - def hvac_modes(self) -> Tuple[str, ...]: + def hvac_modes(self) -> tuple[str, ...]: """Return the list of available HVAC operation modes.""" return SEQ_OF_OPERATION.get(self._thrm.ctrl_seqe_of_oper, (HVAC_MODE_OFF,)) @@ -290,12 +291,12 @@ class Thermostat(ZhaEntity, ClimateEntity): return PRECISION_TENTHS @property - def preset_mode(self) -> Optional[str]: + def preset_mode(self) -> str | None: """Return current preset mode.""" return self._preset @property - def preset_modes(self) -> Optional[List[str]]: + def preset_modes(self) -> list[str] | None: """Return supported preset modes.""" return self._presets @@ -566,7 +567,7 @@ class ZenWithinThermostat(Thermostat): """Zen Within Thermostat implementation.""" @property - def _rm_rs_action(self) -> Optional[str]: + def _rm_rs_action(self) -> str | None: """Return the current HVAC action based on running mode and running state.""" running_state = self._thrm.running_state diff --git a/homeassistant/components/zha/config_flow.py b/homeassistant/components/zha/config_flow.py index f59f53c7995..9835a9a3937 100644 --- a/homeassistant/components/zha/config_flow.py +++ b/homeassistant/components/zha/config_flow.py @@ -1,6 +1,8 @@ """Config flow for ZHA.""" +from __future__ import annotations + import os -from typing import Any, Dict, Optional +from typing import Any import serial.tools.list_ports import voluptuous as vol @@ -127,7 +129,7 @@ class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): ) -async def detect_radios(dev_path: str) -> Optional[Dict[str, Any]]: +async def detect_radios(dev_path: str) -> dict[str, Any] | None: """Probe all radio types on the device port.""" for radio in RadioType: dev_config = radio.controller.SCHEMA_DEVICE({CONF_DEVICE_PATH: dev_path}) diff --git a/homeassistant/components/zha/core/channels/__init__.py b/homeassistant/components/zha/core/channels/__init__.py index d3de5f7004d..289f1c36d4d 100644 --- a/homeassistant/components/zha/core/channels/__init__.py +++ b/homeassistant/components/zha/core/channels/__init__.py @@ -2,7 +2,7 @@ from __future__ import annotations import asyncio -from typing import Any, Dict, List, Optional, Tuple, Union +from typing import Any, Dict import zigpy.zcl.clusters.closures @@ -40,7 +40,7 @@ class Channels: def __init__(self, zha_device: zha_typing.ZhaDeviceType) -> None: """Initialize instance.""" - self._pools: List[zha_typing.ChannelPoolType] = [] + self._pools: list[zha_typing.ChannelPoolType] = [] self._power_config = None self._identify = None self._semaphore = asyncio.Semaphore(3) @@ -49,7 +49,7 @@ class Channels: self._zha_device = zha_device @property - def pools(self) -> List[ChannelPool]: + def pools(self) -> list[ChannelPool]: """Return channel pools list.""" return self._pools @@ -96,7 +96,7 @@ class Channels: return self._unique_id @property - def zigbee_signature(self) -> Dict[int, Dict[str, Any]]: + def zigbee_signature(self) -> dict[int, dict[str, Any]]: """Get the zigbee signatures for the pools in channels.""" return { signature[0]: signature[1] @@ -137,7 +137,7 @@ class Channels: component: str, entity_class: zha_typing.CALLABLE_T, unique_id: str, - channels: List[zha_typing.ChannelType], + channels: list[zha_typing.ChannelType], ): """Signal new entity addition.""" if self.zha_device.status == zha_core_device.DeviceStatus.INITIALIZED: @@ -153,7 +153,7 @@ class Channels: async_dispatcher_send(self.zha_device.hass, signal, *args) @callback - def zha_send_event(self, event_data: Dict[str, Union[str, int]]) -> None: + def zha_send_event(self, event_data: dict[str, str | int]) -> None: """Relay events to hass.""" self.zha_device.hass.bus.async_fire( "zha_event", @@ -175,7 +175,7 @@ class ChannelPool: self._channels: Channels = channels self._claimed_channels: ChannelsDict = {} self._id: int = ep_id - self._client_channels: Dict[str, zha_typing.ClientChannelType] = {} + self._client_channels: dict[str, zha_typing.ClientChannelType] = {} self._unique_id: str = f"{channels.unique_id}-{ep_id}" @property @@ -189,7 +189,7 @@ class ChannelPool: return self._claimed_channels @property - def client_channels(self) -> Dict[str, zha_typing.ClientChannelType]: + def client_channels(self) -> dict[str, zha_typing.ClientChannelType]: """Return a dict of client channels.""" return self._client_channels @@ -214,12 +214,12 @@ class ChannelPool: return self._channels.zha_device.is_mains_powered @property - def manufacturer(self) -> Optional[str]: + def manufacturer(self) -> str | None: """Return device manufacturer.""" return self._channels.zha_device.manufacturer @property - def manufacturer_code(self) -> Optional[int]: + def manufacturer_code(self) -> int | None: """Return device manufacturer.""" return self._channels.zha_device.manufacturer_code @@ -229,7 +229,7 @@ class ChannelPool: return self._channels.zha_device.hass @property - def model(self) -> Optional[str]: + def model(self) -> str | None: """Return device model.""" return self._channels.zha_device.model @@ -244,7 +244,7 @@ class ChannelPool: return self._unique_id @property - def zigbee_signature(self) -> Tuple[int, Dict[str, Any]]: + def zigbee_signature(self) -> tuple[int, dict[str, Any]]: """Get the zigbee signature for the endpoint this pool represents.""" return ( self.endpoint.endpoint_id, @@ -342,7 +342,7 @@ class ChannelPool: component: str, entity_class: zha_typing.CALLABLE_T, unique_id: str, - channels: List[zha_typing.ChannelType], + channels: list[zha_typing.ChannelType], ): """Signal new entity addition.""" self._channels.async_new_entity(component, entity_class, unique_id, channels) @@ -353,19 +353,19 @@ class ChannelPool: self._channels.async_send_signal(signal, *args) @callback - def claim_channels(self, channels: List[zha_typing.ChannelType]) -> None: + def claim_channels(self, channels: list[zha_typing.ChannelType]) -> None: """Claim a channel.""" self.claimed_channels.update({ch.id: ch for ch in channels}) @callback - def unclaimed_channels(self) -> List[zha_typing.ChannelType]: + def unclaimed_channels(self) -> list[zha_typing.ChannelType]: """Return a list of available (unclaimed) channels.""" claimed = set(self.claimed_channels) available = set(self.all_channels) return [self.all_channels[chan_id] for chan_id in (available - claimed)] @callback - def zha_send_event(self, event_data: Dict[str, Union[str, int]]) -> None: + def zha_send_event(self, event_data: dict[str, str | int]) -> None: """Relay events to hass.""" self._channels.zha_send_event( { diff --git a/homeassistant/components/zha/core/channels/base.py b/homeassistant/components/zha/core/channels/base.py index 14774c09550..bc93459dbad 100644 --- a/homeassistant/components/zha/core/channels/base.py +++ b/homeassistant/components/zha/core/channels/base.py @@ -1,10 +1,11 @@ """Base classes for channels.""" +from __future__ import annotations import asyncio from enum import Enum from functools import wraps import logging -from typing import Any, Union +from typing import Any import zigpy.exceptions @@ -238,7 +239,7 @@ class ZigbeeChannel(LogMixin): """Handle ZDO commands on this cluster.""" @callback - def zha_send_event(self, command: str, args: Union[int, dict]) -> None: + def zha_send_event(self, command: str, args: int | dict) -> None: """Relay events to hass.""" self._ch_pool.zha_send_event( { diff --git a/homeassistant/components/zha/core/channels/general.py b/homeassistant/components/zha/core/channels/general.py index 1c1ab28ba5d..626596e1a3e 100644 --- a/homeassistant/components/zha/core/channels/general.py +++ b/homeassistant/components/zha/core/channels/general.py @@ -1,6 +1,8 @@ """General channels module for Zigbee Home Automation.""" +from __future__ import annotations + import asyncio -from typing import Any, Coroutine, List, Optional +from typing import Any, Coroutine import zigpy.exceptions import zigpy.zcl.clusters.general as general @@ -44,42 +46,42 @@ class AnalogOutput(ZigbeeChannel): REPORT_CONFIG = [{"attr": "present_value", "config": REPORT_CONFIG_DEFAULT}] @property - def present_value(self) -> Optional[float]: + def present_value(self) -> float | None: """Return cached value of present_value.""" return self.cluster.get("present_value") @property - def min_present_value(self) -> Optional[float]: + def min_present_value(self) -> float | None: """Return cached value of min_present_value.""" return self.cluster.get("min_present_value") @property - def max_present_value(self) -> Optional[float]: + def max_present_value(self) -> float | None: """Return cached value of max_present_value.""" return self.cluster.get("max_present_value") @property - def resolution(self) -> Optional[float]: + def resolution(self) -> float | None: """Return cached value of resolution.""" return self.cluster.get("resolution") @property - def relinquish_default(self) -> Optional[float]: + def relinquish_default(self) -> float | None: """Return cached value of relinquish_default.""" return self.cluster.get("relinquish_default") @property - def description(self) -> Optional[str]: + def description(self) -> str | None: """Return cached value of description.""" return self.cluster.get("description") @property - def engineering_units(self) -> Optional[int]: + def engineering_units(self) -> int | None: """Return cached value of engineering_units.""" return self.cluster.get("engineering_units") @property - def application_type(self) -> Optional[int]: + def application_type(self) -> int | None: """Return cached value of application_type.""" return self.cluster.get("application_type") @@ -215,7 +217,7 @@ class LevelControlChannel(ZigbeeChannel): REPORT_CONFIG = ({"attr": "current_level", "config": REPORT_CONFIG_ASAP},) @property - def current_level(self) -> Optional[int]: + def current_level(self) -> int | None: """Return cached value of the current_level attribute.""" return self.cluster.get("current_level") @@ -293,7 +295,7 @@ class OnOffChannel(ZigbeeChannel): self._off_listener = None @property - def on_off(self) -> Optional[bool]: + def on_off(self) -> bool | None: """Return cached value of on/off attribute.""" return self.cluster.get("on_off") @@ -367,7 +369,7 @@ class Ota(ZigbeeChannel): @callback def cluster_command( - self, tsn: int, command_id: int, args: Optional[List[Any]] + self, tsn: int, command_id: int, args: list[Any] | None ) -> None: """Handle OTA commands.""" cmd_name = self.cluster.server_commands.get(command_id, [command_id])[0] @@ -402,7 +404,7 @@ class PollControl(ZigbeeChannel): @callback def cluster_command( - self, tsn: int, command_id: int, args: Optional[List[Any]] + self, tsn: int, command_id: int, args: list[Any] | None ) -> None: """Handle commands received to this cluster.""" cmd_name = self.cluster.client_commands.get(command_id, [command_id])[0] diff --git a/homeassistant/components/zha/core/channels/homeautomation.py b/homeassistant/components/zha/core/channels/homeautomation.py index 5b3a4778fcd..989cc17f97d 100644 --- a/homeassistant/components/zha/core/channels/homeautomation.py +++ b/homeassistant/components/zha/core/channels/homeautomation.py @@ -1,5 +1,7 @@ """Home automation channels module for Zigbee Home Automation.""" -from typing import Coroutine, Optional +from __future__ import annotations + +from typing import Coroutine import zigpy.zcl.clusters.homeautomation as homeautomation @@ -76,14 +78,14 @@ class ElectricalMeasurementChannel(ZigbeeChannel): ) @property - def divisor(self) -> Optional[int]: + def divisor(self) -> int | None: """Return active power divisor.""" return self.cluster.get( "ac_power_divisor", self.cluster.get("power_divisor", 1) ) @property - def multiplier(self) -> Optional[int]: + def multiplier(self) -> int | None: """Return active power divisor.""" return self.cluster.get( "ac_power_multiplier", self.cluster.get("power_multiplier", 1) diff --git a/homeassistant/components/zha/core/channels/hvac.py b/homeassistant/components/zha/core/channels/hvac.py index e02f6c01836..76f9c0b4e80 100644 --- a/homeassistant/components/zha/core/channels/hvac.py +++ b/homeassistant/components/zha/core/channels/hvac.py @@ -4,9 +4,11 @@ HVAC channels module for Zigbee Home Automation. For more details about this component, please refer to the documentation at https://home-assistant.io/integrations/zha/ """ +from __future__ import annotations + import asyncio from collections import namedtuple -from typing import Any, Dict, List, Optional, Tuple, Union +from typing import Any from zigpy.exceptions import ZigbeeException import zigpy.zcl.clusters.hvac as hvac @@ -44,7 +46,7 @@ class FanChannel(ZigbeeChannel): REPORT_CONFIG = ({"attr": "fan_mode", "config": REPORT_CONFIG_OP},) @property - def fan_mode(self) -> Optional[int]: + def fan_mode(self) -> int | None: """Return current fan mode.""" return self.cluster.get("fan_mode") @@ -198,22 +200,22 @@ class ThermostatChannel(ZigbeeChannel): return self._min_heat_setpoint_limit @property - def local_temp(self) -> Optional[int]: + def local_temp(self) -> int | None: """Thermostat temperature.""" return self._local_temp @property - def occupancy(self) -> Optional[int]: + def occupancy(self) -> int | None: """Is occupancy detected.""" return self._occupancy @property - def occupied_cooling_setpoint(self) -> Optional[int]: + def occupied_cooling_setpoint(self) -> int | None: """Temperature when room is occupied.""" return self._occupied_cooling_setpoint @property - def occupied_heating_setpoint(self) -> Optional[int]: + def occupied_heating_setpoint(self) -> int | None: """Temperature when room is occupied.""" return self._occupied_heating_setpoint @@ -228,27 +230,27 @@ class ThermostatChannel(ZigbeeChannel): return self._pi_heating_demand @property - def running_mode(self) -> Optional[int]: + def running_mode(self) -> int | None: """Thermostat running mode.""" return self._running_mode @property - def running_state(self) -> Optional[int]: + def running_state(self) -> int | None: """Thermostat running state, state of heat, cool, fan relays.""" return self._running_state @property - def system_mode(self) -> Optional[int]: + def system_mode(self) -> int | None: """System mode.""" return self._system_mode @property - def unoccupied_cooling_setpoint(self) -> Optional[int]: + def unoccupied_cooling_setpoint(self) -> int | None: """Temperature when room is not occupied.""" return self._unoccupied_cooling_setpoint @property - def unoccupied_heating_setpoint(self) -> Optional[int]: + def unoccupied_heating_setpoint(self) -> int | None: """Temperature when room is not occupied.""" return self._unoccupied_heating_setpoint @@ -309,7 +311,7 @@ class ThermostatChannel(ZigbeeChannel): chunk, rest = rest[:4], rest[4:] def _configure_reporting_status( - self, attrs: Dict[Union[int, str], Tuple], res: Union[List, Tuple] + self, attrs: dict[int | str, tuple], res: list | tuple ) -> None: """Parse configure reporting result.""" if not isinstance(res, list): @@ -405,7 +407,7 @@ class ThermostatChannel(ZigbeeChannel): self.debug("set cooling setpoint to %s", temperature) return True - async def get_occupancy(self) -> Optional[bool]: + async def get_occupancy(self) -> bool | None: """Get unreportable occupancy attribute.""" try: res, fail = await self.cluster.read_attributes(["occupancy"]) diff --git a/homeassistant/components/zha/core/channels/lighting.py b/homeassistant/components/zha/core/channels/lighting.py index c8827e20e01..000549b0593 100644 --- a/homeassistant/components/zha/core/channels/lighting.py +++ b/homeassistant/components/zha/core/channels/lighting.py @@ -1,5 +1,7 @@ """Lighting channels module for Zigbee Home Automation.""" -from typing import Coroutine, Optional +from __future__ import annotations + +from typing import Coroutine import zigpy.zcl.clusters.lighting as lighting @@ -46,22 +48,22 @@ class ColorChannel(ZigbeeChannel): return self.CAPABILITIES_COLOR_XY @property - def color_loop_active(self) -> Optional[int]: + def color_loop_active(self) -> int | None: """Return cached value of the color_loop_active attribute.""" return self.cluster.get("color_loop_active") @property - def color_temperature(self) -> Optional[int]: + def color_temperature(self) -> int | None: """Return cached value of color temperature.""" return self.cluster.get("color_temperature") @property - def current_x(self) -> Optional[int]: + def current_x(self) -> int | None: """Return cached value of the current_x attribute.""" return self.cluster.get("current_x") @property - def current_y(self) -> Optional[int]: + def current_y(self) -> int | None: """Return cached value of the current_y attribute.""" return self.cluster.get("current_y") diff --git a/homeassistant/components/zha/core/channels/smartenergy.py b/homeassistant/components/zha/core/channels/smartenergy.py index 32e4902799e..a815c75c8b3 100644 --- a/homeassistant/components/zha/core/channels/smartenergy.py +++ b/homeassistant/components/zha/core/channels/smartenergy.py @@ -1,5 +1,7 @@ """Smart energy channels module for Zigbee Home Automation.""" -from typing import Coroutine, Union +from __future__ import annotations + +from typing import Coroutine import zigpy.zcl.clusters.smartenergy as smartenergy @@ -139,7 +141,7 @@ class Metering(ZigbeeChannel): else: self._format_spec = "{:0" + str(width) + "." + str(r_digits) + "f}" - def formatter_function(self, value: int) -> Union[int, float]: + def formatter_function(self, value: int) -> int | float: """Return formatted value for display.""" value = value * self.multiplier / self.divisor if self.unit_of_measurement == POWER_WATT: diff --git a/homeassistant/components/zha/core/const.py b/homeassistant/components/zha/core/const.py index 672fafdd98f..45453ba7545 100644 --- a/homeassistant/components/zha/core/const.py +++ b/homeassistant/components/zha/core/const.py @@ -1,7 +1,8 @@ """All constants related to the ZHA component.""" +from __future__ import annotations + import enum import logging -from typing import List import bellows.zigbee.application from zigpy.config import CONF_DEVICE_PATH # noqa: F401 # pylint: disable=unused-import @@ -203,7 +204,7 @@ class RadioType(enum.Enum): ) @classmethod - def list(cls) -> List[str]: + def list(cls) -> list[str]: """Return a list of descriptions.""" return [e.description for e in RadioType] diff --git a/homeassistant/components/zha/core/decorators.py b/homeassistant/components/zha/core/decorators.py index c416548dbe9..c3eec07e980 100644 --- a/homeassistant/components/zha/core/decorators.py +++ b/homeassistant/components/zha/core/decorators.py @@ -1,5 +1,7 @@ """Decorators for ZHA core registries.""" -from typing import Callable, TypeVar, Union +from __future__ import annotations + +from typing import Callable, TypeVar CALLABLE_T = TypeVar("CALLABLE_T", bound=Callable) # pylint: disable=invalid-name @@ -8,7 +10,7 @@ class DictRegistry(dict): """Dict Registry of items.""" def register( - self, name: Union[int, str], item: Union[str, CALLABLE_T] = None + self, name: int | str, item: str | CALLABLE_T = None ) -> Callable[[CALLABLE_T], CALLABLE_T]: """Return decorator to register item with a specific name.""" @@ -26,7 +28,7 @@ class DictRegistry(dict): class SetRegistry(set): """Set Registry of items.""" - def register(self, name: Union[int, str]) -> Callable[[CALLABLE_T], CALLABLE_T]: + def register(self, name: int | str) -> Callable[[CALLABLE_T], CALLABLE_T]: """Return decorator to register item with a specific name.""" def decorator(channel: CALLABLE_T) -> CALLABLE_T: diff --git a/homeassistant/components/zha/core/device.py b/homeassistant/components/zha/core/device.py index 4f93a7a95d6..65605b2f7a3 100644 --- a/homeassistant/components/zha/core/device.py +++ b/homeassistant/components/zha/core/device.py @@ -1,11 +1,13 @@ """Device for Zigbee Home Automation.""" +from __future__ import annotations + import asyncio from datetime import timedelta from enum import Enum import logging import random import time -from typing import Any, Dict +from typing import Any from zigpy import types import zigpy.exceptions @@ -275,7 +277,7 @@ class ZHADevice(LogMixin): self._available = new_availability @property - def zigbee_signature(self) -> Dict[str, Any]: + def zigbee_signature(self) -> dict[str, Any]: """Get zigbee signature for this device.""" return { ATTR_NODE_DESCRIPTOR: str(self._zigpy_device.node_desc), diff --git a/homeassistant/components/zha/core/discovery.py b/homeassistant/components/zha/core/discovery.py index ba970570ecb..338796acffe 100644 --- a/homeassistant/components/zha/core/discovery.py +++ b/homeassistant/components/zha/core/discovery.py @@ -1,8 +1,9 @@ """Device discovery functions for Zigbee Home Automation.""" +from __future__ import annotations from collections import Counter import logging -from typing import Callable, List, Tuple +from typing import Callable from homeassistant import const as ha_const from homeassistant.core import callback @@ -34,10 +35,10 @@ _LOGGER = logging.getLogger(__name__) @callback async def async_add_entities( _async_add_entities: Callable, - entities: List[ - Tuple[ + entities: list[ + tuple[ zha_typing.ZhaEntityType, - Tuple[str, zha_typing.ZhaDeviceType, List[zha_typing.ChannelType]], + tuple[str, zha_typing.ZhaDeviceType, list[zha_typing.ChannelType]], ] ], update_before_add: bool = True, @@ -235,9 +236,9 @@ class GroupProbe: @staticmethod def determine_entity_domains( hass: HomeAssistantType, group: zha_typing.ZhaGroupType - ) -> List[str]: + ) -> list[str]: """Determine the entity domains for this group.""" - entity_domains: List[str] = [] + entity_domains: list[str] = [] zha_gateway = hass.data[zha_const.DATA_ZHA][zha_const.DATA_ZHA_GATEWAY] all_domain_occurrences = [] for member in group.members: diff --git a/homeassistant/components/zha/core/gateway.py b/homeassistant/components/zha/core/gateway.py index 3a65787cd48..d8baf89efcf 100644 --- a/homeassistant/components/zha/core/gateway.py +++ b/homeassistant/components/zha/core/gateway.py @@ -1,4 +1,5 @@ """Virtual gateway for Zigbee Home Automation.""" +from __future__ import annotations import asyncio import collections @@ -9,7 +10,6 @@ import logging import os import time import traceback -from typing import List, Optional from serial import SerialException from zigpy.config import CONF_DEVICE @@ -378,12 +378,12 @@ class ZHAGateway: """Return ZHADevice for given ieee.""" return self._devices.get(ieee) - def get_group(self, group_id: str) -> Optional[ZhaGroupType]: + def get_group(self, group_id: str) -> ZhaGroupType | None: """Return Group for given group id.""" return self.groups.get(group_id) @callback - def async_get_group_by_name(self, group_name: str) -> Optional[ZhaGroupType]: + def async_get_group_by_name(self, group_name: str) -> ZhaGroupType | None: """Get ZHA group by name.""" for group in self.groups.values(): if group.name == group_name: @@ -620,7 +620,7 @@ class ZHAGateway: zha_device.update_available(True) async def async_create_zigpy_group( - self, name: str, members: List[GroupMember] + self, name: str, members: list[GroupMember] ) -> ZhaGroupType: """Create a new Zigpy Zigbee group.""" # we start with two to fill any gaps from a user removing existing groups diff --git a/homeassistant/components/zha/core/group.py b/homeassistant/components/zha/core/group.py index 59277a394b3..beaebbe8767 100644 --- a/homeassistant/components/zha/core/group.py +++ b/homeassistant/components/zha/core/group.py @@ -1,8 +1,10 @@ """Group for Zigbee Home Automation.""" +from __future__ import annotations + import asyncio import collections import logging -from typing import Any, Dict, List +from typing import Any import zigpy.exceptions @@ -58,16 +60,16 @@ class ZHAGroupMember(LogMixin): return self._zha_device @property - def member_info(self) -> Dict[str, Any]: + def member_info(self) -> dict[str, Any]: """Get ZHA group info.""" - member_info: Dict[str, Any] = {} + member_info: dict[str, Any] = {} member_info["endpoint_id"] = self.endpoint_id member_info["device"] = self.device.zha_device_info member_info["entities"] = self.associated_entities return member_info @property - def associated_entities(self) -> List[GroupEntityReference]: + def associated_entities(self) -> list[GroupEntityReference]: """Return the list of entities that were derived from this endpoint.""" ha_entity_registry = self.device.gateway.ha_entity_registry zha_device_registry = self.device.gateway.device_registry @@ -136,7 +138,7 @@ class ZHAGroup(LogMixin): return self._zigpy_group.endpoint @property - def members(self) -> List[ZHAGroupMember]: + def members(self) -> list[ZHAGroupMember]: """Return the ZHA devices that are members of this group.""" return [ ZHAGroupMember( @@ -146,7 +148,7 @@ class ZHAGroup(LogMixin): if member_ieee in self._zha_gateway.devices ] - async def async_add_members(self, members: List[GroupMember]) -> None: + async def async_add_members(self, members: list[GroupMember]) -> None: """Add members to this group.""" if len(members) > 1: tasks = [] @@ -162,7 +164,7 @@ class ZHAGroup(LogMixin): members[0].ieee ].async_add_endpoint_to_group(members[0].endpoint_id, self.group_id) - async def async_remove_members(self, members: List[GroupMember]) -> None: + async def async_remove_members(self, members: list[GroupMember]) -> None: """Remove members from this group.""" if len(members) > 1: tasks = [] @@ -181,18 +183,18 @@ class ZHAGroup(LogMixin): ].async_remove_endpoint_from_group(members[0].endpoint_id, self.group_id) @property - def member_entity_ids(self) -> List[str]: + def member_entity_ids(self) -> list[str]: """Return the ZHA entity ids for all entities for the members of this group.""" - all_entity_ids: List[str] = [] + all_entity_ids: list[str] = [] for member in self.members: entity_references = member.associated_entities for entity_reference in entity_references: all_entity_ids.append(entity_reference["entity_id"]) return all_entity_ids - def get_domain_entity_ids(self, domain) -> List[str]: + def get_domain_entity_ids(self, domain) -> list[str]: """Return entity ids from the entity domain for this group.""" - domain_entity_ids: List[str] = [] + domain_entity_ids: list[str] = [] for member in self.members: if member.device.is_coordinator: continue @@ -207,9 +209,9 @@ class ZHAGroup(LogMixin): return domain_entity_ids @property - def group_info(self) -> Dict[str, Any]: + def group_info(self) -> dict[str, Any]: """Get ZHA group info.""" - group_info: Dict[str, Any] = {} + group_info: dict[str, Any] = {} group_info["group_id"] = self.group_id group_info["name"] = self.name group_info["members"] = [member.member_info for member in self.members] diff --git a/homeassistant/components/zha/core/helpers.py b/homeassistant/components/zha/core/helpers.py index 47911fc1078..cf3d040f020 100644 --- a/homeassistant/components/zha/core/helpers.py +++ b/homeassistant/components/zha/core/helpers.py @@ -4,6 +4,7 @@ Helpers for Zigbee Home Automation. For more details about this component, please refer to the documentation at https://home-assistant.io/integrations/zha/ """ +from __future__ import annotations import asyncio import binascii @@ -13,7 +14,7 @@ import itertools import logging from random import uniform import re -from typing import Any, Callable, Iterator, List, Optional, Tuple +from typing import Any, Callable, Iterator import voluptuous as vol import zigpy.exceptions @@ -67,7 +68,7 @@ async def safe_read( async def get_matched_clusters( source_zha_device: ZhaDeviceType, target_zha_device: ZhaDeviceType -) -> List[BindingPair]: +) -> list[BindingPair]: """Get matched input/output cluster pairs for 2 devices.""" source_clusters = source_zha_device.async_get_std_clusters() target_clusters = target_zha_device.async_get_std_clusters() @@ -131,7 +132,7 @@ async def async_get_zha_device(hass, device_id): return zha_gateway.devices[ieee] -def find_state_attributes(states: List[State], key: str) -> Iterator[Any]: +def find_state_attributes(states: list[State], key: str) -> Iterator[Any]: """Find attributes with matching key from states.""" for state in states: value = state.attributes.get(key) @@ -150,9 +151,9 @@ def mean_tuple(*args): def reduce_attribute( - states: List[State], + states: list[State], key: str, - default: Optional[Any] = None, + default: Any | None = None, reduce: Callable[..., Any] = mean_int, ) -> Any: """Find the first attribute matching key from states. @@ -280,7 +281,7 @@ QR_CODES = ( ) -def qr_to_install_code(qr_code: str) -> Tuple[zigpy.types.EUI64, bytes]: +def qr_to_install_code(qr_code: str) -> tuple[zigpy.types.EUI64, bytes]: """Try to parse the QR code. if successful, return a tuple of a EUI64 address and install code. diff --git a/homeassistant/components/zha/core/registries.py b/homeassistant/components/zha/core/registries.py index 66a9a70e752..2f9ed57745a 100644 --- a/homeassistant/components/zha/core/registries.py +++ b/homeassistant/components/zha/core/registries.py @@ -1,6 +1,8 @@ """Mapping registries for Zigbee Home Automation.""" +from __future__ import annotations + import collections -from typing import Callable, Dict, List, Set, Tuple, Union +from typing import Callable, Dict import attr import zigpy.profiles.zha @@ -134,19 +136,19 @@ def set_or_callable(value): class MatchRule: """Match a ZHA Entity to a channel name or generic id.""" - channel_names: Union[Callable, Set[str], str] = attr.ib( + channel_names: Callable | set[str] | str = attr.ib( factory=frozenset, converter=set_or_callable ) - generic_ids: Union[Callable, Set[str], str] = attr.ib( + generic_ids: Callable | set[str] | str = attr.ib( factory=frozenset, converter=set_or_callable ) - manufacturers: Union[Callable, Set[str], str] = attr.ib( + manufacturers: Callable | set[str] | str = attr.ib( factory=frozenset, converter=set_or_callable ) - models: Union[Callable, Set[str], str] = attr.ib( + models: Callable | set[str] | str = attr.ib( factory=frozenset, converter=set_or_callable ) - aux_channels: Union[Callable, Set[str], str] = attr.ib( + aux_channels: Callable | set[str] | str = attr.ib( factory=frozenset, converter=set_or_callable ) @@ -176,7 +178,7 @@ class MatchRule: weight += 1 * len(self.aux_channels) return weight - def claim_channels(self, channel_pool: List[ChannelType]) -> List[ChannelType]: + def claim_channels(self, channel_pool: list[ChannelType]) -> list[ChannelType]: """Return a list of channels this rule matches + aux channels.""" claimed = [] if isinstance(self.channel_names, frozenset): @@ -189,15 +191,15 @@ class MatchRule: claimed.extend([ch for ch in channel_pool if ch.name in self.aux_channels]) return claimed - def strict_matched(self, manufacturer: str, model: str, channels: List) -> bool: + def strict_matched(self, manufacturer: str, model: str, channels: list) -> bool: """Return True if this device matches the criteria.""" return all(self._matched(manufacturer, model, channels)) - def loose_matched(self, manufacturer: str, model: str, channels: List) -> bool: + def loose_matched(self, manufacturer: str, model: str, channels: list) -> bool: """Return True if this device matches the criteria.""" return any(self._matched(manufacturer, model, channels)) - def _matched(self, manufacturer: str, model: str, channels: List) -> list: + def _matched(self, manufacturer: str, model: str, channels: list) -> list: """Return a list of field matches.""" if not any(attr.asdict(self).values()): return [False] @@ -245,9 +247,9 @@ class ZHAEntityRegistry: component: str, manufacturer: str, model: str, - channels: List[ChannelType], + channels: list[ChannelType], default: CALLABLE_T = None, - ) -> Tuple[CALLABLE_T, List[ChannelType]]: + ) -> tuple[CALLABLE_T, list[ChannelType]]: """Match a ZHA Channels to a ZHA Entity class.""" matches = self._strict_registry[component] for match in sorted(matches, key=lambda x: x.weight, reverse=True): @@ -264,11 +266,11 @@ class ZHAEntityRegistry: def strict_match( self, component: str, - channel_names: Union[Callable, Set[str], str] = None, - generic_ids: Union[Callable, Set[str], str] = None, - manufacturers: Union[Callable, Set[str], str] = None, - models: Union[Callable, Set[str], str] = None, - aux_channels: Union[Callable, Set[str], str] = None, + channel_names: Callable | set[str] | str = None, + generic_ids: Callable | set[str] | str = None, + manufacturers: Callable | set[str] | str = None, + models: Callable | set[str] | str = None, + aux_channels: Callable | set[str] | str = None, ) -> Callable[[CALLABLE_T], CALLABLE_T]: """Decorate a strict match rule.""" @@ -289,11 +291,11 @@ class ZHAEntityRegistry: def loose_match( self, component: str, - channel_names: Union[Callable, Set[str], str] = None, - generic_ids: Union[Callable, Set[str], str] = None, - manufacturers: Union[Callable, Set[str], str] = None, - models: Union[Callable, Set[str], str] = None, - aux_channels: Union[Callable, Set[str], str] = None, + channel_names: Callable | set[str] | str = None, + generic_ids: Callable | set[str] | str = None, + manufacturers: Callable | set[str] | str = None, + models: Callable | set[str] | str = None, + aux_channels: Callable | set[str] | str = None, ) -> Callable[[CALLABLE_T], CALLABLE_T]: """Decorate a loose match rule.""" diff --git a/homeassistant/components/zha/core/store.py b/homeassistant/components/zha/core/store.py index 2db803258bc..43b41153657 100644 --- a/homeassistant/components/zha/core/store.py +++ b/homeassistant/components/zha/core/store.py @@ -1,8 +1,10 @@ """Data storage helper for ZHA.""" +from __future__ import annotations + from collections import OrderedDict import datetime import time -from typing import MutableMapping, Optional, cast +from typing import MutableMapping, cast import attr @@ -24,9 +26,9 @@ TOMBSTONE_LIFETIME = datetime.timedelta(days=60).total_seconds() class ZhaDeviceEntry: """Zha Device storage Entry.""" - name: Optional[str] = attr.ib(default=None) - ieee: Optional[str] = attr.ib(default=None) - last_seen: Optional[float] = attr.ib(default=None) + name: str | None = attr.ib(default=None) + ieee: str | None = attr.ib(default=None) + last_seen: float | None = attr.ib(default=None) class ZhaStorage: diff --git a/homeassistant/components/zha/cover.py b/homeassistant/components/zha/cover.py index e202def46c5..5530cd3e3f5 100644 --- a/homeassistant/components/zha/cover.py +++ b/homeassistant/components/zha/cover.py @@ -1,8 +1,9 @@ """Support for ZHA covers.""" +from __future__ import annotations + import asyncio import functools import logging -from typing import List, Optional from zigpy.zcl.foundation import Status @@ -180,7 +181,7 @@ class Shade(ZhaEntity, CoverEntity): self, unique_id: str, zha_device: ZhaDeviceType, - channels: List[ChannelType], + channels: list[ChannelType], **kwargs, ): """Initialize the ZHA light.""" @@ -199,12 +200,12 @@ class Shade(ZhaEntity, CoverEntity): return self._position @property - def device_class(self) -> Optional[str]: + def device_class(self) -> str | None: """Return the class of this device, from component DEVICE_CLASSES.""" return DEVICE_CLASS_SHADE @property - def is_closed(self) -> Optional[bool]: + def is_closed(self) -> bool | None: """Return True if shade is closed.""" if self._is_open is None: return None @@ -289,7 +290,7 @@ class KeenVent(Shade): """Keen vent cover.""" @property - def device_class(self) -> Optional[str]: + def device_class(self) -> str | None: """Return the class of this device, from component DEVICE_CLASSES.""" return DEVICE_CLASS_DAMPER diff --git a/homeassistant/components/zha/device_action.py b/homeassistant/components/zha/device_action.py index 46363939190..9d419b16435 100644 --- a/homeassistant/components/zha/device_action.py +++ b/homeassistant/components/zha/device_action.py @@ -1,5 +1,5 @@ """Provides device actions for ZHA devices.""" -from typing import List +from __future__ import annotations import voluptuous as vol @@ -54,7 +54,7 @@ async def async_call_action_from_config( ) -async def async_get_actions(hass: HomeAssistant, device_id: str) -> List[dict]: +async def async_get_actions(hass: HomeAssistant, device_id: str) -> list[dict]: """List device actions.""" try: zha_device = await async_get_zha_device(hass, device_id) diff --git a/homeassistant/components/zha/entity.py b/homeassistant/components/zha/entity.py index 9a573e92aa4..445151899ee 100644 --- a/homeassistant/components/zha/entity.py +++ b/homeassistant/components/zha/entity.py @@ -1,9 +1,10 @@ """Entity for Zigbee Home Automation.""" +from __future__ import annotations import asyncio import functools import logging -from typing import Any, Awaitable, Dict, List, Optional +from typing import Any, Awaitable from homeassistant.const import ATTR_NAME from homeassistant.core import CALLBACK_TYPE, Event, callback @@ -45,9 +46,9 @@ class BaseZhaEntity(LogMixin, entity.Entity): self._should_poll: bool = False self._unique_id: str = unique_id self._state: Any = None - self._extra_state_attributes: Dict[str, Any] = {} + self._extra_state_attributes: dict[str, Any] = {} self._zha_device: ZhaDeviceType = zha_device - self._unsubs: List[CALLABLE_T] = [] + self._unsubs: list[CALLABLE_T] = [] self.remove_future: Awaitable[None] = None @property @@ -66,7 +67,7 @@ class BaseZhaEntity(LogMixin, entity.Entity): return self._zha_device @property - def extra_state_attributes(self) -> Dict[str, Any]: + def extra_state_attributes(self) -> dict[str, Any]: """Return device specific state attributes.""" return self._extra_state_attributes @@ -81,7 +82,7 @@ class BaseZhaEntity(LogMixin, entity.Entity): return self._should_poll @property - def device_info(self) -> Dict[str, Any]: + def device_info(self) -> dict[str, Any]: """Return a device description for device registry.""" zha_device_info = self._zha_device.device_info ieee = zha_device_info["ieee"] @@ -143,7 +144,7 @@ class ZhaEntity(BaseZhaEntity, RestoreEntity): self, unique_id: str, zha_device: ZhaDeviceType, - channels: List[ChannelType], + channels: list[ChannelType], **kwargs, ): """Init ZHA entity.""" @@ -152,7 +153,7 @@ class ZhaEntity(BaseZhaEntity, RestoreEntity): ch_names = [ch.cluster.ep_attribute for ch in channels] ch_names = ", ".join(sorted(ch_names)) self._name: str = f"{zha_device.name} {ieeetail} {ch_names}" - self.cluster_channels: Dict[str, ChannelType] = {} + self.cluster_channels: dict[str, ChannelType] = {} for channel in channels: self.cluster_channels[channel.name] = channel @@ -217,7 +218,7 @@ class ZhaGroupEntity(BaseZhaEntity): """A base class for ZHA group entities.""" def __init__( - self, entity_ids: List[str], unique_id: str, group_id: int, zha_device, **kwargs + self, entity_ids: list[str], unique_id: str, group_id: int, zha_device, **kwargs ) -> None: """Initialize a light group.""" super().__init__(unique_id, zha_device, **kwargs) @@ -225,8 +226,8 @@ class ZhaGroupEntity(BaseZhaEntity): self._group = zha_device.gateway.groups.get(group_id) self._name = f"{self._group.name}_zha_group_0x{group_id:04x}" self._group_id: int = group_id - self._entity_ids: List[str] = entity_ids - self._async_unsub_state_changed: Optional[CALLBACK_TYPE] = None + self._entity_ids: list[str] = entity_ids + self._async_unsub_state_changed: CALLBACK_TYPE | None = None self._handled_group_membership = False @property diff --git a/homeassistant/components/zha/fan.py b/homeassistant/components/zha/fan.py index ed041f8c6c0..3c50261b565 100644 --- a/homeassistant/components/zha/fan.py +++ b/homeassistant/components/zha/fan.py @@ -1,8 +1,9 @@ """Fans on Zigbee Home Automation networks.""" +from __future__ import annotations + from abc import abstractmethod import functools import math -from typing import List, Optional from zigpy.exceptions import ZigbeeException import zigpy.zcl.clusters.hvac as hvac @@ -77,7 +78,7 @@ class BaseFan(FanEntity): """Base representation of a ZHA fan.""" @property - def preset_modes(self) -> List[str]: + def preset_modes(self) -> list[str]: """Return the available preset modes.""" return PRESET_MODES @@ -103,7 +104,7 @@ class BaseFan(FanEntity): """Turn the entity off.""" await self.async_set_percentage(0) - async def async_set_percentage(self, percentage: Optional[int]) -> None: + async def async_set_percentage(self, percentage: int | None) -> None: """Set the speed percenage of the fan.""" fan_mode = math.ceil(percentage_to_ranged_value(SPEED_RANGE, percentage)) await self._async_set_fan_mode(fan_mode) @@ -142,7 +143,7 @@ class ZhaFan(BaseFan, ZhaEntity): ) @property - def percentage(self) -> Optional[int]: + def percentage(self) -> int | None: """Return the current speed percentage.""" if ( self._fan_channel.fan_mode is None @@ -154,7 +155,7 @@ class ZhaFan(BaseFan, ZhaEntity): return ranged_value_to_percentage(SPEED_RANGE, self._fan_channel.fan_mode) @property - def preset_mode(self) -> Optional[str]: + def preset_mode(self) -> str | None: """Return the current preset mode.""" return PRESET_MODES_TO_NAME.get(self._fan_channel.fan_mode) @@ -174,7 +175,7 @@ class FanGroup(BaseFan, ZhaGroupEntity): """Representation of a fan group.""" def __init__( - self, entity_ids: List[str], unique_id: str, group_id: int, zha_device, **kwargs + self, entity_ids: list[str], unique_id: str, group_id: int, zha_device, **kwargs ) -> None: """Initialize a fan group.""" super().__init__(entity_ids, unique_id, group_id, zha_device, **kwargs) @@ -185,12 +186,12 @@ class FanGroup(BaseFan, ZhaGroupEntity): self._preset_mode = None @property - def percentage(self) -> Optional[int]: + def percentage(self) -> int | None: """Return the current speed percentage.""" return self._percentage @property - def preset_mode(self) -> Optional[str]: + def preset_mode(self) -> str | None: """Return the current preset mode.""" return self._preset_mode @@ -205,11 +206,11 @@ class FanGroup(BaseFan, ZhaGroupEntity): async def async_update(self): """Attempt to retrieve on off state from the fan.""" all_states = [self.hass.states.get(x) for x in self._entity_ids] - states: List[State] = list(filter(None, all_states)) - percentage_states: List[State] = [ + states: list[State] = list(filter(None, all_states)) + percentage_states: list[State] = [ state for state in states if state.attributes.get(ATTR_PERCENTAGE) ] - preset_mode_states: List[State] = [ + preset_mode_states: list[State] = [ state for state in states if state.attributes.get(ATTR_PRESET_MODE) ] self._available = any(state.state != STATE_UNAVAILABLE for state in states) diff --git a/homeassistant/components/zha/light.py b/homeassistant/components/zha/light.py index f81b931a49d..72807458d26 100644 --- a/homeassistant/components/zha/light.py +++ b/homeassistant/components/zha/light.py @@ -1,4 +1,6 @@ """Lights on Zigbee Home Automation networks.""" +from __future__ import annotations + from collections import Counter from datetime import timedelta import enum @@ -6,7 +8,7 @@ import functools import itertools import logging import random -from typing import Any, Dict, List, Optional, Tuple +from typing import Any from zigpy.zcl.clusters.general import Identify, LevelControl, OnOff from zigpy.zcl.clusters.lighting import Color @@ -122,15 +124,15 @@ class BaseLight(LogMixin, light.LightEntity): """Initialize the light.""" super().__init__(*args, **kwargs) self._available: bool = False - self._brightness: Optional[int] = None - self._off_brightness: Optional[int] = None - self._hs_color: Optional[Tuple[float, float]] = None - self._color_temp: Optional[int] = None - self._min_mireds: Optional[int] = 153 - self._max_mireds: Optional[int] = 500 - self._white_value: Optional[int] = None - self._effect_list: Optional[List[str]] = None - self._effect: Optional[str] = None + self._brightness: int | None = None + self._off_brightness: int | None = None + self._hs_color: tuple[float, float] | None = None + self._color_temp: int | None = None + self._min_mireds: int | None = 153 + self._max_mireds: int | None = 500 + self._white_value: int | None = None + self._effect_list: list[str] | None = None + self._effect: str | None = None self._supported_features: int = 0 self._state: bool = False self._on_off_channel = None @@ -139,7 +141,7 @@ class BaseLight(LogMixin, light.LightEntity): self._identify_channel = None @property - def extra_state_attributes(self) -> Dict[str, Any]: + def extra_state_attributes(self) -> dict[str, Any]: """Return state attributes.""" attributes = {"off_brightness": self._off_brightness} return attributes @@ -348,8 +350,8 @@ class Light(BaseLight, ZhaEntity): self._color_channel = self.cluster_channels.get(CHANNEL_COLOR) self._identify_channel = self.zha_device.channels.identify_ch if self._color_channel: - self._min_mireds: Optional[int] = self._color_channel.min_mireds - self._max_mireds: Optional[int] = self._color_channel.max_mireds + self._min_mireds: int | None = self._color_channel.min_mireds + self._max_mireds: int | None = self._color_channel.max_mireds self._cancel_refresh_handle = None effect_list = [] @@ -532,7 +534,7 @@ class LightGroup(BaseLight, ZhaGroupEntity): """Representation of a light group.""" def __init__( - self, entity_ids: List[str], unique_id: str, group_id: int, zha_device, **kwargs + self, entity_ids: list[str], unique_id: str, group_id: int, zha_device, **kwargs ) -> None: """Initialize a light group.""" super().__init__(entity_ids, unique_id, group_id, zha_device, **kwargs) @@ -569,7 +571,7 @@ class LightGroup(BaseLight, ZhaGroupEntity): async def async_update(self) -> None: """Query all members and determine the light group state.""" all_states = [self.hass.states.get(x) for x in self._entity_ids] - states: List[State] = list(filter(None, all_states)) + states: list[State] = list(filter(None, all_states)) on_states = [state for state in states if state.state == STATE_ON] self._state = len(on_states) > 0 diff --git a/homeassistant/components/zha/sensor.py b/homeassistant/components/zha/sensor.py index 926fd4555e1..425a41a1340 100644 --- a/homeassistant/components/zha/sensor.py +++ b/homeassistant/components/zha/sensor.py @@ -1,7 +1,9 @@ """Sensors on Zigbee Home Automation networks.""" +from __future__ import annotations + import functools import numbers -from typing import Any, Callable, Dict, List, Optional, Union +from typing import Any, Callable from homeassistant.components.sensor import ( DEVICE_CLASS_BATTERY, @@ -90,18 +92,18 @@ async def async_setup_entry( class Sensor(ZhaEntity): """Base ZHA sensor.""" - SENSOR_ATTR: Optional[Union[int, str]] = None + SENSOR_ATTR: int | str | None = None _decimals: int = 1 - _device_class: Optional[str] = None + _device_class: str | None = None _divisor: int = 1 _multiplier: int = 1 - _unit: Optional[str] = None + _unit: str | None = None def __init__( self, unique_id: str, zha_device: ZhaDeviceType, - channels: List[ChannelType], + channels: list[ChannelType], **kwargs, ): """Init this sensor.""" @@ -121,7 +123,7 @@ class Sensor(ZhaEntity): return self._device_class @property - def unit_of_measurement(self) -> Optional[str]: + def unit_of_measurement(self) -> str | None: """Return the unit of measurement of this entity.""" return self._unit @@ -139,7 +141,7 @@ class Sensor(ZhaEntity): """Handle state update from channel.""" self.async_write_ha_state() - def formatter(self, value: int) -> Union[int, float]: + def formatter(self, value: int) -> int | float: """Numeric pass-through formatter.""" if self._decimals > 0: return round( @@ -178,7 +180,7 @@ class Battery(Sensor): return value @property - def extra_state_attributes(self) -> Dict[str, Any]: + def extra_state_attributes(self) -> dict[str, Any]: """Return device state attrs for battery sensors.""" state_attrs = {} battery_size = self._channel.cluster.get("battery_size") @@ -208,7 +210,7 @@ class ElectricalMeasurement(Sensor): """Return True if HA needs to poll for state changes.""" return True - def formatter(self, value: int) -> Union[int, float]: + def formatter(self, value: int) -> int | float: """Return 'normalized' value.""" value = value * self._channel.multiplier / self._channel.divisor if value < 100 and self._channel.divisor > 1: @@ -254,7 +256,7 @@ class SmartEnergyMetering(Sensor): SENSOR_ATTR = "instantaneous_demand" _device_class = DEVICE_CLASS_POWER - def formatter(self, value: int) -> Union[int, float]: + def formatter(self, value: int) -> int | float: """Pass through channel formatter.""" return self._channel.formatter_function(value) diff --git a/homeassistant/components/zha/switch.py b/homeassistant/components/zha/switch.py index 60db9ec0a08..75254f631b9 100644 --- a/homeassistant/components/zha/switch.py +++ b/homeassistant/components/zha/switch.py @@ -1,6 +1,8 @@ """Switches on Zigbee Home Automation networks.""" +from __future__ import annotations + import functools -from typing import Any, List +from typing import Any from zigpy.zcl.clusters.general import OnOff from zigpy.zcl.foundation import Status @@ -113,7 +115,7 @@ class SwitchGroup(BaseSwitch, ZhaGroupEntity): """Representation of a switch group.""" def __init__( - self, entity_ids: List[str], unique_id: str, group_id: int, zha_device, **kwargs + self, entity_ids: list[str], unique_id: str, group_id: int, zha_device, **kwargs ) -> None: """Initialize a switch group.""" super().__init__(entity_ids, unique_id, group_id, zha_device, **kwargs) @@ -124,7 +126,7 @@ class SwitchGroup(BaseSwitch, ZhaGroupEntity): async def async_update(self) -> None: """Query all members and determine the light group state.""" all_states = [self.hass.states.get(x) for x in self._entity_ids] - states: List[State] = list(filter(None, all_states)) + states: list[State] = list(filter(None, all_states)) on_states = [state for state in states if state.state == STATE_ON] self._state = len(on_states) > 0 diff --git a/homeassistant/components/zone/__init__.py b/homeassistant/components/zone/__init__.py index e1d48cbe1ff..7e329127d03 100644 --- a/homeassistant/components/zone/__init__.py +++ b/homeassistant/components/zone/__init__.py @@ -2,7 +2,7 @@ from __future__ import annotations import logging -from typing import Any, Dict, Optional, cast +from typing import Any, Dict, cast import voluptuous as vol @@ -92,7 +92,7 @@ STORAGE_VERSION = 1 @bind_hass def async_active_zone( hass: HomeAssistant, latitude: float, longitude: float, radius: int = 0 -) -> Optional[State]: +) -> State | None: """Find the active zone for given latitude, longitude. This method must be run in the event loop. @@ -161,22 +161,22 @@ class ZoneStorageCollection(collection.StorageCollection): CREATE_SCHEMA = vol.Schema(CREATE_FIELDS) UPDATE_SCHEMA = vol.Schema(UPDATE_FIELDS) - async def _process_create_data(self, data: Dict) -> Dict: + async def _process_create_data(self, data: dict) -> dict: """Validate the config is valid.""" return cast(Dict, self.CREATE_SCHEMA(data)) @callback - def _get_suggested_id(self, info: Dict) -> str: + def _get_suggested_id(self, info: dict) -> str: """Suggest an ID based on the config.""" return cast(str, info[CONF_NAME]) - async def _update_data(self, data: dict, update_data: Dict) -> Dict: + async def _update_data(self, data: dict, update_data: dict) -> dict: """Return a new updated data object.""" update_data = self.UPDATE_SCHEMA(update_data) return {**data, **update_data} -async def async_setup(hass: HomeAssistant, config: Dict) -> bool: +async def async_setup(hass: HomeAssistant, config: dict) -> bool: """Set up configured zones as well as Home Assistant zone if necessary.""" component = entity_component.EntityComponent(_LOGGER, DOMAIN, hass) id_manager = collection.IDManager() @@ -240,7 +240,7 @@ async def async_setup(hass: HomeAssistant, config: Dict) -> bool: @callback -def _home_conf(hass: HomeAssistant) -> Dict: +def _home_conf(hass: HomeAssistant) -> dict: """Return the home zone config.""" return { CONF_NAME: hass.config.location_name, @@ -279,15 +279,15 @@ async def async_unload_entry( class Zone(entity.Entity): """Representation of a Zone.""" - def __init__(self, config: Dict): + def __init__(self, config: dict): """Initialize the zone.""" self._config = config self.editable = True - self._attrs: Optional[Dict] = None + self._attrs: dict | None = None self._generate_attrs() @classmethod - def from_yaml(cls, config: Dict) -> Zone: + def from_yaml(cls, config: dict) -> Zone: """Return entity instance initialized from yaml storage.""" zone = cls(config) zone.editable = False @@ -305,17 +305,17 @@ class Zone(entity.Entity): return cast(str, self._config[CONF_NAME]) @property - def unique_id(self) -> Optional[str]: + def unique_id(self) -> str | None: """Return unique ID.""" return self._config.get(CONF_ID) @property - def icon(self) -> Optional[str]: + def icon(self) -> str | None: """Return the icon if any.""" return self._config.get(CONF_ICON) @property - def state_attributes(self) -> Optional[Dict]: + def state_attributes(self) -> dict | None: """Return the state attributes of the zone.""" return self._attrs @@ -324,7 +324,7 @@ class Zone(entity.Entity): """Zone does not poll.""" return False - async def async_update_config(self, config: Dict) -> None: + async def async_update_config(self, config: dict) -> None: """Handle when the config is updated.""" if self._config == config: return diff --git a/homeassistant/components/zwave/climate.py b/homeassistant/components/zwave/climate.py index 20eb52c2c19..ac8c47af9a3 100644 --- a/homeassistant/components/zwave/climate.py +++ b/homeassistant/components/zwave/climate.py @@ -1,7 +1,8 @@ """Support for Z-Wave climate devices.""" # Because we do not compile openzwave on CI +from __future__ import annotations + import logging -from typing import Optional, Tuple from homeassistant.components.climate import ClimateEntity from homeassistant.components.climate.const import ( @@ -191,7 +192,7 @@ class ZWaveClimateBase(ZWaveDeviceEntity, ClimateEntity): """Return thermostat mode Z-Wave value.""" raise NotImplementedError() - def _current_mode_setpoints(self) -> Tuple: + def _current_mode_setpoints(self) -> tuple: """Return a tuple of current setpoint Z-Wave value(s).""" raise NotImplementedError() @@ -483,12 +484,12 @@ class ZWaveClimateBase(ZWaveDeviceEntity, ClimateEntity): return self._target_temperature @property - def target_temperature_low(self) -> Optional[float]: + def target_temperature_low(self) -> float | None: """Return the lowbound target temperature we try to reach.""" return self._target_temperature_range[0] @property - def target_temperature_high(self) -> Optional[float]: + def target_temperature_high(self) -> float | None: """Return the highbound target temperature we try to reach.""" return self._target_temperature_range[1] @@ -590,7 +591,7 @@ class ZWaveClimateSingleSetpoint(ZWaveClimateBase): """Return thermostat mode Z-Wave value.""" return self.values.mode - def _current_mode_setpoints(self) -> Tuple: + def _current_mode_setpoints(self) -> tuple: """Return a tuple of current setpoint Z-Wave value(s).""" return (self.values.primary,) @@ -606,7 +607,7 @@ class ZWaveClimateMultipleSetpoint(ZWaveClimateBase): """Return thermostat mode Z-Wave value.""" return self.values.primary - def _current_mode_setpoints(self) -> Tuple: + def _current_mode_setpoints(self) -> tuple: """Return a tuple of current setpoint Z-Wave value(s).""" current_mode = str(self.values.primary.data).lower() setpoints_names = MODE_SETPOINT_MAPPINGS.get(current_mode, ()) diff --git a/homeassistant/components/zwave_js/__init__.py b/homeassistant/components/zwave_js/__init__.py index 2d4db960886..64dcf9614b2 100644 --- a/homeassistant/components/zwave_js/__init__.py +++ b/homeassistant/components/zwave_js/__init__.py @@ -1,6 +1,8 @@ """The Z-Wave JS integration.""" +from __future__ import annotations + import asyncio -from typing import Callable, List +from typing import Callable from async_timeout import timeout from zwave_js_server.client import Client as ZwaveClient @@ -226,7 +228,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: entry_hass_data[DATA_CONNECT_FAILED_LOGGED] = False entry_hass_data[DATA_INVALID_SERVER_VERSION_LOGGED] = False - unsubscribe_callbacks: List[Callable] = [] + unsubscribe_callbacks: list[Callable] = [] entry_hass_data[DATA_CLIENT] = client entry_hass_data[DATA_UNSUBSCRIBE] = unsubscribe_callbacks diff --git a/homeassistant/components/zwave_js/addon.py b/homeassistant/components/zwave_js/addon.py index 818e46a34aa..0c2fdb17944 100644 --- a/homeassistant/components/zwave_js/addon.py +++ b/homeassistant/components/zwave_js/addon.py @@ -3,7 +3,7 @@ from __future__ import annotations import asyncio from functools import partial -from typing import Any, Callable, Optional, TypeVar, cast +from typing import Any, Callable, TypeVar, cast from homeassistant.components.hassio import ( async_create_snapshot, @@ -66,9 +66,9 @@ class AddonManager: def __init__(self, hass: HomeAssistant) -> None: """Set up the add-on manager.""" self._hass = hass - self._install_task: Optional[asyncio.Task] = None - self._start_task: Optional[asyncio.Task] = None - self._update_task: Optional[asyncio.Task] = None + self._install_task: asyncio.Task | None = None + self._start_task: asyncio.Task | None = None + self._update_task: asyncio.Task | None = None def task_in_progress(self) -> bool: """Return True if any of the add-on tasks are in progress.""" diff --git a/homeassistant/components/zwave_js/api.py b/homeassistant/components/zwave_js/api.py index 055115db7b9..087b94d0060 100644 --- a/homeassistant/components/zwave_js/api.py +++ b/homeassistant/components/zwave_js/api.py @@ -1,7 +1,8 @@ """Websocket API for Z-Wave JS.""" +from __future__ import annotations + import dataclasses import json -from typing import Dict from aiohttp import hdrs, web, web_exceptions import voluptuous as vol @@ -399,7 +400,7 @@ def convert_log_level_to_enum(value: str) -> LogLevel: return LogLevel[value.upper()] -def filename_is_present_if_logging_to_file(obj: Dict) -> Dict: +def filename_is_present_if_logging_to_file(obj: dict) -> dict: """Validate that filename is provided if log_to_file is True.""" if obj.get(LOG_TO_FILE, False) and FILENAME not in obj: raise vol.Invalid("`filename` must be provided if logging to file") diff --git a/homeassistant/components/zwave_js/binary_sensor.py b/homeassistant/components/zwave_js/binary_sensor.py index 8d266c83f22..47c374405b1 100644 --- a/homeassistant/components/zwave_js/binary_sensor.py +++ b/homeassistant/components/zwave_js/binary_sensor.py @@ -1,7 +1,8 @@ """Representation of Z-Wave binary sensors.""" +from __future__ import annotations import logging -from typing import Callable, List, Optional, TypedDict +from typing import Callable, TypedDict from zwave_js_server.client import Client as ZwaveClient from zwave_js_server.const import CommandClass @@ -56,14 +57,14 @@ class NotificationSensorMapping(TypedDict, total=False): """Represent a notification sensor mapping dict type.""" type: int # required - states: List[str] + states: list[str] device_class: str enabled: bool # Mappings for Notification sensors # https://github.com/zwave-js/node-zwave-js/blob/master/packages/config/config/notifications.json -NOTIFICATION_SENSOR_MAPPINGS: List[NotificationSensorMapping] = [ +NOTIFICATION_SENSOR_MAPPINGS: list[NotificationSensorMapping] = [ { # NotificationType 1: Smoke Alarm - State Id's 1 and 2 - Smoke detected "type": NOTIFICATION_SMOKE_ALARM, @@ -201,13 +202,13 @@ class PropertySensorMapping(TypedDict, total=False): """Represent a property sensor mapping dict type.""" property_name: str # required - on_states: List[str] # required + on_states: list[str] # required device_class: str enabled: bool # Mappings for property sensors -PROPERTY_SENSOR_MAPPINGS: List[PropertySensorMapping] = [ +PROPERTY_SENSOR_MAPPINGS: list[PropertySensorMapping] = [ { "property_name": PROPERTY_DOOR_STATUS, "on_states": ["open"], @@ -226,7 +227,7 @@ async def async_setup_entry( @callback def async_add_binary_sensor(info: ZwaveDiscoveryInfo) -> None: """Add Z-Wave Binary Sensor.""" - entities: List[BinarySensorEntity] = [] + entities: list[BinarySensorEntity] = [] if info.platform_hint == "notification": # Get all sensors from Notification CC states @@ -268,14 +269,14 @@ class ZWaveBooleanBinarySensor(ZWaveBaseEntity, BinarySensorEntity): self._name = self.generate_name(include_value_name=True) @property - def is_on(self) -> Optional[bool]: + def is_on(self) -> bool | None: """Return if the sensor is on or off.""" if self.info.primary_value.value is None: return None return bool(self.info.primary_value.value) @property - def device_class(self) -> Optional[str]: + def device_class(self) -> str | None: """Return device class.""" if self.info.primary_value.command_class == CommandClass.BATTERY: return DEVICE_CLASS_BATTERY @@ -314,14 +315,14 @@ class ZWaveNotificationBinarySensor(ZWaveBaseEntity, BinarySensorEntity): self._mapping_info = self._get_sensor_mapping() @property - def is_on(self) -> Optional[bool]: + def is_on(self) -> bool | None: """Return if the sensor is on or off.""" if self.info.primary_value.value is None: return None return int(self.info.primary_value.value) == int(self.state_key) @property - def device_class(self) -> Optional[str]: + def device_class(self) -> str | None: """Return device class.""" return self._mapping_info.get("device_class") @@ -365,14 +366,14 @@ class ZWavePropertyBinarySensor(ZWaveBaseEntity, BinarySensorEntity): self._name = self.generate_name(include_value_name=True) @property - def is_on(self) -> Optional[bool]: + def is_on(self) -> bool | None: """Return if the sensor is on or off.""" if self.info.primary_value.value is None: return None return self.info.primary_value.value in self._mapping_info["on_states"] @property - def device_class(self) -> Optional[str]: + def device_class(self) -> str | None: """Return device class.""" return self._mapping_info.get("device_class") diff --git a/homeassistant/components/zwave_js/climate.py b/homeassistant/components/zwave_js/climate.py index 3f6b26f536f..78d76423378 100644 --- a/homeassistant/components/zwave_js/climate.py +++ b/homeassistant/components/zwave_js/climate.py @@ -1,5 +1,7 @@ """Representation of Z-Wave thermostats.""" -from typing import Any, Callable, Dict, List, Optional, cast +from __future__ import annotations + +from typing import Any, Callable, cast from zwave_js_server.client import Client as ZwaveClient from zwave_js_server.const import ( @@ -50,7 +52,7 @@ from .entity import ZWaveBaseEntity # Map Z-Wave HVAC Mode to Home Assistant value # Note: We treat "auto" as "heat_cool" as most Z-Wave devices # report auto_changeover as auto without schedule support. -ZW_HVAC_MODE_MAP: Dict[int, str] = { +ZW_HVAC_MODE_MAP: dict[int, str] = { ThermostatMode.OFF: HVAC_MODE_OFF, ThermostatMode.HEAT: HVAC_MODE_HEAT, ThermostatMode.COOL: HVAC_MODE_COOL, @@ -67,7 +69,7 @@ ZW_HVAC_MODE_MAP: Dict[int, str] = { ThermostatMode.FULL_POWER: HVAC_MODE_HEAT, } -HVAC_CURRENT_MAP: Dict[int, str] = { +HVAC_CURRENT_MAP: dict[int, str] = { ThermostatOperatingState.IDLE: CURRENT_HVAC_IDLE, ThermostatOperatingState.PENDING_HEAT: CURRENT_HVAC_IDLE, ThermostatOperatingState.HEATING: CURRENT_HVAC_HEAT, @@ -94,7 +96,7 @@ async def async_setup_entry( @callback def async_add_climate(info: ZwaveDiscoveryInfo) -> None: """Add Z-Wave Climate.""" - entities: List[ZWaveBaseEntity] = [] + entities: list[ZWaveBaseEntity] = [] entities.append(ZWaveClimate(config_entry, client, info)) async_add_entities(entities) @@ -116,14 +118,14 @@ class ZWaveClimate(ZWaveBaseEntity, ClimateEntity): ) -> None: """Initialize lock.""" super().__init__(config_entry, client, info) - self._hvac_modes: Dict[str, Optional[int]] = {} - self._hvac_presets: Dict[str, Optional[int]] = {} - self._unit_value: Optional[ZwaveValue] = None + self._hvac_modes: dict[str, int | None] = {} + self._hvac_presets: dict[str, int | None] = {} + self._unit_value: ZwaveValue | None = None self._current_mode = self.get_zwave_value( THERMOSTAT_MODE_PROPERTY, command_class=CommandClass.THERMOSTAT_MODE ) - self._setpoint_values: Dict[ThermostatSetpointType, ZwaveValue] = {} + self._setpoint_values: dict[ThermostatSetpointType, ZwaveValue] = {} for enum in ThermostatSetpointType: self._setpoint_values[enum] = self.get_zwave_value( THERMOSTAT_SETPOINT_PROPERTY, @@ -184,8 +186,8 @@ class ZWaveClimate(ZWaveBaseEntity, ClimateEntity): def _set_modes_and_presets(self) -> None: """Convert Z-Wave Thermostat modes into Home Assistant modes and presets.""" - all_modes: Dict[str, Optional[int]] = {} - all_presets: Dict[str, Optional[int]] = {PRESET_NONE: None} + all_modes: dict[str, int | None] = {} + all_presets: dict[str, int | None] = {PRESET_NONE: None} # Z-Wave uses one list for both modes and presets. # Iterate over all Z-Wave ThermostatModes and extract the hvac modes and presets. @@ -208,7 +210,7 @@ class ZWaveClimate(ZWaveBaseEntity, ClimateEntity): self._hvac_presets = all_presets @property - def _current_mode_setpoint_enums(self) -> List[Optional[ThermostatSetpointType]]: + def _current_mode_setpoint_enums(self) -> list[ThermostatSetpointType | None]: """Return the list of enums that are relevant to the current thermostat mode.""" if self._current_mode is None: # Thermostat(valve) with no support for setting a mode is considered heating-only @@ -238,12 +240,12 @@ class ZWaveClimate(ZWaveBaseEntity, ClimateEntity): return ZW_HVAC_MODE_MAP.get(int(self._current_mode.value), HVAC_MODE_HEAT_COOL) @property - def hvac_modes(self) -> List[str]: + def hvac_modes(self) -> list[str]: """Return the list of available hvac operation modes.""" return list(self._hvac_modes) @property - def hvac_action(self) -> Optional[str]: + def hvac_action(self) -> str | None: """Return the current running hvac operation if supported.""" if not self._operating_state: return None @@ -253,17 +255,17 @@ class ZWaveClimate(ZWaveBaseEntity, ClimateEntity): return HVAC_CURRENT_MAP.get(int(self._operating_state.value)) @property - def current_humidity(self) -> Optional[int]: + def current_humidity(self) -> int | None: """Return the current humidity level.""" return self._current_humidity.value if self._current_humidity else None @property - def current_temperature(self) -> Optional[float]: + def current_temperature(self) -> float | None: """Return the current temperature.""" return self._current_temp.value if self._current_temp else None @property - def target_temperature(self) -> Optional[float]: + def target_temperature(self) -> float | None: """Return the temperature we try to reach.""" if self._current_mode and self._current_mode.value is None: # guard missing value @@ -275,7 +277,7 @@ class ZWaveClimate(ZWaveBaseEntity, ClimateEntity): return temp.value if temp else None @property - def target_temperature_high(self) -> Optional[float]: + def target_temperature_high(self) -> float | None: """Return the highbound target temperature we try to reach.""" if self._current_mode and self._current_mode.value is None: # guard missing value @@ -287,7 +289,7 @@ class ZWaveClimate(ZWaveBaseEntity, ClimateEntity): return temp.value if temp else None @property - def target_temperature_low(self) -> Optional[float]: + def target_temperature_low(self) -> float | None: """Return the lowbound target temperature we try to reach.""" if self._current_mode and self._current_mode.value is None: # guard missing value @@ -297,7 +299,7 @@ class ZWaveClimate(ZWaveBaseEntity, ClimateEntity): return None @property - def preset_mode(self) -> Optional[str]: + def preset_mode(self) -> str | None: """Return the current preset mode, e.g., home, away, temp.""" if self._current_mode and self._current_mode.value is None: # guard missing value @@ -310,12 +312,12 @@ class ZWaveClimate(ZWaveBaseEntity, ClimateEntity): return PRESET_NONE @property - def preset_modes(self) -> Optional[List[str]]: + def preset_modes(self) -> list[str] | None: """Return a list of available preset modes.""" return list(self._hvac_presets) @property - def fan_mode(self) -> Optional[str]: + def fan_mode(self) -> str | None: """Return the fan setting.""" if ( self._fan_mode @@ -326,14 +328,14 @@ class ZWaveClimate(ZWaveBaseEntity, ClimateEntity): return None @property - def fan_modes(self) -> Optional[List[str]]: + def fan_modes(self) -> list[str] | None: """Return the list of available fan modes.""" if self._fan_mode and self._fan_mode.metadata.states: return list(self._fan_mode.metadata.states.values()) return None @property - def extra_state_attributes(self) -> Optional[Dict[str, str]]: + def extra_state_attributes(self) -> dict[str, str] | None: """Return the optional state attributes.""" if ( self._fan_state @@ -373,7 +375,7 @@ class ZWaveClimate(ZWaveBaseEntity, ClimateEntity): async def async_set_temperature(self, **kwargs: Any) -> None: """Set new target temperature.""" - hvac_mode: Optional[str] = kwargs.get(ATTR_HVAC_MODE) + hvac_mode: str | None = kwargs.get(ATTR_HVAC_MODE) if hvac_mode is not None: await self.async_set_hvac_mode(hvac_mode) @@ -381,7 +383,7 @@ class ZWaveClimate(ZWaveBaseEntity, ClimateEntity): setpoint: ZwaveValue = self._setpoint_value( self._current_mode_setpoint_enums[0] ) - target_temp: Optional[float] = kwargs.get(ATTR_TEMPERATURE) + target_temp: float | None = kwargs.get(ATTR_TEMPERATURE) if target_temp is not None: await self.info.node.async_set_value(setpoint, target_temp) elif len(self._current_mode_setpoint_enums) == 2: @@ -391,8 +393,8 @@ class ZWaveClimate(ZWaveBaseEntity, ClimateEntity): setpoint_high: ZwaveValue = self._setpoint_value( self._current_mode_setpoint_enums[1] ) - target_temp_low: Optional[float] = kwargs.get(ATTR_TARGET_TEMP_LOW) - target_temp_high: Optional[float] = kwargs.get(ATTR_TARGET_TEMP_HIGH) + target_temp_low: float | None = kwargs.get(ATTR_TARGET_TEMP_LOW) + target_temp_high: float | None = kwargs.get(ATTR_TARGET_TEMP_HIGH) if target_temp_low is not None: await self.info.node.async_set_value(setpoint_low, target_temp_low) if target_temp_high is not None: diff --git a/homeassistant/components/zwave_js/config_flow.py b/homeassistant/components/zwave_js/config_flow.py index 8ba2909cf20..75e948004cb 100644 --- a/homeassistant/components/zwave_js/config_flow.py +++ b/homeassistant/components/zwave_js/config_flow.py @@ -1,7 +1,9 @@ """Config flow for Z-Wave JS integration.""" +from __future__ import annotations + import asyncio import logging -from typing import Any, Dict, Optional, cast +from typing import Any, cast import aiohttp from async_timeout import timeout @@ -76,18 +78,18 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): def __init__(self) -> None: """Set up flow instance.""" - self.network_key: Optional[str] = None - self.usb_path: Optional[str] = None + self.network_key: str | None = None + self.usb_path: str | None = None self.use_addon = False - self.ws_address: Optional[str] = None + self.ws_address: str | None = None # If we install the add-on we should uninstall it on entry remove. self.integration_created_addon = False - self.install_task: Optional[asyncio.Task] = None - self.start_task: Optional[asyncio.Task] = None + self.install_task: asyncio.Task | None = None + self.start_task: asyncio.Task | None = None async def async_step_user( - self, user_input: Optional[Dict[str, Any]] = None - ) -> Dict[str, Any]: + self, user_input: dict[str, Any] | None = None + ) -> dict[str, Any]: """Handle the initial step.""" if is_hassio(self.hass): # type: ignore # no-untyped-call return await self.async_step_on_supervisor() @@ -95,8 +97,8 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): return await self.async_step_manual() async def async_step_manual( - self, user_input: Optional[Dict[str, Any]] = None - ) -> Dict[str, Any]: + self, user_input: dict[str, Any] | None = None + ) -> dict[str, Any]: """Handle a manual configuration.""" if user_input is None: return self.async_show_form( @@ -133,8 +135,8 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): ) async def async_step_hassio( # type: ignore # override - self, discovery_info: Dict[str, Any] - ) -> Dict[str, Any]: + self, discovery_info: dict[str, Any] + ) -> dict[str, Any]: """Receive configuration from add-on discovery info. This flow is triggered by the Z-Wave JS add-on. @@ -151,8 +153,8 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): return await self.async_step_hassio_confirm() async def async_step_hassio_confirm( - self, user_input: Optional[Dict[str, Any]] = None - ) -> Dict[str, Any]: + self, user_input: dict[str, Any] | None = None + ) -> dict[str, Any]: """Confirm the add-on discovery.""" if user_input is not None: return await self.async_step_on_supervisor( @@ -162,7 +164,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): return self.async_show_form(step_id="hassio_confirm") @callback - def _async_create_entry_from_vars(self) -> Dict[str, Any]: + def _async_create_entry_from_vars(self) -> dict[str, Any]: """Return a config entry for the flow.""" return self.async_create_entry( title=TITLE, @@ -176,8 +178,8 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): ) async def async_step_on_supervisor( - self, user_input: Optional[Dict[str, Any]] = None - ) -> Dict[str, Any]: + self, user_input: dict[str, Any] | None = None + ) -> dict[str, Any]: """Handle logic when on Supervisor host.""" if user_input is None: return self.async_show_form( @@ -200,8 +202,8 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): return await self.async_step_install_addon() async def async_step_install_addon( - self, user_input: Optional[Dict[str, Any]] = None - ) -> Dict[str, Any]: + self, user_input: dict[str, Any] | None = None + ) -> dict[str, Any]: """Install Z-Wave JS add-on.""" if not self.install_task: self.install_task = self.hass.async_create_task(self._async_install_addon()) @@ -220,18 +222,18 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): return self.async_show_progress_done(next_step_id="configure_addon") async def async_step_install_failed( - self, user_input: Optional[Dict[str, Any]] = None - ) -> Dict[str, Any]: + self, user_input: dict[str, Any] | None = None + ) -> dict[str, Any]: """Add-on installation failed.""" return self.async_abort(reason="addon_install_failed") async def async_step_configure_addon( - self, user_input: Optional[Dict[str, Any]] = None - ) -> Dict[str, Any]: + self, user_input: dict[str, Any] | None = None + ) -> dict[str, Any]: """Ask for config for Z-Wave JS add-on.""" addon_config = await self._async_get_addon_config() - errors: Dict[str, str] = {} + errors: dict[str, str] = {} if user_input is not None: self.network_key = user_input[CONF_NETWORK_KEY] @@ -262,8 +264,8 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): ) async def async_step_start_addon( - self, user_input: Optional[Dict[str, Any]] = None - ) -> Dict[str, Any]: + self, user_input: dict[str, Any] | None = None + ) -> dict[str, Any]: """Start Z-Wave JS add-on.""" if not self.start_task: self.start_task = self.hass.async_create_task(self._async_start_addon()) @@ -280,8 +282,8 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): return self.async_show_progress_done(next_step_id="finish_addon_setup") async def async_step_start_failed( - self, user_input: Optional[Dict[str, Any]] = None - ) -> Dict[str, Any]: + self, user_input: dict[str, Any] | None = None + ) -> dict[str, Any]: """Add-on start failed.""" return self.async_abort(reason="addon_start_failed") @@ -317,8 +319,8 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): ) async def async_step_finish_addon_setup( - self, user_input: Optional[Dict[str, Any]] = None - ) -> Dict[str, Any]: + self, user_input: dict[str, Any] | None = None + ) -> dict[str, Any]: """Prepare info needed to complete the config entry. Get add-on discovery info and server version info. diff --git a/homeassistant/components/zwave_js/cover.py b/homeassistant/components/zwave_js/cover.py index ff77bdb408d..25c69335ed1 100644 --- a/homeassistant/components/zwave_js/cover.py +++ b/homeassistant/components/zwave_js/cover.py @@ -1,6 +1,8 @@ """Support for Z-Wave cover devices.""" +from __future__ import annotations + import logging -from typing import Any, Callable, List, Optional +from typing import Any, Callable from zwave_js_server.client import Client as ZwaveClient from zwave_js_server.model.value import Value as ZwaveValue @@ -42,7 +44,7 @@ async def async_setup_entry( @callback def async_add_cover(info: ZwaveDiscoveryInfo) -> None: """Add Z-Wave cover.""" - entities: List[ZWaveBaseEntity] = [] + entities: list[ZWaveBaseEntity] = [] if info.platform_hint == "motorized_barrier": entities.append(ZwaveMotorizedBarrier(config_entry, client, info)) else: @@ -72,7 +74,7 @@ class ZWaveCover(ZWaveBaseEntity, CoverEntity): """Representation of a Z-Wave Cover device.""" @property - def is_closed(self) -> Optional[bool]: + def is_closed(self) -> bool | None: """Return true if cover is closed.""" if self.info.primary_value.value is None: # guard missing value @@ -80,7 +82,7 @@ class ZWaveCover(ZWaveBaseEntity, CoverEntity): return bool(self.info.primary_value.value == 0) @property - def current_cover_position(self) -> Optional[int]: + def current_cover_position(self) -> int | None: """Return the current position of cover where 0 means closed and 100 is fully open.""" if self.info.primary_value.value is None: # guard missing value @@ -130,31 +132,31 @@ class ZwaveMotorizedBarrier(ZWaveBaseEntity, CoverEntity): ) @property - def supported_features(self) -> Optional[int]: + def supported_features(self) -> int | None: """Flag supported features.""" return SUPPORT_OPEN | SUPPORT_CLOSE @property - def device_class(self) -> Optional[str]: + def device_class(self) -> str | None: """Return the class of this device, from component DEVICE_CLASSES.""" return DEVICE_CLASS_GARAGE @property - def is_opening(self) -> Optional[bool]: + def is_opening(self) -> bool | None: """Return if the cover is opening or not.""" if self.info.primary_value.value is None: return None return bool(self.info.primary_value.value == BARRIER_STATE_OPENING) @property - def is_closing(self) -> Optional[bool]: + def is_closing(self) -> bool | None: """Return if the cover is closing or not.""" if self.info.primary_value.value is None: return None return bool(self.info.primary_value.value == BARRIER_STATE_CLOSING) @property - def is_closed(self) -> Optional[bool]: + def is_closed(self) -> bool | None: """Return if the cover is closed or not.""" if self.info.primary_value.value is None: return None diff --git a/homeassistant/components/zwave_js/discovery.py b/homeassistant/components/zwave_js/discovery.py index f27325b87d2..5f1f04274ba 100644 --- a/homeassistant/components/zwave_js/discovery.py +++ b/homeassistant/components/zwave_js/discovery.py @@ -1,7 +1,8 @@ """Map Z-Wave nodes and values to Home Assistant entities.""" +from __future__ import annotations from dataclasses import dataclass -from typing import Generator, List, Optional, Set, Union +from typing import Generator from zwave_js_server.const import CommandClass from zwave_js_server.model.device_class import DeviceClassItem @@ -22,7 +23,7 @@ class ZwaveDiscoveryInfo: # the home assistant platform for which an entity should be created platform: str # hint for the platform about this discovered entity - platform_hint: Optional[str] = "" + platform_hint: str | None = "" @dataclass @@ -35,13 +36,13 @@ class ZWaveValueDiscoverySchema: """ # [optional] the value's command class must match ANY of these values - command_class: Optional[Set[int]] = None + command_class: set[int] | None = None # [optional] the value's endpoint must match ANY of these values - endpoint: Optional[Set[int]] = None + endpoint: set[int] | None = None # [optional] the value's property must match ANY of these values - property: Optional[Set[Union[str, int]]] = None + property: set[str | int] | None = None # [optional] the value's metadata_type must match ANY of these values - type: Optional[Set[str]] = None + type: set[str] | None = None @dataclass @@ -58,25 +59,25 @@ class ZWaveDiscoverySchema: # primary value belonging to this discovery scheme primary_value: ZWaveValueDiscoverySchema # [optional] hint for platform - hint: Optional[str] = None + hint: str | None = None # [optional] the node's manufacturer_id must match ANY of these values - manufacturer_id: Optional[Set[int]] = None + manufacturer_id: set[int] | None = None # [optional] the node's product_id must match ANY of these values - product_id: Optional[Set[int]] = None + product_id: set[int] | None = None # [optional] the node's product_type must match ANY of these values - product_type: Optional[Set[int]] = None + product_type: set[int] | None = None # [optional] the node's firmware_version must match ANY of these values - firmware_version: Optional[Set[str]] = None + firmware_version: set[str] | None = None # [optional] the node's basic device class must match ANY of these values - device_class_basic: Optional[Set[Union[str, int]]] = None + device_class_basic: set[str | int] | None = None # [optional] the node's generic device class must match ANY of these values - device_class_generic: Optional[Set[Union[str, int]]] = None + device_class_generic: set[str | int] | None = None # [optional] the node's specific device class must match ANY of these values - device_class_specific: Optional[Set[Union[str, int]]] = None + device_class_specific: set[str | int] | None = None # [optional] additional values that ALL need to be present on the node for this scheme to pass - required_values: Optional[List[ZWaveValueDiscoverySchema]] = None + required_values: list[ZWaveValueDiscoverySchema] | None = None # [optional] additional values that MAY NOT be present on the node for this scheme to pass - absent_values: Optional[List[ZWaveValueDiscoverySchema]] = None + absent_values: list[ZWaveValueDiscoverySchema] | None = None # [optional] bool to specify if this primary value may be discovered by multiple platforms allow_multi: bool = False @@ -487,7 +488,7 @@ def check_value(value: ZwaveValue, schema: ZWaveValueDiscoverySchema) -> bool: @callback def check_device_class( - device_class: DeviceClassItem, required_value: Optional[Set[Union[str, int]]] + device_class: DeviceClassItem, required_value: set[str | int] | None ) -> bool: """Check if device class id or label matches.""" if required_value is None: diff --git a/homeassistant/components/zwave_js/entity.py b/homeassistant/components/zwave_js/entity.py index 34b5940bfea..854fbe45039 100644 --- a/homeassistant/components/zwave_js/entity.py +++ b/homeassistant/components/zwave_js/entity.py @@ -1,7 +1,7 @@ """Generic Z-Wave Entity Class.""" +from __future__ import annotations import logging -from typing import List, Optional, Union from zwave_js_server.client import Client as ZwaveClient from zwave_js_server.model.value import Value as ZwaveValue, get_value_id @@ -97,8 +97,8 @@ class ZWaveBaseEntity(Entity): def generate_name( self, include_value_name: bool = False, - alternate_value_name: Optional[str] = None, - additional_info: Optional[List[str]] = None, + alternate_value_name: str | None = None, + additional_info: list[str] | None = None, ) -> str: """Generate entity name.""" if additional_info is None: @@ -167,13 +167,13 @@ class ZWaveBaseEntity(Entity): @callback def get_zwave_value( self, - value_property: Union[str, int], - command_class: Optional[int] = None, - endpoint: Optional[int] = None, - value_property_key: Optional[int] = None, + value_property: str | int, + command_class: int | None = None, + endpoint: int | None = None, + value_property_key: int | None = None, add_to_watched_value_ids: bool = True, check_all_endpoints: bool = False, - ) -> Optional[ZwaveValue]: + ) -> ZwaveValue | None: """Return specific ZwaveValue on this ZwaveNode.""" # use commandclass and endpoint from primary value if omitted return_value = None diff --git a/homeassistant/components/zwave_js/fan.py b/homeassistant/components/zwave_js/fan.py index ea17fbe4cff..100e400f9f7 100644 --- a/homeassistant/components/zwave_js/fan.py +++ b/homeassistant/components/zwave_js/fan.py @@ -1,6 +1,8 @@ """Support for Z-Wave fans.""" +from __future__ import annotations + import math -from typing import Any, Callable, List, Optional +from typing import Any, Callable from zwave_js_server.client import Client as ZwaveClient @@ -36,7 +38,7 @@ async def async_setup_entry( @callback def async_add_fan(info: ZwaveDiscoveryInfo) -> None: """Add Z-Wave fan.""" - entities: List[ZWaveBaseEntity] = [] + entities: list[ZWaveBaseEntity] = [] entities.append(ZwaveFan(config_entry, client, info)) async_add_entities(entities) @@ -52,7 +54,7 @@ async def async_setup_entry( class ZwaveFan(ZWaveBaseEntity, FanEntity): """Representation of a Z-Wave fan.""" - async def async_set_percentage(self, percentage: Optional[int]) -> None: + async def async_set_percentage(self, percentage: int | None) -> None: """Set the speed percentage of the fan.""" target_value = self.get_zwave_value("targetValue") @@ -68,9 +70,9 @@ class ZwaveFan(ZWaveBaseEntity, FanEntity): async def async_turn_on( self, - speed: Optional[str] = None, - percentage: Optional[int] = None, - preset_mode: Optional[str] = None, + speed: str | None = None, + percentage: int | None = None, + preset_mode: str | None = None, **kwargs: Any, ) -> None: """Turn the device on.""" @@ -82,7 +84,7 @@ class ZwaveFan(ZWaveBaseEntity, FanEntity): await self.info.node.async_set_value(target_value, 0) @property - def is_on(self) -> Optional[bool]: # type: ignore + def is_on(self) -> bool | None: # type: ignore """Return true if device is on (speed above 0).""" if self.info.primary_value.value is None: # guard missing value @@ -90,7 +92,7 @@ class ZwaveFan(ZWaveBaseEntity, FanEntity): return bool(self.info.primary_value.value > 0) @property - def percentage(self) -> Optional[int]: + def percentage(self) -> int | None: """Return the current speed percentage.""" if self.info.primary_value.value is None: # guard missing value diff --git a/homeassistant/components/zwave_js/helpers.py b/homeassistant/components/zwave_js/helpers.py index 16baeb816c2..d535a22394c 100644 --- a/homeassistant/components/zwave_js/helpers.py +++ b/homeassistant/components/zwave_js/helpers.py @@ -1,5 +1,7 @@ """Helper functions for Z-Wave JS integration.""" -from typing import List, Tuple, cast +from __future__ import annotations + +from typing import cast from zwave_js_server.client import Client as ZwaveClient from zwave_js_server.model.node import Node as ZwaveNode @@ -19,13 +21,13 @@ def get_unique_id(home_id: str, value_id: str) -> str: @callback -def get_device_id(client: ZwaveClient, node: ZwaveNode) -> Tuple[str, str]: +def get_device_id(client: ZwaveClient, node: ZwaveNode) -> tuple[str, str]: """Get device registry identifier for Z-Wave node.""" return (DOMAIN, f"{client.driver.controller.home_id}-{node.node_id}") @callback -def get_home_and_node_id_from_device_id(device_id: Tuple[str, str]) -> List[str]: +def get_home_and_node_id_from_device_id(device_id: tuple[str, str]) -> list[str]: """ Get home ID and node ID for Z-Wave device registry entry. diff --git a/homeassistant/components/zwave_js/light.py b/homeassistant/components/zwave_js/light.py index d66821c9ba3..6b42c0ef08e 100644 --- a/homeassistant/components/zwave_js/light.py +++ b/homeassistant/components/zwave_js/light.py @@ -1,6 +1,8 @@ """Support for Z-Wave lights.""" +from __future__ import annotations + import logging -from typing import Any, Callable, Dict, Optional, Tuple +from typing import Any, Callable from zwave_js_server.client import Client as ZwaveClient from zwave_js_server.const import ColorComponent, CommandClass @@ -85,9 +87,9 @@ class ZwaveLight(ZWaveBaseEntity, LightEntity): self._supports_color = False self._supports_white_value = False self._supports_color_temp = False - self._hs_color: Optional[Tuple[float, float]] = None - self._white_value: Optional[int] = None - self._color_temp: Optional[int] = None + self._hs_color: tuple[float, float] | None = None + self._white_value: int | None = None + self._color_temp: int | None = None self._min_mireds = 153 # 6500K as a safe default self._max_mireds = 370 # 2700K as a safe default self._supported_features = SUPPORT_BRIGHTNESS @@ -126,17 +128,17 @@ class ZwaveLight(ZWaveBaseEntity, LightEntity): return self.brightness > 0 @property - def hs_color(self) -> Optional[Tuple[float, float]]: + def hs_color(self) -> tuple[float, float] | None: """Return the hs color.""" return self._hs_color @property - def white_value(self) -> Optional[int]: + def white_value(self) -> int | None: """Return the white value of this light between 0..255.""" return self._white_value @property - def color_temp(self) -> Optional[int]: + def color_temp(self) -> int | None: """Return the color temperature.""" return self._color_temp @@ -220,7 +222,7 @@ class ZwaveLight(ZWaveBaseEntity, LightEntity): """Turn the light off.""" await self._async_set_brightness(0, kwargs.get(ATTR_TRANSITION)) - async def _async_set_colors(self, colors: Dict[ColorComponent, int]) -> None: + async def _async_set_colors(self, colors: dict[ColorComponent, int]) -> None: """Set (multiple) defined colors to given value(s).""" # prefer the (new) combined color property # https://github.com/zwave-js/node-zwave-js/pull/1782 @@ -258,7 +260,7 @@ class ZwaveLight(ZWaveBaseEntity, LightEntity): await self.info.node.async_set_value(target_zwave_value, new_value) async def _async_set_brightness( - self, brightness: Optional[int], transition: Optional[int] = None + self, brightness: int | None, transition: int | None = None ) -> None: """Set new brightness to light.""" if brightness is None: @@ -273,9 +275,7 @@ class ZwaveLight(ZWaveBaseEntity, LightEntity): # setting a value requires setting targetValue await self.info.node.async_set_value(self._target_value, zwave_brightness) - async def _async_set_transition_duration( - self, duration: Optional[int] = None - ) -> None: + async def _async_set_transition_duration(self, duration: int | None = None) -> None: """Set the transition time for the brightness value.""" if self._dimming_duration is None: return diff --git a/homeassistant/components/zwave_js/lock.py b/homeassistant/components/zwave_js/lock.py index 6f2a1a72c7d..0647885345b 100644 --- a/homeassistant/components/zwave_js/lock.py +++ b/homeassistant/components/zwave_js/lock.py @@ -1,6 +1,8 @@ """Representation of Z-Wave locks.""" +from __future__ import annotations + import logging -from typing import Any, Callable, Dict, List, Optional, Union +from typing import Any, Callable import voluptuous as vol from zwave_js_server.client import Client as ZwaveClient @@ -28,7 +30,7 @@ from .entity import ZWaveBaseEntity LOGGER = logging.getLogger(__name__) -STATE_TO_ZWAVE_MAP: Dict[int, Dict[str, Union[int, bool]]] = { +STATE_TO_ZWAVE_MAP: dict[int, dict[str, int | bool]] = { CommandClass.DOOR_LOCK: { STATE_UNLOCKED: DoorLockMode.UNSECURED, STATE_LOCKED: DoorLockMode.SECURED, @@ -52,7 +54,7 @@ async def async_setup_entry( @callback def async_add_lock(info: ZwaveDiscoveryInfo) -> None: """Add Z-Wave Lock.""" - entities: List[ZWaveBaseEntity] = [] + entities: list[ZWaveBaseEntity] = [] entities.append(ZWaveLock(config_entry, client, info)) async_add_entities(entities) @@ -88,7 +90,7 @@ class ZWaveLock(ZWaveBaseEntity, LockEntity): """Representation of a Z-Wave lock.""" @property - def is_locked(self) -> Optional[bool]: + def is_locked(self) -> bool | None: """Return true if the lock is locked.""" if self.info.primary_value.value is None: # guard missing value @@ -100,7 +102,7 @@ class ZWaveLock(ZWaveBaseEntity, LockEntity): ) == int(self.info.primary_value.value) async def _set_lock_state( - self, target_state: str, **kwargs: Dict[str, Any] + self, target_state: str, **kwargs: dict[str, Any] ) -> None: """Set the lock state.""" target_value: ZwaveValue = self.get_zwave_value( @@ -112,11 +114,11 @@ class ZWaveLock(ZWaveBaseEntity, LockEntity): STATE_TO_ZWAVE_MAP[self.info.primary_value.command_class][target_state], ) - async def async_lock(self, **kwargs: Dict[str, Any]) -> None: + async def async_lock(self, **kwargs: dict[str, Any]) -> None: """Lock the lock.""" await self._set_lock_state(STATE_LOCKED) - async def async_unlock(self, **kwargs: Dict[str, Any]) -> None: + async def async_unlock(self, **kwargs: dict[str, Any]) -> None: """Unlock the lock.""" await self._set_lock_state(STATE_UNLOCKED) diff --git a/homeassistant/components/zwave_js/migrate.py b/homeassistant/components/zwave_js/migrate.py index 49c18073de5..016aa3066d7 100644 --- a/homeassistant/components/zwave_js/migrate.py +++ b/homeassistant/components/zwave_js/migrate.py @@ -1,6 +1,7 @@ """Functions used to migrate unique IDs for Z-Wave JS entities.""" +from __future__ import annotations + import logging -from typing import List from zwave_js_server.client import Client as ZwaveClient from zwave_js_server.model.value import Value as ZwaveValue @@ -85,7 +86,7 @@ def async_migrate_discovered_value( @callback -def get_old_value_ids(value: ZwaveValue) -> List[str]: +def get_old_value_ids(value: ZwaveValue) -> list[str]: """Get old value IDs so we can migrate entity unique ID.""" value_ids = [] diff --git a/homeassistant/components/zwave_js/number.py b/homeassistant/components/zwave_js/number.py index 8f8e894cda2..f418ee3d35b 100644 --- a/homeassistant/components/zwave_js/number.py +++ b/homeassistant/components/zwave_js/number.py @@ -1,5 +1,7 @@ """Support for Z-Wave controls using the number platform.""" -from typing import Callable, List, Optional +from __future__ import annotations + +from typing import Callable from zwave_js_server.client import Client as ZwaveClient @@ -22,7 +24,7 @@ async def async_setup_entry( @callback def async_add_number(info: ZwaveDiscoveryInfo) -> None: """Add Z-Wave number entity.""" - entities: List[ZWaveBaseEntity] = [] + entities: list[ZWaveBaseEntity] = [] entities.append(ZwaveNumberEntity(config_entry, client, info)) async_add_entities(entities) @@ -66,14 +68,14 @@ class ZwaveNumberEntity(ZWaveBaseEntity, NumberEntity): return float(self.info.primary_value.metadata.max) @property - def value(self) -> Optional[float]: # type: ignore + def value(self) -> float | None: # type: ignore """Return the entity value.""" if self.info.primary_value.value is None: return None return float(self.info.primary_value.value) @property - def unit_of_measurement(self) -> Optional[str]: + def unit_of_measurement(self) -> str | None: """Return the unit of measurement of this entity, if any.""" if self.info.primary_value.metadata.unit is None: return None diff --git a/homeassistant/components/zwave_js/sensor.py b/homeassistant/components/zwave_js/sensor.py index d6cc9aabd49..ada6f6b5d06 100644 --- a/homeassistant/components/zwave_js/sensor.py +++ b/homeassistant/components/zwave_js/sensor.py @@ -1,7 +1,8 @@ """Representation of Z-Wave sensors.""" +from __future__ import annotations import logging -from typing import Callable, Dict, List, Optional +from typing import Callable from zwave_js_server.client import Client as ZwaveClient from zwave_js_server.const import CommandClass @@ -39,7 +40,7 @@ async def async_setup_entry( @callback def async_add_sensor(info: ZwaveDiscoveryInfo) -> None: """Add Z-Wave Sensor.""" - entities: List[ZWaveBaseEntity] = [] + entities: list[ZWaveBaseEntity] = [] if info.platform_hint == "string_sensor": entities.append(ZWaveStringSensor(config_entry, client, info)) @@ -80,7 +81,7 @@ class ZwaveSensorBase(ZWaveBaseEntity): self._name = self.generate_name(include_value_name=True) @property - def device_class(self) -> Optional[str]: + def device_class(self) -> str | None: """Return the device class of the sensor.""" if self.info.primary_value.command_class == CommandClass.BATTERY: return DEVICE_CLASS_BATTERY @@ -122,14 +123,14 @@ class ZWaveStringSensor(ZwaveSensorBase): """Representation of a Z-Wave String sensor.""" @property - def state(self) -> Optional[str]: + def state(self) -> str | None: """Return state of the sensor.""" if self.info.primary_value.value is None: return None return str(self.info.primary_value.value) @property - def unit_of_measurement(self) -> Optional[str]: + def unit_of_measurement(self) -> str | None: """Return unit of measurement the value is expressed in.""" if self.info.primary_value.metadata.unit is None: return None @@ -161,7 +162,7 @@ class ZWaveNumericSensor(ZwaveSensorBase): return round(float(self.info.primary_value.value), 2) @property - def unit_of_measurement(self) -> Optional[str]: + def unit_of_measurement(self) -> str | None: """Return unit of measurement the value is expressed in.""" if self.info.primary_value.metadata.unit is None: return None @@ -191,7 +192,7 @@ class ZWaveListSensor(ZwaveSensorBase): ) @property - def state(self) -> Optional[str]: + def state(self) -> str | None: """Return state of the sensor.""" if self.info.primary_value.value is None: return None @@ -205,7 +206,7 @@ class ZWaveListSensor(ZwaveSensorBase): ) @property - def extra_state_attributes(self) -> Optional[Dict[str, str]]: + def extra_state_attributes(self) -> dict[str, str] | None: """Return the device specific state attributes.""" # add the value's int value as property for multi-value (list) items return {"value": self.info.primary_value.value} diff --git a/homeassistant/components/zwave_js/services.py b/homeassistant/components/zwave_js/services.py index c971891b35b..1944e4b3dd0 100644 --- a/homeassistant/components/zwave_js/services.py +++ b/homeassistant/components/zwave_js/services.py @@ -1,7 +1,7 @@ """Methods and classes related to executing Z-Wave commands and publishing these to hass.""" +from __future__ import annotations import logging -from typing import Dict, Set, Union import voluptuous as vol from zwave_js_server.model.node import Node as ZwaveNode @@ -20,8 +20,8 @@ _LOGGER = logging.getLogger(__name__) def parameter_name_does_not_need_bitmask( - val: Dict[str, Union[int, str]] -) -> Dict[str, Union[int, str]]: + val: dict[str, int | str] +) -> dict[str, int | str]: """Validate that if a parameter name is provided, bitmask is not as well.""" if isinstance(val[const.ATTR_CONFIG_PARAMETER], str) and ( val.get(const.ATTR_CONFIG_PARAMETER_BITMASK) @@ -88,7 +88,7 @@ class ZWaveServices: async def async_set_config_parameter(self, service: ServiceCall) -> None: """Set a config value on a node.""" - nodes: Set[ZwaveNode] = set() + nodes: set[ZwaveNode] = set() if ATTR_ENTITY_ID in service.data: nodes |= { async_get_node_from_entity_id(self._hass, entity_id) diff --git a/homeassistant/components/zwave_js/switch.py b/homeassistant/components/zwave_js/switch.py index a325e9821f7..e64ea57703d 100644 --- a/homeassistant/components/zwave_js/switch.py +++ b/homeassistant/components/zwave_js/switch.py @@ -1,7 +1,8 @@ """Representation of Z-Wave switches.""" +from __future__ import annotations import logging -from typing import Any, Callable, List, Optional +from typing import Any, Callable from zwave_js_server.client import Client as ZwaveClient @@ -30,7 +31,7 @@ async def async_setup_entry( @callback def async_add_switch(info: ZwaveDiscoveryInfo) -> None: """Add Z-Wave Switch.""" - entities: List[ZWaveBaseEntity] = [] + entities: list[ZWaveBaseEntity] = [] if info.platform_hint == "barrier_event_signaling_state": entities.append( ZWaveBarrierEventSignalingSwitch(config_entry, client, info) @@ -53,7 +54,7 @@ class ZWaveSwitch(ZWaveBaseEntity, SwitchEntity): """Representation of a Z-Wave switch.""" @property - def is_on(self) -> Optional[bool]: # type: ignore + def is_on(self) -> bool | None: # type: ignore """Return a boolean for the state of the switch.""" if self.info.primary_value.value is None: # guard missing value @@ -85,7 +86,7 @@ class ZWaveBarrierEventSignalingSwitch(ZWaveBaseEntity, SwitchEntity): """Initialize a ZWaveBarrierEventSignalingSwitch entity.""" super().__init__(config_entry, client, info) self._name = self.generate_name(include_value_name=True) - self._state: Optional[bool] = None + self._state: bool | None = None self._update_state() @@ -100,7 +101,7 @@ class ZWaveBarrierEventSignalingSwitch(ZWaveBaseEntity, SwitchEntity): return self._name @property - def is_on(self) -> Optional[bool]: # type: ignore + def is_on(self) -> bool | None: # type: ignore """Return a boolean for the state of the switch.""" return self._state