Update typing 11 (#48072)

This commit is contained in:
Marc Mueller 2021-03-18 13:21:46 +01:00 committed by GitHub
parent 99f9f8dec0
commit 3d2b81a401
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
45 changed files with 206 additions and 154 deletions

View file

@ -1,6 +1,7 @@
"""Config flow for Neato Botvac."""
from __future__ import annotations
import logging
from typing import Optional
import voluptuous as vol
@ -24,7 +25,7 @@ class OAuth2FlowHandler(
"""Return logger."""
return logging.getLogger(__name__)
async def async_step_user(self, user_input: Optional[dict] = None) -> dict:
async def async_step_user(self, user_input: dict | None = None) -> dict:
"""Create an entry for the flow."""
current_entries = self._async_current_entries()
if current_entries and CONF_TOKEN in current_entries[0].data:
@ -37,9 +38,7 @@ class OAuth2FlowHandler(
"""Perform reauth upon migration of old entries."""
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: Optional[dict] = None
) -> dict:
async def async_step_reauth_confirm(self, user_input: dict | None = None) -> dict:
"""Confirm reauth upon migration of old entries."""
if user_input is None:
return self.async_show_form(

View file

@ -1,8 +1,8 @@
"""Support for Google Nest SDM Cameras."""
from __future__ import annotations
import datetime
import logging
from typing import Optional
from google_nest_sdm.camera_traits import (
CameraEventImageTrait,
@ -74,7 +74,7 @@ class NestCamera(Camera):
return False
@property
def unique_id(self) -> Optional[str]:
def unique_id(self) -> str | None:
"""Return a unique ID."""
# The API "name" field is a unique device identifier.
return f"{self._device.name}-camera"

View file

@ -1,6 +1,5 @@
"""Support for Google Nest SDM climate devices."""
from typing import Optional
from __future__ import annotations
from google_nest_sdm.device import Device
from google_nest_sdm.device_traits import FanTrait, TemperatureTrait
@ -111,7 +110,7 @@ class ThermostatEntity(ClimateEntity):
return False
@property
def unique_id(self) -> Optional[str]:
def unique_id(self) -> str | None:
"""Return a unique ID."""
# The API "name" field is a unique device identifier.
return self._device.name

View file

@ -11,12 +11,12 @@ and everything else custom is for the old api. When configured with the
new api via NestFlowHandler.register_sdm_api, the custom methods just
invoke the AbstractOAuth2FlowHandler methods.
"""
from __future__ import annotations
import asyncio
from collections import OrderedDict
import logging
import os
from typing import Dict
import async_timeout
import voluptuous as vol
@ -98,7 +98,7 @@ class NestFlowHandler(
return logging.getLogger(__name__)
@property
def extra_authorize_data(self) -> Dict[str, str]:
def extra_authorize_data(self) -> dict[str, str]:
"""Extra data that needs to be appended to the authorize url."""
return {
"scope": " ".join(SDM_SCOPES),

View file

@ -1,5 +1,5 @@
"""Provides device automations for Nest."""
from typing import List
from __future__ import annotations
import voluptuous as vol
@ -39,7 +39,7 @@ async def async_get_nest_device_id(hass: HomeAssistant, device_id: str) -> str:
async def async_get_device_trigger_types(
hass: HomeAssistant, nest_device_id: str
) -> List[str]:
) -> list[str]:
"""List event triggers supported for a Nest device."""
# All devices should have already been loaded so any failures here are
# "shouldn't happen" cases
@ -58,7 +58,7 @@ async def async_get_device_trigger_types(
return trigger_types
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 a Nest device."""
nest_device_id = await async_get_nest_device_id(hass, device_id)
if not nest_device_id:

View file

@ -1,7 +1,7 @@
"""Support for Google Nest SDM sensors."""
from __future__ import annotations
import logging
from typing import Optional
from google_nest_sdm.device import Device
from google_nest_sdm.device_traits import HumidityTrait, TemperatureTrait
@ -67,7 +67,7 @@ class SensorBase(Entity):
return False
@property
def unique_id(self) -> Optional[str]:
def unique_id(self) -> str | None:
"""Return a unique ID."""
# The API "name" field is a unique device identifier.
return f"{self._device.name}-{self.device_class}"
@ -113,7 +113,7 @@ class HumiditySensor(SensorBase):
"""Representation of a Humidity Sensor."""
@property
def unique_id(self) -> Optional[str]:
def unique_id(self) -> str | None:
"""Return a unique ID."""
# The API returns the identifier under the name field.
return f"{self._device.name}-humidity"

View file

@ -1,6 +1,7 @@
"""Support for Netatmo Smart thermostats."""
from __future__ import annotations
import logging
from typing import List, Optional
import pyatmo
import voluptuous as vol
@ -320,7 +321,7 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
return self._target_temperature
@property
def target_temperature_step(self) -> Optional[float]:
def target_temperature_step(self) -> float | None:
"""Return the supported step of target temperature."""
return PRECISION_HALVES
@ -335,7 +336,7 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
return self._operation_list
@property
def hvac_action(self) -> Optional[str]:
def hvac_action(self) -> str | None:
"""Return the current running hvac operation if supported."""
if self._model == NA_THERM and self._boilerstatus is not None:
return CURRENT_HVAC_MAP_NETATMO[self._boilerstatus]
@ -400,12 +401,12 @@ class NetatmoThermostat(NetatmoBase, 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."""
return self._preset
@property
def preset_modes(self) -> Optional[List[str]]:
def preset_modes(self) -> list[str] | None:
"""Return a list of available preset modes."""
return SUPPORT_PRESET
@ -631,7 +632,7 @@ def interpolate(batterylevel: int, module_type: str) -> int:
return int(pct)
def get_all_home_ids(home_data: pyatmo.HomeData) -> List[str]:
def get_all_home_ids(home_data: pyatmo.HomeData) -> list[str]:
"""Get all the home ids returned by NetAtmo API."""
if home_data is None:
return []

View file

@ -1,11 +1,13 @@
"""The Netatmo data handler."""
from __future__ import annotations
from collections import deque
from datetime import timedelta
from functools import partial
from itertools import islice
import logging
from time import time
from typing import Deque, Dict, List
from typing import Deque
import pyatmo
@ -55,8 +57,8 @@ class NetatmoDataHandler:
"""Initialize self."""
self.hass = hass
self._auth = hass.data[DOMAIN][entry.entry_id][AUTH]
self.listeners: List[CALLBACK_TYPE] = []
self._data_classes: Dict = {}
self.listeners: list[CALLBACK_TYPE] = []
self._data_classes: dict = {}
self.data = {}
self._queue: Deque = deque()
self._webhook: bool = False

View file

@ -1,5 +1,5 @@
"""Provides device automations for Netatmo."""
from typing import List
from __future__ import annotations
import voluptuous as vol
@ -82,7 +82,7 @@ async def async_validate_trigger_config(hass, config):
return 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 Netatmo devices."""
registry = await entity_registry.async_get_registry(hass)
device_registry = await hass.helpers.device_registry.async_get_registry()

View file

@ -1,8 +1,9 @@
"""Netatmo Media Source Implementation."""
from __future__ import annotations
import datetime as dt
import logging
import re
from typing import Optional, Tuple
from homeassistant.components.media_player.const import (
MEDIA_CLASS_DIRECTORY,
@ -53,7 +54,7 @@ class NetatmoSource(MediaSource):
return PlayMedia(url, 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:
@ -156,7 +157,7 @@ def remove_html_tags(text):
@callback
def async_parse_identifier(
item: MediaSourceItem,
) -> Tuple[str, str, Optional[int]]:
) -> tuple[str, str, int | None]:
"""Parse identifier."""
if not item.identifier:
return "events", "", None

View file

@ -1,6 +1,7 @@
"""Base class for Netatmo entities."""
from __future__ import annotations
import logging
from typing import Dict, List
from homeassistant.core import CALLBACK_TYPE, callback
from homeassistant.helpers.entity import Entity
@ -17,8 +18,8 @@ class NetatmoBase(Entity):
def __init__(self, data_handler: NetatmoDataHandler) -> None:
"""Set up Netatmo entity base."""
self.data_handler = data_handler
self._data_classes: List[Dict] = []
self._listeners: List[CALLBACK_TYPE] = []
self._data_classes: list[dict] = []
self._listeners: list[CALLBACK_TYPE] = []
self._device_name = None
self._id = None

View file

@ -1,8 +1,10 @@
"""Support for Nightscout sensors."""
from __future__ import annotations
from asyncio import TimeoutError as AsyncIOTimeoutError
from datetime import timedelta
import logging
from typing import Callable, List
from typing import Callable
from aiohttp import ClientError
from py_nightscout import Api as NightscoutAPI
@ -24,7 +26,7 @@ DEFAULT_NAME = "Blood Glucose"
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: Callable[[List[Entity], bool], None],
async_add_entities: Callable[[list[Entity], bool], None],
) -> None:
"""Set up the Glucose Sensor."""
api = hass.data[DOMAIN][entry.entry_id]

View file

@ -1,8 +1,10 @@
"""Provides functionality to notify people."""
from __future__ import annotations
import asyncio
from functools import partial
import logging
from typing import Any, Dict, cast
from typing import Any, cast
import voluptuous as vol
@ -120,7 +122,7 @@ class BaseNotificationService:
hass: HomeAssistantType = None # type: ignore
# Name => target
registered_targets: Dict[str, str]
registered_targets: dict[str, str]
def send_message(self, message, **kwargs):
"""Send a message.

View file

@ -1,7 +1,8 @@
"""Sensor platform to display the current fuel prices at a NSW fuel station."""
from __future__ import annotations
import datetime
import logging
from typing import Optional
from nsw_fuel import FuelCheckClient, FuelCheckError
import voluptuous as vol
@ -159,7 +160,7 @@ class StationPriceSensor(Entity):
return f"{self._station_data.get_station_name()} {self._fuel_type}"
@property
def state(self) -> Optional[float]:
def state(self) -> float | None:
"""Return the state of the sensor."""
price_info = self._station_data.for_fuel_type(self._fuel_type)
if price_info:

View file

@ -1,7 +1,8 @@
"""Support for NSW Rural Fire Service Feeds."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Optional
from aio_geojson_nsw_rfs_incidents import NswRuralFireServiceIncidentsFeedManager
import voluptuous as vol
@ -259,22 +260,22 @@ class NswRuralFireServiceLocationEvent(GeolocationEvent):
return SOURCE
@property
def name(self) -> Optional[str]:
def name(self) -> str | None:
"""Return the name of the entity."""
return self._name
@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,8 +1,10 @@
"""Component to allow numeric input for platforms."""
from __future__ import annotations
from abc import abstractmethod
from datetime import timedelta
import logging
from typing import Any, Dict
from typing import Any
import voluptuous as vol
@ -66,7 +68,7 @@ class NumberEntity(Entity):
"""Representation of a Number entity."""
@property
def capability_attributes(self) -> Dict[str, Any]:
def capability_attributes(self) -> dict[str, Any]:
"""Return capability attributes."""
return {
ATTR_MIN: self.min_value,

View file

@ -1,5 +1,7 @@
"""Provides device actions for Number."""
from typing import Any, Dict, List, Optional
from __future__ import annotations
from typing import Any
import voluptuous as vol
@ -27,10 +29,10 @@ 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 Number."""
registry = await entity_registry.async_get_registry(hass)
actions: List[Dict[str, Any]] = []
actions: list[dict[str, Any]] = []
# Get all the integrations entities for this device
for entry in entity_registry.async_entries_for_device(registry, device_id):
@ -50,7 +52,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,7 +1,9 @@
"""Reproduce a Number entity 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
@ -16,8 +18,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)
@ -50,8 +52,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 multiple Number states."""
# Reproduce states in parallel.

View file

@ -1,8 +1,10 @@
"""The National Weather Service integration."""
from __future__ import annotations
import asyncio
import datetime
import logging
from typing import Awaitable, Callable, Optional
from typing import Awaitable, Callable
from pynws import SimpleNWS
@ -58,8 +60,8 @@ class NwsDataUpdateCoordinator(DataUpdateCoordinator):
name: str,
update_interval: datetime.timedelta,
failed_update_interval: datetime.timedelta,
update_method: Optional[Callable[[], Awaitable]] = None,
request_refresh_debouncer: Optional[debounce.Debouncer] = None,
update_method: Callable[[], Awaitable] | None = None,
request_refresh_debouncer: debounce.Debouncer | None = None,
):
"""Initialize NWS coordinator."""
super().__init__(

View file

@ -1,6 +1,8 @@
"""Config flow for NZBGet."""
from __future__ import annotations
import logging
from typing import Any, Dict, Optional
from typing import Any
import voluptuous as vol
@ -31,7 +33,7 @@ from .coordinator import NZBGetAPI, NZBGetAPIException
_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.
@ -63,8 +65,8 @@ class NZBGetConfigFlow(ConfigFlow, domain=DOMAIN):
return NZBGetOptionsFlowHandler(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."""
if CONF_SCAN_INTERVAL in user_input:
user_input[CONF_SCAN_INTERVAL] = user_input[CONF_SCAN_INTERVAL].seconds
@ -72,8 +74,8 @@ class NZBGetConfigFlow(ConfigFlow, domain=DOMAIN):
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")
@ -127,7 +129,7 @@ class NZBGetOptionsFlowHandler(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 NZBGet options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)

View file

@ -1,7 +1,9 @@
"""Monitor the NZBGet API."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Callable, List, Optional
from typing import Callable
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
@ -41,7 +43,7 @@ SENSOR_TYPES = {
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 NZBGet sensor based on a config entry."""
coordinator: NZBGetDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
@ -74,7 +76,7 @@ class NZBGetSensor(NZBGetEntity):
entry_name: str,
sensor_type: str,
sensor_name: str,
unit_of_measurement: Optional[str] = None,
unit_of_measurement: str | None = None,
):
"""Initialize a new NZBGet sensor."""
self._sensor_type = sensor_type

View file

@ -1,5 +1,7 @@
"""Support for NZBGet switches."""
from typing import Callable, List
from __future__ import annotations
from typing import Callable
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
@ -15,7 +17,7 @@ from .coordinator import NZBGetDataUpdateCoordinator
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 NZBGet sensor based on a config entry."""
coordinator: NZBGetDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][

View file

@ -1,6 +1,8 @@
"""Support for 1-Wire entities."""
from __future__ import annotations
import logging
from typing import Any, Dict, Optional
from typing import Any
from pyownet import protocol
@ -43,32 +45,32 @@ class OneWireBaseEntity(Entity):
self._unique_id = unique_id or device_file
@property
def name(self) -> Optional[str]:
def name(self) -> str | None:
"""Return the name of the entity."""
return self._name
@property
def device_class(self) -> Optional[str]:
def device_class(self) -> str | None:
"""Return the class of this device."""
return self._device_class
@property
def unit_of_measurement(self) -> Optional[str]:
def unit_of_measurement(self) -> str | None:
"""Return the unit the value is expressed in."""
return self._unit_of_measurement
@property
def extra_state_attributes(self) -> Optional[Dict[str, Any]]:
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return the state attributes of the entity."""
return {"device_file": self._device_file, "raw_value": self._value_raw}
@property
def unique_id(self) -> Optional[str]:
def unique_id(self) -> str | None:
"""Return a unique ID."""
return self._unique_id
@property
def device_info(self) -> Optional[Dict[str, Any]]:
def device_info(self) -> dict[str, Any] | None:
"""Return device specific attributes."""
return self._device_info
@ -85,9 +87,9 @@ class OneWireProxyEntity(OneWireBaseEntity):
self,
device_id: str,
device_name: str,
device_info: Dict[str, Any],
device_info: dict[str, Any],
entity_path: str,
entity_specs: Dict[str, Any],
entity_specs: dict[str, Any],
owproxy: protocol._Proxy,
):
"""Initialize the sensor."""

View file

@ -1,6 +1,7 @@
"""Support for Onkyo Receivers."""
from __future__ import annotations
import logging
from typing import List
import eiscp
from eiscp import eISCP
@ -56,7 +57,7 @@ SUPPORT_ONKYO_WO_VOLUME = (
| SUPPORT_PLAY_MEDIA
)
KNOWN_HOSTS: List[str] = []
KNOWN_HOSTS: list[str] = []
DEFAULT_SOURCES = {
"tv": "TV",
"bd": "Bluray",

View file

@ -1,5 +1,5 @@
"""Support for ONVIF binary sensors."""
from typing import Optional
from __future__ import annotations
from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.core import callback
@ -55,7 +55,7 @@ class ONVIFBinarySensor(ONVIFBaseEntity, BinarySensorEntity):
return self.device.events.get_uid(self.uid).name
@property
def device_class(self) -> Optional[str]:
def device_class(self) -> str | None:
"""Return the class of this device, from component DEVICE_CLASSES."""
return self.device.events.get_uid(self.uid).device_class

View file

@ -1,6 +1,7 @@
"""Config flow for ONVIF."""
from __future__ import annotations
from pprint import pformat
from typing import List
from urllib.parse import urlparse
from onvif.exceptions import ONVIFError
@ -36,7 +37,7 @@ from .device import get_device
CONF_MANUAL_INPUT = "Manually configure ONVIF device"
def wsdiscovery() -> List[Service]:
def wsdiscovery() -> list[Service]:
"""Get ONVIF Profile S devices from network."""
discovery = WSDiscovery(ttl=4)
discovery.start()

View file

@ -1,8 +1,9 @@
"""ONVIF device abstraction."""
from __future__ import annotations
import asyncio
import datetime as dt
import os
from typing import List
from httpx import RequestError
import onvif
@ -50,7 +51,7 @@ class ONVIFDevice:
self.info: DeviceInfo = DeviceInfo()
self.capabilities: Capabilities = Capabilities()
self.profiles: List[Profile] = []
self.profiles: list[Profile] = []
self.max_resolution: int = 0
self._dt_diff_seconds: int = 0
@ -262,7 +263,7 @@ class ONVIFDevice:
return Capabilities(snapshot, pullpoint, ptz)
async def async_get_profiles(self) -> List[Profile]:
async def async_get_profiles(self) -> list[Profile]:
"""Obtain media profiles for this device."""
media_service = self.device.create_media_service()
result = await media_service.GetProfiles()

View file

@ -1,7 +1,9 @@
"""ONVIF event abstraction."""
from __future__ import annotations
import asyncio
import datetime as dt
from typing import Callable, Dict, List, Optional, Set
from typing import Callable
from httpx import RemoteProtocolError, TransportError
from onvif import ONVIFCamera, ONVIFService
@ -34,14 +36,14 @@ class EventManager:
self.started: bool = False
self._subscription: ONVIFService = None
self._events: Dict[str, Event] = {}
self._listeners: List[CALLBACK_TYPE] = []
self._unsub_refresh: Optional[CALLBACK_TYPE] = None
self._events: dict[str, Event] = {}
self._listeners: list[CALLBACK_TYPE] = []
self._unsub_refresh: CALLBACK_TYPE | None = None
super().__init__()
@property
def platforms(self) -> Set[str]:
def platforms(self) -> set[str]:
"""Return platforms to setup."""
return {event.platform for event in self._events.values()}
@ -229,6 +231,6 @@ class EventManager:
"""Retrieve event for given id."""
return self._events[uid]
def get_platform(self, platform) -> List[Event]:
def get_platform(self, platform) -> list[Event]:
"""Retrieve events for given platform."""
return [event for event in self._events.values() if event.platform == platform]

View file

@ -1,6 +1,8 @@
"""ONVIF models."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, List
from typing import Any
@dataclass
@ -37,7 +39,7 @@ class PTZ:
continuous: bool
relative: bool
absolute: bool
presets: List[str] = None
presets: list[str] = None
@dataclass

View file

@ -1,5 +1,5 @@
"""Support for ONVIF binary sensors."""
from typing import Optional, Union
from __future__ import annotations
from homeassistant.core import callback
@ -43,7 +43,7 @@ class ONVIFSensor(ONVIFBaseEntity):
super().__init__(device)
@property
def state(self) -> Union[None, str, int, float]:
def state(self) -> None | str | int | float:
"""Return the state of the entity."""
return self.device.events.get_uid(self.uid).value
@ -53,12 +53,12 @@ class ONVIFSensor(ONVIFBaseEntity):
return self.device.events.get_uid(self.uid).name
@property
def device_class(self) -> Optional[str]:
def device_class(self) -> str | None:
"""Return the class of this device, from component DEVICE_CLASSES."""
return self.device.events.get_uid(self.uid).device_class
@property
def unit_of_measurement(self) -> Optional[str]:
def unit_of_measurement(self) -> str | None:
"""Return the unit of measurement of this entity, if any."""
return self.device.events.get_uid(self.uid).unit_of_measurement

View file

@ -1,7 +1,9 @@
"""Support for OVO Energy."""
from __future__ import annotations
from datetime import datetime, timedelta
import logging
from typing import Any, Dict
from typing import Any
import aiohttp
import async_timeout
@ -148,7 +150,7 @@ class OVOEnergyDeviceEntity(OVOEnergyEntity):
"""Defines a OVO Energy device entity."""
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return device information about this OVO Energy instance."""
return {
"identifiers": {(DOMAIN, self._client.account_id)},

View file

@ -1,7 +1,8 @@
"""Support for Z-Wave climate devices."""
from __future__ import annotations
from enum import IntEnum
import logging
from typing import Optional, Tuple
from homeassistant.components.climate import DOMAIN as CLIMATE_DOMAIN, ClimateEntity
from homeassistant.components.climate.const import (
@ -239,12 +240,12 @@ class ZWaveClimateEntity(ZWaveDeviceEntity, ClimateEntity):
return self._current_mode_setpoint_values[0].value
@property
def target_temperature_low(self) -> Optional[float]:
def target_temperature_low(self) -> float | None:
"""Return the lowbound target temperature we try to reach."""
return self._current_mode_setpoint_values[0].value
@property
def target_temperature_high(self) -> Optional[float]:
def target_temperature_high(self) -> float | None:
"""Return the highbound target temperature we try to reach."""
return self._current_mode_setpoint_values[1].value
@ -333,7 +334,7 @@ class ZWaveClimateEntity(ZWaveDeviceEntity, ClimateEntity):
support |= SUPPORT_PRESET_MODE
return support
def _get_current_mode_setpoint_values(self) -> Tuple:
def _get_current_mode_setpoint_values(self) -> tuple:
"""Return a tuple of current setpoint Z-Wave value(s)."""
if not self.values.mode:
setpoint_names = ("setpoint_heating",)

View file

@ -1,7 +1,9 @@
"""Support for displaying persistent notifications."""
from __future__ import annotations
from collections import OrderedDict
import logging
from typing import Any, Mapping, MutableMapping, Optional
from typing import Any, Mapping, MutableMapping
import voluptuous as vol
@ -71,8 +73,8 @@ def dismiss(hass, notification_id):
def async_create(
hass: HomeAssistant,
message: str,
title: Optional[str] = None,
notification_id: Optional[str] = None,
title: str | None = None,
notification_id: str | None = None,
) -> None:
"""Generate a notification."""
data = {

View file

@ -1,6 +1,8 @@
"""Support for tracking people."""
from __future__ import annotations
import logging
from typing import List, Optional, cast
from typing import cast
import voluptuous as vol
@ -171,7 +173,7 @@ class PersonStorageCollection(collection.StorageCollection):
super().__init__(store, logger, id_manager)
self.yaml_collection = yaml_collection
async def _async_load_data(self) -> Optional[dict]:
async def _async_load_data(self) -> dict | None:
"""Load the data.
A past bug caused onboarding to create invalid person objects.
@ -257,7 +259,7 @@ class PersonStorageCollection(collection.StorageCollection):
raise ValueError("User already taken")
async def filter_yaml_data(hass: HomeAssistantType, persons: List[dict]) -> List[dict]:
async def filter_yaml_data(hass: HomeAssistantType, persons: list[dict]) -> list[dict]:
"""Validate YAML data that we can't validate via schema."""
filtered = []
person_invalid_user = []
@ -380,7 +382,7 @@ class Person(RestoreEntity):
return self._config[CONF_NAME]
@property
def entity_picture(self) -> Optional[str]:
def entity_picture(self) -> str | None:
"""Return entity picture."""
return self._config.get(CONF_PICTURE)
@ -522,7 +524,7 @@ def ws_list_person(
)
def _get_latest(prev: Optional[State], curr: State):
def _get_latest(prev: State | None, curr: State):
"""Get latest state."""
if prev is None or curr.last_updated > prev.last_updated:
return curr

View file

@ -1,5 +1,7 @@
"""Helper to test significant Person 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 new_state != old_state:

View file

@ -1,8 +1,10 @@
"""The Philips TV integration."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
from typing import Any, Callable, Dict, Optional
from typing import Any, Callable
from haphilipsjs import ConnectionFailure, PhilipsTV
@ -77,14 +79,14 @@ class PluggableAction:
def __init__(self, update: Callable[[], None]):
"""Initialize."""
self._update = update
self._actions: Dict[Any, AutomationActionType] = {}
self._actions: dict[Any, AutomationActionType] = {}
def __bool__(self):
"""Return if we have something attached."""
return bool(self._actions)
@callback
def async_attach(self, action: AutomationActionType, variables: Dict[str, Any]):
def async_attach(self, action: AutomationActionType, variables: dict[str, Any]):
"""Attach a device trigger for turn on."""
@callback
@ -99,9 +101,7 @@ class PluggableAction:
return _remove
async def async_run(
self, hass: HomeAssistantType, context: Optional[Context] = None
):
async def async_run(self, hass: HomeAssistantType, context: Context | None = None):
"""Run all turn on triggers."""
for job, variables in self._actions.values():
hass.async_run_hass_job(job, variables, context)
@ -113,7 +113,7 @@ class PhilipsTVDataUpdateCoordinator(DataUpdateCoordinator[None]):
def __init__(self, hass, api: PhilipsTV) -> None:
"""Set up the coordinator."""
self.api = api
self._notify_future: Optional[asyncio.Task] = None
self._notify_future: asyncio.Task | None = None
@callback
def _update_listeners():

View file

@ -1,6 +1,8 @@
"""Config flow for Philips TV integration."""
from __future__ import annotations
import platform
from typing import Any, Dict, Optional, Tuple
from typing import Any
from haphilipsjs import ConnectionFailure, PairingFailure, PhilipsTV
import voluptuous as vol
@ -25,7 +27,7 @@ from .const import ( # pylint:disable=unused-import
async def validate_input(
hass: core.HomeAssistant, host: str, api_version: int
) -> Tuple[Dict, PhilipsTV]:
) -> tuple[dict, PhilipsTV]:
"""Validate the user input allows us to connect."""
hub = PhilipsTV(host, api_version)
@ -48,7 +50,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Initialize flow."""
super().__init__()
self._current = {}
self._hub: Optional[PhilipsTV] = None
self._hub: PhilipsTV | None = None
self._pair_state: Any = None
async def async_step_import(self, conf: dict) -> dict:
@ -72,7 +74,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
data=self._current,
)
async def async_step_pair(self, user_input: Optional[dict] = None) -> dict:
async def async_step_pair(self, user_input: dict | None = None) -> dict:
"""Attempt to pair with device."""
assert self._hub
@ -123,7 +125,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self._current[CONF_PASSWORD] = password
return await self._async_create_current()
async def async_step_user(self, user_input: Optional[dict] = None) -> dict:
async def async_step_user(self, user_input: dict | None = None) -> dict:
"""Handle the initial step."""
errors = {}
if user_input:

View file

@ -1,5 +1,5 @@
"""Provides device automations for control of device."""
from typing import List, Optional
from __future__ import annotations
import voluptuous as vol
@ -23,7 +23,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."""
triggers = []
triggers.append(
@ -43,7 +43,7 @@ async def async_attach_trigger(
config: ConfigType,
action: AutomationActionType,
automation_info: dict,
) -> Optional[CALLBACK_TYPE]:
) -> CALLBACK_TYPE | None:
"""Attach a trigger."""
registry: DeviceRegistry = await async_get_registry(hass)
if config[CONF_TYPE] == TRIGGER_TYPE_TURN_ON:

View file

@ -1,5 +1,7 @@
"""Media Player component to integrate TVs exposing the Joint Space API."""
from typing import Any, Dict, Optional
from __future__ import annotations
from typing import Any
from haphilipsjs import ConnectionFailure
import voluptuous as vol
@ -125,7 +127,7 @@ class PhilipsTVMediaPlayer(CoordinatorEntity, MediaPlayerEntity):
def __init__(
self,
coordinator: PhilipsTVDataUpdateCoordinator,
system: Dict[str, Any],
system: dict[str, Any],
unique_id: str,
):
"""Initialize the Philips TV."""
@ -137,10 +139,10 @@ class PhilipsTVMediaPlayer(CoordinatorEntity, MediaPlayerEntity):
self._system = system
self._unique_id = unique_id
self._state = STATE_OFF
self._media_content_type: Optional[str] = None
self._media_content_id: Optional[str] = None
self._media_title: Optional[str] = None
self._media_channel: Optional[str] = None
self._media_content_type: str | None = None
self._media_content_id: str | None = None
self._media_title: str | None = None
self._media_channel: str | None = None
super().__init__(coordinator)
self._update_from_coordinator()

View file

@ -1,11 +1,13 @@
"""Tracks the latency of a host by sending ICMP echo requests (ping)."""
from __future__ import annotations
import asyncio
from datetime import timedelta
from functools import partial
import logging
import re
import sys
from typing import Any, Dict
from typing import Any
from icmplib import SocketPermissionError, ping as icmp_ping
import voluptuous as vol
@ -105,7 +107,7 @@ class PingBinarySensor(BinarySensorEntity):
return self._ping.available
@property
def extra_state_attributes(self) -> Dict[str, Any]:
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the ICMP checo request."""
if self._ping.data is not False:
return {

View file

@ -1,5 +1,5 @@
"""Support for Plaato Airlock sensors."""
from typing import Optional
from __future__ import annotations
from pyplaato.models.device import PlaatoDevice
from pyplaato.plaato import PlaatoKeg
@ -63,7 +63,7 @@ class PlaatoSensor(PlaatoEntity):
"""Representation of a Plaato Sensor."""
@property
def device_class(self) -> Optional[str]:
def device_class(self) -> str | None:
"""Return the class of this device, from component DEVICE_CLASSES."""
if self._coordinator is not None:
if self._sensor_type == PlaatoKeg.Pins.TEMPERATURE:

View file

@ -1,9 +1,9 @@
"""Plugwise platform for Home Assistant Core."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
from typing import Dict
import async_timeout
from plugwise.exceptions import (
@ -201,7 +201,7 @@ class SmileGateway(CoordinatorEntity):
return self._name
@property
def device_info(self) -> Dict[str, any]:
def device_info(self) -> dict[str, any]:
"""Return the device information."""
device_information = {
"identifiers": {(DOMAIN, self._dev_id)},

View file

@ -1,6 +1,8 @@
"""Config flow for Plum Lightpad."""
from __future__ import annotations
import logging
from typing import Any, Dict, Optional
from typing import Any
from aiohttp import ContentTypeError
from requests.exceptions import ConnectTimeout, HTTPError
@ -34,8 +36,8 @@ class PlumLightpadConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
)
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 initialized by the user or redirected to by import."""
if not user_input:
return self._show_form()
@ -58,7 +60,7 @@ class PlumLightpadConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
)
async def async_step_import(
self, import_config: Optional[ConfigType]
) -> Dict[str, Any]:
self, import_config: ConfigType | None
) -> dict[str, Any]:
"""Import a config entry from configuration.yaml."""
return await self.async_step_user(import_config)

View file

@ -1,6 +1,8 @@
"""Support for Plum Lightpad lights."""
from __future__ import annotations
import asyncio
from typing import Callable, List
from typing import Callable
from plumlightpad import Plum
@ -23,7 +25,7 @@ from .const import DOMAIN
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: Callable[[List[Entity]], None],
async_add_entities: Callable[[list[Entity]], None],
) -> None:
"""Set up Plum Lightpad dimmer lights and glow rings."""

View file

@ -1,7 +1,8 @@
"""Sensor to collect the reference daily prices of electricity ('PVPC') in Spain."""
from __future__ import annotations
import logging
from random import randint
from typing import Optional
from aiopvpc import PVPCData
@ -92,7 +93,7 @@ class ElecPriceSensor(RestoreEntity):
self.update_current_price(dt_util.utcnow())
@property
def unique_id(self) -> Optional[str]:
def unique_id(self) -> str | None:
"""Return a unique ID."""
return self._unique_id