diff --git a/homeassistant/components/acmeda/config_flow.py b/homeassistant/components/acmeda/config_flow.py index f421fa9ca25..b8913e31f2a 100644 --- a/homeassistant/components/acmeda/config_flow.py +++ b/homeassistant/components/acmeda/config_flow.py @@ -1,6 +1,7 @@ """Config flow for Rollease Acmeda Automate Pulse Hub.""" +from __future__ import annotations + import asyncio -from typing import Dict, Optional import aiopulse import async_timeout @@ -19,7 +20,7 @@ class AcmedaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): def __init__(self): """Initialize the config flow.""" - self.discovered_hubs: Optional[Dict[str, aiopulse.Hub]] = None + self.discovered_hubs: dict[str, aiopulse.Hub] | None = None async def async_step_user(self, user_input=None): """Handle a flow initialized by the user.""" diff --git a/homeassistant/components/acmeda/hub.py b/homeassistant/components/acmeda/hub.py index 0b74b874dcc..e156ee5cb78 100644 --- a/homeassistant/components/acmeda/hub.py +++ b/homeassistant/components/acmeda/hub.py @@ -1,6 +1,7 @@ """Code to handle a Pulse Hub.""" +from __future__ import annotations + import asyncio -from typing import Optional import aiopulse @@ -17,7 +18,7 @@ class PulseHub: """Initialize the system.""" self.config_entry = config_entry self.hass = hass - self.api: Optional[aiopulse.Hub] = None + self.api: aiopulse.Hub | None = None self.tasks = [] self.current_rollers = {} self.cleanup_callbacks = [] diff --git a/homeassistant/components/alarm_control_panel/device_action.py b/homeassistant/components/alarm_control_panel/device_action.py index 0dc16fdcf42..67637550db2 100644 --- a/homeassistant/components/alarm_control_panel/device_action.py +++ b/homeassistant/components/alarm_control_panel/device_action.py @@ -1,5 +1,5 @@ """Provides device automations for Alarm control panel.""" -from typing import List, Optional +from __future__ import annotations import voluptuous as vol @@ -41,7 +41,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 Alarm control panel devices.""" registry = await entity_registry.async_get_registry(hass) actions = [] @@ -109,7 +109,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/alarm_control_panel/device_condition.py b/homeassistant/components/alarm_control_panel/device_condition.py index e5b3ec6aeee..3817cf37b45 100644 --- a/homeassistant/components/alarm_control_panel/device_condition.py +++ b/homeassistant/components/alarm_control_panel/device_condition.py @@ -1,5 +1,5 @@ """Provide the device automations for Alarm control panel.""" -from typing import Dict, List +from __future__ import annotations import voluptuous as vol @@ -58,7 +58,7 @@ CONDITION_SCHEMA = DEVICE_CONDITION_BASE_SCHEMA.extend( async def async_get_conditions( hass: HomeAssistant, device_id: str -) -> List[Dict[str, str]]: +) -> list[dict[str, str]]: """List device conditions for Alarm control panel devices.""" registry = await entity_registry.async_get_registry(hass) conditions = [] diff --git a/homeassistant/components/alarm_control_panel/device_trigger.py b/homeassistant/components/alarm_control_panel/device_trigger.py index 5669340c2ce..9ab28e3e863 100644 --- a/homeassistant/components/alarm_control_panel/device_trigger.py +++ b/homeassistant/components/alarm_control_panel/device_trigger.py @@ -1,5 +1,5 @@ """Provides device automations for Alarm control panel.""" -from typing import List +from __future__ import annotations import voluptuous as vol @@ -48,7 +48,7 @@ TRIGGER_SCHEMA = TRIGGER_BASE_SCHEMA.extend( ) -async def async_get_triggers(hass: HomeAssistant, device_id: str) -> List[dict]: +async def async_get_triggers(hass: HomeAssistant, device_id: str) -> list[dict]: """List device triggers for Alarm control panel devices.""" registry = await entity_registry.async_get_registry(hass) triggers = [] diff --git a/homeassistant/components/alarm_control_panel/reproduce_state.py b/homeassistant/components/alarm_control_panel/reproduce_state.py index 9e7d8e6f1a7..3021d4421d9 100644 --- a/homeassistant/components/alarm_control_panel/reproduce_state.py +++ b/homeassistant/components/alarm_control_panel/reproduce_state.py @@ -1,7 +1,9 @@ """Reproduce an Alarm control panel 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, @@ -39,8 +41,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) @@ -83,8 +85,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 Alarm control panel states.""" await asyncio.gather( diff --git a/homeassistant/components/alert/reproduce_state.py b/homeassistant/components/alert/reproduce_state.py index 7645b642d59..de40649854e 100644 --- a/homeassistant/components/alert/reproduce_state.py +++ b/homeassistant/components/alert/reproduce_state.py @@ -1,7 +1,9 @@ """Reproduce an Alert 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, @@ -24,8 +26,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) @@ -61,8 +63,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 Alert states.""" # Reproduce states in parallel. diff --git a/homeassistant/components/alexa/capabilities.py b/homeassistant/components/alexa/capabilities.py index acfba91a933..69acf95e207 100644 --- a/homeassistant/components/alexa/capabilities.py +++ b/homeassistant/components/alexa/capabilities.py @@ -1,6 +1,7 @@ """Alexa capabilities.""" +from __future__ import annotations + import logging -from typing import List, Optional from homeassistant.components import ( cover, @@ -72,7 +73,7 @@ class AlexaCapability: supported_locales = {"en-US"} - def __init__(self, entity: State, instance: Optional[str] = None): + def __init__(self, entity: State, instance: str | None = None): """Initialize an Alexa capability.""" self.entity = entity self.instance = instance @@ -82,7 +83,7 @@ class AlexaCapability: raise NotImplementedError @staticmethod - def properties_supported() -> List[dict]: + def properties_supported() -> list[dict]: """Return what properties this entity supports.""" return [] diff --git a/homeassistant/components/alexa/entities.py b/homeassistant/components/alexa/entities.py index c05d9641b9a..71f4e41a810 100644 --- a/homeassistant/components/alexa/entities.py +++ b/homeassistant/components/alexa/entities.py @@ -1,6 +1,8 @@ """Alexa entity adapters.""" +from __future__ import annotations + import logging -from typing import TYPE_CHECKING, List +from typing import TYPE_CHECKING from homeassistant.components import ( alarm_control_panel, @@ -300,7 +302,7 @@ class AlexaEntity: Raises _UnsupportedInterface. """ - def interfaces(self) -> List[AlexaCapability]: + def interfaces(self) -> list[AlexaCapability]: """Return a list of supported interfaces. Used for discovery. The list should contain AlexaInterface instances. @@ -353,7 +355,7 @@ class AlexaEntity: @callback -def async_get_entities(hass, config) -> List[AlexaEntity]: +def async_get_entities(hass, config) -> list[AlexaEntity]: """Return all entities that are supported by Alexa.""" entities = [] for state in hass.states.async_all(): diff --git a/homeassistant/components/alexa/state_report.py b/homeassistant/components/alexa/state_report.py index c34dc34f0dd..712a08ac6b9 100644 --- a/homeassistant/components/alexa/state_report.py +++ b/homeassistant/components/alexa/state_report.py @@ -1,8 +1,9 @@ """Alexa state report code.""" +from __future__ import annotations + import asyncio import json import logging -from typing import Optional import aiohttp import async_timeout @@ -45,8 +46,8 @@ async def async_enable_proactive_mode(hass, smart_home_config): async def async_entity_state_listener( changed_entity: str, - old_state: Optional[State], - new_state: Optional[State], + old_state: State | None, + new_state: State | None, ): if not hass.is_running: return diff --git a/homeassistant/components/almond/__init__.py b/homeassistant/components/almond/__init__.py index b9f75ff8c6b..554a4aa47bc 100644 --- a/homeassistant/components/almond/__init__.py +++ b/homeassistant/components/almond/__init__.py @@ -1,9 +1,10 @@ """Support for Almond.""" +from __future__ import annotations + import asyncio from datetime import timedelta import logging import time -from typing import Optional from aiohttp import ClientError, ClientSession import async_timeout @@ -281,7 +282,7 @@ class AlmondAgent(conversation.AbstractConversationAgent): return True async def async_process( - self, text: str, context: Context, conversation_id: Optional[str] = None + self, text: str, context: Context, conversation_id: str | None = None ) -> intent.IntentResponse: """Process a sentence.""" response = await self.api.async_converse_text(text, conversation_id) diff --git a/homeassistant/components/arcam_fmj/device_trigger.py b/homeassistant/components/arcam_fmj/device_trigger.py index c03a082c149..060c56e5953 100644 --- a/homeassistant/components/arcam_fmj/device_trigger.py +++ b/homeassistant/components/arcam_fmj/device_trigger.py @@ -1,5 +1,5 @@ """Provides device automations for Arcam FMJ Receiver control.""" -from typing import List +from __future__ import annotations import voluptuous as vol @@ -28,7 +28,7 @@ TRIGGER_SCHEMA = TRIGGER_BASE_SCHEMA.extend( ) -async def async_get_triggers(hass: HomeAssistant, device_id: str) -> List[dict]: +async def async_get_triggers(hass: HomeAssistant, device_id: str) -> list[dict]: """List device triggers for Arcam FMJ Receiver control devices.""" registry = await entity_registry.async_get_registry(hass) triggers = [] diff --git a/homeassistant/components/arris_tg2492lg/device_tracker.py b/homeassistant/components/arris_tg2492lg/device_tracker.py index e63bef9c108..1011d76f8aa 100644 --- a/homeassistant/components/arris_tg2492lg/device_tracker.py +++ b/homeassistant/components/arris_tg2492lg/device_tracker.py @@ -1,5 +1,5 @@ """Support for Arris TG2492LG router.""" -from typing import List +from __future__ import annotations from arris_tg2492lg import ConnectBox, Device import voluptuous as vol @@ -36,7 +36,7 @@ class ArrisDeviceScanner(DeviceScanner): def __init__(self, connect_box: ConnectBox): """Initialize the scanner.""" self.connect_box = connect_box - self.last_results: List[Device] = [] + self.last_results: list[Device] = [] def scan_devices(self): """Scan for new devices and return a list with found device IDs.""" diff --git a/homeassistant/components/asuswrt/device_tracker.py b/homeassistant/components/asuswrt/device_tracker.py index f5cb9b934e3..1ef68f9d65d 100644 --- a/homeassistant/components/asuswrt/device_tracker.py +++ b/homeassistant/components/asuswrt/device_tracker.py @@ -1,5 +1,5 @@ """Support for ASUSWRT routers.""" -from typing import Dict +from __future__ import annotations from homeassistant.components.device_tracker import SOURCE_TYPE_ROUTER from homeassistant.components.device_tracker.config_entry import ScannerEntity @@ -103,12 +103,12 @@ class AsusWrtDevice(ScannerEntity): return self._icon @property - def extra_state_attributes(self) -> Dict[str, any]: + def extra_state_attributes(self) -> dict[str, any]: """Return the attributes.""" return self._attrs @property - def device_info(self) -> Dict[str, any]: + def device_info(self) -> dict[str, any]: """Return the device information.""" return { "connections": {(CONNECTION_NETWORK_MAC, self._mac)}, diff --git a/homeassistant/components/asuswrt/router.py b/homeassistant/components/asuswrt/router.py index 4c92ee2ef67..550e4c8fc16 100644 --- a/homeassistant/components/asuswrt/router.py +++ b/homeassistant/components/asuswrt/router.py @@ -1,7 +1,9 @@ """Represent the AsusWrt router.""" +from __future__ import annotations + from datetime import datetime, timedelta import logging -from typing import Any, Dict, Optional +from typing import Any from aioasuswrt.asuswrt import AsusWrt @@ -74,7 +76,7 @@ class AsusWrtSensorDataHandler: async def _get_bytes(self): """Fetch byte information from the router.""" - ret_dict: Dict[str, Any] = {} + ret_dict: dict[str, Any] = {} try: datas = await self._api.async_get_bytes_total() except OSError as exc: @@ -87,7 +89,7 @@ class AsusWrtSensorDataHandler: async def _get_rates(self): """Fetch rates information from the router.""" - ret_dict: Dict[str, Any] = {} + ret_dict: dict[str, Any] = {} try: rates = await self._api.async_get_current_transfer_rates() except OSError as exc: @@ -194,12 +196,12 @@ class AsusWrtRouter: self._protocol = entry.data[CONF_PROTOCOL] self._host = entry.data[CONF_HOST] - self._devices: Dict[str, Any] = {} + self._devices: dict[str, Any] = {} self._connected_devices = 0 self._connect_error = False self._sensors_data_handler: AsusWrtSensorDataHandler = None - self._sensors_coordinator: Dict[str, Any] = {} + self._sensors_coordinator: dict[str, Any] = {} self._on_close = [] @@ -245,7 +247,7 @@ class AsusWrtRouter: async_track_time_interval(self.hass, self.update_all, SCAN_INTERVAL) ) - async def update_all(self, now: Optional[datetime] = None) -> None: + async def update_all(self, now: datetime | None = None) -> None: """Update all AsusWrt platforms.""" await self.update_devices() @@ -353,7 +355,7 @@ class AsusWrtRouter: """Add a function to call when router is closed.""" self._on_close.append(func) - def update_options(self, new_options: Dict) -> bool: + def update_options(self, new_options: dict) -> bool: """Update router options.""" req_reload = False for name, new_opt in new_options.items(): @@ -367,7 +369,7 @@ class AsusWrtRouter: return req_reload @property - def device_info(self) -> Dict[str, Any]: + def device_info(self) -> dict[str, Any]: """Return the device information.""" return { "identifiers": {(DOMAIN, "AsusWRT")}, @@ -392,12 +394,12 @@ class AsusWrtRouter: return self._host @property - def devices(self) -> Dict[str, Any]: + def devices(self) -> dict[str, Any]: """Return devices.""" return self._devices @property - def sensors_coordinator(self) -> Dict[str, Any]: + def sensors_coordinator(self) -> dict[str, Any]: """Return sensors coordinators.""" return self._sensors_coordinator @@ -407,7 +409,7 @@ class AsusWrtRouter: return self._api -def get_api(conf: Dict, options: Optional[Dict] = None) -> AsusWrt: +def get_api(conf: dict, options: dict | None = None) -> AsusWrt: """Get the AsusWrt API.""" opt = options or {} diff --git a/homeassistant/components/asuswrt/sensor.py b/homeassistant/components/asuswrt/sensor.py index 0cd427d3b64..7dc8208ee67 100644 --- a/homeassistant/components/asuswrt/sensor.py +++ b/homeassistant/components/asuswrt/sensor.py @@ -1,7 +1,8 @@ """Asuswrt status sensors.""" +from __future__ import annotations + import logging from numbers import Number -from typing import Dict from homeassistant.config_entries import ConfigEntry from homeassistant.const import DATA_GIGABYTES, DATA_RATE_MEGABITS_PER_SECOND @@ -104,7 +105,7 @@ class AsusWrtSensor(CoordinatorEntity): coordinator: DataUpdateCoordinator, router: AsusWrtRouter, sensor_type: str, - sensor: Dict[str, any], + sensor: dict[str, any], ) -> None: """Initialize a AsusWrt sensor.""" super().__init__(coordinator) @@ -159,11 +160,11 @@ class AsusWrtSensor(CoordinatorEntity): return self._device_class @property - def extra_state_attributes(self) -> Dict[str, any]: + def extra_state_attributes(self) -> dict[str, any]: """Return the attributes.""" return {"hostname": self._router.host} @property - def device_info(self) -> Dict[str, any]: + def device_info(self) -> dict[str, any]: """Return the device information.""" return self._router.device_info diff --git a/homeassistant/components/atag/climate.py b/homeassistant/components/atag/climate.py index a2aa5cf16e4..da7e6a14a73 100644 --- a/homeassistant/components/atag/climate.py +++ b/homeassistant/components/atag/climate.py @@ -1,5 +1,5 @@ """Initialization of ATAG One climate platform.""" -from typing import List, Optional +from __future__ import annotations from homeassistant.components.climate import ClimateEntity from homeassistant.components.climate.const import ( @@ -43,46 +43,46 @@ class AtagThermostat(AtagEntity, ClimateEntity): return SUPPORT_FLAGS @property - def hvac_mode(self) -> Optional[str]: + def hvac_mode(self) -> str | None: """Return hvac operation ie. heat, cool mode.""" if self.coordinator.data.climate.hvac_mode in HVAC_MODES: return self.coordinator.data.climate.hvac_mode return None @property - def hvac_modes(self) -> List[str]: + def hvac_modes(self) -> list[str]: """Return the list of available hvac operation modes.""" return HVAC_MODES @property - def hvac_action(self) -> Optional[str]: + def hvac_action(self) -> str | None: """Return the current running hvac operation.""" is_active = self.coordinator.data.climate.status return CURRENT_HVAC_HEAT if is_active else CURRENT_HVAC_IDLE @property - def temperature_unit(self) -> Optional[str]: + def temperature_unit(self) -> str | None: """Return the unit of measurement.""" return self.coordinator.data.climate.temp_unit @property - def current_temperature(self) -> Optional[float]: + def current_temperature(self) -> float | None: """Return the current temperature.""" return self.coordinator.data.climate.temperature @property - def target_temperature(self) -> Optional[float]: + def target_temperature(self) -> float | None: """Return the temperature we try to reach.""" return self.coordinator.data.climate.target_temperature @property - def preset_mode(self) -> Optional[str]: + def preset_mode(self) -> str | None: """Return the current preset mode, e.g., auto, manual, fireplace, extend, etc.""" preset = self.coordinator.data.climate.preset_mode return PRESET_INVERTED.get(preset) @property - def preset_modes(self) -> Optional[List[str]]: + def preset_modes(self) -> list[str] | None: """Return a list of available preset modes.""" return list(PRESET_MAP.keys()) diff --git a/homeassistant/components/auth/__init__.py b/homeassistant/components/auth/__init__.py index 4ddf82cc022..7381be5e9de 100644 --- a/homeassistant/components/auth/__init__.py +++ b/homeassistant/components/auth/__init__.py @@ -114,8 +114,9 @@ Result will be a long-lived access token: } """ +from __future__ import annotations + from datetime import timedelta -from typing import Union import uuid from aiohttp import web @@ -183,7 +184,7 @@ RESULT_TYPE_USER = "user" @bind_hass def create_auth_code( - hass, client_id: str, credential_or_user: Union[Credentials, User] + hass, client_id: str, credential_or_user: Credentials | User ) -> str: """Create an authorization code to fetch tokens.""" return hass.data[DOMAIN](client_id, credential_or_user) diff --git a/homeassistant/components/automation/__init__.py b/homeassistant/components/automation/__init__.py index b769b0329cc..425ffe17979 100644 --- a/homeassistant/components/automation/__init__.py +++ b/homeassistant/components/automation/__init__.py @@ -1,6 +1,8 @@ """Allow to set up simple automation rules via the config file.""" +from __future__ import annotations + import logging -from typing import Any, Awaitable, Callable, Dict, List, Optional, Set, Union, cast +from typing import Any, Awaitable, Callable, Dict, cast import voluptuous as vol from voluptuous.humanize import humanize_error @@ -109,7 +111,7 @@ def is_on(hass, entity_id): @callback -def automations_with_entity(hass: HomeAssistant, entity_id: str) -> List[str]: +def automations_with_entity(hass: HomeAssistant, entity_id: str) -> list[str]: """Return all automations that reference the entity.""" if DOMAIN not in hass.data: return [] @@ -124,7 +126,7 @@ def automations_with_entity(hass: HomeAssistant, entity_id: str) -> List[str]: @callback -def entities_in_automation(hass: HomeAssistant, entity_id: str) -> List[str]: +def entities_in_automation(hass: HomeAssistant, entity_id: str) -> list[str]: """Return all entities in a scene.""" if DOMAIN not in hass.data: return [] @@ -140,7 +142,7 @@ def entities_in_automation(hass: HomeAssistant, entity_id: str) -> List[str]: @callback -def automations_with_device(hass: HomeAssistant, device_id: str) -> List[str]: +def automations_with_device(hass: HomeAssistant, device_id: str) -> list[str]: """Return all automations that reference the device.""" if DOMAIN not in hass.data: return [] @@ -155,7 +157,7 @@ def automations_with_device(hass: HomeAssistant, device_id: str) -> List[str]: @callback -def devices_in_automation(hass: HomeAssistant, entity_id: str) -> List[str]: +def devices_in_automation(hass: HomeAssistant, entity_id: str) -> list[str]: """Return all devices in a scene.""" if DOMAIN not in hass.data: return [] @@ -249,8 +251,8 @@ class AutomationEntity(ToggleEntity, RestoreEntity): self.action_script.change_listener = self.async_write_ha_state self._initial_state = initial_state self._is_enabled = False - self._referenced_entities: Optional[Set[str]] = None - self._referenced_devices: Optional[Set[str]] = None + self._referenced_entities: set[str] | None = None + self._referenced_devices: set[str] | None = None self._logger = LOGGER self._variables: ScriptVariables = variables self._trigger_variables: ScriptVariables = trigger_variables @@ -509,7 +511,7 @@ class AutomationEntity(ToggleEntity, RestoreEntity): async def _async_attach_triggers( self, home_assistant_start: bool - ) -> Optional[Callable[[], None]]: + ) -> Callable[[], None] | None: """Set up the triggers.""" def log_cb(level, msg, **kwargs): @@ -539,7 +541,7 @@ class AutomationEntity(ToggleEntity, RestoreEntity): async def _async_process_config( hass: HomeAssistant, - config: Dict[str, Any], + config: dict[str, Any], component: EntityComponent, ) -> bool: """Process config and add automations. @@ -550,7 +552,7 @@ async def _async_process_config( blueprints_used = False for config_key in extract_domain_configs(config, DOMAIN): - conf: List[Union[Dict[str, Any], blueprint.BlueprintInputs]] = config[ # type: ignore + conf: list[dict[str, Any] | blueprint.BlueprintInputs] = config[ # type: ignore config_key ] @@ -680,7 +682,7 @@ async def _async_process_if(hass, name, config, p_config): @callback -def _trigger_extract_device(trigger_conf: dict) -> Optional[str]: +def _trigger_extract_device(trigger_conf: dict) -> str | None: """Extract devices from a trigger config.""" if trigger_conf[CONF_PLATFORM] != "device": return None @@ -689,7 +691,7 @@ def _trigger_extract_device(trigger_conf: dict) -> Optional[str]: @callback -def _trigger_extract_entities(trigger_conf: dict) -> List[str]: +def _trigger_extract_entities(trigger_conf: dict) -> list[str]: """Extract entities from a trigger config.""" if trigger_conf[CONF_PLATFORM] in ("state", "numeric_state"): return trigger_conf[CONF_ENTITY_ID] diff --git a/homeassistant/components/automation/reproduce_state.py b/homeassistant/components/automation/reproduce_state.py index bcd0cc4e585..efe83960f41 100644 --- a/homeassistant/components/automation/reproduce_state.py +++ b/homeassistant/components/automation/reproduce_state.py @@ -1,7 +1,9 @@ """Reproduce an Automation 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, @@ -24,8 +26,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) @@ -60,8 +62,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 Automation states.""" await asyncio.gather( diff --git a/homeassistant/components/automation/trace.py b/homeassistant/components/automation/trace.py index 68cca5a4a41..79fa4c844bc 100644 --- a/homeassistant/components/automation/trace.py +++ b/homeassistant/components/automation/trace.py @@ -1,11 +1,13 @@ """Trace support for automation.""" +from __future__ import annotations + from collections import OrderedDict from contextlib import contextmanager import datetime as dt from datetime import timedelta from itertools import count import logging -from typing import Any, Awaitable, Callable, Deque, Dict, Optional +from typing import Any, Awaitable, Callable, Deque from homeassistant.core import Context, HomeAssistant, callback from homeassistant.helpers.json import JSONEncoder as HAJSONEncoder @@ -30,28 +32,28 @@ class AutomationTrace: def __init__( self, - unique_id: Optional[str], - config: Dict[str, Any], + unique_id: str | None, + config: dict[str, Any], context: Context, ): """Container for automation trace.""" - self._action_trace: Optional[Dict[str, Deque[TraceElement]]] = None - self._condition_trace: Optional[Dict[str, Deque[TraceElement]]] = None - self._config: Dict[str, Any] = config + self._action_trace: dict[str, Deque[TraceElement]] | None = None + self._condition_trace: dict[str, Deque[TraceElement]] | None = None + self._config: dict[str, Any] = config self.context: Context = context - self._error: Optional[Exception] = None + self._error: Exception | None = None self._state: str = "running" self.run_id: str = str(next(self._run_ids)) - self._timestamp_finish: Optional[dt.datetime] = None + self._timestamp_finish: dt.datetime | None = None self._timestamp_start: dt.datetime = dt_util.utcnow() - self._unique_id: Optional[str] = unique_id - self._variables: Optional[Dict[str, Any]] = None + self._unique_id: str | None = unique_id + self._variables: dict[str, Any] | None = None - def set_action_trace(self, trace: Dict[str, Deque[TraceElement]]) -> None: + def set_action_trace(self, trace: dict[str, Deque[TraceElement]]) -> None: """Set action trace.""" self._action_trace = trace - def set_condition_trace(self, trace: Dict[str, Deque[TraceElement]]) -> None: + def set_condition_trace(self, trace: dict[str, Deque[TraceElement]]) -> None: """Set condition trace.""" self._condition_trace = trace @@ -59,7 +61,7 @@ class AutomationTrace: """Set error.""" self._error = ex - def set_variables(self, variables: Dict[str, Any]) -> None: + def set_variables(self, variables: dict[str, Any]) -> None: """Set variables.""" self._variables = variables @@ -68,7 +70,7 @@ class AutomationTrace: self._timestamp_finish = dt_util.utcnow() self._state = "stopped" - def as_dict(self) -> Dict[str, Any]: + def as_dict(self) -> dict[str, Any]: """Return dictionary version of this AutomationTrace.""" result = self.as_short_dict() @@ -96,7 +98,7 @@ class AutomationTrace: result["error"] = str(self._error) return result - def as_short_dict(self) -> Dict[str, Any]: + def as_short_dict(self) -> dict[str, Any]: """Return a brief dictionary version of this AutomationTrace.""" last_action = None diff --git a/homeassistant/components/awair/__init__.py b/homeassistant/components/awair/__init__.py index 56af5d2b662..eefc7445d36 100644 --- a/homeassistant/components/awair/__init__.py +++ b/homeassistant/components/awair/__init__.py @@ -1,7 +1,8 @@ """The awair component.""" +from __future__ import annotations from asyncio import gather -from typing import Any, Optional +from typing import Any from async_timeout import timeout from python_awair import Awair @@ -70,7 +71,7 @@ class AwairDataUpdateCoordinator(DataUpdateCoordinator): super().__init__(hass, LOGGER, name=DOMAIN, update_interval=UPDATE_INTERVAL) - async def _async_update_data(self) -> Optional[Any]: + async def _async_update_data(self) -> Any | None: """Update data via Awair client library.""" with timeout(API_TIMEOUT): try: diff --git a/homeassistant/components/awair/config_flow.py b/homeassistant/components/awair/config_flow.py index c28ac55f216..76c7cbca3a9 100644 --- a/homeassistant/components/awair/config_flow.py +++ b/homeassistant/components/awair/config_flow.py @@ -1,6 +1,5 @@ """Config flow for Awair.""" - -from typing import Optional +from __future__ import annotations from python_awair import Awair from python_awair.exceptions import AuthError, AwairError @@ -36,7 +35,7 @@ class AwairFlowHandler(ConfigFlow, domain=DOMAIN): data={CONF_ACCESS_TOKEN: conf[CONF_ACCESS_TOKEN]}, ) - async def async_step_user(self, user_input: Optional[dict] = None): + async def async_step_user(self, user_input: dict | None = None): """Handle a flow initialized by the user.""" errors = {} @@ -61,7 +60,7 @@ class AwairFlowHandler(ConfigFlow, domain=DOMAIN): errors=errors, ) - async def async_step_reauth(self, user_input: Optional[dict] = None): + async def async_step_reauth(self, user_input: dict | None = None): """Handle re-auth if token invalid.""" errors = {} diff --git a/homeassistant/components/awair/sensor.py b/homeassistant/components/awair/sensor.py index d685b3ec17b..81dc0562fc4 100644 --- a/homeassistant/components/awair/sensor.py +++ b/homeassistant/components/awair/sensor.py @@ -1,6 +1,7 @@ """Support for Awair sensors.""" +from __future__ import annotations -from typing import Callable, List, Optional +from typing import Callable from python_awair.devices import AwairDevice import voluptuous as vol @@ -55,13 +56,13 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info= async def async_setup_entry( hass: HomeAssistantType, config_entry: ConfigType, - async_add_entities: Callable[[List[Entity], bool], None], + async_add_entities: Callable[[list[Entity], bool], None], ): """Set up Awair sensor entity based on a config entry.""" coordinator = hass.data[DOMAIN][config_entry.entry_id] sensors = [] - data: List[AwairResult] = coordinator.data.values() + data: list[AwairResult] = coordinator.data.values() for result in data: if result.air_data: sensors.append(AwairSensor(API_SCORE, result.device, coordinator)) @@ -228,9 +229,9 @@ class AwairSensor(CoordinatorEntity): return info @property - def _air_data(self) -> Optional[AwairResult]: + def _air_data(self) -> AwairResult | None: """Return the latest data for our device, or None.""" - result: Optional[AwairResult] = self.coordinator.data.get(self._device.uuid) + result: AwairResult | None = self.coordinator.data.get(self._device.uuid) if result: return result.air_data diff --git a/homeassistant/components/azure_devops/__init__.py b/homeassistant/components/azure_devops/__init__.py index f72a4c44918..b856dc5aa00 100644 --- a/homeassistant/components/azure_devops/__init__.py +++ b/homeassistant/components/azure_devops/__init__.py @@ -1,6 +1,8 @@ """Support for Azure DevOps.""" +from __future__ import annotations + import logging -from typing import Any, Dict +from typing import Any from aioazuredevops.client import DevOpsClient import aiohttp @@ -114,7 +116,7 @@ class AzureDevOpsDeviceEntity(AzureDevOpsEntity): """Defines a Azure DevOps device entity.""" @property - def device_info(self) -> Dict[str, Any]: + def device_info(self) -> dict[str, Any]: """Return device information about this Azure DevOps instance.""" return { "identifiers": { diff --git a/homeassistant/components/azure_devops/sensor.py b/homeassistant/components/azure_devops/sensor.py index a26f0c65f9c..1d30bfcb9f9 100644 --- a/homeassistant/components/azure_devops/sensor.py +++ b/homeassistant/components/azure_devops/sensor.py @@ -1,7 +1,8 @@ """Support for Azure DevOps sensors.""" +from __future__ import annotations + from datetime import timedelta import logging -from typing import List from aioazuredevops.builds import DevOpsBuild from aioazuredevops.client import DevOpsClient @@ -39,7 +40,7 @@ async def async_setup_entry( sensors = [] try: - builds: List[DevOpsBuild] = await client.get_builds( + builds: list[DevOpsBuild] = await client.get_builds( organization, project, BUILDS_QUERY ) except aiohttp.ClientError as exception: diff --git a/homeassistant/components/azure_event_hub/__init__.py b/homeassistant/components/azure_event_hub/__init__.py index 3b44c6423be..0473c4ff5a7 100644 --- a/homeassistant/components/azure_event_hub/__init__.py +++ b/homeassistant/components/azure_event_hub/__init__.py @@ -1,9 +1,11 @@ """Support for Azure Event Hubs.""" +from __future__ import annotations + import asyncio import json import logging import time -from typing import Any, Dict +from typing import Any from azure.eventhub import EventData from azure.eventhub.aio import EventHubProducerClient, EventHubSharedKeyCredential @@ -95,7 +97,7 @@ class AzureEventHub: def __init__( self, hass: HomeAssistant, - client_args: Dict[str, Any], + client_args: dict[str, Any], conn_str_client: bool, entities_filter: vol.Schema, send_interval: int, diff --git a/homeassistant/components/bbox/device_tracker.py b/homeassistant/components/bbox/device_tracker.py index 8097c11eb89..130d315197b 100644 --- a/homeassistant/components/bbox/device_tracker.py +++ b/homeassistant/components/bbox/device_tracker.py @@ -1,8 +1,9 @@ """Support for French FAI Bouygues Bbox routers.""" +from __future__ import annotations + from collections import namedtuple from datetime import timedelta import logging -from typing import List import pybbox import voluptuous as vol @@ -47,7 +48,7 @@ class BboxDeviceScanner(DeviceScanner): self.host = config[CONF_HOST] """Initialize the scanner.""" - self.last_results: List[Device] = [] + self.last_results: list[Device] = [] self.success_init = self._update_info() _LOGGER.info("Scanner initialized") diff --git a/homeassistant/components/binary_sensor/device_condition.py b/homeassistant/components/binary_sensor/device_condition.py index 999a62b3a80..8c506634200 100644 --- a/homeassistant/components/binary_sensor/device_condition.py +++ b/homeassistant/components/binary_sensor/device_condition.py @@ -1,5 +1,5 @@ """Implement device conditions for binary sensor.""" -from typing import Dict, List +from __future__ import annotations import voluptuous as vol @@ -205,9 +205,9 @@ CONDITION_SCHEMA = cv.DEVICE_CONDITION_BASE_SCHEMA.extend( async def async_get_conditions( hass: HomeAssistant, device_id: str -) -> List[Dict[str, str]]: +) -> list[dict[str, str]]: """List device conditions.""" - conditions: List[Dict[str, str]] = [] + conditions: list[dict[str, str]] = [] entity_registry = await async_get_registry(hass) entries = [ entry diff --git a/homeassistant/components/binary_sensor/significant_change.py b/homeassistant/components/binary_sensor/significant_change.py index bc2dba04f09..8421483ba0c 100644 --- a/homeassistant/components/binary_sensor/significant_change.py +++ b/homeassistant/components/binary_sensor/significant_change.py @@ -1,5 +1,7 @@ """Helper to test significant Binary Sensor state changes.""" -from typing import Any, Optional +from __future__ import annotations + +from typing import Any from homeassistant.core import HomeAssistant, callback @@ -12,7 +14,7 @@ def async_check_significant_change( new_state: str, new_attrs: dict, **kwargs: Any, -) -> Optional[bool]: +) -> bool | None: """Test if state significantly changed.""" if old_state != new_state: return True diff --git a/homeassistant/components/blueprint/importer.py b/homeassistant/components/blueprint/importer.py index 217851df980..99dffb114e1 100644 --- a/homeassistant/components/blueprint/importer.py +++ b/homeassistant/components/blueprint/importer.py @@ -1,8 +1,9 @@ """Import logic for blueprint.""" +from __future__ import annotations + from dataclasses import dataclass import html import re -from typing import Optional import voluptuous as vol import yarl @@ -93,7 +94,7 @@ def _get_community_post_import_url(url: str) -> str: def _extract_blueprint_from_community_topic( url: str, topic: dict, -) -> Optional[ImportedBlueprint]: +) -> ImportedBlueprint | None: """Extract a blueprint from a community post JSON. Async friendly. @@ -136,7 +137,7 @@ def _extract_blueprint_from_community_topic( async def fetch_blueprint_from_community_post( hass: HomeAssistant, url: str -) -> Optional[ImportedBlueprint]: +) -> ImportedBlueprint | None: """Get blueprints from a community post url. Method can raise aiohttp client exceptions, vol.Invalid. diff --git a/homeassistant/components/blueprint/models.py b/homeassistant/components/blueprint/models.py index 84931a04310..797f9bd1512 100644 --- a/homeassistant/components/blueprint/models.py +++ b/homeassistant/components/blueprint/models.py @@ -1,9 +1,11 @@ """Blueprint models.""" +from __future__ import annotations + import asyncio import logging import pathlib import shutil -from typing import Any, Dict, List, Optional, Union +from typing import Any from awesomeversion import AwesomeVersion import voluptuous as vol @@ -49,8 +51,8 @@ class Blueprint: self, data: dict, *, - path: Optional[str] = None, - expected_domain: Optional[str] = None, + path: str | None = None, + expected_domain: str | None = None, ) -> None: """Initialize a blueprint.""" try: @@ -95,7 +97,7 @@ class Blueprint: """Return blueprint metadata.""" return self.data[CONF_BLUEPRINT] - def update_metadata(self, *, source_url: Optional[str] = None) -> None: + def update_metadata(self, *, source_url: str | None = None) -> None: """Update metadata.""" if source_url is not None: self.data[CONF_BLUEPRINT][CONF_SOURCE_URL] = source_url @@ -105,7 +107,7 @@ class Blueprint: return yaml.dump(self.data) @callback - def validate(self) -> Optional[List[str]]: + def validate(self) -> list[str] | None: """Test if the Home Assistant installation supports this blueprint. Return list of errors if not valid. @@ -126,7 +128,7 @@ class BlueprintInputs: """Inputs for a blueprint.""" def __init__( - self, blueprint: Blueprint, config_with_inputs: Dict[str, Any] + self, blueprint: Blueprint, config_with_inputs: dict[str, Any] ) -> None: """Instantiate a blueprint inputs object.""" self.blueprint = blueprint @@ -218,7 +220,7 @@ class DomainBlueprints: blueprint_data, expected_domain=self.domain, path=blueprint_path ) - def _load_blueprints(self) -> Dict[str, Union[Blueprint, BlueprintException]]: + def _load_blueprints(self) -> dict[str, Blueprint | BlueprintException]: """Load all the blueprints.""" blueprint_folder = pathlib.Path( self.hass.config.path(BLUEPRINT_FOLDER, self.domain) @@ -243,7 +245,7 @@ class DomainBlueprints: async def async_get_blueprints( self, - ) -> Dict[str, Union[Blueprint, BlueprintException]]: + ) -> dict[str, Blueprint | BlueprintException]: """Get all the blueprints.""" async with self._load_lock: return await self.hass.async_add_executor_job(self._load_blueprints) diff --git a/homeassistant/components/blueprint/websocket_api.py b/homeassistant/components/blueprint/websocket_api.py index 05ae2816696..b8a4c214a2e 100644 --- a/homeassistant/components/blueprint/websocket_api.py +++ b/homeassistant/components/blueprint/websocket_api.py @@ -1,5 +1,5 @@ """Websocket API for blueprint.""" -from typing import Dict, Optional +from __future__ import annotations import async_timeout import voluptuous as vol @@ -33,7 +33,7 @@ def async_setup(hass: HomeAssistant): ) async def ws_list_blueprints(hass, connection, msg): """List available blueprints.""" - domain_blueprints: Optional[Dict[str, models.DomainBlueprints]] = hass.data.get( + domain_blueprints: dict[str, models.DomainBlueprints] | None = hass.data.get( DOMAIN, {} ) results = {} @@ -102,7 +102,7 @@ async def ws_save_blueprint(hass, connection, msg): path = msg["path"] domain = msg["domain"] - domain_blueprints: Optional[Dict[str, models.DomainBlueprints]] = hass.data.get( + domain_blueprints: dict[str, models.DomainBlueprints] | None = hass.data.get( DOMAIN, {} ) @@ -149,7 +149,7 @@ async def ws_delete_blueprint(hass, connection, msg): path = msg["path"] domain = msg["domain"] - domain_blueprints: Optional[Dict[str, models.DomainBlueprints]] = hass.data.get( + domain_blueprints: dict[str, models.DomainBlueprints] | None = hass.data.get( DOMAIN, {} ) diff --git a/homeassistant/components/bluetooth_tracker/device_tracker.py b/homeassistant/components/bluetooth_tracker/device_tracker.py index 9bc2e630ba0..f00bd672892 100644 --- a/homeassistant/components/bluetooth_tracker/device_tracker.py +++ b/homeassistant/components/bluetooth_tracker/device_tracker.py @@ -1,7 +1,8 @@ """Tracking for bluetooth devices.""" +from __future__ import annotations + import asyncio import logging -from typing import List, Optional, Set, Tuple import bluetooth # pylint: disable=import-error from bt_proximity import BluetoothRSSI @@ -50,7 +51,7 @@ def is_bluetooth_device(device) -> bool: return device.mac and device.mac[:3].upper() == BT_PREFIX -def discover_devices(device_id: int) -> List[Tuple[str, str]]: +def discover_devices(device_id: int) -> list[tuple[str, str]]: """Discover Bluetooth devices.""" result = bluetooth.discover_devices( duration=8, @@ -79,7 +80,7 @@ async def see_device( ) -async def get_tracking_devices(hass: HomeAssistantType) -> Tuple[Set[str], Set[str]]: +async def get_tracking_devices(hass: HomeAssistantType) -> tuple[set[str], set[str]]: """ Load all known devices. @@ -90,17 +91,17 @@ async def get_tracking_devices(hass: HomeAssistantType) -> Tuple[Set[str], Set[s devices = await async_load_config(yaml_path, hass, 0) bluetooth_devices = [device for device in devices if is_bluetooth_device(device)] - devices_to_track: Set[str] = { + devices_to_track: set[str] = { device.mac[3:] for device in bluetooth_devices if device.track } - devices_to_not_track: Set[str] = { + devices_to_not_track: set[str] = { device.mac[3:] for device in bluetooth_devices if not device.track } return devices_to_track, devices_to_not_track -def lookup_name(mac: str) -> Optional[str]: +def lookup_name(mac: str) -> str | None: """Lookup a Bluetooth device name.""" _LOGGER.debug("Scanning %s", mac) return bluetooth.lookup_name(mac, timeout=5) diff --git a/homeassistant/components/bond/config_flow.py b/homeassistant/components/bond/config_flow.py index af889b803b5..71ad54ee35c 100644 --- a/homeassistant/components/bond/config_flow.py +++ b/homeassistant/components/bond/config_flow.py @@ -1,6 +1,8 @@ """Config flow for Bond integration.""" +from __future__ import annotations + import logging -from typing import Any, Dict, Optional, Tuple +from typing import Any from aiohttp import ClientConnectionError, ClientResponseError from bond_api import Bond @@ -28,7 +30,7 @@ DISCOVERY_SCHEMA = vol.Schema({vol.Required(CONF_ACCESS_TOKEN): str}) TOKEN_SCHEMA = vol.Schema({}) -async def _validate_input(data: Dict[str, Any]) -> Tuple[str, str]: +async def _validate_input(data: dict[str, Any]) -> tuple[str, str]: """Validate the user input allows us to connect.""" bond = Bond(data[CONF_HOST], data[CONF_ACCESS_TOKEN]) @@ -60,7 +62,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): def __init__(self) -> None: """Initialize config flow.""" - self._discovered: Dict[str, str] = {} + self._discovered: dict[str, str] = {} async def _async_try_automatic_configure(self) -> None: """Try to auto configure the device. @@ -83,7 +85,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): _, hub_name = await _validate_input(self._discovered) self._discovered[CONF_NAME] = hub_name - async def async_step_zeroconf(self, discovery_info: DiscoveryInfoType) -> Dict[str, Any]: # type: ignore + async def async_step_zeroconf(self, discovery_info: DiscoveryInfoType) -> dict[str, Any]: # type: ignore """Handle a flow initialized by zeroconf discovery.""" name: str = discovery_info[CONF_NAME] host: str = discovery_info[CONF_HOST] @@ -106,8 +108,8 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): return await self.async_step_confirm() async def async_step_confirm( - self, user_input: Optional[Dict[str, Any]] = None - ) -> Dict[str, Any]: + self, user_input: dict[str, Any] | None = None + ) -> dict[str, Any]: """Handle confirmation flow for discovered bond hub.""" errors = {} if user_input is not None: @@ -147,8 +149,8 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): ) 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 a flow initialized by the user.""" errors = {} if user_input is not None: diff --git a/homeassistant/components/bond/cover.py b/homeassistant/components/bond/cover.py index 0c73bdbc8f9..60dcc4ec1f0 100644 --- a/homeassistant/components/bond/cover.py +++ b/homeassistant/components/bond/cover.py @@ -1,5 +1,7 @@ """Support for Bond covers.""" -from typing import Any, Callable, List, Optional +from __future__ import annotations + +from typing import Any, Callable from bond_api import Action, BPUPSubscriptions, DeviceType @@ -16,14 +18,14 @@ from .utils import BondDevice, BondHub 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 Bond cover devices.""" data = hass.data[DOMAIN][entry.entry_id] hub: BondHub = data[HUB] bpup_subs: BPUPSubscriptions = data[BPUP_SUBS] - covers: List[Entity] = [ + covers: list[Entity] = [ BondCover(hub, device, bpup_subs) for device in hub.devices if device.type == DeviceType.MOTORIZED_SHADES @@ -41,19 +43,19 @@ class BondCover(BondEntity, CoverEntity): """Create HA entity representing Bond cover.""" super().__init__(hub, device, bpup_subs) - self._closed: Optional[bool] = None + self._closed: bool | None = None def _apply_state(self, state: dict) -> None: cover_open = state.get("open") self._closed = True if cover_open == 0 else False if cover_open == 1 else None @property - def device_class(self) -> Optional[str]: + def device_class(self) -> str | None: """Get device class.""" return DEVICE_CLASS_SHADE @property - def is_closed(self) -> Optional[bool]: + def is_closed(self) -> bool | None: """Return if the cover is closed or not.""" return self._closed diff --git a/homeassistant/components/bond/entity.py b/homeassistant/components/bond/entity.py index b56c87f692f..a676d99e9ad 100644 --- a/homeassistant/components/bond/entity.py +++ b/homeassistant/components/bond/entity.py @@ -1,9 +1,11 @@ """An abstract class common to all Bond entities.""" +from __future__ import annotations + from abc import abstractmethod from asyncio import Lock, TimeoutError as AsyncIOTimeoutError from datetime import timedelta import logging -from typing import Any, Dict, Optional +from typing import Any from aiohttp import ClientError from bond_api import BPUPSubscriptions @@ -29,7 +31,7 @@ class BondEntity(Entity): hub: BondHub, device: BondDevice, bpup_subs: BPUPSubscriptions, - sub_device: Optional[str] = None, + sub_device: str | None = None, ): """Initialize entity with API and device info.""" self._hub = hub @@ -38,11 +40,11 @@ class BondEntity(Entity): self._sub_device = sub_device self._available = True self._bpup_subs = bpup_subs - self._update_lock: Optional[Lock] = None + self._update_lock: Lock | None = None self._initialized = False @property - def unique_id(self) -> Optional[str]: + def unique_id(self) -> str | None: """Get unique ID for the entity.""" hub_id = self._hub.bond_id device_id = self._device_id @@ -50,7 +52,7 @@ class BondEntity(Entity): return f"{hub_id}_{device_id}{sub_device_id}" @property - def name(self) -> Optional[str]: + def name(self) -> str | None: """Get entity name.""" if self._sub_device: sub_device_name = self._sub_device.replace("_", " ").title() @@ -63,7 +65,7 @@ class BondEntity(Entity): return False @property - def device_info(self) -> Optional[Dict[str, Any]]: + def device_info(self) -> dict[str, Any] | None: """Get a an HA device representing this Bond controlled device.""" device_info = { ATTR_NAME: self.name, diff --git a/homeassistant/components/bond/fan.py b/homeassistant/components/bond/fan.py index 1c94a6f3e9a..817cf0f99a2 100644 --- a/homeassistant/components/bond/fan.py +++ b/homeassistant/components/bond/fan.py @@ -1,7 +1,9 @@ """Support for Bond fans.""" +from __future__ import annotations + import logging import math -from typing import Any, Callable, List, Optional, Tuple +from typing import Any, Callable from bond_api import Action, BPUPSubscriptions, DeviceType, Direction @@ -31,14 +33,14 @@ _LOGGER = logging.getLogger(__name__) 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 Bond fan devices.""" data = hass.data[DOMAIN][entry.entry_id] hub: BondHub = data[HUB] bpup_subs: BPUPSubscriptions = data[BPUP_SUBS] - fans: List[Entity] = [ + fans: list[Entity] = [ BondFan(hub, device, bpup_subs) for device in hub.devices if DeviceType.is_fan(device.type) @@ -54,9 +56,9 @@ class BondFan(BondEntity, FanEntity): """Create HA entity representing Bond fan.""" super().__init__(hub, device, bpup_subs) - self._power: Optional[bool] = None - self._speed: Optional[int] = None - self._direction: Optional[int] = None + self._power: bool | None = None + self._speed: int | None = None + self._direction: int | None = None def _apply_state(self, state: dict) -> None: self._power = state.get("power") @@ -75,7 +77,7 @@ class BondFan(BondEntity, FanEntity): return features @property - def _speed_range(self) -> Tuple[int, int]: + def _speed_range(self) -> tuple[int, int]: """Return the range of speeds.""" return (1, self._device.props.get("max_speed", 3)) @@ -92,7 +94,7 @@ class BondFan(BondEntity, FanEntity): return int_states_in_range(self._speed_range) @property - def current_direction(self) -> Optional[str]: + def current_direction(self) -> str | None: """Return fan rotation direction.""" direction = None if self._direction == Direction.FORWARD: @@ -125,9 +127,9 @@ class BondFan(BondEntity, 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 on the fan.""" diff --git a/homeassistant/components/bond/light.py b/homeassistant/components/bond/light.py index d2b06012ed3..8faab26f785 100644 --- a/homeassistant/components/bond/light.py +++ b/homeassistant/components/bond/light.py @@ -1,6 +1,8 @@ """Support for Bond lights.""" +from __future__ import annotations + import logging -from typing import Any, Callable, List, Optional +from typing import Any, Callable from bond_api import Action, BPUPSubscriptions, DeviceType @@ -24,14 +26,14 @@ _LOGGER = logging.getLogger(__name__) 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 Bond light devices.""" data = hass.data[DOMAIN][entry.entry_id] hub: BondHub = data[HUB] bpup_subs: BPUPSubscriptions = data[BPUP_SUBS] - fan_lights: List[Entity] = [ + fan_lights: list[Entity] = [ BondLight(hub, device, bpup_subs) for device in hub.devices if DeviceType.is_fan(device.type) @@ -39,31 +41,31 @@ async def async_setup_entry( and not (device.supports_up_light() and device.supports_down_light()) ] - fan_up_lights: List[Entity] = [ + fan_up_lights: list[Entity] = [ BondUpLight(hub, device, bpup_subs, "up_light") for device in hub.devices if DeviceType.is_fan(device.type) and device.supports_up_light() ] - fan_down_lights: List[Entity] = [ + fan_down_lights: list[Entity] = [ BondDownLight(hub, device, bpup_subs, "down_light") for device in hub.devices if DeviceType.is_fan(device.type) and device.supports_down_light() ] - fireplaces: List[Entity] = [ + fireplaces: list[Entity] = [ BondFireplace(hub, device, bpup_subs) for device in hub.devices if DeviceType.is_fireplace(device.type) ] - fp_lights: List[Entity] = [ + fp_lights: list[Entity] = [ BondLight(hub, device, bpup_subs, "light") for device in hub.devices if DeviceType.is_fireplace(device.type) and device.supports_light() ] - lights: List[Entity] = [ + lights: list[Entity] = [ BondLight(hub, device, bpup_subs) for device in hub.devices if DeviceType.is_light(device.type) @@ -83,11 +85,11 @@ class BondBaseLight(BondEntity, LightEntity): hub: BondHub, device: BondDevice, bpup_subs: BPUPSubscriptions, - sub_device: Optional[str] = None, + sub_device: str | None = None, ): """Create HA entity representing Bond light.""" super().__init__(hub, device, bpup_subs, sub_device) - self._light: Optional[int] = None + self._light: int | None = None @property def is_on(self) -> bool: @@ -108,11 +110,11 @@ class BondLight(BondBaseLight, BondEntity, LightEntity): hub: BondHub, device: BondDevice, bpup_subs: BPUPSubscriptions, - sub_device: Optional[str] = None, + sub_device: str | None = None, ): """Create HA entity representing Bond light.""" super().__init__(hub, device, bpup_subs, sub_device) - self._brightness: Optional[int] = None + self._brightness: int | None = None def _apply_state(self, state: dict) -> None: self._light = state.get("light") @@ -126,7 +128,7 @@ class BondLight(BondBaseLight, BondEntity, LightEntity): return 0 @property - def brightness(self) -> Optional[int]: + def brightness(self) -> int | None: """Return the brightness of this light between 1..255.""" brightness_value = ( round(self._brightness * 255 / 100) if self._brightness else None @@ -194,9 +196,9 @@ class BondFireplace(BondEntity, LightEntity): """Create HA entity representing Bond fireplace.""" super().__init__(hub, device, bpup_subs) - self._power: Optional[bool] = None + self._power: bool | None = None # Bond flame level, 0-100 - self._flame: Optional[int] = None + self._flame: int | None = None def _apply_state(self, state: dict) -> None: self._power = state.get("power") @@ -230,11 +232,11 @@ class BondFireplace(BondEntity, LightEntity): await self._hub.bond.action(self._device.device_id, Action.turn_off()) @property - def brightness(self) -> Optional[int]: + def brightness(self) -> int | None: """Return the flame of this fireplace converted to HA brightness between 0..255.""" return round(self._flame * 255 / 100) if self._flame else None @property - def icon(self) -> Optional[str]: + def icon(self) -> str | None: """Show fireplace icon for the entity.""" return "mdi:fireplace" if self._power == 1 else "mdi:fireplace-off" diff --git a/homeassistant/components/bond/switch.py b/homeassistant/components/bond/switch.py index abbc2e2b44c..23e99d6af30 100644 --- a/homeassistant/components/bond/switch.py +++ b/homeassistant/components/bond/switch.py @@ -1,5 +1,7 @@ """Support for Bond generic devices.""" -from typing import Any, Callable, List, Optional +from __future__ import annotations + +from typing import Any, Callable from bond_api import Action, BPUPSubscriptions, DeviceType @@ -16,14 +18,14 @@ from .utils import BondDevice, BondHub 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 Bond generic devices.""" data = hass.data[DOMAIN][entry.entry_id] hub: BondHub = data[HUB] bpup_subs: BPUPSubscriptions = data[BPUP_SUBS] - switches: List[Entity] = [ + switches: list[Entity] = [ BondSwitch(hub, device, bpup_subs) for device in hub.devices if DeviceType.is_generic(device.type) @@ -39,7 +41,7 @@ class BondSwitch(BondEntity, SwitchEntity): """Create HA entity representing Bond generic device (switch).""" super().__init__(hub, device, bpup_subs) - self._power: Optional[bool] = None + self._power: bool | None = None def _apply_state(self, state: dict) -> None: self._power = state.get("power") diff --git a/homeassistant/components/bond/utils.py b/homeassistant/components/bond/utils.py index 28580ae415e..e3d951b8137 100644 --- a/homeassistant/components/bond/utils.py +++ b/homeassistant/components/bond/utils.py @@ -1,7 +1,9 @@ """Reusable utilities for the Bond component.""" +from __future__ import annotations + import asyncio import logging -from typing import Any, Dict, List, Optional, Set, cast +from typing import Any, cast from aiohttp import ClientResponseError from bond_api import Action, Bond @@ -15,7 +17,7 @@ class BondDevice: """Helper device class to hold ID and attributes together.""" def __init__( - self, device_id: str, attrs: Dict[str, Any], props: Dict[str, Any] + self, device_id: str, attrs: dict[str, Any], props: dict[str, Any] ) -> None: """Create a helper device from ID and attributes returned by API.""" self.device_id = device_id @@ -41,17 +43,17 @@ class BondDevice: return cast(str, self._attrs["type"]) @property - def location(self) -> Optional[str]: + def location(self) -> str | None: """Get the location of this device.""" return self._attrs.get("location") @property - def template(self) -> Optional[str]: + def template(self) -> str | None: """Return this model template.""" return self._attrs.get("template") @property - def branding_profile(self) -> Optional[str]: + def branding_profile(self) -> str | None: """Return this branding profile.""" return self.props.get("branding_profile") @@ -60,9 +62,9 @@ class BondDevice: """Check if Trust State is turned on.""" return self.props.get("trust_state", False) - def _has_any_action(self, actions: Set[str]) -> bool: + def _has_any_action(self, actions: set[str]) -> bool: """Check to see if the device supports any of the actions.""" - supported_actions: List[str] = self._attrs["actions"] + supported_actions: list[str] = self._attrs["actions"] for action in supported_actions: if action in actions: return True @@ -101,11 +103,11 @@ class BondHub: def __init__(self, bond: Bond): """Initialize Bond Hub.""" self.bond: Bond = bond - self._bridge: Dict[str, Any] = {} - self._version: Dict[str, Any] = {} - self._devices: List[BondDevice] = [] + self._bridge: dict[str, Any] = {} + self._version: dict[str, Any] = {} + self._devices: list[BondDevice] = [] - async def setup(self, max_devices: Optional[int] = None) -> None: + async def setup(self, max_devices: int | None = None) -> None: """Read hub version information.""" self._version = await self.bond.version() _LOGGER.debug("Bond reported the following version info: %s", self._version) @@ -131,18 +133,18 @@ class BondHub: _LOGGER.debug("Bond reported the following bridge info: %s", self._bridge) @property - def bond_id(self) -> Optional[str]: + def bond_id(self) -> str | None: """Return unique Bond ID for this hub.""" # Old firmwares are missing the bondid return self._version.get("bondid") @property - def target(self) -> Optional[str]: + def target(self) -> str | None: """Return this hub target.""" return self._version.get("target") @property - def model(self) -> Optional[str]: + def model(self) -> str | None: """Return this hub model.""" return self._version.get("model") @@ -159,19 +161,19 @@ class BondHub: return cast(str, self._bridge["name"]) @property - def location(self) -> Optional[str]: + def location(self) -> str | None: """Get the location of this bridge.""" if not self.is_bridge and self._devices: return self._devices[0].location return self._bridge.get("location") @property - def fw_ver(self) -> Optional[str]: + def fw_ver(self) -> str | None: """Return this hub firmware version.""" return self._version.get("fw_ver") @property - def devices(self) -> List[BondDevice]: + def devices(self) -> list[BondDevice]: """Return a list of all devices controlled by this hub.""" return self._devices diff --git a/homeassistant/components/bsblan/climate.py b/homeassistant/components/bsblan/climate.py index a97c13c3424..4d83fb04dbe 100644 --- a/homeassistant/components/bsblan/climate.py +++ b/homeassistant/components/bsblan/climate.py @@ -1,7 +1,9 @@ """BSBLAN platform to control a compatible Climate Device.""" +from __future__ import annotations + from datetime import timedelta import logging -from typing import Any, Callable, Dict, List, Optional +from typing import Any, Callable from bsblan import BSBLan, BSBLanError, Info, State @@ -74,7 +76,7 @@ BSBLAN_TO_HA_PRESET = { 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 BSBLan device based on a config entry.""" bsblan: BSBLan = hass.data[DOMAIN][entry.entry_id][DATA_BSBLAN_CLIENT] @@ -92,10 +94,10 @@ class BSBLanClimate(ClimateEntity): info: Info, ): """Initialize BSBLan climate device.""" - self._current_temperature: Optional[float] = None + self._current_temperature: float | None = None self._available = True - self._hvac_mode: Optional[str] = None - self._target_temperature: Optional[float] = None + self._hvac_mode: str | None = None + self._target_temperature: float | None = None self._temperature_unit = None self._preset_mode = None self._store_hvac_mode = None @@ -229,7 +231,7 @@ class BSBLanClimate(ClimateEntity): self._temperature_unit = state.current_temperature.unit @property - def device_info(self) -> Dict[str, Any]: + def device_info(self) -> dict[str, Any]: """Return device information about this BSBLan device.""" return { ATTR_IDENTIFIERS: {(DOMAIN, self._info.device_identification)}, diff --git a/homeassistant/components/bsblan/config_flow.py b/homeassistant/components/bsblan/config_flow.py index dee04e6ef85..8ea597d2386 100644 --- a/homeassistant/components/bsblan/config_flow.py +++ b/homeassistant/components/bsblan/config_flow.py @@ -1,6 +1,8 @@ """Config flow for BSB-Lan integration.""" +from __future__ import annotations + import logging -from typing import Any, Dict, Optional +from typing import Any from bsblan import BSBLan, BSBLanError, Info import voluptuous as vol @@ -26,8 +28,8 @@ class BSBLanFlowHandler(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.""" if user_input is None: return self._show_setup_form() @@ -59,7 +61,7 @@ class BSBLanFlowHandler(ConfigFlow, domain=DOMAIN): }, ) - 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", @@ -78,9 +80,9 @@ class BSBLanFlowHandler(ConfigFlow, domain=DOMAIN): async def _get_bsblan_info( self, host: str, - username: Optional[str], - password: Optional[str], - passkey: Optional[str], + username: str | None, + password: str | None, + passkey: str | None, port: int, ) -> Info: """Get device information from an BSBLan device.""" diff --git a/homeassistant/components/buienradar/camera.py b/homeassistant/components/buienradar/camera.py index 44f86589b27..92f25b7ffc6 100644 --- a/homeassistant/components/buienradar/camera.py +++ b/homeassistant/components/buienradar/camera.py @@ -1,8 +1,9 @@ """Provide animated GIF loops of Buienradar imagery.""" +from __future__ import annotations + import asyncio from datetime import datetime, timedelta import logging -from typing import Optional import aiohttp import voluptuous as vol @@ -85,13 +86,13 @@ class BuienradarCam(Camera): # invariant: this condition is private to and owned by this instance. self._condition = asyncio.Condition() - self._last_image: Optional[bytes] = None + self._last_image: bytes | None = None # value of the last seen last modified header - self._last_modified: Optional[str] = None + self._last_modified: str | None = None # loading status self._loading = False # deadline for image refresh - self.delta after last successful load - self._deadline: Optional[datetime] = None + self._deadline: datetime | None = None self._unique_id = f"{self._dimension}_{self._country}" @@ -140,7 +141,7 @@ class BuienradarCam(Camera): _LOGGER.error("Failed to fetch image, %s", type(err)) return False - async def async_camera_image(self) -> Optional[bytes]: + async def async_camera_image(self) -> bytes | None: """ Return a still image response from the camera.