diff --git a/homeassistant/components/jewish_calendar/__init__.py b/homeassistant/components/jewish_calendar/__init__.py index d1474c3cf5f..35c1505561d 100644 --- a/homeassistant/components/jewish_calendar/__init__.py +++ b/homeassistant/components/jewish_calendar/__init__.py @@ -1,5 +1,5 @@ """The jewish_calendar component.""" -from typing import Optional +from __future__ import annotations import hdate import voluptuous as vol @@ -78,8 +78,8 @@ CONFIG_SCHEMA = vol.Schema( def get_unique_prefix( location: hdate.Location, language: str, - candle_lighting_offset: Optional[int], - havdalah_offset: Optional[int], + candle_lighting_offset: int | None, + havdalah_offset: int | None, ) -> str: """Create a prefix for unique ids.""" config_properties = [ diff --git a/homeassistant/components/keenetic_ndms2/config_flow.py b/homeassistant/components/keenetic_ndms2/config_flow.py index 9338cb05935..a832f68e017 100644 --- a/homeassistant/components/keenetic_ndms2/config_flow.py +++ b/homeassistant/components/keenetic_ndms2/config_flow.py @@ -1,5 +1,5 @@ """Config flow for Keenetic NDMS2.""" -from typing import List +from __future__ import annotations from ndms2_client import Client, ConnectionException, InterfaceInfo, TelnetConnection import voluptuous as vol @@ -103,7 +103,7 @@ class KeeneticOptionsFlowHandler(config_entries.OptionsFlow): ROUTER ] - interfaces: List[InterfaceInfo] = await self.hass.async_add_executor_job( + interfaces: list[InterfaceInfo] = await self.hass.async_add_executor_job( router.client.get_interfaces ) diff --git a/homeassistant/components/keenetic_ndms2/device_tracker.py b/homeassistant/components/keenetic_ndms2/device_tracker.py index a8b0f943dd6..461814b1917 100644 --- a/homeassistant/components/keenetic_ndms2/device_tracker.py +++ b/homeassistant/components/keenetic_ndms2/device_tracker.py @@ -1,7 +1,8 @@ """Support for Keenetic routers as device tracker.""" +from __future__ import annotations + from datetime import timedelta import logging -from typing import List, Optional, Set from ndms2_client import Device import voluptuous as vol @@ -57,8 +58,8 @@ async def async_get_scanner(hass: HomeAssistant, config): """Import legacy configuration from YAML.""" scanner_config = config[DEVICE_TRACKER_DOMAIN] - scan_interval: Optional[timedelta] = scanner_config.get(CONF_SCAN_INTERVAL) - consider_home: Optional[timedelta] = scanner_config.get(CONF_CONSIDER_HOME) + scan_interval: timedelta | None = scanner_config.get(CONF_SCAN_INTERVAL) + consider_home: timedelta | None = scanner_config.get(CONF_CONSIDER_HOME) host: str = scanner_config[CONF_HOST] hass.data[DOMAIN][f"imported_options_{host}"] = { @@ -139,9 +140,9 @@ async def async_setup_entry( @callback -def update_items(router: KeeneticRouter, async_add_entities, tracked: Set[str]): +def update_items(router: KeeneticRouter, async_add_entities, tracked: set[str]): """Update tracked device state from the hub.""" - new_tracked: List[KeeneticTracker] = [] + new_tracked: list[KeeneticTracker] = [] for mac, device in router.last_devices.items(): if mac not in tracked: tracked.add(mac) diff --git a/homeassistant/components/keenetic_ndms2/router.py b/homeassistant/components/keenetic_ndms2/router.py index 340b25ff725..0066b49223b 100644 --- a/homeassistant/components/keenetic_ndms2/router.py +++ b/homeassistant/components/keenetic_ndms2/router.py @@ -1,7 +1,9 @@ """The Keenetic Client class.""" +from __future__ import annotations + from datetime import timedelta import logging -from typing import Callable, Dict, Optional +from typing import Callable from ndms2_client import Client, ConnectionException, Device, TelnetConnection from ndms2_client.client import RouterInfo @@ -39,11 +41,11 @@ class KeeneticRouter: """Initialize the Client.""" self.hass = hass self.config_entry = config_entry - self._last_devices: Dict[str, Device] = {} - self._router_info: Optional[RouterInfo] = None - self._connection: Optional[TelnetConnection] = None - self._client: Optional[Client] = None - self._cancel_periodic_update: Optional[Callable] = None + self._last_devices: dict[str, Device] = {} + self._router_info: RouterInfo | None = None + self._connection: TelnetConnection | None = None + self._client: Client | None = None + self._cancel_periodic_update: Callable | None = None self._available = False self._progress = None diff --git a/homeassistant/components/knx/__init__.py b/homeassistant/components/knx/__init__.py index e05c18e5d5c..348eac8f40e 100644 --- a/homeassistant/components/knx/__init__.py +++ b/homeassistant/components/knx/__init__.py @@ -1,7 +1,8 @@ """Support KNX devices.""" +from __future__ import annotations + import asyncio import logging -from typing import Union import voluptuous as vol from xknx import XKNX @@ -466,7 +467,7 @@ class KNXModule: attr_payload = call.data.get(SERVICE_KNX_ATTR_PAYLOAD) attr_type = call.data.get(SERVICE_KNX_ATTR_TYPE) - payload: Union[DPTBinary, DPTArray] + payload: DPTBinary | DPTArray if attr_type is not None: transcoder = DPTBase.parse_transcoder(attr_type) if transcoder is None: diff --git a/homeassistant/components/knx/binary_sensor.py b/homeassistant/components/knx/binary_sensor.py index ecb79664afd..6ee37abee18 100644 --- a/homeassistant/components/knx/binary_sensor.py +++ b/homeassistant/components/knx/binary_sensor.py @@ -1,5 +1,7 @@ """Support for KNX/IP binary sensors.""" -from typing import Any, Dict, Optional +from __future__ import annotations + +from typing import Any from xknx.devices import BinarySensor as XknxBinarySensor @@ -38,7 +40,7 @@ class KNXBinarySensor(KnxEntity, BinarySensorEntity): return self._device.is_on() @property - def extra_state_attributes(self) -> Optional[Dict[str, Any]]: + def extra_state_attributes(self) -> dict[str, Any] | None: """Return device specific state attributes.""" if self._device.counter is not None: return {ATTR_COUNTER: self._device.counter} diff --git a/homeassistant/components/knx/climate.py b/homeassistant/components/knx/climate.py index 565c41298a3..e90371e3282 100644 --- a/homeassistant/components/knx/climate.py +++ b/homeassistant/components/knx/climate.py @@ -1,5 +1,5 @@ """Support for KNX/IP climate devices.""" -from typing import List, Optional +from __future__ import annotations from xknx.devices import Climate as XknxClimate from xknx.dpt.dpt_hvac_mode import HVACControllerMode, HVACOperationMode @@ -88,7 +88,7 @@ class KNXClimate(KnxEntity, ClimateEntity): self.async_write_ha_state() @property - def hvac_mode(self) -> Optional[str]: + def hvac_mode(self) -> str | None: """Return current operation ie. heat, cool, idle.""" if self._device.supports_on_off and not self._device.is_on: return HVAC_MODE_OFF @@ -100,7 +100,7 @@ class KNXClimate(KnxEntity, ClimateEntity): return HVAC_MODE_HEAT @property - def hvac_modes(self) -> Optional[List[str]]: + def hvac_modes(self) -> list[str] | None: """Return the list of available operation/controller modes.""" _controller_modes = [ CONTROLLER_MODES.get(controller_mode.value) @@ -131,7 +131,7 @@ class KNXClimate(KnxEntity, ClimateEntity): self.async_write_ha_state() @property - def preset_mode(self) -> Optional[str]: + def preset_mode(self) -> str | None: """Return the current preset mode, e.g., home, away, temp. Requires SUPPORT_PRESET_MODE. @@ -141,7 +141,7 @@ class KNXClimate(KnxEntity, ClimateEntity): return None @property - def preset_modes(self) -> Optional[List[str]]: + def preset_modes(self) -> list[str] | None: """Return a list of available preset modes. Requires SUPPORT_PRESET_MODE. diff --git a/homeassistant/components/knx/expose.py b/homeassistant/components/knx/expose.py index 3a010810862..8bdd3d1d1d1 100644 --- a/homeassistant/components/knx/expose.py +++ b/homeassistant/components/knx/expose.py @@ -1,5 +1,5 @@ """Exposures to KNX bus.""" -from typing import Union +from __future__ import annotations from xknx import XKNX from xknx.devices import DateTime, ExposeSensor @@ -22,7 +22,7 @@ from .schema import ExposeSchema @callback def create_knx_exposure( hass: HomeAssistant, xknx: XKNX, config: ConfigType -) -> Union["KNXExposeSensor", "KNXExposeTime"]: +) -> KNXExposeSensor | KNXExposeTime: """Create exposures from config.""" address = config[KNX_ADDRESS] attribute = config.get(ExposeSchema.CONF_KNX_EXPOSE_ATTRIBUTE) @@ -30,7 +30,7 @@ def create_knx_exposure( expose_type = config.get(ExposeSchema.CONF_KNX_EXPOSE_TYPE) default = config.get(ExposeSchema.CONF_KNX_EXPOSE_DEFAULT) - exposure: Union["KNXExposeSensor", "KNXExposeTime"] + exposure: KNXExposeSensor | KNXExposeTime if expose_type.lower() in ["time", "date", "datetime"]: exposure = KNXExposeTime(xknx, expose_type, address) else: diff --git a/homeassistant/components/knx/factory.py b/homeassistant/components/knx/factory.py index 893ac1e55e3..0543119cadf 100644 --- a/homeassistant/components/knx/factory.py +++ b/homeassistant/components/knx/factory.py @@ -1,5 +1,5 @@ """Factory function to initialize KNX devices from config.""" -from typing import Optional, Tuple +from __future__ import annotations from xknx import XKNX from xknx.devices import ( @@ -95,7 +95,7 @@ def _create_cover(knx_module: XKNX, config: ConfigType) -> XknxCover: def _create_light_color( color: str, config: ConfigType -) -> Tuple[Optional[str], Optional[str], Optional[str], Optional[str]]: +) -> tuple[str | None, str | None, str | None, str | None]: """Load color configuration from configuration structure.""" if "individual_colors" in config and color in config["individual_colors"]: sub_config = config["individual_colors"][color] diff --git a/homeassistant/components/knx/fan.py b/homeassistant/components/knx/fan.py index 43d1cd7d6f2..2d9f48fe804 100644 --- a/homeassistant/components/knx/fan.py +++ b/homeassistant/components/knx/fan.py @@ -1,6 +1,8 @@ """Support for KNX/IP fans.""" +from __future__ import annotations + import math -from typing import Any, Optional +from typing import Any from xknx.devices import Fan as XknxFan from xknx.devices.fan import FanSpeedMode @@ -58,7 +60,7 @@ class KNXFan(KnxEntity, FanEntity): return flags @property - def percentage(self) -> Optional[int]: + def percentage(self) -> int | None: """Return the current speed as a percentage.""" if self._device.current_speed is None: return None @@ -78,9 +80,9 @@ class KNXFan(KnxEntity, 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, ) -> None: """Turn on the fan.""" diff --git a/homeassistant/components/knx/notify.py b/homeassistant/components/knx/notify.py index 7210795bd71..9d6ed35e36b 100644 --- a/homeassistant/components/knx/notify.py +++ b/homeassistant/components/knx/notify.py @@ -1,5 +1,5 @@ """Support for KNX/IP notification services.""" -from typing import List +from __future__ import annotations from xknx.devices import Notification as XknxNotification @@ -22,7 +22,7 @@ async def async_get_service(hass, config, discovery_info=None): class KNXNotificationService(BaseNotificationService): """Implement demo notification service.""" - def __init__(self, devices: List[XknxNotification]): + def __init__(self, devices: list[XknxNotification]): """Initialize the service.""" self.devices = devices diff --git a/homeassistant/components/kodi/device_trigger.py b/homeassistant/components/kodi/device_trigger.py index 314f73a927e..3454fc122ed 100644 --- a/homeassistant/components/kodi/device_trigger.py +++ b/homeassistant/components/kodi/device_trigger.py @@ -1,5 +1,5 @@ """Provides device automations for Kodi.""" -from typing import List +from __future__ import annotations import voluptuous as vol @@ -29,7 +29,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 Kodi devices.""" registry = await entity_registry.async_get_registry(hass) triggers = [] diff --git a/homeassistant/components/kulersky/light.py b/homeassistant/components/kulersky/light.py index 9098975d500..599f99a83e8 100644 --- a/homeassistant/components/kulersky/light.py +++ b/homeassistant/components/kulersky/light.py @@ -1,7 +1,9 @@ """Kuler Sky light platform.""" +from __future__ import annotations + from datetime import timedelta import logging -from typing import Callable, List +from typing import Callable import pykulersky @@ -34,7 +36,7 @@ DISCOVERY_INTERVAL = timedelta(seconds=60) 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 Kuler sky light devices.""" if DOMAIN not in hass.data: diff --git a/homeassistant/components/launch_library/sensor.py b/homeassistant/components/launch_library/sensor.py index 366cd4e7d44..7663bf86fc5 100644 --- a/homeassistant/components/launch_library/sensor.py +++ b/homeassistant/components/launch_library/sensor.py @@ -1,7 +1,8 @@ """A sensor platform that give you information about the next space launch.""" +from __future__ import annotations + from datetime import timedelta import logging -from typing import Optional from pylaunches import PyLaunches, PyLaunchesException import voluptuous as vol @@ -64,7 +65,7 @@ class LaunchLibrarySensor(Entity): return self._name @property - def state(self) -> Optional[str]: + def state(self) -> str | None: """Return the state of the sensor.""" if self.next_launch: return self.next_launch.name @@ -76,7 +77,7 @@ class LaunchLibrarySensor(Entity): return "mdi:rocket" @property - def extra_state_attributes(self) -> Optional[dict]: + def extra_state_attributes(self) -> dict | None: """Return attributes for the sensor.""" if self.next_launch: return { diff --git a/homeassistant/components/light/__init__.py b/homeassistant/components/light/__init__.py index 56fd5841388..c2e2fdbeaa9 100644 --- a/homeassistant/components/light/__init__.py +++ b/homeassistant/components/light/__init__.py @@ -6,7 +6,7 @@ import dataclasses from datetime import timedelta import logging import os -from typing import Dict, List, Optional, Set, Tuple, cast +from typing import cast import voluptuous as vol @@ -364,11 +364,11 @@ class Profile: """Representation of a profile.""" name: str - color_x: Optional[float] = dataclasses.field(repr=False) - color_y: Optional[float] = dataclasses.field(repr=False) - brightness: Optional[int] - transition: Optional[int] = None - hs_color: Optional[Tuple[float, float]] = dataclasses.field(init=False) + color_x: float | None = dataclasses.field(repr=False) + color_y: float | None = dataclasses.field(repr=False) + brightness: int | None + transition: int | None = None + hs_color: tuple[float, float] | None = dataclasses.field(init=False) SCHEMA = vol.Schema( # pylint: disable=invalid-name vol.Any( @@ -403,7 +403,7 @@ class Profile: ) @classmethod - def from_csv_row(cls, csv_row: List[str]) -> Profile: + def from_csv_row(cls, csv_row: list[str]) -> Profile: """Create profile from a CSV row tuple.""" return cls(*cls.SCHEMA(csv_row)) @@ -414,9 +414,9 @@ class Profiles: def __init__(self, hass: HomeAssistantType): """Initialize profiles.""" self.hass = hass - self.data: Dict[str, Profile] = {} + self.data: dict[str, Profile] = {} - def _load_profile_data(self) -> Dict[str, Profile]: + def _load_profile_data(self) -> dict[str, Profile]: """Load built-in profiles and custom profiles.""" profile_paths = [ os.path.join(os.path.dirname(__file__), LIGHT_PROFILES_FILE), @@ -453,7 +453,7 @@ class Profiles: self.data = await self.hass.async_add_executor_job(self._load_profile_data) @callback - def apply_default(self, entity_id: str, params: Dict) -> None: + def apply_default(self, entity_id: str, params: dict) -> None: """Return the default turn-on profile for the given light.""" for _entity_id in (entity_id, "group.all_lights"): name = f"{_entity_id}.default" @@ -462,7 +462,7 @@ class Profiles: return @callback - def apply_profile(self, name: str, params: Dict) -> None: + def apply_profile(self, name: str, params: dict) -> None: """Apply a profile.""" profile = self.data.get(name) @@ -481,12 +481,12 @@ class LightEntity(ToggleEntity): """Representation of a light.""" @property - def brightness(self) -> Optional[int]: + def brightness(self) -> int | None: """Return the brightness of this light between 0..255.""" return None @property - def color_mode(self) -> Optional[str]: + def color_mode(self) -> str | None: """Return the color mode of the light.""" return None @@ -519,27 +519,27 @@ class LightEntity(ToggleEntity): return color_mode @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].""" return None @property - def xy_color(self) -> Optional[Tuple[float, float]]: + def xy_color(self) -> tuple[float, float] | None: """Return the xy color value [float, float].""" return None @property - def rgb_color(self) -> Optional[Tuple[int, int, int]]: + def rgb_color(self) -> tuple[int, int, int] | None: """Return the rgb color value [int, int, int].""" return None @property - def rgbw_color(self) -> Optional[Tuple[int, int, int, int]]: + def rgbw_color(self) -> tuple[int, int, int, int] | None: """Return the rgbw color value [int, int, int, int].""" return None @property - def _light_internal_rgbw_color(self) -> Optional[Tuple[int, int, int, int]]: + def _light_internal_rgbw_color(self) -> tuple[int, int, int, int] | None: """Return the rgbw color value [int, int, int, int].""" rgbw_color = self.rgbw_color if ( @@ -558,12 +558,12 @@ class LightEntity(ToggleEntity): return rgbw_color @property - def rgbww_color(self) -> Optional[Tuple[int, int, int, int, int]]: + def rgbww_color(self) -> tuple[int, int, int, int, int] | None: """Return the rgbww color value [int, int, int, int, int].""" return None @property - def color_temp(self) -> Optional[int]: + def color_temp(self) -> int | None: """Return the CT color value in mireds.""" return None @@ -582,17 +582,17 @@ class LightEntity(ToggleEntity): return 500 @property - def white_value(self) -> Optional[int]: + def white_value(self) -> int | None: """Return the white value of this light between 0..255.""" return None @property - def effect_list(self) -> Optional[List[str]]: + def effect_list(self) -> list[str] | None: """Return the list of supported effects.""" return None @property - def effect(self) -> Optional[str]: + def effect(self) -> str | None: """Return the current effect.""" return None @@ -616,7 +616,7 @@ class LightEntity(ToggleEntity): return data def _light_internal_convert_color(self, color_mode: str) -> dict: - data: Dict[str, Tuple] = {} + data: dict[str, tuple] = {} if color_mode == COLOR_MODE_HS and self.hs_color: hs_color = self.hs_color data[ATTR_HS_COLOR] = (round(hs_color[0], 3), round(hs_color[1], 3)) @@ -692,7 +692,7 @@ class LightEntity(ToggleEntity): return {key: val for key, val in data.items() if val is not None} @property - def _light_internal_supported_color_modes(self) -> Set: + def _light_internal_supported_color_modes(self) -> set: """Calculate supported color modes with backwards compatibility.""" supported_color_modes = self.supported_color_modes @@ -717,7 +717,7 @@ class LightEntity(ToggleEntity): return supported_color_modes @property - def supported_color_modes(self) -> Optional[Set]: + def supported_color_modes(self) -> set | None: """Flag supported color modes.""" return None diff --git a/homeassistant/components/light/device_action.py b/homeassistant/components/light/device_action.py index d499bc0c2a2..4c37647f168 100644 --- a/homeassistant/components/light/device_action.py +++ b/homeassistant/components/light/device_action.py @@ -1,5 +1,5 @@ """Provides device actions for lights.""" -from typing import List +from __future__ import annotations import voluptuous as vol @@ -78,7 +78,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.""" actions = await toggle_entity.async_get_actions(hass, device_id, DOMAIN) diff --git a/homeassistant/components/light/device_condition.py b/homeassistant/components/light/device_condition.py index 1d9323907f2..7396ddeea31 100644 --- a/homeassistant/components/light/device_condition.py +++ b/homeassistant/components/light/device_condition.py @@ -1,5 +1,5 @@ """Provides device conditions for lights.""" -from typing import Dict, List +from __future__ import annotations import voluptuous as vol @@ -28,7 +28,7 @@ def async_condition_from_config( async def async_get_conditions( hass: HomeAssistant, device_id: str -) -> List[Dict[str, str]]: +) -> list[dict[str, str]]: """List device conditions.""" return await toggle_entity.async_get_conditions(hass, device_id, DOMAIN) diff --git a/homeassistant/components/light/device_trigger.py b/homeassistant/components/light/device_trigger.py index 066d1f4c020..e1b14124831 100644 --- a/homeassistant/components/light/device_trigger.py +++ b/homeassistant/components/light/device_trigger.py @@ -1,5 +1,5 @@ """Provides device trigger for lights.""" -from typing import List +from __future__ import annotations import voluptuous as vol @@ -28,7 +28,7 @@ async def async_attach_trigger( ) -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.""" return await toggle_entity.async_get_triggers(hass, device_id, DOMAIN) diff --git a/homeassistant/components/light/reproduce_state.py b/homeassistant/components/light/reproduce_state.py index 863790cff71..13a014426a0 100644 --- a/homeassistant/components/light/reproduce_state.py +++ b/homeassistant/components/light/reproduce_state.py @@ -1,8 +1,10 @@ """Reproduce an Light state.""" +from __future__ import annotations + import asyncio import logging from types import MappingProxyType -from typing import Any, Dict, Iterable, Optional +from typing import Any, Iterable from homeassistant.const import ( ATTR_ENTITY_ID, @@ -95,8 +97,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) @@ -123,7 +125,7 @@ async def _async_reproduce_state( ): return - service_data: Dict[str, Any] = {ATTR_ENTITY_ID: state.entity_id} + service_data: dict[str, Any] = {ATTR_ENTITY_ID: state.entity_id} if reproduce_options is not None and ATTR_TRANSITION in reproduce_options: service_data[ATTR_TRANSITION] = reproduce_options[ATTR_TRANSITION] @@ -171,8 +173,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 Light states.""" await asyncio.gather( diff --git a/homeassistant/components/light/significant_change.py b/homeassistant/components/light/significant_change.py index a0bd5203101..9e0f10fae47 100644 --- a/homeassistant/components/light/significant_change.py +++ b/homeassistant/components/light/significant_change.py @@ -1,5 +1,7 @@ """Helper to test significant Light state changes.""" -from typing import Any, Optional +from __future__ import annotations + +from typing import Any from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.significant_change import ( @@ -24,7 +26,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/litejet/config_flow.py b/homeassistant/components/litejet/config_flow.py index e1c7d8ab7b9..124b229c786 100644 --- a/homeassistant/components/litejet/config_flow.py +++ b/homeassistant/components/litejet/config_flow.py @@ -1,6 +1,8 @@ """Config flow for the LiteJet lighting system.""" +from __future__ import annotations + import logging -from typing import Any, Dict, Optional +from typing import Any import pylitejet from serial import SerialException @@ -18,8 +20,8 @@ class LiteJetConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): """LiteJet config flow.""" 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]: """Create a LiteJet config entry based upon user input.""" if self.hass.config_entries.async_entries(DOMAIN): return self.async_abort(reason="single_instance_allowed") diff --git a/homeassistant/components/litterrobot/hub.py b/homeassistant/components/litterrobot/hub.py index 1847f37ed64..86c3aff5462 100644 --- a/homeassistant/components/litterrobot/hub.py +++ b/homeassistant/components/litterrobot/hub.py @@ -1,8 +1,10 @@ """A wrapper 'hub' for the Litter-Robot API and base entity for common attributes.""" +from __future__ import annotations + from datetime import time, timedelta import logging from types import MethodType -from typing import Any, Optional +from typing import Any import pylitterbot from pylitterbot.exceptions import LitterRobotException, LitterRobotLoginException @@ -106,7 +108,7 @@ class LitterRobotEntity(CoordinatorEntity): async_call_later(self.hass, REFRESH_WAIT_TIME, async_call_later_callback) @staticmethod - def parse_time_at_default_timezone(time_str: str) -> Optional[time]: + def parse_time_at_default_timezone(time_str: str) -> time | None: """Parse a time string and add default timezone.""" parsed_time = dt_util.parse_time(time_str) diff --git a/homeassistant/components/litterrobot/sensor.py b/homeassistant/components/litterrobot/sensor.py index 8900c6c54ca..8ae512fa801 100644 --- a/homeassistant/components/litterrobot/sensor.py +++ b/homeassistant/components/litterrobot/sensor.py @@ -1,5 +1,5 @@ """Support for Litter-Robot sensors.""" -from typing import Optional +from __future__ import annotations from pylitterbot.robot import Robot @@ -10,7 +10,7 @@ from .const import DOMAIN from .hub import LitterRobotEntity, LitterRobotHub -def icon_for_gauge_level(gauge_level: Optional[int] = None, offset: int = 0) -> str: +def icon_for_gauge_level(gauge_level: int | None = None, offset: int = 0) -> str: """Return a gauge icon valid identifier.""" if gauge_level is None or gauge_level <= 0 + offset: return "mdi:gauge-empty" diff --git a/homeassistant/components/locative/__init__.py b/homeassistant/components/locative/__init__.py index 28af822ae63..bb2a19c6380 100644 --- a/homeassistant/components/locative/__init__.py +++ b/homeassistant/components/locative/__init__.py @@ -1,6 +1,7 @@ """Support for Locative.""" +from __future__ import annotations + import logging -from typing import Dict from aiohttp import web import voluptuous as vol @@ -34,7 +35,7 @@ def _id(value: str) -> str: return value.replace("-", "") -def _validate_test_mode(obj: Dict) -> Dict: +def _validate_test_mode(obj: dict) -> dict: """Validate that id is provided outside of test mode.""" if ATTR_ID not in obj and obj[ATTR_TRIGGER] != "test": raise vol.Invalid("Location id not specified") diff --git a/homeassistant/components/lock/device_action.py b/homeassistant/components/lock/device_action.py index efdb5e352cf..639947f3b88 100644 --- a/homeassistant/components/lock/device_action.py +++ b/homeassistant/components/lock/device_action.py @@ -1,5 +1,5 @@ """Provides device automations for Lock.""" -from typing import List, Optional +from __future__ import annotations import voluptuous as vol @@ -30,7 +30,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 Lock devices.""" registry = await entity_registry.async_get_registry(hass) actions = [] @@ -75,7 +75,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/lock/device_condition.py b/homeassistant/components/lock/device_condition.py index a25018dc709..0fae680f829 100644 --- a/homeassistant/components/lock/device_condition.py +++ b/homeassistant/components/lock/device_condition.py @@ -1,5 +1,5 @@ """Provides device automations for Lock.""" -from typing import List +from __future__ import annotations import voluptuous as vol @@ -30,7 +30,7 @@ CONDITION_SCHEMA = DEVICE_CONDITION_BASE_SCHEMA.extend( ) -async def async_get_conditions(hass: HomeAssistant, device_id: str) -> List[dict]: +async def async_get_conditions(hass: HomeAssistant, device_id: str) -> list[dict]: """List device conditions for Lock devices.""" registry = await entity_registry.async_get_registry(hass) conditions = [] diff --git a/homeassistant/components/lock/device_trigger.py b/homeassistant/components/lock/device_trigger.py index 05d5041ca65..3e5bee49a22 100644 --- a/homeassistant/components/lock/device_trigger.py +++ b/homeassistant/components/lock/device_trigger.py @@ -1,5 +1,5 @@ """Provides device automations for Lock.""" -from typing import List +from __future__ import annotations import voluptuous as vol @@ -31,7 +31,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 Lock devices.""" registry = await entity_registry.async_get_registry(hass) triggers = [] diff --git a/homeassistant/components/lock/reproduce_state.py b/homeassistant/components/lock/reproduce_state.py index 812b9bf04df..0d575964b2b 100644 --- a/homeassistant/components/lock/reproduce_state.py +++ b/homeassistant/components/lock/reproduce_state.py @@ -1,7 +1,9 @@ """Reproduce an Lock 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 Lock states.""" await asyncio.gather( diff --git a/homeassistant/components/lock/significant_change.py b/homeassistant/components/lock/significant_change.py index 59a3b1a95c5..172bf2559c5 100644 --- a/homeassistant/components/lock/significant_change.py +++ b/homeassistant/components/lock/significant_change.py @@ -1,5 +1,7 @@ """Helper to test significant Lock 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/lovelace/resources.py b/homeassistant/components/lovelace/resources.py index 0a3e36892d5..6a97d5c4192 100644 --- a/homeassistant/components/lovelace/resources.py +++ b/homeassistant/components/lovelace/resources.py @@ -1,6 +1,8 @@ """Lovelace resources support.""" +from __future__ import annotations + import logging -from typing import List, Optional, cast +from typing import Optional, cast import uuid import voluptuous as vol @@ -38,7 +40,7 @@ class ResourceYAMLCollection: return {"resources": len(self.async_items() or [])} @callback - def async_items(self) -> List[dict]: + def async_items(self) -> list[dict]: """Return list of items in collection.""" return self.data @@ -66,7 +68,7 @@ class ResourceStorageCollection(collection.StorageCollection): return {"resources": len(self.async_items() or [])} - async def _async_load_data(self) -> Optional[dict]: + async def _async_load_data(self) -> dict | None: """Load the data.""" data = await self.store.async_load() diff --git a/homeassistant/components/lutron_caseta/device_trigger.py b/homeassistant/components/lutron_caseta/device_trigger.py index 86ee5e46b51..230301c12f2 100644 --- a/homeassistant/components/lutron_caseta/device_trigger.py +++ b/homeassistant/components/lutron_caseta/device_trigger.py @@ -1,5 +1,5 @@ """Provides device triggers for lutron caseta.""" -from typing import List +from __future__ import annotations import voluptuous as vol @@ -225,7 +225,7 @@ async def async_validate_trigger_config(hass: HomeAssistant, config: ConfigType) return schema(config) -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 lutron caseta devices.""" triggers = [] diff --git a/homeassistant/components/lutron_caseta/fan.py b/homeassistant/components/lutron_caseta/fan.py index de2b2a2ae8c..edca88c10fc 100644 --- a/homeassistant/components/lutron_caseta/fan.py +++ b/homeassistant/components/lutron_caseta/fan.py @@ -1,6 +1,7 @@ """Support for Lutron Caseta fans.""" +from __future__ import annotations + import logging -from typing import Optional from pylutron_caseta import FAN_HIGH, FAN_LOW, FAN_MEDIUM, FAN_MEDIUM_HIGH, FAN_OFF @@ -42,7 +43,7 @@ class LutronCasetaFan(LutronCasetaDevice, FanEntity): """Representation of a Lutron Caseta fan. Including Fan Speed.""" @property - def percentage(self) -> Optional[int]: + def percentage(self) -> int | None: """Return the current speed percentage.""" if self._device["fan_speed"] is None: return None diff --git a/homeassistant/components/lyric/__init__.py b/homeassistant/components/lyric/__init__.py index 8b23d6f85cf..8536fa03e8a 100644 --- a/homeassistant/components/lyric/__init__.py +++ b/homeassistant/components/lyric/__init__.py @@ -1,8 +1,10 @@ """The Honeywell Lyric integration.""" +from __future__ import annotations + import asyncio from datetime import timedelta import logging -from typing import Any, Dict, Optional +from typing import Any from aiolyric import Lyric from aiolyric.objects.device import LyricDevice @@ -147,7 +149,7 @@ class LyricEntity(CoordinatorEntity): device: LyricDevice, key: str, name: str, - icon: Optional[str], + icon: str | None, ) -> None: """Initialize the Honeywell Lyric entity.""" super().__init__(coordinator) @@ -190,7 +192,7 @@ class LyricDeviceEntity(LyricEntity): """Defines a Honeywell Lyric device entity.""" @property - def device_info(self) -> Dict[str, Any]: + def device_info(self) -> dict[str, Any]: """Return device information about this Honeywell Lyric instance.""" return { "connections": {(dr.CONNECTION_NETWORK_MAC, self._mac_id)}, diff --git a/homeassistant/components/lyric/climate.py b/homeassistant/components/lyric/climate.py index 4338a6e92cc..0e3672f952e 100644 --- a/homeassistant/components/lyric/climate.py +++ b/homeassistant/components/lyric/climate.py @@ -1,7 +1,8 @@ """Support for Honeywell Lyric climate platform.""" +from __future__ import annotations + import logging from time import gmtime, strftime, time -from typing import List, Optional from aiolyric.objects.device import LyricDevice from aiolyric.objects.location import LyricLocation @@ -162,7 +163,7 @@ class LyricClimate(LyricDeviceEntity, ClimateEntity): return self._temperature_unit @property - def current_temperature(self) -> Optional[float]: + def current_temperature(self) -> float | None: """Return the current temperature.""" return self.device.indoorTemperature @@ -180,12 +181,12 @@ class LyricClimate(LyricDeviceEntity, ClimateEntity): return HVAC_MODES[self.device.changeableValues.mode] @property - def hvac_modes(self) -> List[str]: + def hvac_modes(self) -> list[str]: """List of available hvac modes.""" return self._hvac_modes @property - def target_temperature(self) -> Optional[float]: + def target_temperature(self) -> float | None: """Return the temperature we try to reach.""" device = self.device if not device.hasDualSetpointStatus: @@ -193,7 +194,7 @@ class LyricClimate(LyricDeviceEntity, ClimateEntity): return None @property - def target_temperature_low(self) -> Optional[float]: + def target_temperature_low(self) -> float | None: """Return the upper bound temperature we try to reach.""" device = self.device if device.hasDualSetpointStatus: @@ -201,7 +202,7 @@ class LyricClimate(LyricDeviceEntity, ClimateEntity): return None @property - def target_temperature_high(self) -> Optional[float]: + def target_temperature_high(self) -> float | None: """Return the upper bound temperature we try to reach.""" device = self.device if device.hasDualSetpointStatus: @@ -209,12 +210,12 @@ class LyricClimate(LyricDeviceEntity, ClimateEntity): return None @property - def preset_mode(self) -> Optional[str]: + def preset_mode(self) -> str | None: """Return current preset mode.""" return self.device.changeableValues.thermostatSetpointStatus @property - def preset_modes(self) -> Optional[List[str]]: + def preset_modes(self) -> list[str] | None: """Return preset modes.""" return [ PRESET_NO_HOLD, diff --git a/homeassistant/components/media_player/__init__.py b/homeassistant/components/media_player/__init__.py index 87ecff7a54c..1e0a9b9f6bb 100644 --- a/homeassistant/components/media_player/__init__.py +++ b/homeassistant/components/media_player/__init__.py @@ -9,7 +9,6 @@ import functools as ft import hashlib import logging import secrets -from typing import List, Optional, Tuple from urllib.parse import urlparse from aiohttp import web @@ -355,7 +354,7 @@ async def async_unload_entry(hass, entry): class MediaPlayerEntity(Entity): """ABC for media player entities.""" - _access_token: Optional[str] = None + _access_token: str | None = None # Implement these for your media player @property @@ -439,8 +438,8 @@ class MediaPlayerEntity(Entity): self, media_content_type: str, media_content_id: str, - media_image_id: Optional[str] = None, - ) -> Tuple[Optional[str], Optional[str]]: + media_image_id: str | None = None, + ) -> tuple[str | None, str | None]: """ Optionally fetch internally accessible image for media browser. @@ -851,8 +850,8 @@ class MediaPlayerEntity(Entity): async def async_browse_media( self, - media_content_type: Optional[str] = None, - media_content_id: Optional[str] = None, + media_content_type: str | None = None, + media_content_id: str | None = None, ) -> BrowseMedia: """Return a BrowseMedia instance. @@ -914,7 +913,7 @@ class MediaPlayerEntity(Entity): self, media_content_type: str, media_content_id: str, - media_image_id: Optional[str] = None, + media_image_id: str | None = None, ) -> str: """Generate an url for a media browser image.""" url_path = ( @@ -947,8 +946,8 @@ class MediaPlayerImageView(HomeAssistantView): self, request: web.Request, entity_id: str, - media_content_type: Optional[str] = None, - media_content_id: Optional[str] = None, + media_content_type: str | None = None, + media_content_id: str | None = None, ) -> web.Response: """Start a get request.""" player = self.component.get_entity(entity_id) @@ -1047,7 +1046,7 @@ async def websocket_browse_media(hass, connection, msg): To use, media_player integrations can implement MediaPlayerEntity.async_browse_media() """ component = hass.data[DOMAIN] - player: Optional[MediaPlayerDevice] = component.get_entity(msg["entity_id"]) + player: MediaPlayerDevice | None = component.get_entity(msg["entity_id"]) if player is None: connection.send_error(msg["id"], "entity_not_found", "Entity not found") @@ -1119,9 +1118,9 @@ class BrowseMedia: title: str, can_play: bool, can_expand: bool, - children: Optional[List["BrowseMedia"]] = None, - children_media_class: Optional[str] = None, - thumbnail: Optional[str] = None, + children: list[BrowseMedia] | None = None, + children_media_class: str | None = None, + thumbnail: str | None = None, ): """Initialize browse media item.""" self.media_class = media_class diff --git a/homeassistant/components/media_player/device_condition.py b/homeassistant/components/media_player/device_condition.py index 6faa6521b70..0e6e0f96c40 100644 --- a/homeassistant/components/media_player/device_condition.py +++ b/homeassistant/components/media_player/device_condition.py @@ -1,5 +1,5 @@ """Provides device automations for Media player.""" -from typing import Dict, List +from __future__ import annotations import voluptuous as vol @@ -35,7 +35,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 Media player devices.""" registry = await entity_registry.async_get_registry(hass) conditions = [] diff --git a/homeassistant/components/media_player/device_trigger.py b/homeassistant/components/media_player/device_trigger.py index 6db5f16cf01..03c165412e9 100644 --- a/homeassistant/components/media_player/device_trigger.py +++ b/homeassistant/components/media_player/device_trigger.py @@ -1,5 +1,5 @@ """Provides device automations for Media player.""" -from typing import List +from __future__ import annotations import voluptuous as vol @@ -34,7 +34,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 Media player entities.""" registry = await entity_registry.async_get_registry(hass) triggers = [] diff --git a/homeassistant/components/media_player/reproduce_state.py b/homeassistant/components/media_player/reproduce_state.py index 64955d1913b..1707109197f 100644 --- a/homeassistant/components/media_player/reproduce_state.py +++ b/homeassistant/components/media_player/reproduce_state.py @@ -1,6 +1,8 @@ """Module that groups code required to handle state restore for component.""" +from __future__ import annotations + import asyncio -from typing import Any, Dict, Iterable, Optional +from typing import Any, Iterable from homeassistant.const import ( SERVICE_MEDIA_PAUSE, @@ -40,8 +42,8 @@ async def _async_reproduce_states( 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 component states.""" @@ -104,8 +106,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 component states.""" await asyncio.gather( diff --git a/homeassistant/components/media_source/__init__.py b/homeassistant/components/media_source/__init__.py index 3dff949d5dd..6aa01403a5f 100644 --- a/homeassistant/components/media_source/__init__.py +++ b/homeassistant/components/media_source/__init__.py @@ -1,6 +1,7 @@ """The media_source integration.""" +from __future__ import annotations + from datetime import timedelta -from typing import Optional import voluptuous as vol @@ -54,7 +55,7 @@ async def _process_media_source_platform(hass, domain, platform): @callback def _get_media_item( - hass: HomeAssistant, media_content_id: Optional[str] + hass: HomeAssistant, media_content_id: str | None ) -> models.MediaSourceItem: """Return media item.""" if media_content_id: diff --git a/homeassistant/components/media_source/local_source.py b/homeassistant/components/media_source/local_source.py index fa62ba48c5f..fb5e9094dfb 100644 --- a/homeassistant/components/media_source/local_source.py +++ b/homeassistant/components/media_source/local_source.py @@ -1,7 +1,8 @@ """Local Media Source Implementation.""" +from __future__ import annotations + import mimetypes from pathlib import Path -from typing import Tuple from aiohttp import web @@ -40,7 +41,7 @@ class LocalSource(MediaSource): return Path(self.hass.config.media_dirs[source_dir_id], location) @callback - def async_parse_identifier(self, item: MediaSourceItem) -> Tuple[str, str]: + def async_parse_identifier(self, item: MediaSourceItem) -> tuple[str, str]: """Parse identifier.""" if not item.identifier: # Empty source_dir_id and location @@ -69,7 +70,7 @@ class LocalSource(MediaSource): return PlayMedia(f"/media/{item.identifier}", mime_type) 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.""" try: diff --git a/homeassistant/components/media_source/models.py b/homeassistant/components/media_source/models.py index 98b817344d9..aa17fff320e 100644 --- a/homeassistant/components/media_source/models.py +++ b/homeassistant/components/media_source/models.py @@ -3,7 +3,6 @@ from __future__ import annotations from abc import ABC from dataclasses import dataclass -from typing import List, Optional, Tuple from homeassistant.components.media_player import BrowseMedia from homeassistant.components.media_player.const import ( @@ -28,9 +27,9 @@ class PlayMedia: class BrowseMediaSource(BrowseMedia): """Represent a browsable media file.""" - children: Optional[List["BrowseMediaSource"]] + children: list[BrowseMediaSource] | None - def __init__(self, *, domain: Optional[str], identifier: Optional[str], **kwargs): + def __init__(self, *, domain: str | None, identifier: str | None, **kwargs): """Initialize media source browse media.""" media_content_id = f"{URI_SCHEME}{domain or ''}" if identifier: @@ -47,7 +46,7 @@ class MediaSourceItem: """A parsed media item.""" hass: HomeAssistant - domain: Optional[str] + domain: str | None identifier: str async def async_browse(self) -> BrowseMediaSource: @@ -118,7 +117,7 @@ class MediaSource(ABC): raise NotImplementedError async def async_browse_media( - self, item: MediaSourceItem, media_types: Tuple[str] + self, item: MediaSourceItem, media_types: tuple[str] ) -> BrowseMediaSource: """Browse media.""" raise NotImplementedError diff --git a/homeassistant/components/melcloud/__init__.py b/homeassistant/components/melcloud/__init__.py index 0e81d6101b3..0f48db96bf8 100644 --- a/homeassistant/components/melcloud/__init__.py +++ b/homeassistant/components/melcloud/__init__.py @@ -1,8 +1,10 @@ """The MELCloud Climate integration.""" +from __future__ import annotations + import asyncio from datetime import timedelta import logging -from typing import Any, Dict, List +from typing import Any from aiohttp import ClientConnectionError from async_timeout import timeout @@ -101,7 +103,7 @@ class MelCloudDevice: _LOGGER.warning("Connection failed for %s", self.name) self._available = False - async def async_set(self, properties: Dict[str, Any]): + async def async_set(self, properties: dict[str, Any]): """Write state changes to the MELCloud API.""" try: await self.device.set(properties) @@ -142,7 +144,7 @@ class MelCloudDevice: return _device_info -async def mel_devices_setup(hass, token) -> List[MelCloudDevice]: +async def mel_devices_setup(hass, token) -> list[MelCloudDevice]: """Query connected devices from MELCloud.""" session = hass.helpers.aiohttp_client.async_get_clientsession() try: diff --git a/homeassistant/components/melcloud/climate.py b/homeassistant/components/melcloud/climate.py index 1abb86cf5e5..8e45cc3d9a4 100644 --- a/homeassistant/components/melcloud/climate.py +++ b/homeassistant/components/melcloud/climate.py @@ -1,6 +1,8 @@ """Platform for climate integration.""" +from __future__ import annotations + from datetime import timedelta -from typing import Any, Dict, List, Optional +from typing import Any from pymelcloud import DEVICE_TYPE_ATA, DEVICE_TYPE_ATW, AtaDevice, AtwDevice import pymelcloud.ata_device as ata @@ -114,7 +116,7 @@ class MelCloudClimate(ClimateEntity): return self.api.device_info @property - def target_temperature_step(self) -> Optional[float]: + def target_temperature_step(self) -> float | None: """Return the supported step of target temperature.""" return self._base_device.temperature_increment @@ -128,7 +130,7 @@ class AtaDeviceClimate(MelCloudClimate): self._device = ata_device @property - def unique_id(self) -> Optional[str]: + def unique_id(self) -> str | None: """Return a unique ID.""" return f"{self.api.device.serial}-{self.api.device.mac}" @@ -138,7 +140,7 @@ class AtaDeviceClimate(MelCloudClimate): return self._name @property - def extra_state_attributes(self) -> Optional[Dict[str, Any]]: + def extra_state_attributes(self) -> dict[str, Any] | None: """Return the optional state attributes with device specific additions.""" attr = {} @@ -190,19 +192,19 @@ class AtaDeviceClimate(MelCloudClimate): await self._device.set(props) @property - def hvac_modes(self) -> List[str]: + def hvac_modes(self) -> list[str]: """Return the list of available hvac operation modes.""" return [HVAC_MODE_OFF] + [ ATA_HVAC_MODE_LOOKUP.get(mode) for mode in self._device.operation_modes ] @property - def current_temperature(self) -> Optional[float]: + def current_temperature(self) -> float | None: """Return the current temperature.""" return self._device.room_temperature @property - def target_temperature(self) -> Optional[float]: + def target_temperature(self) -> float | None: """Return the temperature we try to reach.""" return self._device.target_temperature @@ -213,7 +215,7 @@ class AtaDeviceClimate(MelCloudClimate): ) @property - def fan_mode(self) -> Optional[str]: + def fan_mode(self) -> str | None: """Return the fan setting.""" return self._device.fan_speed @@ -222,7 +224,7 @@ class AtaDeviceClimate(MelCloudClimate): await self._device.set({"fan_speed": fan_mode}) @property - def fan_modes(self) -> Optional[List[str]]: + def fan_modes(self) -> list[str] | None: """Return the list of available fan modes.""" return self._device.fan_speeds @@ -243,7 +245,7 @@ class AtaDeviceClimate(MelCloudClimate): await self._device.set({ata.PROPERTY_VANE_VERTICAL: position}) @property - def swing_mode(self) -> Optional[str]: + def swing_mode(self) -> str | None: """Return vertical vane position or mode.""" return self._device.vane_vertical @@ -252,7 +254,7 @@ class AtaDeviceClimate(MelCloudClimate): await self.async_set_vane_vertical(swing_mode) @property - def swing_modes(self) -> Optional[str]: + def swing_modes(self) -> str | None: """Return a list of available vertical vane positions and modes.""" return self._device.vane_vertical_positions @@ -300,7 +302,7 @@ class AtwDeviceZoneClimate(MelCloudClimate): self._zone = atw_zone @property - def unique_id(self) -> Optional[str]: + def unique_id(self) -> str | None: """Return a unique ID.""" return f"{self.api.device.serial}-{self._zone.zone_index}" @@ -310,7 +312,7 @@ class AtwDeviceZoneClimate(MelCloudClimate): return f"{self._name} {self._zone.name}" @property - def extra_state_attributes(self) -> Dict[str, Any]: + def extra_state_attributes(self) -> dict[str, Any]: """Return the optional state attributes with device specific additions.""" data = { ATTR_STATUS: ATW_ZONE_HVAC_MODE_LOOKUP.get( @@ -351,17 +353,17 @@ class AtwDeviceZoneClimate(MelCloudClimate): await self._device.set(props) @property - def hvac_modes(self) -> List[str]: + def hvac_modes(self) -> list[str]: """Return the list of available hvac operation modes.""" return [self.hvac_mode] @property - def current_temperature(self) -> Optional[float]: + def current_temperature(self) -> float | None: """Return the current temperature.""" return self._zone.room_temperature @property - def target_temperature(self) -> Optional[float]: + def target_temperature(self) -> float | None: """Return the temperature we try to reach.""" return self._zone.target_temperature diff --git a/homeassistant/components/melcloud/config_flow.py b/homeassistant/components/melcloud/config_flow.py index 41ce24989a5..a487a446d9b 100644 --- a/homeassistant/components/melcloud/config_flow.py +++ b/homeassistant/components/melcloud/config_flow.py @@ -1,6 +1,7 @@ """Config flow for the MELCloud platform.""" +from __future__ import annotations + import asyncio -from typing import Optional from aiohttp import ClientError, ClientResponseError from async_timeout import timeout @@ -37,8 +38,8 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): self, username: str, *, - password: Optional[str] = None, - token: Optional[str] = None, + password: str | None = None, + token: str | None = None, ): """Create client.""" if password is None and token is None: diff --git a/homeassistant/components/melcloud/water_heater.py b/homeassistant/components/melcloud/water_heater.py index 3474fc07540..e01d78a5270 100644 --- a/homeassistant/components/melcloud/water_heater.py +++ b/homeassistant/components/melcloud/water_heater.py @@ -1,5 +1,5 @@ """Platform for water_heater integration.""" -from typing import List, Optional +from __future__ import annotations from pymelcloud import DEVICE_TYPE_ATW, AtwDevice from pymelcloud.atw_device import ( @@ -49,7 +49,7 @@ class AtwWaterHeater(WaterHeaterEntity): await self._api.async_update() @property - def unique_id(self) -> Optional[str]: + def unique_id(self) -> str | None: """Return a unique ID.""" return f"{self._api.device.serial}" @@ -83,17 +83,17 @@ class AtwWaterHeater(WaterHeaterEntity): return TEMP_CELSIUS @property - def current_operation(self) -> Optional[str]: + def current_operation(self) -> str | None: """Return current operation as reported by pymelcloud.""" return self._device.operation_mode @property - def operation_list(self) -> List[str]: + def operation_list(self) -> list[str]: """Return the list of available operation modes as reported by pymelcloud.""" return self._device.operation_modes @property - def current_temperature(self) -> Optional[float]: + def current_temperature(self) -> float | None: """Return the current temperature.""" return self._device.tank_temperature @@ -122,11 +122,11 @@ class AtwWaterHeater(WaterHeaterEntity): return SUPPORT_TARGET_TEMPERATURE | SUPPORT_OPERATION_MODE @property - def min_temp(self) -> Optional[float]: + def min_temp(self) -> float | None: """Return the minimum temperature.""" return self._device.target_tank_temperature_min @property - def max_temp(self) -> Optional[float]: + def max_temp(self) -> float | None: """Return the maximum temperature.""" return self._device.target_tank_temperature_max diff --git a/homeassistant/components/met/config_flow.py b/homeassistant/components/met/config_flow.py index 6b3d6735f46..b9d50ba59a5 100644 --- a/homeassistant/components/met/config_flow.py +++ b/homeassistant/components/met/config_flow.py @@ -1,5 +1,7 @@ """Config flow to configure Met component.""" -from typing import Any, Dict, Optional +from __future__ import annotations + +from typing import Any import voluptuous as vol @@ -73,9 +75,7 @@ class MetFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): errors=self._errors, ) - async def async_step_import( - self, user_input: Optional[Dict] = None - ) -> Dict[str, Any]: + async def async_step_import(self, user_input: dict | None = None) -> dict[str, Any]: """Handle configuration by yaml file.""" return await self.async_step_user(user_input) diff --git a/homeassistant/components/minecraft_server/__init__.py b/homeassistant/components/minecraft_server/__init__.py index 0e7096881f5..f76e8e8467e 100644 --- a/homeassistant/components/minecraft_server/__init__.py +++ b/homeassistant/components/minecraft_server/__init__.py @@ -1,9 +1,10 @@ """The Minecraft Server integration.""" +from __future__ import annotations import asyncio from datetime import datetime, timedelta import logging -from typing import Any, Dict +from typing import Any from mcstatus.server import MinecraftServer as MCStatus @@ -260,7 +261,7 @@ class MinecraftServerEntity(Entity): return self._unique_id @property - def device_info(self) -> Dict[str, Any]: + def device_info(self) -> dict[str, Any]: """Return device information.""" return self._device_info diff --git a/homeassistant/components/minecraft_server/helpers.py b/homeassistant/components/minecraft_server/helpers.py index 7f9380cdec2..f6409ce525d 100644 --- a/homeassistant/components/minecraft_server/helpers.py +++ b/homeassistant/components/minecraft_server/helpers.py @@ -1,6 +1,7 @@ """Helper functions for the Minecraft Server integration.""" +from __future__ import annotations -from typing import Any, Dict +from typing import Any import aiodns @@ -10,7 +11,7 @@ from homeassistant.helpers.typing import HomeAssistantType from .const import SRV_RECORD_PREFIX -async def async_check_srv_record(hass: HomeAssistantType, host: str) -> Dict[str, Any]: +async def async_check_srv_record(hass: HomeAssistantType, host: str) -> dict[str, Any]: """Check if the given host is a valid Minecraft SRV record.""" # Check if 'host' is a valid SRV record. return_value = None diff --git a/homeassistant/components/minecraft_server/sensor.py b/homeassistant/components/minecraft_server/sensor.py index 8cc626390dd..4cc9e9746bb 100644 --- a/homeassistant/components/minecraft_server/sensor.py +++ b/homeassistant/components/minecraft_server/sensor.py @@ -1,5 +1,7 @@ """The Minecraft Server sensor platform.""" -from typing import Any, Dict +from __future__ import annotations + +from typing import Any from homeassistant.config_entries import ConfigEntry from homeassistant.const import TIME_MILLISECONDS @@ -151,7 +153,7 @@ class MinecraftServerPlayersOnlineSensor(MinecraftServerSensorEntity): self._extra_state_attributes = extra_state_attributes @property - def extra_state_attributes(self) -> Dict[str, Any]: + def extra_state_attributes(self) -> dict[str, Any]: """Return players list in device state attributes.""" return self._extra_state_attributes diff --git a/homeassistant/components/minio/__init__.py b/homeassistant/components/minio/__init__.py index 178058986cc..6e7174b60ee 100644 --- a/homeassistant/components/minio/__init__.py +++ b/homeassistant/components/minio/__init__.py @@ -1,9 +1,10 @@ """Minio component.""" +from __future__ import annotations + import logging import os from queue import Queue import threading -from typing import List import voluptuous as vol @@ -230,7 +231,7 @@ class MinioListener: bucket_name: str, prefix: str, suffix: str, - events: List[str], + events: list[str], ): """Create Listener.""" self._queue = queue diff --git a/homeassistant/components/minio/minio_helper.py b/homeassistant/components/minio/minio_helper.py index 2aaba9d4085..f2d86067552 100644 --- a/homeassistant/components/minio/minio_helper.py +++ b/homeassistant/components/minio/minio_helper.py @@ -1,4 +1,6 @@ """Minio helper methods.""" +from __future__ import annotations + from collections.abc import Iterable import json import logging @@ -6,7 +8,7 @@ from queue import Queue import re import threading import time -from typing import Iterator, List +from typing import Iterator from urllib.parse import unquote from minio import Minio @@ -38,7 +40,7 @@ def create_minio_client( def get_minio_notification_response( - minio_client, bucket_name: str, prefix: str, suffix: str, events: List[str] + minio_client, bucket_name: str, prefix: str, suffix: str, events: list[str] ): """Start listening to minio events. Copied from minio-py.""" query = {"prefix": prefix, "suffix": suffix, "events": events} @@ -87,7 +89,7 @@ class MinioEventThread(threading.Thread): bucket_name: str, prefix: str, suffix: str, - events: List[str], + events: list[str], ): """Copy over all Minio client options.""" super().__init__() diff --git a/homeassistant/components/mobile_app/device_action.py b/homeassistant/components/mobile_app/device_action.py index 2592d4b486b..33a7510da21 100644 --- a/homeassistant/components/mobile_app/device_action.py +++ b/homeassistant/components/mobile_app/device_action.py @@ -1,5 +1,5 @@ """Provides device actions for Mobile App.""" -from typing import List, Optional +from __future__ import annotations import voluptuous as vol @@ -22,7 +22,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 Mobile App devices.""" webhook_id = webhook_id_from_device_id(hass, device_id) @@ -33,7 +33,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.""" webhook_id = webhook_id_from_device_id(hass, config[CONF_DEVICE_ID]) diff --git a/homeassistant/components/mobile_app/helpers.py b/homeassistant/components/mobile_app/helpers.py index fed322df464..63d638cd9e5 100644 --- a/homeassistant/components/mobile_app/helpers.py +++ b/homeassistant/components/mobile_app/helpers.py @@ -1,7 +1,9 @@ """Helpers for mobile_app.""" +from __future__ import annotations + import json import logging -from typing import Callable, Dict, Tuple +from typing import Callable from aiohttp.web import Response, json_response from nacl.encoding import Base64Encoder @@ -36,7 +38,7 @@ from .const import ( _LOGGER = logging.getLogger(__name__) -def setup_decrypt() -> Tuple[int, Callable]: +def setup_decrypt() -> tuple[int, Callable]: """Return decryption function and length of key. Async friendly. @@ -49,7 +51,7 @@ def setup_decrypt() -> Tuple[int, Callable]: return (SecretBox.KEY_SIZE, decrypt) -def setup_encrypt() -> Tuple[int, Callable]: +def setup_encrypt() -> tuple[int, Callable]: """Return encryption function and length of key. Async friendly. @@ -62,7 +64,7 @@ def setup_encrypt() -> Tuple[int, Callable]: return (SecretBox.KEY_SIZE, encrypt) -def _decrypt_payload(key: str, ciphertext: str) -> Dict[str, str]: +def _decrypt_payload(key: str, ciphertext: str) -> dict[str, str]: """Decrypt encrypted payload.""" try: keylen, decrypt = setup_decrypt() @@ -88,12 +90,12 @@ def _decrypt_payload(key: str, ciphertext: str) -> Dict[str, str]: return None -def registration_context(registration: Dict) -> Context: +def registration_context(registration: dict) -> Context: """Generate a context from a request.""" return Context(user_id=registration[CONF_USER_ID]) -def empty_okay_response(headers: Dict = None, status: int = HTTP_OK) -> Response: +def empty_okay_response(headers: dict = None, status: int = HTTP_OK) -> Response: """Return a Response with empty JSON object and a 200.""" return Response( text="{}", status=status, content_type=CONTENT_TYPE_JSON, headers=headers @@ -121,7 +123,7 @@ def supports_encryption() -> bool: return False -def safe_registration(registration: Dict) -> Dict: +def safe_registration(registration: dict) -> dict: """Return a registration without sensitive values.""" # Sensitive values: webhook_id, secret, cloudhook_url return { @@ -137,7 +139,7 @@ def safe_registration(registration: Dict) -> Dict: } -def savable_state(hass: HomeAssistantType) -> Dict: +def savable_state(hass: HomeAssistantType) -> dict: """Return a clean object containing things that should be saved.""" return { DATA_DELETED_IDS: hass.data[DOMAIN][DATA_DELETED_IDS], @@ -145,7 +147,7 @@ def savable_state(hass: HomeAssistantType) -> Dict: def webhook_response( - data, *, registration: Dict, status: int = HTTP_OK, headers: Dict = None + data, *, registration: dict, status: int = HTTP_OK, headers: dict = None ) -> Response: """Return a encrypted response if registration supports it.""" data = json.dumps(data, cls=JSONEncoder) @@ -165,7 +167,7 @@ def webhook_response( ) -def device_info(registration: Dict) -> Dict: +def device_info(registration: dict) -> dict: """Return the device info for this registration.""" return { "identifiers": {(DOMAIN, registration[ATTR_DEVICE_ID])}, diff --git a/homeassistant/components/mobile_app/http_api.py b/homeassistant/components/mobile_app/http_api.py index 4bd8d0cd76b..5583b7c58d1 100644 --- a/homeassistant/components/mobile_app/http_api.py +++ b/homeassistant/components/mobile_app/http_api.py @@ -1,6 +1,7 @@ """Provides an HTTP API for mobile_app.""" +from __future__ import annotations + import secrets -from typing import Dict from aiohttp.web import Request, Response import emoji @@ -58,7 +59,7 @@ class RegistrationsView(HomeAssistantView): extra=vol.REMOVE_EXTRA, ) ) - async def post(self, request: Request, data: Dict) -> Response: + async def post(self, request: Request, data: dict) -> Response: """Handle the POST request for registration.""" hass = request.app["hass"] diff --git a/homeassistant/components/mobile_app/util.py b/homeassistant/components/mobile_app/util.py index 60dfe242e04..b0a3f52e394 100644 --- a/homeassistant/components/mobile_app/util.py +++ b/homeassistant/components/mobile_app/util.py @@ -1,5 +1,7 @@ """Mobile app utility functions.""" -from typing import TYPE_CHECKING, Optional +from __future__ import annotations + +from typing import TYPE_CHECKING from homeassistant.core import callback @@ -18,7 +20,7 @@ if TYPE_CHECKING: @callback -def webhook_id_from_device_id(hass, device_id: str) -> Optional[str]: +def webhook_id_from_device_id(hass, device_id: str) -> str | None: """Get webhook ID from device ID.""" if DOMAIN not in hass.data: return None @@ -39,7 +41,7 @@ def supports_push(hass, webhook_id: str) -> bool: @callback -def get_notify_service(hass, webhook_id: str) -> Optional[str]: +def get_notify_service(hass, webhook_id: str) -> str | None: """Return the notify service for this webhook ID.""" notify_service: "MobileAppNotificationService" = hass.data[DOMAIN][DATA_NOTIFY] diff --git a/homeassistant/components/modbus/binary_sensor.py b/homeassistant/components/modbus/binary_sensor.py index 8e91945d073..16cd191bba7 100644 --- a/homeassistant/components/modbus/binary_sensor.py +++ b/homeassistant/components/modbus/binary_sensor.py @@ -1,5 +1,5 @@ """Support for Modbus Coil and Discrete Input sensors.""" -from typing import Optional +from __future__ import annotations from pymodbus.exceptions import ConnectionException, ModbusException from pymodbus.pdu import ExceptionResponse @@ -94,7 +94,7 @@ class ModbusBinarySensor(BinarySensorEntity): return self._value @property - def device_class(self) -> Optional[str]: + def device_class(self) -> str | None: """Return the device class of the sensor.""" return self._device_class diff --git a/homeassistant/components/modbus/climate.py b/homeassistant/components/modbus/climate.py index 45cfbf5eb57..d32f2ae3cf5 100644 --- a/homeassistant/components/modbus/climate.py +++ b/homeassistant/components/modbus/climate.py @@ -1,8 +1,10 @@ """Support for Generic Modbus Thermostats.""" +from __future__ import annotations + from datetime import timedelta import logging import struct -from typing import Any, Dict, Optional +from typing import Any from pymodbus.exceptions import ConnectionException, ModbusException from pymodbus.pdu import ExceptionResponse @@ -57,7 +59,7 @@ async def async_setup_platform( hass: HomeAssistantType, config: ConfigType, async_add_entities, - discovery_info: Optional[DiscoveryInfoType] = None, + discovery_info: DiscoveryInfoType | None = None, ): """Read configuration and create Modbus climate.""" if discovery_info is None: @@ -108,7 +110,7 @@ class ModbusThermostat(ClimateEntity): def __init__( self, hub: ModbusHub, - config: Dict[str, Any], + config: dict[str, Any], ): """Initialize the modbus thermostat.""" self._hub: ModbusHub = hub @@ -232,7 +234,7 @@ class ModbusThermostat(ClimateEntity): self.schedule_update_ha_state() - def _read_register(self, register_type, register) -> Optional[float]: + def _read_register(self, register_type, register) -> float | None: """Read register using the Modbus hub slave.""" try: if register_type == CALL_TYPE_REGISTER_INPUT: diff --git a/homeassistant/components/modbus/cover.py b/homeassistant/components/modbus/cover.py index 09a465a2cdd..a8003676640 100644 --- a/homeassistant/components/modbus/cover.py +++ b/homeassistant/components/modbus/cover.py @@ -1,6 +1,8 @@ """Support for Modbus covers.""" +from __future__ import annotations + from datetime import timedelta -from typing import Any, Dict, Optional +from typing import Any from pymodbus.exceptions import ConnectionException, ModbusException from pymodbus.pdu import ExceptionResponse @@ -41,7 +43,7 @@ async def async_setup_platform( hass: HomeAssistantType, config: ConfigType, async_add_entities, - discovery_info: Optional[DiscoveryInfoType] = None, + discovery_info: DiscoveryInfoType | None = None, ): """Read configuration and create Modbus cover.""" if discovery_info is None: @@ -61,7 +63,7 @@ class ModbusCover(CoverEntity, RestoreEntity): def __init__( self, hub: ModbusHub, - config: Dict[str, Any], + config: dict[str, Any], ): """Initialize the modbus cover.""" self._hub: ModbusHub = hub @@ -108,7 +110,7 @@ class ModbusCover(CoverEntity, RestoreEntity): ) @property - def device_class(self) -> Optional[str]: + def device_class(self) -> str | None: """Return the device class of the sensor.""" return self._device_class @@ -178,7 +180,7 @@ class ModbusCover(CoverEntity, RestoreEntity): self.schedule_update_ha_state() - def _read_status_register(self) -> Optional[int]: + def _read_status_register(self) -> int | None: """Read status register using the Modbus hub slave.""" try: if self._status_register_type == CALL_TYPE_REGISTER_INPUT: @@ -212,7 +214,7 @@ class ModbusCover(CoverEntity, RestoreEntity): self._available = True - def _read_coil(self) -> Optional[bool]: + def _read_coil(self) -> bool | None: """Read coil using the Modbus hub slave.""" try: result = self._hub.read_coils(self._slave, self._coil, 1) diff --git a/homeassistant/components/modbus/sensor.py b/homeassistant/components/modbus/sensor.py index 656e5e2986d..d0fad973802 100644 --- a/homeassistant/components/modbus/sensor.py +++ b/homeassistant/components/modbus/sensor.py @@ -1,7 +1,9 @@ """Support for Modbus Register sensors.""" +from __future__ import annotations + import logging import struct -from typing import Any, Optional, Union +from typing import Any from pymodbus.exceptions import ConnectionException, ModbusException from pymodbus.pdu import ExceptionResponse @@ -44,7 +46,7 @@ from .const import ( _LOGGER = logging.getLogger(__name__) -def number(value: Any) -> Union[int, float]: +def number(value: Any) -> int | float: """Coerce a value to number without losing precision.""" if isinstance(value, int): return value @@ -217,7 +219,7 @@ class ModbusRegisterSensor(RestoreEntity): return self._unit_of_measurement @property - def device_class(self) -> Optional[str]: + def device_class(self) -> str | None: """Return the device class of the sensor.""" return self._device_class diff --git a/homeassistant/components/modbus/switch.py b/homeassistant/components/modbus/switch.py index 36fbef08428..bf2233e2407 100644 --- a/homeassistant/components/modbus/switch.py +++ b/homeassistant/components/modbus/switch.py @@ -1,7 +1,9 @@ """Support for Modbus switches.""" +from __future__ import annotations + from abc import ABC import logging -from typing import Any, Dict, Optional +from typing import Any from pymodbus.exceptions import ConnectionException, ModbusException from pymodbus.pdu import ExceptionResponse @@ -99,7 +101,7 @@ async def async_setup_platform( class ModbusBaseSwitch(ToggleEntity, RestoreEntity, ABC): """Base class representing a Modbus switch.""" - def __init__(self, hub: ModbusHub, config: Dict[str, Any]): + def __init__(self, hub: ModbusHub, config: dict[str, Any]): """Initialize the switch.""" self._hub: ModbusHub = hub self._name = config[CONF_NAME] @@ -133,7 +135,7 @@ class ModbusBaseSwitch(ToggleEntity, RestoreEntity, ABC): class ModbusCoilSwitch(ModbusBaseSwitch, SwitchEntity): """Representation of a Modbus coil switch.""" - def __init__(self, hub: ModbusHub, config: Dict[str, Any]): + def __init__(self, hub: ModbusHub, config: dict[str, Any]): """Initialize the coil switch.""" super().__init__(hub, config) self._coil = config[CALL_TYPE_COIL] @@ -184,7 +186,7 @@ class ModbusCoilSwitch(ModbusBaseSwitch, SwitchEntity): class ModbusRegisterSwitch(ModbusBaseSwitch, SwitchEntity): """Representation of a Modbus register switch.""" - def __init__(self, hub: ModbusHub, config: Dict[str, Any]): + def __init__(self, hub: ModbusHub, config: dict[str, Any]): """Initialize the register switch.""" super().__init__(hub, config) self._register = config[CONF_REGISTER] @@ -238,7 +240,7 @@ class ModbusRegisterSwitch(ModbusBaseSwitch, SwitchEntity): value, ) - def _read_register(self) -> Optional[int]: + def _read_register(self) -> int | None: try: if self._register_type == CALL_TYPE_REGISTER_INPUT: result = self._hub.read_input_registers( diff --git a/homeassistant/components/mqtt/__init__.py b/homeassistant/components/mqtt/__init__.py index 503083b0067..ce2d413e1b6 100644 --- a/homeassistant/components/mqtt/__init__.py +++ b/homeassistant/components/mqtt/__init__.py @@ -1,4 +1,6 @@ """Support for MQTT message handling.""" +from __future__ import annotations + import asyncio from functools import lru_cache, partial, wraps import inspect @@ -8,7 +10,7 @@ from operator import attrgetter import os import ssl import time -from typing import Any, Callable, List, Optional, Union +from typing import Any, Callable, Union import uuid import attr @@ -310,7 +312,7 @@ async def async_subscribe( topic: str, msg_callback: MessageCallbackType, qos: int = DEFAULT_QOS, - encoding: Optional[str] = "utf-8", + encoding: str | None = "utf-8", ): """Subscribe to an MQTT topic. @@ -385,7 +387,7 @@ async def _async_setup_discovery( async def async_setup(hass: HomeAssistantType, config: ConfigType) -> bool: """Start the MQTT protocol service.""" - conf: Optional[ConfigType] = config.get(DOMAIN) + conf: ConfigType | None = config.get(DOMAIN) websocket_api.async_register_command(hass, websocket_subscribe) websocket_api.async_register_command(hass, websocket_remove_device) @@ -552,7 +554,7 @@ class MQTT: self.hass = hass self.config_entry = config_entry self.conf = conf - self.subscriptions: List[Subscription] = [] + self.subscriptions: list[Subscription] = [] self.connected = False self._ha_started = asyncio.Event() self._last_subscribe = time.time() @@ -730,7 +732,7 @@ class MQTT: topic: str, msg_callback: MessageCallbackType, qos: int, - encoding: Optional[str] = None, + encoding: str | None = None, ) -> Callable[[], None]: """Set up a subscription to a topic with the provided qos. diff --git a/homeassistant/components/mqtt/device_trigger.py b/homeassistant/components/mqtt/device_trigger.py index d6e2ee0fc65..1e4981d7d6f 100644 --- a/homeassistant/components/mqtt/device_trigger.py +++ b/homeassistant/components/mqtt/device_trigger.py @@ -1,6 +1,8 @@ """Provides device automations for MQTT.""" +from __future__ import annotations + import logging -from typing import Callable, List, Optional +from typing import Callable import attr import voluptuous as vol @@ -86,7 +88,7 @@ class TriggerInstance: action: AutomationActionType = attr.ib() automation_info: dict = attr.ib() trigger: "Trigger" = attr.ib() - remove: Optional[CALLBACK_TYPE] = attr.ib(default=None) + remove: CALLBACK_TYPE | None = attr.ib(default=None) async def async_attach_trigger(self): """Attach MQTT trigger.""" @@ -126,7 +128,7 @@ class Trigger: topic: str = attr.ib() type: str = attr.ib() value_template: str = attr.ib() - trigger_instances: List[TriggerInstance] = attr.ib(factory=list) + trigger_instances: list[TriggerInstance] = attr.ib(factory=list) async def add_trigger(self, action, automation_info): """Add MQTT trigger.""" @@ -285,7 +287,7 @@ async def async_device_removed(hass: HomeAssistant, device_id: str): ) -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 MQTT devices.""" triggers = [] diff --git a/homeassistant/components/mqtt/mixins.py b/homeassistant/components/mqtt/mixins.py index 5737dad255c..332632f4e0f 100644 --- a/homeassistant/components/mqtt/mixins.py +++ b/homeassistant/components/mqtt/mixins.py @@ -1,8 +1,9 @@ """MQTT component mixins and helpers.""" +from __future__ import annotations + from abc import abstractmethod import json import logging -from typing import Optional import voluptuous as vol @@ -502,7 +503,7 @@ def device_info_from_config(config): class MqttEntityDeviceInfo(Entity): """Mixin used for mqtt platforms that support the device registry.""" - def __init__(self, device_config: Optional[ConfigType], config_entry=None) -> None: + def __init__(self, device_config: ConfigType | None, config_entry=None) -> None: """Initialize the device mixin.""" self._device_config = device_config self._config_entry = config_entry diff --git a/homeassistant/components/mqtt/models.py b/homeassistant/components/mqtt/models.py index 202f457372c..7cdafeef98d 100644 --- a/homeassistant/components/mqtt/models.py +++ b/homeassistant/components/mqtt/models.py @@ -1,6 +1,8 @@ """Modesl used by multiple MQTT modules.""" +from __future__ import annotations + import datetime as dt -from typing import Callable, Optional, Union +from typing import Callable, Union import attr @@ -15,8 +17,8 @@ class Message: payload: PublishPayloadType = attr.ib() qos: int = attr.ib() retain: bool = attr.ib() - subscribed_topic: Optional[str] = attr.ib(default=None) - timestamp: Optional[dt.datetime] = attr.ib(default=None) + subscribed_topic: str | None = attr.ib(default=None) + timestamp: dt.datetime | None = attr.ib(default=None) MessageCallbackType = Callable[[Message], None] diff --git a/homeassistant/components/mqtt/sensor.py b/homeassistant/components/mqtt/sensor.py index d32f4c42383..b20595922cd 100644 --- a/homeassistant/components/mqtt/sensor.py +++ b/homeassistant/components/mqtt/sensor.py @@ -1,7 +1,8 @@ """Support for MQTT sensors.""" +from __future__ import annotations + from datetime import timedelta import functools -from typing import Optional import voluptuous as vol @@ -166,7 +167,7 @@ class MqttSensor(MqttEntity, Entity): return self._state @property - def device_class(self) -> Optional[str]: + def device_class(self) -> str | None: """Return the device class of the sensor.""" return self._config.get(CONF_DEVICE_CLASS) diff --git a/homeassistant/components/mqtt/subscription.py b/homeassistant/components/mqtt/subscription.py index 5c2efabc266..e6c99c09fd5 100644 --- a/homeassistant/components/mqtt/subscription.py +++ b/homeassistant/components/mqtt/subscription.py @@ -1,5 +1,7 @@ """Helper to handle a set of topics to subscribe to.""" -from typing import Any, Callable, Dict, Optional +from __future__ import annotations + +from typing import Any, Callable import attr @@ -19,7 +21,7 @@ class EntitySubscription: hass: HomeAssistantType = attr.ib() topic: str = attr.ib() message_callback: MessageCallbackType = attr.ib() - unsubscribe_callback: Optional[Callable[[], None]] = attr.ib() + unsubscribe_callback: Callable[[], None] | None = attr.ib() qos: int = attr.ib(default=0) encoding: str = attr.ib(default="utf-8") @@ -62,8 +64,8 @@ class EntitySubscription: @bind_hass async def async_subscribe_topics( hass: HomeAssistantType, - new_state: Optional[Dict[str, EntitySubscription]], - topics: Dict[str, Any], + new_state: dict[str, EntitySubscription] | None, + topics: dict[str, Any], ): """(Re)Subscribe to a set of MQTT topics. diff --git a/homeassistant/components/mysensors/__init__.py b/homeassistant/components/mysensors/__init__.py index 7dd118099f7..c9ad496762d 100644 --- a/homeassistant/components/mysensors/__init__.py +++ b/homeassistant/components/mysensors/__init__.py @@ -1,8 +1,10 @@ """Connect to a MySensors gateway via pymysensors API.""" +from __future__ import annotations + import asyncio from functools import partial import logging -from typing import Callable, Dict, List, Optional, Tuple, Type, Union +from typing import Callable from mysensors import BaseAsyncGateway import voluptuous as vol @@ -265,13 +267,13 @@ async def async_unload_entry(hass: HomeAssistantType, entry: ConfigEntry) -> boo def setup_mysensors_platform( hass: HomeAssistant, domain: str, # hass platform name - discovery_info: Dict[str, List[DevId]], - device_class: Union[Type[MySensorsDevice], Dict[SensorType, Type[MySensorsEntity]]], - device_args: Optional[ - Tuple - ] = None, # extra arguments that will be given to the entity constructor - async_add_entities: Optional[Callable] = None, -) -> Optional[List[MySensorsDevice]]: + discovery_info: dict[str, list[DevId]], + device_class: type[MySensorsDevice] | dict[SensorType, type[MySensorsEntity]], + device_args: ( + None | tuple + ) = None, # extra arguments that will be given to the entity constructor + async_add_entities: Callable | None = None, +) -> list[MySensorsDevice] | None: """Set up a MySensors platform. Sets up a bunch of instances of a single platform that is supported by this integration. @@ -281,10 +283,10 @@ def setup_mysensors_platform( """ if device_args is None: device_args = () - new_devices: List[MySensorsDevice] = [] - new_dev_ids: List[DevId] = discovery_info[ATTR_DEVICES] + new_devices: list[MySensorsDevice] = [] + new_dev_ids: list[DevId] = discovery_info[ATTR_DEVICES] for dev_id in new_dev_ids: - devices: Dict[DevId, MySensorsDevice] = get_mysensors_devices(hass, domain) + devices: dict[DevId, MySensorsDevice] = get_mysensors_devices(hass, domain) if dev_id in devices: _LOGGER.debug( "Skipping setup of %s for platform %s as it already exists", @@ -293,7 +295,7 @@ def setup_mysensors_platform( ) continue gateway_id, node_id, child_id, value_type = dev_id - gateway: Optional[BaseAsyncGateway] = get_mysensors_gateway(hass, gateway_id) + gateway: BaseAsyncGateway | None = get_mysensors_gateway(hass, gateway_id) if not gateway: _LOGGER.warning("Skipping setup of %s, no gateway found", dev_id) continue diff --git a/homeassistant/components/mysensors/config_flow.py b/homeassistant/components/mysensors/config_flow.py index 7bf30e4cab5..3a37799106a 100644 --- a/homeassistant/components/mysensors/config_flow.py +++ b/homeassistant/components/mysensors/config_flow.py @@ -1,7 +1,9 @@ """Config flow for MySensors.""" +from __future__ import annotations + import logging import os -from typing import Any, Dict, Optional +from typing import Any from awesomeversion import ( AwesomeVersion, @@ -42,7 +44,7 @@ from .gateway import MQTT_COMPONENT, is_serial_port, is_socket_address, try_conn _LOGGER = logging.getLogger(__name__) -def _get_schema_common(user_input: Dict[str, str]) -> dict: +def _get_schema_common(user_input: dict[str, str]) -> dict: """Create a schema with options common to all gateway types.""" schema = { vol.Required( @@ -57,7 +59,7 @@ def _get_schema_common(user_input: Dict[str, str]) -> dict: return schema -def _validate_version(version: str) -> Dict[str, str]: +def _validate_version(version: str) -> dict[str, str]: """Validate a version string from the user.""" version_okay = False try: @@ -75,7 +77,7 @@ def _validate_version(version: str) -> Dict[str, str]: def _is_same_device( - gw_type: ConfGatewayType, user_input: Dict[str, str], entry: ConfigEntry + gw_type: ConfGatewayType, user_input: dict[str, str], entry: ConfigEntry ): """Check if another ConfigDevice is actually the same as user_input. @@ -102,9 +104,9 @@ class MySensorsConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): def __init__(self) -> None: """Set up config flow.""" - self._gw_type: Optional[str] = None + self._gw_type: str | None = None - async def async_step_import(self, user_input: Optional[Dict[str, str]] = None): + async def async_step_import(self, user_input: dict[str, str] | None = None): """Import a config entry. This method is called by async_setup and it has already @@ -124,12 +126,12 @@ class MySensorsConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): else: user_input[CONF_GATEWAY_TYPE] = CONF_GATEWAY_TYPE_SERIAL - result: Dict[str, Any] = await self.async_step_user(user_input=user_input) + result: dict[str, Any] = await self.async_step_user(user_input=user_input) if result["type"] == "form": return self.async_abort(reason=next(iter(result["errors"].values()))) return result - async def async_step_user(self, user_input: Optional[Dict[str, str]] = None): + async def async_step_user(self, user_input: dict[str, str] | None = None): """Create a config entry from frontend user input.""" schema = {vol.Required(CONF_GATEWAY_TYPE): vol.In(CONF_GATEWAY_TYPE_ALL)} schema = vol.Schema(schema) @@ -146,7 +148,7 @@ class MySensorsConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): return self.async_show_form(step_id="user", data_schema=schema) - async def async_step_gw_serial(self, user_input: Optional[Dict[str, str]] = None): + async def async_step_gw_serial(self, user_input: dict[str, str] | None = None): """Create config entry for a serial gateway.""" errors = {} if user_input is not None: @@ -175,7 +177,7 @@ class MySensorsConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): step_id="gw_serial", data_schema=schema, errors=errors ) - async def async_step_gw_tcp(self, user_input: Optional[Dict[str, str]] = None): + async def async_step_gw_tcp(self, user_input: dict[str, str] | None = None): """Create a config entry for a tcp gateway.""" errors = {} if user_input is not None: @@ -213,7 +215,7 @@ class MySensorsConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): return True return False - async def async_step_gw_mqtt(self, user_input: Optional[Dict[str, str]] = None): + async def async_step_gw_mqtt(self, user_input: dict[str, str] | None = None): """Create a config entry for a mqtt gateway.""" errors = {} if user_input is not None: @@ -269,8 +271,8 @@ class MySensorsConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): @callback def _async_create_entry( - self, user_input: Optional[Dict[str, str]] = None - ) -> Dict[str, Any]: + self, user_input: dict[str, str] | None = None + ) -> dict[str, Any]: """Create the config entry.""" return self.async_create_entry( title=f"{user_input[CONF_DEVICE]}", @@ -283,9 +285,9 @@ class MySensorsConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): async def validate_common( self, gw_type: ConfGatewayType, - errors: Dict[str, str], - user_input: Optional[Dict[str, str]] = None, - ) -> Dict[str, str]: + errors: dict[str, str], + user_input: dict[str, str] | None = None, + ) -> dict[str, str]: """Validate parameters common to all gateway types.""" if user_input is not None: errors.update(_validate_version(user_input.get(CONF_VERSION))) diff --git a/homeassistant/components/mysensors/const.py b/homeassistant/components/mysensors/const.py index 1b8fa5e24e8..7a9027d9b72 100644 --- a/homeassistant/components/mysensors/const.py +++ b/homeassistant/components/mysensors/const.py @@ -1,6 +1,8 @@ """MySensors constants.""" +from __future__ import annotations + from collections import defaultdict -from typing import Dict, List, Literal, Set, Tuple +from typing import Literal, Tuple ATTR_DEVICES: str = "devices" ATTR_GATEWAY_ID: str = "gateway_id" @@ -21,7 +23,7 @@ ConfGatewayType = Literal["Serial", "TCP", "MQTT"] CONF_GATEWAY_TYPE_SERIAL: ConfGatewayType = "Serial" CONF_GATEWAY_TYPE_TCP: ConfGatewayType = "TCP" CONF_GATEWAY_TYPE_MQTT: ConfGatewayType = "MQTT" -CONF_GATEWAY_TYPE_ALL: List[str] = [ +CONF_GATEWAY_TYPE_ALL: list[str] = [ CONF_GATEWAY_TYPE_MQTT, CONF_GATEWAY_TYPE_SERIAL, CONF_GATEWAY_TYPE_TCP, @@ -62,7 +64,7 @@ DevId = Tuple[GatewayId, int, int, int] # The MySensors integration brings these together by creating an entity for every v_type of every child_id of every node. # The DevId tuple perfectly captures this. -BINARY_SENSOR_TYPES: Dict[SensorType, Set[ValueType]] = { +BINARY_SENSOR_TYPES: dict[SensorType, set[ValueType]] = { "S_DOOR": {"V_TRIPPED"}, "S_MOTION": {"V_TRIPPED"}, "S_SMOKE": {"V_TRIPPED"}, @@ -73,23 +75,23 @@ BINARY_SENSOR_TYPES: Dict[SensorType, Set[ValueType]] = { "S_MOISTURE": {"V_TRIPPED"}, } -CLIMATE_TYPES: Dict[SensorType, Set[ValueType]] = {"S_HVAC": {"V_HVAC_FLOW_STATE"}} +CLIMATE_TYPES: dict[SensorType, set[ValueType]] = {"S_HVAC": {"V_HVAC_FLOW_STATE"}} -COVER_TYPES: Dict[SensorType, Set[ValueType]] = { +COVER_TYPES: dict[SensorType, set[ValueType]] = { "S_COVER": {"V_DIMMER", "V_PERCENTAGE", "V_LIGHT", "V_STATUS"} } -DEVICE_TRACKER_TYPES: Dict[SensorType, Set[ValueType]] = {"S_GPS": {"V_POSITION"}} +DEVICE_TRACKER_TYPES: dict[SensorType, set[ValueType]] = {"S_GPS": {"V_POSITION"}} -LIGHT_TYPES: Dict[SensorType, Set[ValueType]] = { +LIGHT_TYPES: dict[SensorType, set[ValueType]] = { "S_DIMMER": {"V_DIMMER", "V_PERCENTAGE"}, "S_RGB_LIGHT": {"V_RGB"}, "S_RGBW_LIGHT": {"V_RGBW"}, } -NOTIFY_TYPES: Dict[SensorType, Set[ValueType]] = {"S_INFO": {"V_TEXT"}} +NOTIFY_TYPES: dict[SensorType, set[ValueType]] = {"S_INFO": {"V_TEXT"}} -SENSOR_TYPES: Dict[SensorType, Set[ValueType]] = { +SENSOR_TYPES: dict[SensorType, set[ValueType]] = { "S_SOUND": {"V_LEVEL"}, "S_VIBRATION": {"V_LEVEL"}, "S_MOISTURE": {"V_LEVEL"}, @@ -117,7 +119,7 @@ SENSOR_TYPES: Dict[SensorType, Set[ValueType]] = { "S_DUST": {"V_DUST_LEVEL", "V_LEVEL"}, } -SWITCH_TYPES: Dict[SensorType, Set[ValueType]] = { +SWITCH_TYPES: dict[SensorType, set[ValueType]] = { "S_LIGHT": {"V_LIGHT"}, "S_BINARY": {"V_STATUS"}, "S_DOOR": {"V_ARMED"}, @@ -134,7 +136,7 @@ SWITCH_TYPES: Dict[SensorType, Set[ValueType]] = { } -PLATFORM_TYPES: Dict[str, Dict[SensorType, Set[ValueType]]] = { +PLATFORM_TYPES: dict[str, dict[SensorType, set[ValueType]]] = { "binary_sensor": BINARY_SENSOR_TYPES, "climate": CLIMATE_TYPES, "cover": COVER_TYPES, @@ -145,13 +147,13 @@ PLATFORM_TYPES: Dict[str, Dict[SensorType, Set[ValueType]]] = { "switch": SWITCH_TYPES, } -FLAT_PLATFORM_TYPES: Dict[Tuple[str, SensorType], Set[ValueType]] = { +FLAT_PLATFORM_TYPES: dict[tuple[str, SensorType], set[ValueType]] = { (platform, s_type_name): v_type_name for platform, platform_types in PLATFORM_TYPES.items() for s_type_name, v_type_name in platform_types.items() } -TYPE_TO_PLATFORMS: Dict[SensorType, List[str]] = defaultdict(list) +TYPE_TO_PLATFORMS: dict[SensorType, list[str]] = defaultdict(list) for platform, platform_types in PLATFORM_TYPES.items(): for s_type_name in platform_types: diff --git a/homeassistant/components/mysensors/device.py b/homeassistant/components/mysensors/device.py index d5ae6c0c156..4e770f70bf0 100644 --- a/homeassistant/components/mysensors/device.py +++ b/homeassistant/components/mysensors/device.py @@ -1,7 +1,9 @@ """Handle MySensors devices.""" +from __future__ import annotations + from functools import partial import logging -from typing import Any, Dict, Optional +from typing import Any from mysensors import BaseAsyncGateway, Sensor from mysensors.sensor import ChildSensor @@ -107,7 +109,7 @@ class MySensorsDevice: return f"{self.gateway_id}-{self.node_id}-{self.child_id}-{self.value_type}" @property - def device_info(self) -> Optional[Dict[str, Any]]: + def device_info(self) -> dict[str, Any] | None: """Return a dict that allows home assistant to puzzle all entities belonging to a node together.""" return { "identifiers": {(DOMAIN, f"{self.gateway_id}-{self.node_id}")}, @@ -196,7 +198,7 @@ class MySensorsDevice: self.hass.loop.call_later(UPDATE_DELAY, delayed_update) -def get_mysensors_devices(hass, domain: str) -> Dict[DevId, MySensorsDevice]: +def get_mysensors_devices(hass, domain: str) -> dict[DevId, MySensorsDevice]: """Return MySensors devices for a hass platform name.""" if MYSENSORS_PLATFORM_DEVICES.format(domain) not in hass.data[DOMAIN]: hass.data[DOMAIN][MYSENSORS_PLATFORM_DEVICES.format(domain)] = {} diff --git a/homeassistant/components/mysensors/gateway.py b/homeassistant/components/mysensors/gateway.py index a7f3a053d3f..6cf8e7d7383 100644 --- a/homeassistant/components/mysensors/gateway.py +++ b/homeassistant/components/mysensors/gateway.py @@ -1,10 +1,12 @@ """Handle MySensors gateways.""" +from __future__ import annotations + import asyncio from collections import defaultdict import logging import socket import sys -from typing import Any, Callable, Coroutine, Dict, Optional +from typing import Any, Callable, Coroutine import async_timeout from mysensors import BaseAsyncGateway, Message, Sensor, mysensors @@ -63,7 +65,7 @@ def is_socket_address(value): raise vol.Invalid("Device is not a valid domain name or ip address") from err -async def try_connect(hass: HomeAssistantType, user_input: Dict[str, str]) -> bool: +async def try_connect(hass: HomeAssistantType, user_input: dict[str, str]) -> bool: """Try to connect to a gateway and report if it worked.""" if user_input[CONF_DEVICE] == MQTT_COMPONENT: return True # dont validate mqtt. mqtt gateways dont send ready messages :( @@ -73,7 +75,7 @@ async def try_connect(hass: HomeAssistantType, user_input: Dict[str, str]) -> bo def on_conn_made(_: BaseAsyncGateway) -> None: gateway_ready.set() - gateway: Optional[BaseAsyncGateway] = await _get_gateway( + gateway: BaseAsyncGateway | None = await _get_gateway( hass, device=user_input[CONF_DEVICE], version=user_input[CONF_VERSION], @@ -110,7 +112,7 @@ async def try_connect(hass: HomeAssistantType, user_input: Dict[str, str]) -> bo def get_mysensors_gateway( hass: HomeAssistantType, gateway_id: GatewayId -) -> Optional[BaseAsyncGateway]: +) -> BaseAsyncGateway | None: """Return the Gateway for a given GatewayId.""" if MYSENSORS_GATEWAYS not in hass.data[DOMAIN]: hass.data[DOMAIN][MYSENSORS_GATEWAYS] = {} @@ -120,7 +122,7 @@ def get_mysensors_gateway( async def setup_gateway( hass: HomeAssistantType, entry: ConfigEntry -) -> Optional[BaseAsyncGateway]: +) -> BaseAsyncGateway | None: """Set up the Gateway for the given ConfigEntry.""" ready_gateway = await _get_gateway( @@ -145,14 +147,14 @@ async def _get_gateway( device: str, version: str, event_callback: Callable[[Message], None], - persistence_file: Optional[str] = None, - baud_rate: Optional[int] = None, - tcp_port: Optional[int] = None, - topic_in_prefix: Optional[str] = None, - topic_out_prefix: Optional[str] = None, + persistence_file: str | None = None, + baud_rate: int | None = None, + tcp_port: int | None = None, + topic_in_prefix: str | None = None, + topic_out_prefix: str | None = None, retain: bool = False, persistence: bool = True, # old persistence option has been deprecated. kwarg is here so we can run try_connect() without persistence -) -> Optional[BaseAsyncGateway]: +) -> BaseAsyncGateway | None: """Return gateway after setup of the gateway.""" if persistence_file is not None: diff --git a/homeassistant/components/mysensors/handler.py b/homeassistant/components/mysensors/handler.py index a47c9174b23..d21140701f9 100644 --- a/homeassistant/components/mysensors/handler.py +++ b/homeassistant/components/mysensors/handler.py @@ -1,5 +1,5 @@ """Handle MySensors messages.""" -from typing import Dict, List +from __future__ import annotations from mysensors import Message @@ -70,16 +70,16 @@ async def handle_sketch_version( @callback def _handle_child_update( - hass: HomeAssistantType, gateway_id: GatewayId, validated: Dict[str, List[DevId]] + hass: HomeAssistantType, gateway_id: GatewayId, validated: dict[str, list[DevId]] ): """Handle a child update.""" - signals: List[str] = [] + signals: list[str] = [] # Update all platforms for the device via dispatcher. # Add/update entity for validated children. for platform, dev_ids in validated.items(): devices = get_mysensors_devices(hass, platform) - new_dev_ids: List[DevId] = [] + new_dev_ids: list[DevId] = [] for dev_id in dev_ids: if dev_id in devices: signals.append(CHILD_CALLBACK.format(*dev_id)) diff --git a/homeassistant/components/mysensors/helpers.py b/homeassistant/components/mysensors/helpers.py index 0b8dc361158..0d18b243520 100644 --- a/homeassistant/components/mysensors/helpers.py +++ b/homeassistant/components/mysensors/helpers.py @@ -1,8 +1,10 @@ """Helper functions for mysensors package.""" +from __future__ import annotations + from collections import defaultdict from enum import IntEnum import logging -from typing import Callable, DefaultDict, Dict, List, Optional, Set, Union +from typing import Callable, DefaultDict from mysensors import BaseAsyncGateway, Message from mysensors.sensor import ChildSensor @@ -35,7 +37,7 @@ SCHEMAS = Registry() async def on_unload( - hass: HomeAssistantType, entry: Union[ConfigEntry, GatewayId], fnct: Callable + hass: HomeAssistantType, entry: ConfigEntry | GatewayId, fnct: Callable ) -> None: """Register a callback to be called when entry is unloaded. @@ -53,7 +55,7 @@ async def on_unload( @callback def discover_mysensors_platform( - hass: HomeAssistant, gateway_id: GatewayId, platform: str, new_devices: List[DevId] + hass: HomeAssistant, gateway_id: GatewayId, platform: str, new_devices: list[DevId] ) -> None: """Discover a MySensors platform.""" _LOGGER.debug("Discovering platform %s with devIds: %s", platform, new_devices) @@ -150,7 +152,7 @@ def invalid_msg( ) -def validate_set_msg(gateway_id: GatewayId, msg: Message) -> Dict[str, List[DevId]]: +def validate_set_msg(gateway_id: GatewayId, msg: Message) -> dict[str, list[DevId]]: """Validate a set message.""" if not validate_node(msg.gateway, msg.node_id): return {} @@ -171,34 +173,34 @@ def validate_child( gateway: BaseAsyncGateway, node_id: int, child: ChildSensor, - value_type: Optional[int] = None, -) -> DefaultDict[str, List[DevId]]: + value_type: int | None = None, +) -> DefaultDict[str, list[DevId]]: """Validate a child. Returns a dict mapping hass platform names to list of DevId.""" - validated: DefaultDict[str, List[DevId]] = defaultdict(list) + validated: DefaultDict[str, list[DevId]] = defaultdict(list) pres: IntEnum = gateway.const.Presentation set_req: IntEnum = gateway.const.SetReq - child_type_name: Optional[SensorType] = next( + child_type_name: SensorType | None = next( (member.name for member in pres if member.value == child.type), None ) - value_types: Set[int] = {value_type} if value_type else {*child.values} - value_type_names: Set[ValueType] = { + value_types: set[int] = {value_type} if value_type else {*child.values} + value_type_names: set[ValueType] = { member.name for member in set_req if member.value in value_types } - platforms: List[str] = TYPE_TO_PLATFORMS.get(child_type_name, []) + platforms: list[str] = TYPE_TO_PLATFORMS.get(child_type_name, []) if not platforms: _LOGGER.warning("Child type %s is not supported", child.type) return validated for platform in platforms: - platform_v_names: Set[ValueType] = FLAT_PLATFORM_TYPES[ + platform_v_names: set[ValueType] = FLAT_PLATFORM_TYPES[ platform, child_type_name ] - v_names: Set[ValueType] = platform_v_names & value_type_names + v_names: set[ValueType] = platform_v_names & value_type_names if not v_names: - child_value_names: Set[ValueType] = { + child_value_names: set[ValueType] = { member.name for member in set_req if member.value in child.values } - v_names: Set[ValueType] = platform_v_names & child_value_names + v_names: set[ValueType] = platform_v_names & child_value_names for v_name in v_names: child_schema_gen = SCHEMAS.get((platform, v_name), default_schema)