Update typing 05 (#48038)

This commit is contained in:
Marc Mueller 2021-03-17 23:43:55 +01:00 committed by GitHub
parent 76199c0eb2
commit 7c0734bdd5
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
57 changed files with 315 additions and 251 deletions

View file

@ -1,8 +1,10 @@
"""Support for Google Calendar event device sensors."""
from __future__ import annotations
from datetime import timedelta
import logging
import re
from typing import Dict, List, cast
from typing import cast
from aiohttp import web
@ -218,7 +220,7 @@ class CalendarListView(http.HomeAssistantView):
async def get(self, request: web.Request) -> web.Response:
"""Retrieve calendar list."""
hass = request.app["hass"]
calendar_list: List[Dict[str, str]] = []
calendar_list: list[dict[str, str]] = []
for entity in self.component.entities:
state = hass.states.get(entity.entity_id)

View file

@ -1,5 +1,7 @@
"""Support for Canary alarm."""
from typing import Callable, List
from __future__ import annotations
from typing import Callable
from canary.api import LOCATION_MODE_AWAY, LOCATION_MODE_HOME, LOCATION_MODE_NIGHT
@ -27,7 +29,7 @@ from .coordinator import CanaryDataUpdateCoordinator
async def async_setup_entry(
hass: HomeAssistantType,
entry: ConfigEntry,
async_add_entities: Callable[[List[Entity], bool], None],
async_add_entities: Callable[[list[Entity], bool], None],
) -> None:
"""Set up Canary alarm control panels based on a config entry."""
coordinator: CanaryDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][

View file

@ -1,7 +1,9 @@
"""Support for Canary camera."""
from __future__ import annotations
import asyncio
from datetime import timedelta
from typing import Callable, List
from typing import Callable
from haffmpeg.camera import CameraMjpeg
from haffmpeg.tools import IMAGE_JPEG, ImageFrame
@ -44,7 +46,7 @@ PLATFORM_SCHEMA = vol.All(
async def async_setup_entry(
hass: HomeAssistantType,
entry: ConfigEntry,
async_add_entities: Callable[[List[Entity], bool], None],
async_add_entities: Callable[[list[Entity], bool], None],
) -> None:
"""Set up Canary sensors based on a config entry."""
coordinator: CanaryDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][

View file

@ -1,6 +1,8 @@
"""Config flow for Canary."""
from __future__ import annotations
import logging
from typing import Any, Dict, Optional
from typing import Any
from canary.api import Api
from requests import ConnectTimeout, HTTPError
@ -17,7 +19,7 @@ from .const import DOMAIN # pylint: disable=unused-import
_LOGGER = logging.getLogger(__name__)
def validate_input(hass: HomeAssistantType, data: dict) -> Dict[str, Any]:
def validate_input(hass: HomeAssistantType, data: dict) -> dict[str, Any]:
"""Validate the user input allows us to connect.
Data has the keys from DATA_SCHEMA with values provided by the user.
@ -45,14 +47,14 @@ class CanaryConfigFlow(ConfigFlow, domain=DOMAIN):
return CanaryOptionsFlowHandler(config_entry)
async def async_step_import(
self, user_input: Optional[ConfigType] = None
) -> Dict[str, Any]:
self, user_input: ConfigType | None = None
) -> dict[str, Any]:
"""Handle a flow initiated by configuration file."""
return await self.async_step_user(user_input)
async def async_step_user(
self, user_input: Optional[ConfigType] = None
) -> Dict[str, Any]:
self, user_input: ConfigType | None = None
) -> dict[str, Any]:
"""Handle a flow initiated by the user."""
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
@ -100,7 +102,7 @@ class CanaryOptionsFlowHandler(OptionsFlow):
"""Initialize options flow."""
self.config_entry = config_entry
async def async_step_init(self, user_input: Optional[ConfigType] = None):
async def async_step_init(self, user_input: ConfigType | None = None):
"""Manage Canary options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)

View file

@ -1,5 +1,7 @@
"""Support for Canary sensors."""
from typing import Callable, List
from __future__ import annotations
from typing import Callable
from canary.api import SensorType
@ -54,7 +56,7 @@ STATE_AIR_QUALITY_VERY_ABNORMAL = "very_abnormal"
async def async_setup_entry(
hass: HomeAssistantType,
entry: ConfigEntry,
async_add_entities: Callable[[List[Entity], bool], None],
async_add_entities: Callable[[list[Entity], bool], None],
) -> None:
"""Set up Canary sensors based on a config entry."""
coordinator: CanaryDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][

View file

@ -1,5 +1,5 @@
"""Home Assistant Cast integration for Cast."""
from typing import Optional
from __future__ import annotations
from pychromecast.controllers.homeassistant import HomeAssistantController
import voluptuous as vol
@ -20,8 +20,8 @@ async def async_setup_ha_cast(
hass: core.HomeAssistant, entry: config_entries.ConfigEntry
):
"""Set up Home Assistant Cast."""
user_id: Optional[str] = entry.data.get("user_id")
user: Optional[auth.models.User] = None
user_id: str | None = entry.data.get("user_id")
user: auth.models.User | None = None
if user_id is not None:
user = await hass.auth.async_get_user(user_id)
@ -78,7 +78,7 @@ async def async_remove_user(
hass: core.HomeAssistant, entry: config_entries.ConfigEntry
):
"""Remove Home Assistant Cast user."""
user_id: Optional[str] = entry.data.get("user_id")
user_id: str | None = entry.data.get("user_id")
if user_id is not None:
user = await hass.auth.async_get_user(user_id)

View file

@ -1,10 +1,11 @@
"""Provide functionality to interact with Cast devices on the network."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import functools as ft
import json
import logging
from typing import Optional
import pychromecast
from pychromecast.controllers.homeassistant import HomeAssistantController
@ -195,7 +196,7 @@ class CastDevice(MediaPlayerEntity):
self._cast_info = cast_info
self.services = cast_info.services
self._chromecast: Optional[pychromecast.Chromecast] = None
self._chromecast: pychromecast.Chromecast | None = None
self.cast_status = None
self.media_status = None
self.media_status_received = None
@ -203,8 +204,8 @@ class CastDevice(MediaPlayerEntity):
self.mz_media_status_received = {}
self.mz_mgr = None
self._available = False
self._status_listener: Optional[CastStatusListener] = None
self._hass_cast_controller: Optional[HomeAssistantController] = None
self._status_listener: CastStatusListener | None = None
self._hass_cast_controller: HomeAssistantController | None = None
self._add_remove_handler = None
self._cast_view_remove_handler = None
@ -783,7 +784,7 @@ class CastDevice(MediaPlayerEntity):
return media_status_recevied
@property
def unique_id(self) -> Optional[str]:
def unique_id(self) -> str | None:
"""Return a unique ID."""
return self._cast_info.uuid
@ -805,7 +806,7 @@ class CastDevice(MediaPlayerEntity):
controller: HomeAssistantController,
entity_id: str,
view_path: str,
url_path: Optional[str],
url_path: str | None,
):
"""Handle a show view signal."""
if entity_id != self.entity_id:
@ -827,9 +828,9 @@ class DynamicCastGroup:
self.hass = hass
self._cast_info = cast_info
self.services = cast_info.services
self._chromecast: Optional[pychromecast.Chromecast] = None
self._chromecast: pychromecast.Chromecast | None = None
self.mz_mgr = None
self._status_listener: Optional[CastStatusListener] = None
self._status_listener: CastStatusListener | None = None
self._add_remove_handler = None
self._del_remove_handler = None

View file

@ -1,7 +1,8 @@
"""The cert_expiry component."""
from __future__ import annotations
from datetime import datetime, timedelta
import logging
from typing import Optional
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PORT
@ -71,7 +72,7 @@ class CertExpiryDataUpdateCoordinator(DataUpdateCoordinator[datetime]):
update_interval=SCAN_INTERVAL,
)
async def _async_update_data(self) -> Optional[datetime]:
async def _async_update_data(self) -> datetime | None:
"""Fetch certificate."""
try:
timestamp = await get_cert_expiry_timestamp(self.hass, self.host, self.port)

View file

@ -1,9 +1,11 @@
"""The ClimaCell integration."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
from math import ceil
from typing import Any, Dict, Optional, Union
from typing import Any
from pyclimacell import ClimaCell
from pyclimacell.const import (
@ -169,7 +171,7 @@ class ClimaCellDataUpdateCoordinator(DataUpdateCoordinator):
update_interval=update_interval,
)
async def _async_update_data(self) -> Dict[str, Any]:
async def _async_update_data(self) -> dict[str, Any]:
"""Update data via library."""
data = {FORECASTS: {}}
try:
@ -217,8 +219,8 @@ class ClimaCellEntity(CoordinatorEntity):
@staticmethod
def _get_cc_value(
weather_dict: Dict[str, Any], key: str
) -> Optional[Union[int, float, str]]:
weather_dict: dict[str, Any], key: str
) -> int | float | str | None:
"""Return property from weather_dict."""
items = weather_dict.get(key, {})
# Handle cases where value returned is a list.
@ -252,7 +254,7 @@ class ClimaCellEntity(CoordinatorEntity):
return ATTRIBUTION
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return device registry information."""
return {
"identifiers": {(DOMAIN, self._config_entry.data[CONF_API_KEY])},

View file

@ -1,6 +1,8 @@
"""Config flow for ClimaCell integration."""
from __future__ import annotations
import logging
from typing import Any, Dict
from typing import Any
from pyclimacell import ClimaCell
from pyclimacell.const import REALTIME
@ -25,7 +27,7 @@ _LOGGER = logging.getLogger(__name__)
def _get_config_schema(
hass: core.HomeAssistant, input_dict: Dict[str, Any] = None
hass: core.HomeAssistant, input_dict: dict[str, Any] = None
) -> vol.Schema:
"""
Return schema defaults for init step based on user input/config dict.
@ -57,7 +59,7 @@ def _get_config_schema(
)
def _get_unique_id(hass: HomeAssistantType, input_dict: Dict[str, Any]):
def _get_unique_id(hass: HomeAssistantType, input_dict: dict[str, Any]):
"""Return unique ID from config data."""
return (
f"{input_dict[CONF_API_KEY]}"
@ -74,8 +76,8 @@ class ClimaCellOptionsConfigFlow(config_entries.OptionsFlow):
self._config_entry = config_entry
async def async_step_init(
self, user_input: Dict[str, Any] = None
) -> Dict[str, Any]:
self, user_input: dict[str, Any] = None
) -> dict[str, Any]:
"""Manage the ClimaCell options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
@ -107,8 +109,8 @@ class ClimaCellConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return ClimaCellOptionsConfigFlow(config_entry)
async def async_step_user(
self, user_input: Dict[str, Any] = None
) -> Dict[str, Any]:
self, user_input: dict[str, Any] = None
) -> dict[str, Any]:
"""Handle the initial step."""
errors = {}
if user_input is not None:

View file

@ -1,7 +1,9 @@
"""Weather component that handles meteorological data for your location."""
from __future__ import annotations
from datetime import datetime
import logging
from typing import Any, Callable, Dict, List, Optional
from typing import Any, Callable
from homeassistant.components.weather import (
ATTR_FORECAST_CONDITION,
@ -64,9 +66,7 @@ from .const import (
_LOGGER = logging.getLogger(__name__)
def _translate_condition(
condition: Optional[str], sun_is_up: bool = True
) -> Optional[str]:
def _translate_condition(condition: str | None, sun_is_up: bool = True) -> str | None:
"""Translate ClimaCell condition into an HA condition."""
if not condition:
return None
@ -82,13 +82,13 @@ def _forecast_dict(
forecast_dt: datetime,
use_datetime: bool,
condition: str,
precipitation: Optional[float],
precipitation_probability: Optional[float],
temp: Optional[float],
temp_low: Optional[float],
wind_direction: Optional[float],
wind_speed: Optional[float],
) -> Dict[str, Any]:
precipitation: float | None,
precipitation_probability: float | None,
temp: float | None,
temp_low: float | None,
wind_direction: float | None,
wind_speed: float | None,
) -> dict[str, Any]:
"""Return formatted Forecast dict from ClimaCell forecast data."""
if use_datetime:
translated_condition = _translate_condition(condition, is_up(hass, forecast_dt))
@ -120,7 +120,7 @@ def _forecast_dict(
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 a config entry."""
coordinator = hass.data[DOMAIN][config_entry.entry_id]

