evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Update typing 10 (#48071)

Browse source
This commit is contained in:
Marc Mueller 2021-03-18 13:07:04 +01:00 • committed by GitHub
parent 00dca88024
commit 25a13d1554
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
73 changed files with 412 additions and 333 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
homeassistant/components/jewish_calendar/__init__.py
View file

@ -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 = [

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

@ -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
)

11
homeassistant/components/keenetic_ndms2/device_tracker.py
View file

@ -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)

14
homeassistant/components/keenetic_ndms2/router.py
View file

@ -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

5
homeassistant/components/knx/__init__.py
View file

@ -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:

6
homeassistant/components/knx/binary_sensor.py
View file

@ -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}

10
homeassistant/components/knx/climate.py
View file

@ -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.

6
homeassistant/components/knx/expose.py
View file

@ -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:

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

@ -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]

12
homeassistant/components/knx/fan.py
View file

@ -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."""

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

@ -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

4
homeassistant/components/kodi/device_trigger.py
View file

@ -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 = []

6
homeassistant/components/kulersky/light.py
View file

@ -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:

7
homeassistant/components/launch_library/sensor.py
View file

@ -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 {

52
homeassistant/components/light/__init__.py
View file

@ -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

4
homeassistant/components/light/device_action.py
View file

@ -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)

4
homeassistant/components/light/device_condition.py
View file

@ -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)

4
homeassistant/components/light/device_trigger.py
View file

@ -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)

14
homeassistant/components/light/reproduce_state.py
View file

@ -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(

6
homeassistant/components/light/significant_change.py
View file

@ -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

8
homeassistant/components/litejet/config_flow.py
View file

@ -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")

6
homeassistant/components/litterrobot/hub.py
View file

@ -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)

4
homeassistant/components/litterrobot/sensor.py
View file

@ -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"

5
homeassistant/components/locative/__init__.py
View file

@ -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")

6
homeassistant/components/lock/device_action.py
View file

@ -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)

4
homeassistant/components/lock/device_condition.py
View file

@ -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 = []

4
homeassistant/components/lock/device_trigger.py
View file

@ -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 = []

12
homeassistant/components/lock/reproduce_state.py
View file

@ -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(

6
homeassistant/components/lock/significant_change.py
View file

@ -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

8
homeassistant/components/lovelace/resources.py
View file

@ -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()

4
homeassistant/components/lutron_caseta/device_trigger.py
View file

@ -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 = []

5
homeassistant/components/lutron_caseta/fan.py
View file

@ -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

8
homeassistant/components/lyric/__init__.py
View file

@ -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)},

17
homeassistant/components/lyric/climate.py
View file

@ -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,

25
homeassistant/components/media_player/__init__.py
View file

@ -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

4
homeassistant/components/media_player/device_condition.py
View file

@ -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 = []

4
homeassistant/components/media_player/device_trigger.py
View file

@ -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 = []

12
homeassistant/components/media_player/reproduce_state.py
View file

@ -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(

5
homeassistant/components/media_source/__init__.py
View file

@ -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:

7
homeassistant/components/media_source/local_source.py
View file

@ -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:

9
homeassistant/components/media_source/models.py
View file

@ -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

8
homeassistant/components/melcloud/__init__.py
View file

@ -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:

34
homeassistant/components/melcloud/climate.py
View file

@ -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

7
homeassistant/components/melcloud/config_flow.py
View file

@ -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:

14
homeassistant/components/melcloud/water_heater.py
View file

@ -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

8
homeassistant/components/met/config_flow.py
View file

@ -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)

5
homeassistant/components/minecraft_server/__init__.py
View file

@ -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

5
homeassistant/components/minecraft_server/helpers.py
View file

@ -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

6
homeassistant/components/minecraft_server/sensor.py
View file

@ -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

5
homeassistant/components/minio/__init__.py
View file

@ -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

8
homeassistant/components/minio/minio_helper.py
View file

@ -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__()

6
homeassistant/components/mobile_app/device_action.py
View file

@ -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])

22
homeassistant/components/mobile_app/helpers.py
View file

@ -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])},

5
homeassistant/components/mobile_app/http_api.py
View file

@ -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"]

8
homeassistant/components/mobile_app/util.py
View file

@ -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]

4
homeassistant/components/modbus/binary_sensor.py
View file

@ -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

10
homeassistant/components/modbus/climate.py
View file

@ -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:

14
homeassistant/components/modbus/cover.py
View file

@ -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)

8
homeassistant/components/modbus/sensor.py
View file

@ -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

12
homeassistant/components/modbus/switch.py
View file

@ -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(

12
homeassistant/components/mqtt/__init__.py
View file

@ -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.

10
homeassistant/components/mqtt/device_trigger.py
View file

@ -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 = []

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

@ -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

8
homeassistant/components/mqtt/models.py
View file

@ -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]

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

@ -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)

10
homeassistant/components/mqtt/subscription.py
View file

@ -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.

26
homeassistant/components/mysensors/__init__.py
View file

@ -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

34
homeassistant/components/mysensors/config_flow.py
View file

@ -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)))

28
homeassistant/components/mysensors/const.py
View file

@ -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:

8
homeassistant/components/mysensors/device.py
View file

@ -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)] = {}

24
homeassistant/components/mysensors/gateway.py
View file

@ -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:

8
homeassistant/components/mysensors/handler.py
View file

@ -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))

32
homeassistant/components/mysensors/helpers.py
View file

@ -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)