Update typing 13 (#48077)

This commit is contained in:
Marc Mueller 2021-03-18 14:43:52 +01:00 committed by GitHub
parent a3cd1854f6
commit b67b9b94f9
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
51 changed files with 287 additions and 226 deletions

View file

@ -1,6 +1,7 @@
"""The Tag integration."""
from __future__ import annotations
import logging
import typing
import uuid
import voluptuous as vol
@ -63,7 +64,7 @@ class TagStorageCollection(collection.StorageCollection):
CREATE_SCHEMA = vol.Schema(CREATE_FIELDS)
UPDATE_SCHEMA = vol.Schema(UPDATE_FIELDS)
async def _process_create_data(self, data: typing.Dict) -> typing.Dict:
async def _process_create_data(self, data: dict) -> dict:
"""Validate the config is valid."""
data = self.CREATE_SCHEMA(data)
if not data[TAG_ID]:
@ -74,11 +75,11 @@ class TagStorageCollection(collection.StorageCollection):
return data
@callback
def _get_suggested_id(self, info: typing.Dict) -> str:
def _get_suggested_id(self, info: dict) -> str:
"""Suggest an ID based on the config."""
return info[TAG_ID]
async def _update_data(self, data: dict, update_data: typing.Dict) -> typing.Dict:
async def _update_data(self, data: dict, update_data: dict) -> dict:
"""Return a new updated data object."""
data = {**data, **self.UPDATE_SCHEMA(update_data)}
# make last_scanned JSON serializeable

View file

@ -1,6 +1,8 @@
"""Provides device automations for Tasmota."""
from __future__ import annotations
import logging
from typing import Callable, List, Optional
from typing import Callable
import attr
from hatasmota.trigger import TasmotaTrigger
@ -47,7 +49,7 @@ class TriggerInstance:
action: AutomationActionType = attr.ib()
automation_info: dict = attr.ib()
trigger: "Trigger" = attr.ib()
remove: Optional[CALLBACK_TYPE] = attr.ib(default=None)
remove: CALLBACK_TYPE | None = attr.ib(default=None)
async def async_attach_trigger(self):
"""Attach event trigger."""
@ -85,7 +87,7 @@ class Trigger:
subtype: str = attr.ib()
tasmota_trigger: TasmotaTrigger = attr.ib()
type: str = attr.ib()
trigger_instances: List[TriggerInstance] = attr.ib(factory=list)
trigger_instances: list[TriggerInstance] = attr.ib(factory=list)
async def add_trigger(self, action, automation_info):
"""Add Tasmota trigger."""
@ -238,7 +240,7 @@ async def async_remove_triggers(hass: HomeAssistant, device_id: str):
device_trigger.remove_update_signal()
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 Tasmota device."""
triggers = []

View file

@ -1,5 +1,5 @@
"""Support for Tasmota sensors."""
from typing import Optional
from __future__ import annotations
from hatasmota import const as hc, status_sensor
@ -163,7 +163,7 @@ class TasmotaSensor(TasmotaAvailability, TasmotaDiscoveryUpdate, Entity):
self.async_write_ha_state()
@property
def device_class(self) -> Optional[str]:
def device_class(self) -> str | None:
"""Return the device class of the sensor."""
class_or_icon = SENSOR_DEVICE_CLASS_ICON_MAP.get(
self._tasmota_entity.quantity, {}

View file

@ -1,5 +1,5 @@
"""Allows the creation of a sensor that breaks out state_attributes."""
from typing import Optional
from __future__ import annotations
import voluptuous as vol
@ -165,7 +165,7 @@ class SensorTemplate(TemplateEntity, Entity):
return self._state
@property
def device_class(self) -> Optional[str]:
def device_class(self) -> str | None:
"""Return the device class of the sensor."""
return self._device_class

View file

@ -1,7 +1,8 @@
"""TemplateEntity utility class."""
from __future__ import annotations
import logging
from typing import Any, Callable, List, Optional, Union
from typing import Any, Callable
import voluptuous as vol
@ -30,8 +31,8 @@ class _TemplateAttribute:
attribute: str,
template: Template,
validator: Callable[[Any], Any] = None,
on_update: Optional[Callable[[Any], None]] = None,
none_on_template_error: Optional[bool] = False,
on_update: Callable[[Any], None] | None = None,
none_on_template_error: bool | None = False,
):
"""Template attribute."""
self._entity = entity
@ -61,10 +62,10 @@ class _TemplateAttribute:
@callback
def handle_result(
self,
event: Optional[Event],
event: Event | None,
template: Template,
last_result: Union[str, None, TemplateError],
result: Union[str, TemplateError],
last_result: str | None | TemplateError,
result: str | TemplateError,
) -> None:
"""Handle a template result event callback."""
if isinstance(result, TemplateError):
@ -189,7 +190,7 @@ class TemplateEntity(Entity):
attribute: str,
template: Template,
validator: Callable[[Any], Any] = None,
on_update: Optional[Callable[[Any], None]] = None,
on_update: Callable[[Any], None] | None = None,
none_on_template_error: bool = False,
) -> None:
"""
@ -219,8 +220,8 @@ class TemplateEntity(Entity):
@callback
def _handle_results(
self,
event: Optional[Event],
updates: List[TrackTemplateResult],
event: Event | None,
updates: list[TrackTemplateResult],
) -> None:
"""Call back the results to the attributes."""
if event:

View file

@ -1,6 +1,7 @@
"""Support for Tesla HVAC system."""
from __future__ import annotations
import logging
from typing import List, Optional
from teslajsonpy.exceptions import UnknownPresetMode
@ -103,7 +104,7 @@ class TeslaThermostat(TeslaDevice, ClimateEntity):
_LOGGER.error("%s", ex.message)
@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.
@ -111,7 +112,7 @@ class TeslaThermostat(TeslaDevice, ClimateEntity):
return self.tesla_device.preset_mode
@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.

View file

@ -1,5 +1,5 @@
"""Support for tracking Tesla cars."""
from typing import Optional
from __future__ import annotations
from homeassistant.components.device_tracker import SOURCE_TYPE_GPS
from homeassistant.components.device_tracker.config_entry import TrackerEntity
@ -25,13 +25,13 @@ class TeslaDeviceEntity(TeslaDevice, TrackerEntity):
"""A class representing a Tesla device."""
@property
def latitude(self) -> Optional[float]:
def latitude(self) -> float | None:
"""Return latitude value of the device."""
location = self.tesla_device.get_location()
return self.tesla_device.get_location().get("latitude") if location else None
@property
def longitude(self) -> Optional[float]:
def longitude(self) -> float | None:
"""Return longitude value of the device."""
location = self.tesla_device.get_location()
return self.tesla_device.get_location().get("longitude") if location else None

View file

@ -1,5 +1,5 @@
"""Support for the Tesla sensors."""
from typing import Optional
from __future__ import annotations
from homeassistant.components.sensor import DEVICE_CLASSES
from homeassistant.const import (
@ -39,7 +39,7 @@ class TeslaSensor(TeslaDevice, Entity):
self._unique_id = f"{super().unique_id}_{self.type}"
@property
def state(self) -> Optional[float]:
def state(self) -> float | None:
"""Return the state of the sensor."""
if self.tesla_device.type == "temperature sensor":
if self.type == "outside":
@ -58,7 +58,7 @@ class TeslaSensor(TeslaDevice, Entity):
return self.tesla_device.get_value()
@property
def unit_of_measurement(self) -> Optional[str]:
def unit_of_measurement(self) -> str | None:
"""Return the unit_of_measurement of the device."""
units = self.tesla_device.measurement
if units == "F":
@ -72,7 +72,7 @@ class TeslaSensor(TeslaDevice, Entity):
return units
@property
def device_class(self) -> Optional[str]:
def device_class(self) -> str | None:
"""Return the device_class of the device."""
return (
self.tesla_device.device_class

View file

@ -3,7 +3,6 @@ from __future__ import annotations
from datetime import datetime, timedelta
import logging
from typing import Dict, Optional
import voluptuous as vol
@ -165,7 +164,7 @@ class TimerStorageCollection(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."""
data = self.CREATE_SCHEMA(data)
# make duration JSON serializeable
@ -173,11 +172,11 @@ class TimerStorageCollection(collection.StorageCollection):
return 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."""
data = {**data, **self.UPDATE_SCHEMA(update_data)}
# make duration JSON serializeable
@ -189,18 +188,18 @@ class TimerStorageCollection(collection.StorageCollection):
class Timer(RestoreEntity):
"""Representation of a timer."""
def __init__(self, config: Dict):
def __init__(self, config: dict):
"""Initialize a timer."""
self._config: dict = config
self.editable: bool = True
self._state: str = STATUS_IDLE
self._duration = cv.time_period_str(config[CONF_DURATION])
self._remaining: Optional[timedelta] = None
self._end: Optional[datetime] = None
self._remaining: timedelta | None = None
self._end: datetime | None = None
self._listener = None
@classmethod
def from_yaml(cls, config: Dict) -> Timer:
def from_yaml(cls, config: dict) -> Timer:
"""Return entity instance initialized from yaml storage."""
timer = cls(config)
timer.entity_id = ENTITY_ID_FORMAT.format(config[CONF_ID])
@ -247,7 +246,7 @@ class Timer(RestoreEntity):
return attrs
@property
def unique_id(self) -> Optional[str]:
def unique_id(self) -> str | None:
"""Return unique id for the entity."""
return self._config[CONF_ID]
@ -348,7 +347,7 @@ class Timer(RestoreEntity):
self.hass.bus.async_fire(EVENT_TIMER_FINISHED, {"entity_id": self.entity_id})
self.async_write_ha_state()
async def async_update_config(self, config: Dict) -> None:
async def async_update_config(self, config: dict) -> None:
"""Handle when the config is updated."""
self._config = config
self._duration = cv.time_period_str(config[CONF_DURATION])

View file

@ -1,7 +1,9 @@
"""Reproduce an Timer 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
@ -27,8 +29,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)
@ -69,8 +71,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 Timer states."""
await asyncio.gather(

View file

@ -1,5 +1,5 @@
"""Support for Toon binary sensors."""
from typing import Optional
from __future__ import annotations
from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.config_entries import ConfigEntry
@ -84,7 +84,7 @@ class ToonBinarySensor(ToonEntity, BinarySensorEntity):
return BINARY_SENSOR_ENTITIES[self.key][ATTR_DEVICE_CLASS]
@property
def is_on(self) -> Optional[bool]:
def is_on(self) -> bool | None:
"""Return the status of the binary sensor."""
section = getattr(
self.coordinator.data, BINARY_SENSOR_ENTITIES[self.key][ATTR_SECTION]

View file

@ -1,5 +1,7 @@
"""Support for Toon thermostat."""
from typing import Any, Dict, List, Optional
from __future__ import annotations
from typing import Any
from toonapi import (
ACTIVE_STATE_AWAY,
@ -61,12 +63,12 @@ class ToonThermostatDevice(ToonDisplayDeviceEntity, ClimateEntity):
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]
@property
def hvac_action(self) -> Optional[str]:
def hvac_action(self) -> str | None:
"""Return the current running hvac operation."""
if self.coordinator.data.thermostat.heating:
return CURRENT_HVAC_HEAT
@ -78,7 +80,7 @@ class ToonThermostatDevice(ToonDisplayDeviceEntity, ClimateEntity):
return TEMP_CELSIUS
@property
def preset_mode(self) -> Optional[str]:
def preset_mode(self) -> str | None:
"""Return the current preset mode, e.g., home, away, temp."""
mapping = {
ACTIVE_STATE_AWAY: PRESET_AWAY,
@ -89,17 +91,17 @@ class ToonThermostatDevice(ToonDisplayDeviceEntity, ClimateEntity):
return mapping.get(self.coordinator.data.thermostat.active_state)
@property
def preset_modes(self) -> List[str]:
def preset_modes(self) -> list[str]:
"""Return a list of available preset modes."""
return [PRESET_AWAY, PRESET_COMFORT, PRESET_HOME, PRESET_SLEEP]
@property
def current_temperature(self) -> Optional[float]:
def current_temperature(self) -> float | None:
"""Return the current temperature."""
return self.coordinator.data.thermostat.current_display_temperature
@property
def target_temperature(self) -> Optional[float]:
def target_temperature(self) -> float | None:
"""Return the temperature we try to reach."""
return self.coordinator.data.thermostat.current_setpoint
@ -114,7 +116,7 @@ class ToonThermostatDevice(ToonDisplayDeviceEntity, ClimateEntity):
return DEFAULT_MAX_TEMP
@property
def extra_state_attributes(self) -> Dict[str, Any]:
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the current state of the burner."""
return {"heating_type": self.coordinator.data.agreement.heating_type}

View file

@ -1,6 +1,8 @@
"""Config flow to configure the Toon component."""
from __future__ import annotations
import logging
from typing import Any, Dict, List, Optional
from typing import Any
from toonapi import Agreement, Toon, ToonError
import voluptuous as vol
@ -19,15 +21,15 @@ class ToonFlowHandler(AbstractOAuth2FlowHandler, domain=DOMAIN):
DOMAIN = DOMAIN
VERSION = 2
agreements: Optional[List[Agreement]] = None
data: Optional[Dict[str, Any]] = None
agreements: list[Agreement] | None = None
data: dict[str, Any] | None = None
@property
def logger(self) -> logging.Logger:
"""Return logger."""
return logging.getLogger(__name__)
async def async_oauth_create_entry(self, data: Dict[str, Any]) -> Dict[str, Any]:
async def async_oauth_create_entry(self, data: dict[str, Any]) -> dict[str, Any]:
"""Test connection and load up agreements."""
self.data = data
@ -46,8 +48,8 @@ class ToonFlowHandler(AbstractOAuth2FlowHandler, domain=DOMAIN):
return await self.async_step_agreement()
async def async_step_import(
self, config: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, config: dict[str, Any] | None = None
) -> dict[str, Any]:
"""Start a configuration flow based on imported data.
This step is merely here to trigger "discovery" when the `toon`
@ -63,8 +65,8 @@ class ToonFlowHandler(AbstractOAuth2FlowHandler, domain=DOMAIN):
return await self.async_step_user()
async def async_step_agreement(
self, user_input: Dict[str, Any] = None
) -> Dict[str, Any]:
self, user_input: dict[str, Any] = None
) -> dict[str, Any]:
"""Select Toon agreement to add."""
if len(self.agreements) == 1:
return await self._create_entry(self.agreements[0])
@ -85,7 +87,7 @@ class ToonFlowHandler(AbstractOAuth2FlowHandler, domain=DOMAIN):
agreement_index = agreements_list.index(user_input[CONF_AGREEMENT])
return await self._create_entry(self.agreements[agreement_index])
async def _create_entry(self, agreement: Agreement) -> Dict[str, Any]:
async def _create_entry(self, agreement: Agreement) -> dict[str, Any]:
if CONF_MIGRATE in self.context:
await self.hass.config_entries.async_remove(self.context[CONF_MIGRATE])

View file

@ -1,7 +1,8 @@
"""Provides the Toon DataUpdateCoordinator."""
from __future__ import annotations
import logging
import secrets
from typing import Optional
from toonapi import Status, Toon, ToonError
@ -50,7 +51,7 @@ class ToonDataUpdateCoordinator(DataUpdateCoordinator[Status]):
for update_callback in self._listeners:
update_callback()
async def register_webhook(self, event: Optional[Event] = None) -> None:
async def register_webhook(self, event: Event | None = None) -> None:
"""Register a webhook with Toon to get live updates."""
if CONF_WEBHOOK_ID not in self.entry.data:
data = {**self.entry.data, CONF_WEBHOOK_ID: secrets.token_hex()}
@ -124,7 +125,7 @@ class ToonDataUpdateCoordinator(DataUpdateCoordinator[Status]):
except ToonError as err:
_LOGGER.error("Could not process data received from Toon webhook - %s", err)
async def unregister_webhook(self, event: Optional[Event] = None) -> None:
async def unregister_webhook(self, event: Event | None = None) -> None:
"""Remove / Unregister webhook for toon."""
_LOGGER.debug(
"Unregistering Toon webhook (%s)", self.entry.data[CONF_WEBHOOK_ID]

View file

@ -1,5 +1,7 @@
"""DataUpdate Coordinator, and base Entity and Device models for Toon."""
from typing import Any, Dict, Optional
from __future__ import annotations
from typing import Any
from homeassistant.helpers.update_coordinator import CoordinatorEntity
@ -31,7 +33,7 @@ class ToonEntity(CoordinatorEntity):
return self._name
@property
def icon(self) -> Optional[str]:
def icon(self) -> str | None:
"""Return the mdi icon of the entity."""
return self._icon
@ -45,7 +47,7 @@ class ToonDisplayDeviceEntity(ToonEntity):
"""Defines a Toon display device entity."""
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return device information about this thermostat."""
agreement = self.coordinator.data.agreement
model = agreement.display_hardware_version.rpartition("/")[0]
@ -63,7 +65,7 @@ class ToonElectricityMeterDeviceEntity(ToonEntity):
"""Defines a Electricity Meter device entity."""
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return device information about this entity."""
agreement_id = self.coordinator.data.agreement.agreement_id
return {
@ -77,7 +79,7 @@ class ToonGasMeterDeviceEntity(ToonEntity):
"""Defines a Gas Meter device entity."""
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return device information about this entity."""
agreement_id = self.coordinator.data.agreement.agreement_id
return {
@ -91,7 +93,7 @@ class ToonWaterMeterDeviceEntity(ToonEntity):
"""Defines a Water Meter device entity."""
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return device information about this entity."""
agreement_id = self.coordinator.data.agreement.agreement_id
return {
@ -105,7 +107,7 @@ class ToonSolarDeviceEntity(ToonEntity):
"""Defines a Solar Device device entity."""
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return device information about this entity."""
agreement_id = self.coordinator.data.agreement.agreement_id
return {
@ -119,7 +121,7 @@ class ToonBoilerModuleDeviceEntity(ToonEntity):
"""Defines a Boiler Module device entity."""
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return device information about this entity."""
agreement_id = self.coordinator.data.agreement.agreement_id
return {
@ -134,7 +136,7 @@ class ToonBoilerDeviceEntity(ToonEntity):
"""Defines a Boiler device entity."""
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return device information about this entity."""
agreement_id = self.coordinator.data.agreement.agreement_id
return {

View file

@ -1,5 +1,7 @@
"""OAuth2 implementations for Toon."""
from typing import Any, Optional, cast
from __future__ import annotations
from typing import Any, cast
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_entry_oauth2_flow
@ -55,7 +57,7 @@ class ToonLocalOAuth2Implementation(config_entry_oauth2_flow.LocalOAuth2Implemen
client_secret: str,
name: str,
tenant_id: str,
issuer: Optional[str] = None,
issuer: str | None = None,
):
"""Local Toon Oauth Implementation."""
self._name = name

View file

@ -1,5 +1,5 @@
"""Support for Toon sensors."""
from typing import Optional
from __future__ import annotations
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
@ -132,7 +132,7 @@ class ToonSensor(ToonEntity):
return f"{DOMAIN}_{agreement_id}_sensor_{self.key}"
@property
def state(self) -> Optional[str]:
def state(self) -> str | None:
"""Return the state of the sensor."""
section = getattr(
self.coordinator.data, SENSOR_ENTITIES[self.key][ATTR_SECTION]
@ -140,12 +140,12 @@ class ToonSensor(ToonEntity):
return getattr(section, SENSOR_ENTITIES[self.key][ATTR_MEASUREMENT])
@property
def unit_of_measurement(self) -> Optional[str]:
def unit_of_measurement(self) -> str | None:
"""Return the unit this state is expressed in."""
return SENSOR_ENTITIES[self.key][ATTR_UNIT_OF_MEASUREMENT]
@property
def device_class(self) -> Optional[str]:
def device_class(self) -> str | None:
"""Return the device class."""
return SENSOR_ENTITIES[self.key][ATTR_DEVICE_CLASS]

View file

@ -1,6 +1,7 @@
"""Common code for tplink."""
from __future__ import annotations
import logging
from typing import List
from pyHS100 import (
Discover,
@ -30,7 +31,7 @@ class SmartDevices:
"""Hold different kinds of devices."""
def __init__(
self, lights: List[SmartDevice] = None, switches: List[SmartDevice] = None
self, lights: list[SmartDevice] = None, switches: list[SmartDevice] = None
):
"""Initialize device holder."""
self._lights = lights or []

View file

@ -1,9 +1,11 @@
"""Support for TPLink lights."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
import time
from typing import Any, Dict, NamedTuple, Tuple, cast
from typing import Any, NamedTuple, cast
from pyHS100 import SmartBulb, SmartDeviceException
@ -88,7 +90,7 @@ class LightState(NamedTuple):
state: bool
brightness: int
color_temp: float
hs: Tuple[int, int]
hs: tuple[int, int]
def to_param(self):
"""Return a version that we can send to the bulb."""
@ -109,7 +111,7 @@ class LightState(NamedTuple):
class LightFeatures(NamedTuple):
"""Light features."""
sysinfo: Dict[str, Any]
sysinfo: dict[str, Any]
mac: str
alias: str
model: str

View file

@ -1,5 +1,5 @@
"""Support for monitoring the Transmission BitTorrent client API."""
from typing import List
from __future__ import annotations
from transmissionrpc.torrent import Torrent
@ -168,7 +168,7 @@ class TransmissionTorrentsSensor(TransmissionSensor):
self._state = len(torrents)
def _filter_torrents(torrents: List[Torrent], statuses=None) -> List[Torrent]:
def _filter_torrents(torrents: list[Torrent], statuses=None) -> list[Torrent]:
return [
torrent
for torrent in torrents

View file

@ -1,4 +1,6 @@
"""Provide functionality for TTS."""
from __future__ import annotations
import asyncio
import functools as ft
import hashlib
@ -7,7 +9,7 @@ import logging
import mimetypes
import os
import re
from typing import Dict, Optional, cast
from typing import cast
from aiohttp import web
import mutagen
@ -243,7 +245,7 @@ async def async_setup(hass, config):
return True
def _hash_options(options: Dict) -> str:
def _hash_options(options: dict) -> str:
"""Hashes an options dictionary."""
opts_hash = hashlib.blake2s(digest_size=5)
for key, value in sorted(options.items()):
@ -512,8 +514,8 @@ class SpeechManager:
class Provider:
"""Represent a single TTS provider."""
hass: Optional[HomeAssistantType] = None
name: Optional[str] = None
hass: HomeAssistantType | None = None
name: str | None = None
@property
def default_language(self):

View file

@ -1,6 +1,7 @@
"""Support for Tuya fans."""
from __future__ import annotations
from datetime import timedelta
from typing import Optional
from homeassistant.components.fan import (
DOMAIN as SENSOR_DOMAIN,
@ -124,7 +125,7 @@ class TuyaFanDevice(TuyaDevice, FanEntity):
return self._tuya.state()
@property
def percentage(self) -> Optional[int]:
def percentage(self) -> int | None:
"""Return the current speed."""
if not self.is_on:
return 0

View file

@ -1,8 +1,9 @@
"""The Twinkly light component."""
from __future__ import annotations
import asyncio
import logging
from typing import Any, Dict, Optional
from typing import Any
from aiohttp import ClientError
@ -84,7 +85,7 @@ class TwinklyLight(LightEntity):
return self._is_available
@property
def unique_id(self) -> Optional[str]:
def unique_id(self) -> str | None:
"""Id of the device."""
return self._id
@ -104,7 +105,7 @@ class TwinklyLight(LightEntity):
return "mdi:string-lights"
@property
def device_info(self) -> Optional[Dict[str, Any]]:
def device_info(self) -> dict[str, Any] | None:
"""Get device specific attributes."""
return (
{
@ -123,7 +124,7 @@ class TwinklyLight(LightEntity):
return self._is_on
@property
def brightness(self) -> Optional[int]:
def brightness(self) -> int | None:
"""Return the brightness of the light."""
return self._brightness

View file

@ -1,8 +1,9 @@
"""UniFi Controller abstraction."""
from __future__ import annotations
import asyncio
from datetime import datetime, timedelta
import ssl
from typing import Optional
from aiohttp import CookieJar
import aiounifi
@ -385,7 +386,7 @@ class UniFiController:
@callback
def async_heartbeat(
self, unique_id: str, heartbeat_expire_time: Optional[datetime] = None
self, unique_id: str, heartbeat_expire_time: datetime | None = None
) -> None:
"""Signal when a device has fresh home state."""
if heartbeat_expire_time is not None:

View file

@ -1,6 +1,7 @@
"""Combination of multiple media players for a universal controller."""
from __future__ import annotations
from copy import copy
from typing import Optional
import voluptuous as vol
@ -270,7 +271,7 @@ class UniversalMediaPlayer(MediaPlayerEntity):
return False
@property
def device_class(self) -> Optional[str]:
def device_class(self) -> str | None:
"""Return the class of this device."""
return self._device_class

View file

@ -1,6 +1,7 @@
"""Support for UPC ConnectBox router."""
from __future__ import annotations
import logging
from typing import List, Optional
from connect_box import ConnectBox
from connect_box.exceptions import ConnectBoxError, ConnectBoxLoginError
@ -57,7 +58,7 @@ class UPCDeviceScanner(DeviceScanner):
"""Initialize the scanner."""
self.connect_box: ConnectBox = connect_box
async def async_scan_devices(self) -> List[str]:
async def async_scan_devices(self) -> list[str]:
"""Scan for new devices and return a list with found device IDs."""
try:
await self.connect_box.async_get_devices()
@ -66,7 +67,7 @@ class UPCDeviceScanner(DeviceScanner):
return [device.mac for device in self.connect_box.devices]
async def async_get_device_name(self, device: str) -> Optional[str]:
async def async_get_device_name(self, device: str) -> str | None:
"""Get the device name (the name of the wireless device not used)."""
for connected_device in self.connect_box.devices:
if (

View file

@ -1,9 +1,10 @@
"""Support for UpCloud."""
from __future__ import annotations
import dataclasses
from datetime import timedelta
import logging
from typing import Dict, List
from typing import Dict
import requests.exceptions
import upcloud_api
@ -91,7 +92,7 @@ class UpCloudDataUpdateCoordinator(
hass, _LOGGER, name=f"{username}@UpCloud", update_interval=update_interval
)
self.cloud_manager = cloud_manager
self.unsub_handlers: List[CALLBACK_TYPE] = []
self.unsub_handlers: list[CALLBACK_TYPE] = []
async def async_update_config(self, config_entry: ConfigEntry) -> None:
"""Handle config update."""
@ -99,7 +100,7 @@ class UpCloudDataUpdateCoordinator(
seconds=config_entry.options[CONF_SCAN_INTERVAL]
)
async def _async_update_data(self) -> Dict[str, upcloud_api.Server]:
async def _async_update_data(self) -> dict[str, upcloud_api.Server]:
return {
x.uuid: x
for x in await self.hass.async_add_executor_job(
@ -112,10 +113,10 @@ class UpCloudDataUpdateCoordinator(
class UpCloudHassData:
"""Home Assistant UpCloud runtime data."""
coordinators: Dict[str, UpCloudDataUpdateCoordinator] = dataclasses.field(
coordinators: dict[str, UpCloudDataUpdateCoordinator] = dataclasses.field(
default_factory=dict
)
scan_interval_migrations: Dict[str, int] = dataclasses.field(default_factory=dict)
scan_interval_migrations: dict[str, int] = dataclasses.field(default_factory=dict)
async def async_setup(hass: HomeAssistantType, config) -> bool:

View file

@ -1,6 +1,8 @@
"""Config flow for UPNP."""
from __future__ import annotations
from datetime import timedelta
from typing import Any, Mapping, Optional
from typing import Any, Mapping
import voluptuous as vol
@ -55,7 +57,7 @@ class UpnpFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
self._discoveries: Mapping = None
async def async_step_user(
self, user_input: Optional[Mapping] = None
self, user_input: Mapping | None = None
) -> Mapping[str, Any]:
"""Handle a flow start."""
_LOGGER.debug("async_step_user: user_input: %s", user_input)
@ -111,9 +113,7 @@ class UpnpFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
data_schema=data_schema,
)
async def async_step_import(
self, import_info: Optional[Mapping]
) -> Mapping[str, Any]:
async def async_step_import(self, import_info: Mapping | None) -> Mapping[str, Any]:
"""Import a new UPnP/IGD device as a config entry.
This flow is triggered by `async_setup`. If no device has been
@ -204,7 +204,7 @@ class UpnpFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return await self.async_step_ssdp_confirm()
async def async_step_ssdp_confirm(
self, user_input: Optional[Mapping] = None
self, user_input: Mapping | None = None
) -> Mapping[str, Any]:
"""Confirm integration via SSDP."""
_LOGGER.debug("async_step_ssdp_confirm: user_input: %s", user_input)

View file

@ -3,7 +3,7 @@ from __future__ import annotations
import asyncio
from ipaddress import IPv4Address
from typing import List, Mapping
from typing import Mapping
from urllib.parse import urlparse
from async_upnp_client import UpnpFactory
@ -42,7 +42,7 @@ class Device:
self._igd_device: IgdDevice = igd_device
@classmethod
async def async_discover(cls, hass: HomeAssistantType) -> List[Mapping]:
async def async_discover(cls, hass: HomeAssistantType) -> list[Mapping]:
"""Discover UPnP/IGD devices."""
_LOGGER.debug("Discovering UPnP/IGD devices")
local_ip = None

View file

@ -1,6 +1,8 @@
"""Support for UPnP/IGD Sensors."""
from __future__ import annotations
from datetime import timedelta
from typing import Any, Mapping, Optional
from typing import Any, Mapping
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import DATA_BYTES, DATA_RATE_KIBIBYTES_PER_SECOND
@ -176,7 +178,7 @@ class RawUpnpSensor(UpnpSensor):
"""Representation of a UPnP/IGD sensor."""
@property
def state(self) -> Optional[str]:
def state(self) -> str | None:
"""Return the state of the device."""
device_value_key = self._sensor_type["device_value_key"]
value = self.coordinator.data[device_value_key]
@ -214,7 +216,7 @@ class DerivedUpnpSensor(UpnpSensor):
return current_value < self._last_value
@property
def state(self) -> Optional[str]:
def state(self) -> str | None:
"""Return the state of the device."""
# Can't calculate any derivative if we have only one value.
device_value_key = self._sensor_type["device_value_key"]

View file

@ -1,7 +1,8 @@
"""Support for U.S. Geological Survey Earthquake Hazards Program Feeds."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Optional
from geojson_client.usgs_earthquake_hazards_program_feed import (
UsgsEarthquakeHazardsProgramFeedManager,
@ -255,22 +256,22 @@ class UsgsEarthquakesEvent(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,9 @@
"""Support for Ubiquiti's UVC cameras."""
from __future__ import annotations
from datetime import datetime
import logging
import re
from typing import Optional
import requests
from uvcclient import camera as uvc_camera, nvr
@ -255,7 +256,7 @@ class UnifiVideoCamera(Camera):
self._caminfo = self._nvr.get_camera(self._uuid)
def timestamp_ms_to_date(epoch_ms: int) -> Optional[datetime]:
def timestamp_ms_to_date(epoch_ms: int) -> datetime | None:
"""Convert millisecond timestamp to datetime."""
if epoch_ms:
return datetime.fromtimestamp(epoch_ms / 1000)

View file

@ -1,5 +1,5 @@
"""Provides device automations for Vacuum."""
from typing import List, Optional
from __future__ import annotations
import voluptuous as vol
@ -26,7 +26,7 @@ ACTION_SCHEMA = cv.DEVICE_ACTION_BASE_SCHEMA.extend(
)
async def async_get_actions(hass: HomeAssistant, device_id: str) -> List[dict]:
async def async_get_actions(hass: HomeAssistant, device_id: str) -> list[dict]:
"""List device actions for Vacuum devices."""
registry = await entity_registry.async_get_registry(hass)
actions = []
@ -57,7 +57,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 Vacuum."""
from typing import Dict, List
from __future__ import annotations
import voluptuous as vol
@ -30,7 +30,7 @@ CONDITION_SCHEMA = DEVICE_CONDITION_BASE_SCHEMA.extend(
async def async_get_conditions(
hass: HomeAssistant, device_id: str
) -> List[Dict[str, str]]:
) -> list[dict[str, str]]:
"""List device conditions for Vacuum devices."""
registry = await entity_registry.async_get_registry(hass)
conditions = []

View file

@ -1,5 +1,5 @@
"""Provides device automations for Vacuum."""
from typing import List
from __future__ import annotations
import voluptuous as vol
@ -29,7 +29,7 @@ TRIGGER_SCHEMA = TRIGGER_BASE_SCHEMA.extend(
)
async def async_get_triggers(hass: HomeAssistant, device_id: str) -> List[dict]:
async def async_get_triggers(hass: HomeAssistant, device_id: str) -> list[dict]:
"""List device triggers for Vacuum devices."""
registry = await entity_registry.async_get_registry(hass)
triggers = []

View file

@ -1,7 +1,9 @@
"""Reproduce an Vacuum 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,
@ -44,8 +46,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)
@ -99,8 +101,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 Vacuum states."""
# Reproduce states in parallel.

View file

@ -1,8 +1,10 @@
"""Support for Vera devices."""
from __future__ import annotations
import asyncio
from collections import defaultdict
import logging
from typing import Any, Dict, Generic, List, Optional, Type, TypeVar
from typing import Any, Generic, TypeVar
import pyvera as veraApi
from requests.exceptions import RequestException
@ -172,7 +174,7 @@ async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) ->
return True
def map_vera_device(vera_device: veraApi.VeraDevice, remap: List[int]) -> str:
def map_vera_device(vera_device: veraApi.VeraDevice, remap: list[int]) -> str:
"""Map vera classes to Home Assistant types."""
type_map = {
@ -187,7 +189,7 @@ def map_vera_device(vera_device: veraApi.VeraDevice, remap: List[int]) -> str:
veraApi.VeraSwitch: "switch",
}
def map_special_case(instance_class: Type, entity_type: str) -> str:
def map_special_case(instance_class: type, entity_type: str) -> str:
if instance_class is veraApi.VeraSwitch and vera_device.device_id in remap:
return "light"
return entity_type
@ -248,7 +250,7 @@ class VeraDevice(Generic[DeviceType], Entity):
return self.vera_device.should_poll
@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 device."""
attr = {}

View file

@ -1,5 +1,7 @@
"""Support for Vera binary sensors."""
from typing import Callable, List, Optional
from __future__ import annotations
from typing import Callable
import pyvera as veraApi
@ -19,7 +21,7 @@ from .common import ControllerData, get_controller_data
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 sensor config entry."""
controller_data = get_controller_data(hass, entry)
@ -44,7 +46,7 @@ class VeraBinarySensor(VeraDevice[veraApi.VeraBinarySensor], BinarySensorEntity)
self.entity_id = ENTITY_ID_FORMAT.format(self.vera_id)
@property
def is_on(self) -> Optional[bool]:
def is_on(self) -> bool | None:
"""Return true if sensor is on."""
return self._state

View file

@ -1,5 +1,7 @@
"""Support for Vera thermostats."""
from typing import Any, Callable, List, Optional
from __future__ import annotations
from typing import Any, Callable
import pyvera as veraApi
@ -36,7 +38,7 @@ SUPPORT_HVAC = [HVAC_MODE_COOL, HVAC_MODE_HEAT, HVAC_MODE_HEAT_COOL, HVAC_MODE_O
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 sensor config entry."""
controller_data = get_controller_data(hass, entry)
@ -60,7 +62,7 @@ class VeraThermostat(VeraDevice[veraApi.VeraThermostat], ClimateEntity):
self.entity_id = ENTITY_ID_FORMAT.format(self.vera_id)
@property
def supported_features(self) -> Optional[int]:
def supported_features(self) -> int | None:
"""Return the list of supported features."""
return SUPPORT_FLAGS
@ -80,7 +82,7 @@ class VeraThermostat(VeraDevice[veraApi.VeraThermostat], ClimateEntity):
return HVAC_MODE_OFF
@property
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.
@ -88,7 +90,7 @@ class VeraThermostat(VeraDevice[veraApi.VeraThermostat], ClimateEntity):
return SUPPORT_HVAC
@property
def fan_mode(self) -> Optional[str]:
def fan_mode(self) -> str | None:
"""Return the fan setting."""
mode = self.vera_device.get_fan_mode()
if mode == "ContinuousOn":
@ -96,7 +98,7 @@ class VeraThermostat(VeraDevice[veraApi.VeraThermostat], ClimateEntity):
return FAN_AUTO
@property
def fan_modes(self) -> Optional[List[str]]:
def fan_modes(self) -> list[str] | None:
"""Return a list of available fan modes."""
return FAN_OPERATION_LIST
@ -110,7 +112,7 @@ class VeraThermostat(VeraDevice[veraApi.VeraThermostat], ClimateEntity):
self.schedule_update_ha_state()
@property
def current_power_w(self) -> Optional[float]:
def current_power_w(self) -> float | None:
"""Return the current power usage in W."""
power = self.vera_device.power
if power:
@ -127,7 +129,7 @@ class VeraThermostat(VeraDevice[veraApi.VeraThermostat], ClimateEntity):
return TEMP_CELSIUS
@property
def current_temperature(self) -> Optional[float]:
def current_temperature(self) -> float | None:
"""Return the current temperature."""
return self.vera_device.get_current_temperature()
@ -137,7 +139,7 @@ class VeraThermostat(VeraDevice[veraApi.VeraThermostat], ClimateEntity):
return self.vera_device.get_hvac_mode()
@property
def target_temperature(self) -> Optional[float]:
def target_temperature(self) -> float | None:
"""Return the temperature we try to reach."""
return self.vera_device.get_current_goal_temperature()

View file

@ -1,5 +1,7 @@
"""Common vera code."""
from typing import DefaultDict, List, NamedTuple, Set
from __future__ import annotations
from typing import DefaultDict, NamedTuple
import pyvera as pv
@ -15,12 +17,12 @@ class ControllerData(NamedTuple):
"""Controller data."""
controller: pv.VeraController
devices: DefaultDict[str, List[pv.VeraDevice]]
scenes: List[pv.VeraScene]
devices: DefaultDict[str, list[pv.VeraDevice]]
scenes: list[pv.VeraScene]
config_entry: ConfigEntry
def get_configured_platforms(controller_data: ControllerData) -> Set[str]:
def get_configured_platforms(controller_data: ControllerData) -> set[str]:
"""Get configured platforms for a controller."""
platforms = []
for platform in controller_data.devices:

View file

@ -1,7 +1,9 @@
"""Config flow for Vera."""
from __future__ import annotations
import logging
import re
from typing import Any, List
from typing import Any
import pyvera as pv
from requests.exceptions import RequestException
@ -19,22 +21,22 @@ LIST_REGEX = re.compile("[^0-9]+")
_LOGGER = logging.getLogger(__name__)
def fix_device_id_list(data: List[Any]) -> List[int]:
def fix_device_id_list(data: list[Any]) -> list[int]:
"""Fix the id list by converting it to a supported int list."""
return str_to_int_list(list_to_str(data))
def str_to_int_list(data: str) -> List[int]:
def str_to_int_list(data: str) -> list[int]:
"""Convert a string to an int list."""
return [int(s) for s in LIST_REGEX.split(data) if len(s) > 0]
def list_to_str(data: List[Any]) -> str:
def list_to_str(data: list[Any]) -> str:
"""Convert an int list to a string."""
return " ".join([str(i) for i in data])
def new_options(lights: List[int], exclude: List[int]) -> dict:
def new_options(lights: list[int], exclude: list[int]) -> dict:
"""Create a standard options object."""
return {CONF_LIGHTS: lights, CONF_EXCLUDE: exclude}

View file

@ -1,5 +1,7 @@
"""Support for Vera cover - curtains, rollershutters etc."""
from typing import Any, Callable, List
from __future__ import annotations
from typing import Any, Callable
import pyvera as veraApi
@ -20,7 +22,7 @@ from .common import ControllerData, get_controller_data
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 sensor config entry."""
controller_data = get_controller_data(hass, entry)

View file

@ -1,5 +1,7 @@
"""Support for Vera lights."""
from typing import Any, Callable, List, Optional, Tuple
from __future__ import annotations
from typing import Any, Callable
import pyvera as veraApi
@ -24,7 +26,7 @@ from .common import ControllerData, get_controller_data
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 sensor config entry."""
controller_data = get_controller_data(hass, entry)
@ -51,12 +53,12 @@ class VeraLight(VeraDevice[veraApi.VeraDimmer], LightEntity):
self.entity_id = ENTITY_ID_FORMAT.format(self.vera_id)
@property
def brightness(self) -> Optional[int]:
def brightness(self) -> int | None:
"""Return the brightness of the light."""
return self._brightness
@property
def hs_color(self) -> Optional[Tuple[float, float]]:
def hs_color(self) -> tuple[float, float] | None:
"""Return the color of the light."""
return self._color

View file

@ -1,5 +1,7 @@
"""Support for Vera locks."""
from typing import Any, Callable, Dict, List, Optional
from __future__ import annotations
from typing import Any, Callable
import pyvera as veraApi
@ -23,7 +25,7 @@ ATTR_LOW_BATTERY = "low_battery"
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 sensor config entry."""
controller_data = get_controller_data(hass, entry)
@ -56,12 +58,12 @@ class VeraLock(VeraDevice[veraApi.VeraLock], LockEntity):
self._state = STATE_UNLOCKED
@property
def is_locked(self) -> Optional[bool]:
def is_locked(self) -> bool | None:
"""Return true if device is on."""
return self._state == STATE_LOCKED
@property
def extra_state_attributes(self) -> Optional[Dict[str, Any]]:
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Who unlocked the lock and did a low battery alert fire.
Reports on the previous poll cycle.
@ -78,7 +80,7 @@ class VeraLock(VeraDevice[veraApi.VeraLock], LockEntity):
return data
@property
def changed_by(self) -> Optional[str]:
def changed_by(self) -> str | None:
"""Who unlocked the lock.
Reports on the previous poll cycle.

View file

@ -1,5 +1,7 @@
"""Support for Vera scenes."""
from typing import Any, Callable, Dict, List, Optional
from __future__ import annotations
from typing import Any, Callable
import pyvera as veraApi
@ -16,7 +18,7 @@ from .const import VERA_ID_FORMAT
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 sensor config entry."""
controller_data = get_controller_data(hass, entry)
@ -53,6 +55,6 @@ class VeraScene(Scene):
return self._name
@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 scene."""
return {"vera_scene_id": self.vera_scene.vera_scene_id}

View file

@ -1,6 +1,8 @@
"""Support for Vera sensors."""
from __future__ import annotations
from datetime import timedelta
from typing import Callable, List, Optional, cast
from typing import Callable, cast
import pyvera as veraApi
@ -20,7 +22,7 @@ SCAN_INTERVAL = timedelta(seconds=5)
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 sensor config entry."""
controller_data = get_controller_data(hass, entry)
@ -52,7 +54,7 @@ class VeraSensor(VeraDevice[veraApi.VeraSensor], Entity):
return self.current_value
@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."""
if self.vera_device.category == veraApi.CATEGORY_TEMPERATURE_SENSOR:

View file

@ -1,5 +1,7 @@
"""Support for Vera switches."""
from typing import Any, Callable, List, Optional
from __future__ import annotations
from typing import Any, Callable
import pyvera as veraApi
@ -20,7 +22,7 @@ from .common import ControllerData, get_controller_data
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 sensor config entry."""
controller_data = get_controller_data(hass, entry)
@ -57,7 +59,7 @@ class VeraSwitch(VeraDevice[veraApi.VeraSwitch], SwitchEntity):
self.schedule_update_ha_state()
@property
def current_power_w(self) -> Optional[float]:
def current_power_w(self) -> float | None:
"""Return the current power usage in W."""
power = self.vera_device.power
if power:

View file

@ -1,8 +1,10 @@
"""The vizio component."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
from typing import Any, Dict, List
from typing import Any
from pyvizio.const import APPS
from pyvizio.util import gen_apps_list_from_url
@ -116,7 +118,7 @@ class VizioAppsDataUpdateCoordinator(DataUpdateCoordinator):
)
self.data = APPS
async def _async_update_data(self) -> List[Dict[str, Any]]:
async def _async_update_data(self) -> list[dict[str, Any]]:
"""Update data via library."""
data = await gen_apps_list_from_url(session=async_get_clientsession(self.hass))
if not data:

View file

@ -1,8 +1,10 @@
"""Config flow for Vizio."""
from __future__ import annotations
import copy
import logging
import socket
from typing import Any, Dict, Optional
from typing import Any
from pyvizio import VizioAsync, async_guess_device_type
from pyvizio.const import APP_HOME
@ -48,7 +50,7 @@ from .const import (
_LOGGER = logging.getLogger(__name__)
def _get_config_schema(input_dict: Dict[str, Any] = None) -> vol.Schema:
def _get_config_schema(input_dict: dict[str, Any] = None) -> vol.Schema:
"""
Return schema defaults for init step based on user input/config dict.
@ -76,7 +78,7 @@ def _get_config_schema(input_dict: Dict[str, Any] = None) -> vol.Schema:
)
def _get_pairing_schema(input_dict: Dict[str, Any] = None) -> vol.Schema:
def _get_pairing_schema(input_dict: dict[str, Any] = None) -> vol.Schema:
"""
Return schema defaults for pairing data based on user input.
@ -108,8 +110,8 @@ class VizioOptionsConfigFlow(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 vizio options."""
if user_input is not None:
if user_input.get(CONF_APPS_TO_INCLUDE_OR_EXCLUDE):
@ -191,7 +193,7 @@ class VizioConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self._data = None
self._apps = {}
async def _create_entry(self, input_dict: Dict[str, Any]) -> Dict[str, Any]:
async def _create_entry(self, input_dict: dict[str, Any]) -> dict[str, Any]:
"""Create vizio config entry."""
# Remove extra keys that will not be used by entry setup
input_dict.pop(CONF_APPS_TO_INCLUDE_OR_EXCLUDE, None)
@ -203,8 +205,8 @@ class VizioConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_create_entry(title=input_dict[CONF_NAME], data=input_dict)
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 a flow initialized by the user."""
errors = {}
@ -276,7 +278,7 @@ class VizioConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_show_form(step_id="user", data_schema=schema, errors=errors)
async def async_step_import(self, import_config: Dict[str, Any]) -> Dict[str, Any]:
async def async_step_import(self, import_config: dict[str, Any]) -> dict[str, Any]:
"""Import a config entry from configuration.yaml."""
# Check if new config entry matches any existing config entries
for entry in self.hass.config_entries.async_entries(DOMAIN):
@ -339,8 +341,8 @@ class VizioConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return await self.async_step_user(user_input=import_config)
async def async_step_zeroconf(
self, discovery_info: Optional[DiscoveryInfoType] = None
) -> Dict[str, Any]:
self, discovery_info: DiscoveryInfoType | None = None
) -> dict[str, Any]:
"""Handle zeroconf discovery."""
# If host already has port, no need to add it again
if ":" not in discovery_info[CONF_HOST]:
@ -376,8 +378,8 @@ class VizioConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return await self.async_step_user(user_input=discovery_info)
async def async_step_pair_tv(
self, user_input: Dict[str, Any] = None
) -> Dict[str, Any]:
self, user_input: dict[str, Any] = None
) -> dict[str, Any]:
"""
Start pairing process for TV.
@ -442,7 +444,7 @@ class VizioConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def _pairing_complete(self, step_id: str) -> Dict[str, Any]:
async def _pairing_complete(self, step_id: str) -> dict[str, Any]:
"""Handle config flow completion."""
if not self._must_show_form:
return await self._create_entry(self._data)
@ -455,8 +457,8 @@ class VizioConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
)
async def async_step_pairing_complete(
self, user_input: Dict[str, Any] = None
) -> Dict[str, Any]:
self, user_input: dict[str, Any] = None
) -> dict[str, Any]:
"""
Complete non-import sourced config flow.
@ -465,8 +467,8 @@ class VizioConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return await self._pairing_complete("pairing_complete")
async def async_step_pairing_complete_import(
self, user_input: Dict[str, Any] = None
) -> Dict[str, Any]:
self, user_input: dict[str, Any] = None
) -> dict[str, Any]:
"""
Complete import sourced config flow.

View file

@ -1,7 +1,9 @@
"""Vizio SmartCast Device support."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any, Callable, Dict, List, Optional, Union
from typing import Any, Callable
from pyvizio import VizioAsync
from pyvizio.api.apps import find_app_name
@ -64,7 +66,7 @@ PARALLEL_UPDATES = 0
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 Vizio media player entry."""
host = config_entry.data[CONF_HOST]
@ -166,7 +168,7 @@ class VizioDevice(MediaPlayerEntity):
self._model = None
self._sw_version = None
def _apps_list(self, apps: List[str]) -> List[str]:
def _apps_list(self, apps: list[str]) -> list[str]:
"""Return process apps list based on configured filters."""
if self._conf_apps.get(CONF_INCLUDE):
return [app for app in apps if app in self._conf_apps[CONF_INCLUDE]]
@ -274,7 +276,7 @@ class VizioDevice(MediaPlayerEntity):
if self._current_app == NO_APP_RUNNING:
self._current_app = None
def _get_additional_app_names(self) -> List[Dict[str, Any]]:
def _get_additional_app_names(self) -> list[dict[str, Any]]:
"""Return list of additional apps that were included in configuration.yaml."""
return [
additional_app["name"] for additional_app in self._additional_app_configs
@ -296,7 +298,7 @@ class VizioDevice(MediaPlayerEntity):
self._conf_apps.update(config_entry.options.get(CONF_APPS, {}))
async def async_update_setting(
self, setting_type: str, setting_name: str, new_value: Union[int, str]
self, setting_type: str, setting_name: str, new_value: int | str
) -> None:
"""Update a setting when update_setting service is called."""
await self._device.set_setting(
@ -340,7 +342,7 @@ class VizioDevice(MediaPlayerEntity):
return self._available
@property
def state(self) -> Optional[str]:
def state(self) -> str | None:
"""Return the state of the device."""
return self._state
@ -355,7 +357,7 @@ class VizioDevice(MediaPlayerEntity):
return self._icon
@property
def volume_level(self) -> Optional[float]:
def volume_level(self) -> float | None:
"""Return the volume level of the device."""
return self._volume_level
@ -365,7 +367,7 @@ class VizioDevice(MediaPlayerEntity):
return self._is_volume_muted
@property
def source(self) -> Optional[str]:
def source(self) -> str | None:
"""Return current input of the device."""
if self._current_app is not None and self._current_input in INPUT_APPS:
return self._current_app
@ -373,7 +375,7 @@ class VizioDevice(MediaPlayerEntity):
return self._current_input
@property
def source_list(self) -> List[str]:
def source_list(self) -> list[str]:
"""Return list of available inputs of the device."""
# If Smartcast app is in input list, and the app list has been retrieved,
# show the combination with , otherwise just return inputs
@ -395,7 +397,7 @@ class VizioDevice(MediaPlayerEntity):
return self._available_inputs
@property
def app_id(self) -> Optional[str]:
def app_id(self) -> str | None:
"""Return the ID of the current app if it is unknown by pyvizio."""
if self._current_app_config and self.app_name == UNKNOWN_APP:
return {
@ -407,7 +409,7 @@ class VizioDevice(MediaPlayerEntity):
return None
@property
def app_name(self) -> Optional[str]:
def app_name(self) -> str | None:
"""Return the friendly name of the current app."""
return self._current_app
@ -422,7 +424,7 @@ class VizioDevice(MediaPlayerEntity):
return self._config_entry.unique_id
@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.unique_id)},
@ -438,12 +440,12 @@ class VizioDevice(MediaPlayerEntity):
return self._device_class
@property
def sound_mode(self) -> Optional[str]:
def sound_mode(self) -> str | None:
"""Name of the current sound mode."""
return self._current_sound_mode
@property
def sound_mode_list(self) -> Optional[List[str]]:
def sound_mode_list(self) -> list[str] | None:
"""List of available sound modes."""
return self._available_sound_modes

View file

@ -1,6 +1,7 @@
"""Config flow for Volumio integration."""
from __future__ import annotations
import logging
from typing import Optional
from pyvolumio import CannotConnectError, Volumio
import voluptuous as vol
@ -39,10 +40,10 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
def __init__(self):
"""Initialize flow."""
self._host: Optional[str] = None
self._port: Optional[int] = None
self._name: Optional[str] = None
self._uuid: Optional[str] = None
self._host: str | None = None
self._port: int | None = None
self._name: str | None = None
self._uuid: str | None = None
@callback
def _async_get_entry(self):