View file

@ -1,9 +1,11 @@
"""Provides functionality to interact with climate devices."""
from __future__ import annotations
from abc import abstractmethod
from datetime import timedelta
import functools as ft
import logging
from typing import Any, Dict, List, Optional
from typing import Any
import voluptuous as vol
@ -180,7 +182,7 @@ class ClimateEntity(Entity):
return PRECISION_WHOLE
@property
def capability_attributes(self) -> Optional[Dict[str, Any]]:
def capability_attributes(self) -> dict[str, Any] | None:
"""Return the capability attributes."""
supported_features = self.supported_features
data = {
@ -212,7 +214,7 @@ class ClimateEntity(Entity):
return data
@property
def state_attributes(self) -> Dict[str, Any]:
def state_attributes(self) -> dict[str, Any]:
"""Return the optional state attributes."""
supported_features = self.supported_features
data = {
@ -275,12 +277,12 @@ class ClimateEntity(Entity):
raise NotImplementedError()
@property
def current_humidity(self) -> Optional[int]:
def current_humidity(self) -> int | None:
"""Return the current humidity."""
return None
@property
def target_humidity(self) -> Optional[int]:
def target_humidity(self) -> int | None:
"""Return the humidity we try to reach."""
return None
@ -294,14 +296,14 @@ class ClimateEntity(Entity):
@property
@abstractmethod
def hvac_modes(self) -> List[str]:
def hvac_modes(self) -> list[str]:
"""Return the list of available hvac operation modes.
Need to be a subset of HVAC_MODES.
"""
@property
def hvac_action(self) -> Optional[str]:
def hvac_action(self) -> str | None:
"""Return the current running hvac operation if supported.
Need to be one of CURRENT_HVAC_*.
@ -309,22 +311,22 @@ class ClimateEntity(Entity):
return None
@property
def current_temperature(self) -> Optional[float]:
def current_temperature(self) -> float | None:
"""Return the current temperature."""
return None
@property
def target_temperature(self) -> Optional[float]:
def target_temperature(self) -> float | None:
"""Return the temperature we try to reach."""
return None
@property
def target_temperature_step(self) -> Optional[float]:
def target_temperature_step(self) -> float | None:
"""Return the supported step of target temperature."""
return None
@property
def target_temperature_high(self) -> Optional[float]:
def target_temperature_high(self) -> float | None:
"""Return the highbound target temperature we try to reach.
Requires SUPPORT_TARGET_TEMPERATURE_RANGE.
@ -332,7 +334,7 @@ class ClimateEntity(Entity):
raise NotImplementedError
@property
def target_temperature_low(self) -> Optional[float]:
def target_temperature_low(self) -> float | None:
"""Return the lowbound target temperature we try to reach.
Requires SUPPORT_TARGET_TEMPERATURE_RANGE.
@ -340,7 +342,7 @@ class ClimateEntity(Entity):
raise NotImplementedError
@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.
@ -348,7 +350,7 @@ class ClimateEntity(Entity):
raise NotImplementedError
@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.
@ -356,7 +358,7 @@ class ClimateEntity(Entity):
raise NotImplementedError
@property
def is_aux_heat(self) -> Optional[bool]:
def is_aux_heat(self) -> bool | None:
"""Return true if aux heater.
Requires SUPPORT_AUX_HEAT.
@ -364,7 +366,7 @@ class ClimateEntity(Entity):
raise NotImplementedError
@property
def fan_mode(self) -> Optional[str]:
def fan_mode(self) -> str | None:
"""Return the fan setting.
Requires SUPPORT_FAN_MODE.
@ -372,7 +374,7 @@ class ClimateEntity(Entity):
raise NotImplementedError
@property
def fan_modes(self) -> Optional[List[str]]:
def fan_modes(self) -> list[str] | None:
"""Return the list of available fan modes.
Requires SUPPORT_FAN_MODE.
@ -380,7 +382,7 @@ class ClimateEntity(Entity):
raise NotImplementedError
@property
def swing_mode(self) -> Optional[str]:
def swing_mode(self) -> str | None:
"""Return the swing setting.
Requires SUPPORT_SWING_MODE.
@ -388,7 +390,7 @@ class ClimateEntity(Entity):
raise NotImplementedError
@property
def swing_modes(self) -> Optional[List[str]]:
def swing_modes(self) -> list[str] | None:
"""Return the list of available swing modes.
Requires SUPPORT_SWING_MODE.

View file

@ -1,5 +1,5 @@
"""Provides device automations for Climate."""
from typing import List, Optional
from __future__ import annotations
import voluptuous as vol
@ -38,7 +38,7 @@ SET_PRESET_MODE_SCHEMA = cv.DEVICE_ACTION_BASE_SCHEMA.extend(
ACTION_SCHEMA = vol.Any(SET_HVAC_MODE_SCHEMA, SET_PRESET_MODE_SCHEMA)
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 Climate devices."""
registry = await entity_registry.async_get_registry(hass)
actions = []
@ -76,7 +76,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)

View file

@ -1,5 +1,5 @@
"""Provide the device automations for Climate."""
from typing import Dict, List
from __future__ import annotations
import voluptuous as vol
@ -42,7 +42,7 @@ CONDITION_SCHEMA = vol.Any(HVAC_MODE_CONDITION, PRESET_MODE_CONDITION)
async def async_get_conditions(
hass: HomeAssistant, device_id: str
) -> List[Dict[str, str]]:
) -> list[dict[str, str]]:
"""List device conditions for Climate devices."""
registry = await entity_registry.async_get_registry(hass)
conditions = []

View file

@ -1,5 +1,5 @@
"""Provides device automations for Climate."""
from typing import List
from __future__ import annotations
import voluptuous as vol
@ -58,7 +58,7 @@ CURRENT_TRIGGER_SCHEMA = vol.All(
TRIGGER_SCHEMA = vol.Any(HVAC_MODE_TRIGGER_SCHEMA, CURRENT_TRIGGER_SCHEMA)
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 Climate devices."""
registry = await entity_registry.async_get_registry(hass)
triggers = []

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 ATTR_TEMPERATURE
from homeassistant.core import Context, State
@ -29,8 +31,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."""
@ -76,8 +78,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(

View file

@ -1,8 +1,10 @@
"""Interface implementation for cloud client."""
from __future__ import annotations
import asyncio
import logging
from pathlib import Path
from typing import Any, Dict
from typing import Any
import aiohttp
from hass_nabucasa.client import CloudClient as Interface
@ -32,8 +34,8 @@ class CloudClient(Interface):
hass: HomeAssistantType,
prefs: CloudPreferences,
websession: aiohttp.ClientSession,
alexa_user_config: Dict[str, Any],
google_user_config: Dict[str, Any],
alexa_user_config: dict[str, Any],
google_user_config: dict[str, Any],
):
"""Initialize client interface to Cloud."""
self._hass = hass
@ -70,7 +72,7 @@ class CloudClient(Interface):
return self._hass.http.runner
@property
def cloudhooks(self) -> Dict[str, Dict[str, str]]:
def cloudhooks(self) -> dict[str, dict[str, str]]:
"""Return list of cloudhooks."""
return self._prefs.cloudhooks
@ -164,7 +166,7 @@ class CloudClient(Interface):
if identifier.startswith("remote_"):
async_dispatcher_send(self._hass, DISPATCHER_REMOTE_UPDATE, data)
async def async_alexa_message(self, payload: Dict[Any, Any]) -> Dict[Any, Any]:
async def async_alexa_message(self, payload: dict[Any, Any]) -> dict[Any, Any]:
"""Process cloud alexa message to client."""
cloud_user = await self._prefs.get_cloud_user()
aconfig = await self.get_alexa_config()
@ -176,7 +178,7 @@ class CloudClient(Interface):
enabled=self._prefs.alexa_enabled,
)
async def async_google_message(self, payload: Dict[Any, Any]) -> Dict[Any, Any]:
async def async_google_message(self, payload: dict[Any, Any]) -> dict[Any, Any]:
"""Process cloud google message to client."""
if not self._prefs.google_enabled:
return ga.turned_off_response(payload)
@ -187,7 +189,7 @@ class CloudClient(Interface):
self._hass, gconf, gconf.cloud_user, payload, gc.SOURCE_CLOUD
)
async def async_webhook_message(self, payload: Dict[Any, Any]) -> Dict[Any, Any]:
async def async_webhook_message(self, payload: dict[Any, Any]) -> dict[Any, Any]:
"""Process cloud webhook message to client."""
cloudhook_id = payload["cloudhook_id"]
@ -221,6 +223,6 @@ class CloudClient(Interface):
"headers": {"Content-Type": response.content_type},
}
async def async_cloudhooks_update(self, data: Dict[str, Dict[str, str]]) -> None:
async def async_cloudhooks_update(self, data: dict[str, dict[str, str]]) -> None:
"""Update local list of cloudhooks."""
await self._prefs.async_update(cloudhooks=data)

View file

@ -1,6 +1,7 @@
"""Preference management for cloud."""
from __future__ import annotations
from ipaddress import ip_address
from typing import List, Optional
from homeassistant.auth.const import GROUP_ID_ADMIN
from homeassistant.auth.models import User
@ -234,7 +235,7 @@ class CloudPreferences:
return self._prefs.get(PREF_ALEXA_REPORT_STATE, DEFAULT_ALEXA_REPORT_STATE)
@property
def alexa_default_expose(self) -> Optional[List[str]]:
def alexa_default_expose(self) -> list[str] | None:
"""Return array of entity domains that are exposed by default to Alexa.
Can return None, in which case for backwards should be interpreted as allow all domains.
@ -272,7 +273,7 @@ class CloudPreferences:
return self._prefs[PREF_GOOGLE_LOCAL_WEBHOOK_ID]
@property
def google_default_expose(self) -> Optional[List[str]]:
def google_default_expose(self) -> list[str] | None:
"""Return array of entity domains that are exposed by default to Google.
Can return None, in which case for backwards should be interpreted as allow all domains.
@ -302,7 +303,7 @@ class CloudPreferences:
await self.async_update(cloud_user=user.id)
return user.id
async def _load_cloud_user(self) -> Optional[User]:
async def _load_cloud_user(self) -> User | None:
"""Load cloud user if available."""
user_id = self._prefs.get(PREF_CLOUD_USER)

View file

@ -1,5 +1,5 @@
"""Support for the cloud for speech to text service."""
from typing import List
from __future__ import annotations
from aiohttp import StreamReader
from hass_nabucasa import Cloud
@ -56,32 +56,32 @@ class CloudProvider(Provider):
self.cloud = cloud
@property
def supported_languages(self) -> List[str]:
def supported_languages(self) -> list[str]:
"""Return a list of supported languages."""
return SUPPORT_LANGUAGES
@property
def supported_formats(self) -> List[AudioFormats]:
def supported_formats(self) -> list[AudioFormats]:
"""Return a list of supported formats."""
return [AudioFormats.WAV, AudioFormats.OGG]
@property
def supported_codecs(self) -> List[AudioCodecs]:
def supported_codecs(self) -> list[AudioCodecs]:
"""Return a list of supported codecs."""
return [AudioCodecs.PCM, AudioCodecs.OPUS]
@property
def supported_bit_rates(self) -> List[AudioBitRates]:
def supported_bit_rates(self) -> list[AudioBitRates]:
"""Return a list of supported bitrates."""
return [AudioBitRates.BITRATE_16]
@property
def supported_sample_rates(self) -> List[AudioSampleRates]:
def supported_sample_rates(self) -> list[AudioSampleRates]:
"""Return a list of supported samplerates."""
return [AudioSampleRates.SAMPLERATE_16000]
@property
def supported_channels(self) -> List[AudioChannels]:
def supported_channels(self) -> list[AudioChannels]:
"""Return a list of supported channels."""
return [AudioChannels.CHANNEL_MONO]

View file

@ -1,10 +1,12 @@
"""Helper functions for cloud components."""
from typing import Any, Dict
from __future__ import annotations
from typing import Any
from aiohttp import payload, web
def aiohttp_serialize_response(response: web.Response) -> Dict[str, Any]:
def aiohttp_serialize_response(response: web.Response) -> dict[str, Any]:
"""Serialize an aiohttp response to a dictionary."""
body = response.body

View file

@ -1,7 +1,8 @@
"""Update the IP addresses of your Cloudflare DNS records."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Dict
from pycfdns import CloudflareUpdater
from pycfdns.exceptions import (
@ -51,7 +52,7 @@ CONFIG_SCHEMA = vol.Schema(
)
async def async_setup(hass: HomeAssistant, config: Dict) -> bool:
async def async_setup(hass: HomeAssistant, config: dict) -> bool:
"""Set up the component."""
hass.data.setdefault(DOMAIN, {})

View file

@ -1,6 +1,7 @@
"""Config flow for Cloudflare integration."""
from __future__ import annotations
import logging
from typing import Dict, List, Optional
from pycfdns import CloudflareUpdater
from pycfdns.exceptions import (
@ -30,7 +31,7 @@ DATA_SCHEMA = vol.Schema(
)
def _zone_schema(zones: Optional[List] = None):
def _zone_schema(zones: list | None = None):
"""Zone selection schema."""
zones_list = []
@ -40,7 +41,7 @@ def _zone_schema(zones: Optional[List] = None):
return vol.Schema({vol.Required(CONF_ZONE): vol.In(zones_list)})
def _records_schema(records: Optional[List] = None):
def _records_schema(records: list | None = None):
"""Zone records selection schema."""
records_dict = {}
@ -50,7 +51,7 @@ def _records_schema(records: Optional[List] = None):
return vol.Schema({vol.Required(CONF_RECORDS): cv.multi_select(records_dict)})
async def validate_input(hass: HomeAssistant, data: Dict):
async def validate_input(hass: HomeAssistant, data: dict):
"""Validate the user input allows us to connect.
Data has the keys from DATA_SCHEMA with values provided by the user.
@ -92,7 +93,7 @@ class CloudflareConfigFlow(ConfigFlow, domain=DOMAIN):
self.zones = None
self.records = None
async def async_step_user(self, user_input: Optional[Dict] = None):
async def async_step_user(self, user_input: dict | None = None):
"""Handle a flow initiated by the user."""
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
@ -113,7 +114,7 @@ class CloudflareConfigFlow(ConfigFlow, domain=DOMAIN):
step_id="user", data_schema=DATA_SCHEMA, errors=errors
)
async def async_step_zone(self, user_input: Optional[Dict] = None):
async def async_step_zone(self, user_input: dict | None = None):
"""Handle the picking the zone."""
errors = {}
@ -133,7 +134,7 @@ class CloudflareConfigFlow(ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def async_step_records(self, user_input: Optional[Dict] = None):
async def async_step_records(self, user_input: dict | None = None):
"""Handle the picking the zone records."""
errors = {}

View file

@ -1,7 +1,8 @@
"""Platform to control a Zehnder ComfoAir Q350/450/600 ventilation unit."""
from __future__ import annotations
import logging
import math
from typing import Optional
from pycomfoconnect import (
CMD_FAN_MODE_AWAY,
@ -96,7 +97,7 @@ class ComfoConnectFan(FanEntity):
return SUPPORT_SET_SPEED
@property
def percentage(self) -> Optional[int]:
def percentage(self) -> int | None:
"""Return the current speed percentage."""
speed = self._ccb.data.get(SENSOR_FAN_SPEED_MODE)
if speed is None:

View file

@ -1,6 +1,7 @@
"""Agent foundation for conversation integration."""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Optional
from homeassistant.core import Context
from homeassistant.helpers import intent
@ -24,6 +25,6 @@ class AbstractConversationAgent(ABC):
@abstractmethod
async def async_process(
self, text: str, context: Context, conversation_id: Optional[str] = None
self, text: str, context: Context, conversation_id: str | None = None
) -> intent.IntentResponse:
"""Process a sentence."""

View file

@ -1,6 +1,7 @@
"""Standard conversastion implementation for Home Assistant."""
from __future__ import annotations
import re
from typing import Optional
from homeassistant import core, setup
from homeassistant.components.cover.intent import INTENT_CLOSE_COVER, INTENT_OPEN_COVER
@ -112,7 +113,7 @@ class DefaultAgent(AbstractConversationAgent):
async_register(self.hass, intent_type, sentences)
async def async_process(
self, text: str, context: core.Context, conversation_id: Optional[str] = None
self, text: str, context: core.Context, conversation_id: str | None = None
) -> intent.IntentResponse:
"""Process a sentence."""
intents = self.hass.data[DOMAIN]

View file

@ -2,7 +2,6 @@
from __future__ import annotations
import logging
from typing import Dict, Optional
import voluptuous as vol
@ -156,16 +155,16 @@ class CounterStorageCollection(collection.StorageCollection):
CREATE_SCHEMA = vol.Schema(CREATE_FIELDS)
UPDATE_SCHEMA = vol.Schema(UPDATE_FIELDS)
async def _process_create_data(self, data: Dict) -> Dict:
async def _process_create_data(self, data: dict) -> dict:
"""Validate the config is valid."""
return self.CREATE_SCHEMA(data)
@callback
def _get_suggested_id(self, info: Dict) -> str:
def _get_suggested_id(self, info: dict) -> str:
"""Suggest an ID based on the config."""
return info[CONF_NAME]
async def _update_data(self, data: dict, update_data: Dict) -> Dict:
async def _update_data(self, data: dict, update_data: dict) -> dict:
"""Return a new updated data object."""
update_data = self.UPDATE_SCHEMA(update_data)
return {**data, **update_data}
@ -174,14 +173,14 @@ class CounterStorageCollection(collection.StorageCollection):
class Counter(RestoreEntity):
"""Representation of a counter."""
def __init__(self, config: Dict):
def __init__(self, config: dict):
"""Initialize a counter."""
self._config: Dict = config
self._state: Optional[int] = config[CONF_INITIAL]
self._config: dict = config
self._state: int | None = config[CONF_INITIAL]
self.editable: bool = True
@classmethod
def from_yaml(cls, config: Dict) -> Counter:
def from_yaml(cls, config: dict) -> Counter:
"""Create counter instance from yaml config."""
counter = cls(config)
counter.editable = False
@ -194,22 +193,22 @@ class Counter(RestoreEntity):
return False
@property
def name(self) -> Optional[str]:
def name(self) -> str | None:
"""Return name of the counter."""
return self._config.get(CONF_NAME)
@property
def icon(self) -> Optional[str]:
def icon(self) -> str | None:
"""Return the icon to be used for this entity."""
return self._config.get(CONF_ICON)
@property
def state(self) -> Optional[int]:
def state(self) -> int | None:
"""Return the current value of the counter."""
return self._state
@property
def state_attributes(self) -> Dict:
def state_attributes(self) -> dict:
"""Return the state attributes."""
ret = {
ATTR_EDITABLE: self.editable,
@ -223,7 +222,7 @@ class Counter(RestoreEntity):
return ret
@property
def unique_id(self) -> Optional[str]:
def unique_id(self) -> str | None:
"""Return unique id of the entity."""
return self._config[CONF_ID]
@ -276,7 +275,7 @@ class Counter(RestoreEntity):
self._state = self.compute_next_state(new_state)
self.async_write_ha_state()
async def async_update_config(self, config: Dict) -> None:
async def async_update_config(self, config: dict) -> None:
"""Change the counter's settings WS CRUD."""
self._config = config
self._state = self.compute_next_state(self._state)

View file

@ -1,7 +1,9 @@
"""Reproduce an Counter 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
from homeassistant.core import Context, State
@ -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)
@ -70,8 +72,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 Counter states."""
await asyncio.gather(

View file

@ -1,5 +1,5 @@
"""Provides device automations for Cover."""
from typing import List, Optional
from __future__ import annotations
import voluptuous as vol
@ -58,7 +58,7 @@ POSITION_ACTION_SCHEMA = cv.DEVICE_ACTION_BASE_SCHEMA.extend(
ACTION_SCHEMA = vol.Any(CMD_ACTION_SCHEMA, POSITION_ACTION_SCHEMA)
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 Cover devices."""
registry = await entity_registry.async_get_registry(hass)
actions = []
@ -162,7 +162,7 @@ async def async_get_action_capabilities(hass: HomeAssistant, config: dict) -> di
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)

View file

@ -1,5 +1,7 @@
"""Provides device automations for Cover."""
from typing import Any, Dict, List
from __future__ import annotations
from typing import Any
import voluptuous as vol
@ -65,10 +67,10 @@ STATE_CONDITION_SCHEMA = DEVICE_CONDITION_BASE_SCHEMA.extend(
CONDITION_SCHEMA = vol.Any(POSITION_CONDITION_SCHEMA, STATE_CONDITION_SCHEMA)
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 Cover devices."""
registry = await entity_registry.async_get_registry(hass)
conditions: List[Dict[str, Any]] = []
conditions: list[dict[str, Any]] = []
# Get all the integrations entities for this device
for entry in entity_registry.async_entries_for_device(registry, device_id):

View file

@ -1,5 +1,5 @@
"""Provides device automations for Cover."""
from typing import List
from __future__ import annotations
import voluptuous as vol
@ -65,7 +65,7 @@ STATE_TRIGGER_SCHEMA = TRIGGER_BASE_SCHEMA.extend(
TRIGGER_SCHEMA = vol.Any(POSITION_TRIGGER_SCHEMA, STATE_TRIGGER_SCHEMA)
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 Cover devices."""
registry = await entity_registry.async_get_registry(hass)
triggers = []

View file

@ -1,7 +1,9 @@
"""Reproduce an Cover state."""
from __future__ import annotations
import asyncio
import logging
from typing import Any, Dict, Iterable, Optional
from typing import Any, Iterable
from homeassistant.components.cover import (
ATTR_CURRENT_POSITION,
@ -36,8 +38,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)
@ -117,8 +119,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 Cover states."""
# Reproduce states in parallel.

View file

@ -1,8 +1,9 @@
"""The Remote Python Debugger integration."""
from __future__ import annotations
from asyncio import Event
import logging
from threading import Thread
from typing import Optional
import debugpy
import voluptuous as vol
@ -40,7 +41,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
conf = config[DOMAIN]
async def debug_start(
call: Optional[ServiceCall] = None, *, wait: bool = True
call: ServiceCall | None = None, *, wait: bool = True
) -> None:
"""Start the debugger."""
debugpy.listen((conf[CONF_HOST], conf[CONF_PORT]))

View file

@ -1,5 +1,5 @@
"""Support for deCONZ climate devices."""
from typing import Optional
from __future__ import annotations
from pydeconz.sensor import Thermostat
@ -195,7 +195,7 @@ class DeconzThermostat(DeconzDevice, ClimateEntity):
# Preset control
@property
def preset_mode(self) -> Optional[str]:
def preset_mode(self) -> str | None:
"""Return preset mode."""
return DECONZ_TO_PRESET_MODE.get(self._device.preset)

View file

@ -1,6 +1,7 @@
"""Describe deCONZ logbook events."""
from __future__ import annotations
from typing import Callable, Optional
from typing import Callable
from homeassistant.const import ATTR_DEVICE_ID, CONF_EVENT
from homeassistant.core import HomeAssistant, callback
@ -125,7 +126,7 @@ def async_describe_events(
@callback
def async_describe_deconz_event(event: Event) -> dict:
"""Describe deCONZ logbook event."""
deconz_event: Optional[DeconzEvent] = _get_deconz_event_from_device_id(
deconz_event: DeconzEvent | None = _get_deconz_event_from_device_id(
hass, event.data[ATTR_DEVICE_ID]
)

View file

@ -1,5 +1,5 @@
"""Demo fan platform that has a fake fan."""
from typing import List, Optional
from __future__ import annotations
from homeassistant.components.fan import (
SPEED_HIGH,
@ -110,8 +110,8 @@ class BaseDemoFan(FanEntity):
unique_id: str,
name: str,
supported_features: int,
preset_modes: Optional[List[str]],
speed_list: Optional[List[str]],
preset_modes: list[str] | None,
speed_list: list[str] | None,
) -> None:
"""Initialize the entity."""
self.hass = hass
@ -211,7 +211,7 @@ class DemoPercentageFan(BaseDemoFan, FanEntity):
"""A demonstration fan component that uses percentages."""
@property
def percentage(self) -> Optional[int]:
def percentage(self) -> int | None:
"""Return the current speed."""
return self._percentage
@ -227,12 +227,12 @@ class DemoPercentageFan(BaseDemoFan, FanEntity):
self.schedule_update_ha_state()
@property
def preset_mode(self) -> Optional[str]:
def preset_mode(self) -> str | None:
"""Return the current preset mode, e.g., auto, smart, interval, favorite."""
return self._preset_mode
@property
def preset_modes(self) -> Optional[List[str]]:
def preset_modes(self) -> list[str] | None:
"""Return a list of available preset modes."""
return self._preset_modes
@ -271,7 +271,7 @@ class AsyncDemoPercentageFan(BaseDemoFan, FanEntity):
"""An async demonstration fan component that uses percentages."""
@property
def percentage(self) -> Optional[int]:
def percentage(self) -> int | None:
"""Return the current speed."""
return self._percentage
@ -287,12 +287,12 @@ class AsyncDemoPercentageFan(BaseDemoFan, FanEntity):
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., auto, smart, interval, favorite."""
return self._preset_mode
@property
def preset_modes(self) -> Optional[List[str]]:
def preset_modes(self) -> list[str] | None:
"""Return a list of available preset modes."""
return self._preset_modes

View file

@ -1,9 +1,10 @@
"""Demo platform for the geolocation component."""
from __future__ import annotations
from datetime import timedelta
import logging
from math import cos, pi, radians, sin
import random
from typing import Optional
from homeassistant.components.geo_location import GeolocationEvent
from homeassistant.const import LENGTH_KILOMETERS
@ -117,7 +118,7 @@ class DemoGeolocationEvent(GeolocationEvent):
return SOURCE
@property
def name(self) -> Optional[str]:
def name(self) -> str | None:
"""Return the name of the event."""
return self._name
@ -127,17 +128,17 @@ class DemoGeolocationEvent(GeolocationEvent):
return False
@property
def distance(self) -> Optional[float]:
def distance(self) -> float | None:
"""Return distance value of this external event."""
return self._distance
@property
def latitude(self) -> Optional[float]:
def latitude(self) -> float | None:
"""Return latitude value of this external event."""
return self._latitude
@property
def longitude(self) -> Optional[float]:
def longitude(self) -> float | None:
"""Return longitude value of this external event."""
return self._longitude

View file

@ -1,5 +1,5 @@
"""Support for the demo for speech to text service."""
from typing import List
from __future__ import annotations
from aiohttp import StreamReader
@ -25,32 +25,32 @@ class DemoProvider(Provider):
"""Demo speech API provider."""
@property
def supported_languages(self) -> List[str]:
def supported_languages(self) -> list[str]:
"""Return a list of supported languages."""
return SUPPORT_LANGUAGES
@property
def supported_formats(self) -> List[AudioFormats]:
def supported_formats(self) -> list[AudioFormats]:
"""Return a list of supported formats."""
return [AudioFormats.WAV]
@property
def supported_codecs(self) -> List[AudioCodecs]:
def supported_codecs(self) -> list[AudioCodecs]:
"""Return a list of supported codecs."""
return [AudioCodecs.PCM]
@property
def supported_bit_rates(self) -> List[AudioBitRates]:
def supported_bit_rates(self) -> list[AudioBitRates]:
"""Return a list of supported bit rates."""
return [AudioBitRates.BITRATE_16]
@property
def supported_sample_rates(self) -> List[AudioSampleRates]:
def supported_sample_rates(self) -> list[AudioSampleRates]:
"""Return a list of supported sample rates."""
return [AudioSampleRates.SAMPLERATE_16000, AudioSampleRates.SAMPLERATE_44100]
@property
def supported_channels(self) -> List[AudioChannels]:
def supported_channels(self) -> list[AudioChannels]:
"""Return a list of supported channels."""
return [AudioChannels.CHANNEL_STEREO]

View file

@ -1,8 +1,10 @@
"""Helpers for device automations."""
from __future__ import annotations
import asyncio
from functools import wraps
from types import ModuleType
from typing import Any, List, MutableMapping
from typing import Any, MutableMapping
import voluptuous as vol
import voluptuous_serialize
@ -116,7 +118,7 @@ async def _async_get_device_automations(hass, automation_type, device_id):
)
domains = set()
automations: List[MutableMapping[str, Any]] = []
automations: list[MutableMapping[str, Any]] = []
device = device_registry.async_get(device_id)
if device is None:

View file

@ -1,5 +1,7 @@
"""Device automation helpers for toggle entity."""
from typing import Any, Dict, List
from __future__ import annotations
from typing import Any
import voluptuous as vol
@ -170,10 +172,10 @@ async def async_attach_trigger(
async def _async_get_automations(
hass: HomeAssistant, device_id: str, automation_templates: List[dict], domain: str
) -> List[dict]:
hass: HomeAssistant, device_id: str, automation_templates: list[dict], domain: str
) -> list[dict]:
"""List device automations."""
automations: List[Dict[str, Any]] = []
automations: list[dict[str, Any]] = []
entity_registry = await hass.helpers.entity_registry.async_get_registry()
entries = [
@ -198,21 +200,21 @@ async def _async_get_automations(
async def async_get_actions(
hass: HomeAssistant, device_id: str, domain: str
) -> List[dict]:
) -> list[dict]:
"""List device actions."""
return await _async_get_automations(hass, device_id, ENTITY_ACTIONS, domain)
async def async_get_conditions(
hass: HomeAssistant, device_id: str, domain: str
) -> List[Dict[str, str]]:
) -> list[dict[str, str]]:
"""List device conditions."""
return await _async_get_automations(hass, device_id, ENTITY_CONDITIONS, domain)
async def async_get_triggers(
hass: HomeAssistant, device_id: str, domain: str
) -> List[dict]:
) -> list[dict]:
"""List device triggers."""
return await _async_get_automations(hass, device_id, ENTITY_TRIGGERS, domain)

View file

@ -1,5 +1,5 @@
"""Code to set up a device tracker platform using a config entry."""
from typing import Optional
from __future__ import annotations
from homeassistant.components import zone
from homeassistant.const import (
@ -18,7 +18,7 @@ from .const import ATTR_HOST_NAME, ATTR_IP, ATTR_MAC, ATTR_SOURCE_TYPE, DOMAIN,
async def async_setup_entry(hass, entry):
"""Set up an entry."""
component: Optional[EntityComponent] = hass.data.get(DOMAIN)
component: EntityComponent | None = hass.data.get(DOMAIN)
if component is None:
component = hass.data[DOMAIN] = EntityComponent(LOGGER, DOMAIN, hass)

View file

@ -1,5 +1,5 @@
"""Provides device automations for Device tracker."""
from typing import Dict, List
from __future__ import annotations
import voluptuous as vol
@ -32,7 +32,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 Device tracker devices."""
registry = await entity_registry.async_get_registry(hass)
conditions = []

View file

@ -1,5 +1,5 @@
"""Provides device automations for Device Tracker."""
from typing import List
from __future__ import annotations
import voluptuous as vol
@ -32,7 +32,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 Device Tracker devices."""
registry = await entity_registry.async_get_registry(hass)
triggers = []

View file

@ -1,9 +1,11 @@
"""Legacy device tracker classes."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import hashlib
from types import ModuleType
from typing import Any, Callable, Dict, List, Optional, Sequence
from typing import Any, Callable, Sequence
import attr
import voluptuous as vol
@ -205,7 +207,7 @@ class DeviceTrackerPlatform:
name: str = attr.ib()
platform: ModuleType = attr.ib()
config: Dict = attr.ib()
config: dict = attr.ib()
@property
def type(self):
@ -285,7 +287,7 @@ async def async_extract_config(hass, config):
async def async_create_platform_type(
hass, config, p_type, p_config
) -> Optional[DeviceTrackerPlatform]:
) -> DeviceTrackerPlatform | None:
"""Determine type of platform."""
platform = await async_prepare_setup_platform(hass, config, DOMAIN, p_type)
@ -786,7 +788,7 @@ class DeviceScanner:
hass: HomeAssistantType = None
def scan_devices(self) -> List[str]:
def scan_devices(self) -> list[str]:
"""Scan for devices."""
raise NotImplementedError()

View file

@ -1,5 +1,5 @@
"""Platform for climate integration."""
from typing import List, Optional
from __future__ import annotations
from homeassistant.components.climate import (
ATTR_TEMPERATURE,
@ -45,7 +45,7 @@ class DevoloClimateDeviceEntity(DevoloMultiLevelSwitchDeviceEntity, ClimateEntit
"""Representation of a climate/thermostat device within devolo Home Control."""
@property
def current_temperature(self) -> Optional[float]:
def current_temperature(self) -> float | None:
"""Return the current temperature."""
if hasattr(self._device_instance, "multi_level_sensor_property"):
return next(
@ -60,7 +60,7 @@ class DevoloClimateDeviceEntity(DevoloMultiLevelSwitchDeviceEntity, ClimateEntit
return None
@property
def target_temperature(self) -> Optional[float]:
def target_temperature(self) -> float | None:
"""Return the target temperature."""
return self._value
@ -75,7 +75,7 @@ class DevoloClimateDeviceEntity(DevoloMultiLevelSwitchDeviceEntity, ClimateEntit
return HVAC_MODE_HEAT
@property
def hvac_modes(self) -> List[str]:
def hvac_modes(self) -> list[str]:
"""Return the list of available hvac operation modes."""
return [HVAC_MODE_HEAT]

View file

@ -1,7 +1,9 @@
"""The DirecTV integration."""
from __future__ import annotations
import asyncio
from datetime import timedelta
from typing import Any, Dict
from typing import Any
from directv import DIRECTV, DIRECTVError
@ -28,7 +30,7 @@ PLATFORMS = ["media_player", "remote"]
SCAN_INTERVAL = timedelta(seconds=30)
async def async_setup(hass: HomeAssistant, config: Dict) -> bool:
async def async_setup(hass: HomeAssistant, config: dict) -> bool:
"""Set up the DirecTV component."""
hass.data.setdefault(DOMAIN, {})
return True
@ -87,7 +89,7 @@ class DIRECTVEntity(Entity):
return self._name
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return device information about this DirecTV receiver."""
return {
ATTR_IDENTIFIERS: {(DOMAIN, self._device_id)},

View file

@ -1,6 +1,8 @@
"""Config flow for DirecTV."""
from __future__ import annotations
import logging
from typing import Any, Dict, Optional
from typing import Any
from urllib.parse import urlparse
from directv import DIRECTV, DIRECTVError
@ -25,7 +27,7 @@ ERROR_CANNOT_CONNECT = "cannot_connect"
ERROR_UNKNOWN = "unknown"
async def validate_input(hass: HomeAssistantType, data: dict) -> Dict[str, Any]:
async def validate_input(hass: HomeAssistantType, data: dict) -> dict[str, Any]:
"""Validate the user input allows us to connect.
Data has the keys from DATA_SCHEMA with values provided by the user.
@ -48,8 +50,8 @@ class DirecTVConfigFlow(ConfigFlow, domain=DOMAIN):
self.discovery_info = {}
async def async_step_user(
self, user_input: Optional[ConfigType] = None
) -> Dict[str, Any]:
self, user_input: ConfigType | None = None
) -> dict[str, Any]:
"""Handle a flow initiated by the user."""
if user_input is None:
return self._show_setup_form()
@ -71,7 +73,7 @@ class DirecTVConfigFlow(ConfigFlow, domain=DOMAIN):
async def async_step_ssdp(
self, discovery_info: DiscoveryInfoType
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Handle SSDP discovery."""
host = urlparse(discovery_info[ATTR_SSDP_LOCATION]).hostname
receiver_id = None
@ -104,7 +106,7 @@ class DirecTVConfigFlow(ConfigFlow, domain=DOMAIN):
async def async_step_ssdp_confirm(
self, user_input: ConfigType = None
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Handle a confirmation flow initiated by SSDP."""
if user_input is None:
return self.async_show_form(
@ -118,7 +120,7 @@ class DirecTVConfigFlow(ConfigFlow, domain=DOMAIN):
data=self.discovery_info,
)
def _show_setup_form(self, errors: Optional[Dict] = None) -> Dict[str, Any]:
def _show_setup_form(self, errors: dict | None = None) -> dict[str, Any]:
"""Show the setup form to the user."""
return self.async_show_form(
step_id="user",

View file

@ -1,6 +1,8 @@
"""Support for the DirecTV receivers."""
from __future__ import annotations
import logging
from typing import Callable, List, Optional
from typing import Callable
from directv import DIRECTV
@ -64,7 +66,7 @@ SUPPORT_DTV_CLIENT = (
async def async_setup_entry(
hass: HomeAssistantType,
entry: ConfigEntry,
async_add_entities: Callable[[List, bool], None],
async_add_entities: Callable[[list, bool], None],
) -> bool:
"""Set up the DirecTV config entry."""
dtv = hass.data[DOMAIN][entry.entry_id]
@ -141,7 +143,7 @@ class DIRECTVMediaPlayer(DIRECTVEntity, MediaPlayerEntity):
return self._name
@property
def device_class(self) -> Optional[str]:
def device_class(self) -> str | None:
"""Return the class of this device."""
return DEVICE_CLASS_RECEIVER

View file

@ -1,7 +1,9 @@
"""Support for the DIRECTV remote."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any, Callable, Iterable, List
from typing import Any, Callable, Iterable
from directv import DIRECTV, DIRECTVError
@ -20,7 +22,7 @@ SCAN_INTERVAL = timedelta(minutes=2)
async def async_setup_entry(
hass: HomeAssistantType,
entry: ConfigEntry,
async_add_entities: Callable[[List, bool], None],
async_add_entities: Callable[[list, bool], None],
) -> bool:
"""Load DirecTV remote based on a config entry."""
dtv = hass.data[DOMAIN][entry.entry_id]

View file

@ -1,9 +1,10 @@
"""Support for DLNA DMR (Device Media Renderer)."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import functools
import logging
from typing import Optional
import aiohttp
from async_upnp_client import UpnpFactory
@ -116,7 +117,7 @@ async def async_start_event_handler(
server_host: str,
server_port: int,
requester,
callback_url_override: Optional[str] = None,
callback_url_override: str | None = None,
):
"""Register notify view."""
hass_data = hass.data[DLNA_DMR_DATA]

View file

@ -1,8 +1,10 @@
"""Config flow for DSMR integration."""
from __future__ import annotations
import asyncio
from functools import partial
import logging
from typing import Any, Dict, Optional
from typing import Any
from async_timeout import timeout
from dsmr_parser import obis_references as obis_ref
@ -133,7 +135,7 @@ class DSMRFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
self,
port: str,
host: str = None,
updates: Optional[Dict[Any, Any]] = None,
updates: dict[Any, Any] | None = None,
reload_on_update: bool = True,
):
"""Test if host and port are already configured."""

View file

@ -1,10 +1,11 @@
"""Support for Dutch Smart Meter (also known as Smartmeter or P1 port)."""
from __future__ import annotations
import asyncio
from asyncio import CancelledError
from datetime import timedelta
from functools import partial
import logging
from typing import Dict
from dsmr_parser import obis_references as obis_ref
from dsmr_parser.clients.protocol import create_dsmr_reader, create_tcp_dsmr_reader
@ -363,7 +364,7 @@ class DSMREntity(Entity):
return self._unique_id
@property
def device_info(self) -> Dict[str, any]:
def device_info(self) -> dict[str, any]:
"""Return the device information."""
return {
"identifiers": {(DOMAIN, self._device_serial)},

View file

@ -1,7 +1,8 @@
"""Support for the Dynalite networks."""
from __future__ import annotations
import asyncio
from typing import Any, Dict, Union
from typing import Any
import voluptuous as vol
@ -54,7 +55,7 @@ from .const import (
)
def num_string(value: Union[int, str]) -> str:
def num_string(value: int | str) -> str:
"""Test if value is a string of digits, aka an integer."""
new_value = str(value)
if new_value.isdigit():
@ -105,7 +106,7 @@ TEMPLATE_DATA_SCHEMA = vol.Any(TEMPLATE_ROOM_SCHEMA, TEMPLATE_TIMECOVER_SCHEMA)
TEMPLATE_SCHEMA = vol.Schema({str: TEMPLATE_DATA_SCHEMA})
def validate_area(config: Dict[str, Any]) -> Dict[str, Any]:
def validate_area(config: dict[str, Any]) -> dict[str, Any]:
"""Validate that template parameters are only used if area is using the relevant template."""
conf_set = set()
for template in DEFAULT_TEMPLATES:
@ -178,7 +179,7 @@ CONFIG_SCHEMA = vol.Schema(
)
async def async_setup(hass: HomeAssistant, config: Dict[str, Any]) -> bool:
async def async_setup(hass: HomeAssistant, config: dict[str, Any]) -> bool:
"""Set up the Dynalite platform."""
conf = config.get(DOMAIN)
LOGGER.debug("Setting up dynalite component config = %s", conf)

View file

@ -1,6 +1,7 @@
"""Code to handle a Dynalite bridge."""
from __future__ import annotations
from typing import Any, Callable, Dict, List, Optional
from typing import Any, Callable
from dynalite_devices_lib.dynalite_devices import (
CONF_AREA as dyn_CONF_AREA,
@ -23,7 +24,7 @@ from .convert_config import convert_config
class DynaliteBridge:
"""Manages a single Dynalite bridge."""
def __init__(self, hass: HomeAssistant, config: Dict[str, Any]) -> None:
def __init__(self, hass: HomeAssistant, config: dict[str, Any]) -> None:
"""Initialize the system based on host parameter."""
self.hass = hass
self.area = {}
@ -44,12 +45,12 @@ class DynaliteBridge:
LOGGER.debug("Setting up bridge - host %s", self.host)
return await self.dynalite_devices.async_setup()
def reload_config(self, config: Dict[str, Any]) -> None:
def reload_config(self, config: dict[str, Any]) -> None:
"""Reconfigure a bridge when config changes."""
LOGGER.debug("Reloading bridge - host %s, config %s", self.host, config)
self.dynalite_devices.configure(convert_config(config))
def update_signal(self, device: Optional[DynaliteBaseDevice] = None) -> str:
def update_signal(self, device: DynaliteBaseDevice | None = None) -> str:
"""Create signal to use to trigger entity update."""
if device:
signal = f"dynalite-update-{self.host}-{device.unique_id}"
@ -58,7 +59,7 @@ class DynaliteBridge:
return signal
@callback
def update_device(self, device: Optional[DynaliteBaseDevice] = None) -> None:
def update_device(self, device: DynaliteBaseDevice | None = None) -> None:
"""Call when a device or all devices should be updated."""
if not device:
# This is used to signal connection or disconnection, so all devices may become available or not.
@ -98,7 +99,7 @@ class DynaliteBridge:
if platform in self.waiting_devices:
self.async_add_devices[platform](self.waiting_devices[platform])
def add_devices_when_registered(self, devices: List[DynaliteBaseDevice]) -> None:
def add_devices_when_registered(self, devices: list[DynaliteBaseDevice]) -> None:
"""Add the devices to HA if the add devices callback was registered, otherwise queue until it is."""
for platform in PLATFORMS:
platform_devices = [

View file

@ -1,5 +1,7 @@
"""Config flow to configure Dynalite hub."""
from typing import Any, Dict
from __future__ import annotations
from typing import Any
from homeassistant import config_entries
from homeassistant.const import CONF_HOST
@ -18,7 +20,7 @@ class DynaliteFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Initialize the Dynalite flow."""
self.host = None
async def async_step_import(self, import_info: Dict[str, Any]) -> Any:
async def async_step_import(self, import_info: dict[str, Any]) -> Any:
"""Import a new bridge as a config entry."""
LOGGER.debug("Starting async_step_import - %s", import_info)
host = import_info[CONF_HOST]

View file

@ -1,6 +1,7 @@
"""Convert the HA config to the dynalite config."""
from __future__ import annotations
from typing import Any, Dict
from typing import Any
from dynalite_devices_lib import const as dyn_const
@ -62,7 +63,7 @@ def convert_with_map(config, conf_map):
return result
def convert_channel(config: Dict[str, Any]) -> Dict[str, Any]:
def convert_channel(config: dict[str, Any]) -> dict[str, Any]:
"""Convert the config for a channel."""
my_map = {
CONF_NAME: dyn_const.CONF_NAME,
@ -72,7 +73,7 @@ def convert_channel(config: Dict[str, Any]) -> Dict[str, Any]:
return convert_with_map(config, my_map)
def convert_preset(config: Dict[str, Any]) -> Dict[str, Any]:
def convert_preset(config: dict[str, Any]) -> dict[str, Any]:
"""Convert the config for a preset."""
my_map = {
CONF_NAME: dyn_const.CONF_NAME,
@ -82,7 +83,7 @@ def convert_preset(config: Dict[str, Any]) -> Dict[str, Any]:
return convert_with_map(config, my_map)
def convert_area(config: Dict[str, Any]) -> Dict[str, Any]:
def convert_area(config: dict[str, Any]) -> dict[str, Any]:
"""Convert the config for an area."""
my_map = {
CONF_NAME: dyn_const.CONF_NAME,
@ -114,12 +115,12 @@ def convert_area(config: Dict[str, Any]) -> Dict[str, Any]:
return result
def convert_default(config: Dict[str, Any]) -> Dict[str, Any]:
def convert_default(config: dict[str, Any]) -> dict[str, Any]:
"""Convert the config for the platform defaults."""
return convert_with_map(config, {CONF_FADE: dyn_const.CONF_FADE})
def convert_template(config: Dict[str, Any]) -> Dict[str, Any]:
def convert_template(config: dict[str, Any]) -> dict[str, Any]:
"""Convert the config for a template."""
my_map = {
CONF_ROOM_ON: dyn_const.CONF_ROOM_ON,
@ -135,7 +136,7 @@ def convert_template(config: Dict[str, Any]) -> Dict[str, Any]:
return convert_with_map(config, my_map)
def convert_config(config: Dict[str, Any]) -> Dict[str, Any]:
def convert_config(config: dict[str, Any]) -> dict[str, Any]:
"""Convert a config dict by replacing component consts with library consts."""
my_map = {
CONF_NAME: dyn_const.CONF_NAME,

View file

@ -1,5 +1,7 @@
"""Support for the Dynalite devices as entities."""
from typing import Any, Callable, Dict
from __future__ import annotations
from typing import Any, Callable
from homeassistant.components.dynalite.bridge import DynaliteBridge
from homeassistant.config_entries import ConfigEntry
@ -58,7 +60,7 @@ class DynaliteBase(Entity):
return self._device.available
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Device info for this entity."""
return {
"identifiers": {(DOMAIN, self._device.unique_id)},

View file

@ -1,7 +1,8 @@
"""Support for Dyson Pure Cool link fan."""
from __future__ import annotations
import logging
import math
from typing import Optional
from libpurecool.const import FanMode, FanSpeed, NightMode, Oscillation
from libpurecool.dyson_pure_cool import DysonPureCool
@ -243,9 +244,9 @@ class DysonFanEntity(DysonEntity, FanEntity):
def 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."""
@ -325,9 +326,9 @@ class DysonPureCoolEntity(DysonFanEntity):
def 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."""