Update typing 04 (#48037)

This commit is contained in:
Marc Mueller 2021-03-17 23:34:25 +01:00 committed by GitHub
parent 02619ca2cd
commit 76199c0eb2
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
44 changed files with 282 additions and 229 deletions

View file

@ -1,6 +1,7 @@
"""Config flow for Rollease Acmeda Automate Pulse Hub."""
from __future__ import annotations
import asyncio
from typing import Dict, Optional
import aiopulse
import async_timeout
@ -19,7 +20,7 @@ class AcmedaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
def __init__(self):
"""Initialize the config flow."""
self.discovered_hubs: Optional[Dict[str, aiopulse.Hub]] = None
self.discovered_hubs: dict[str, aiopulse.Hub] | None = None
async def async_step_user(self, user_input=None):
"""Handle a flow initialized by the user."""

View file

@ -1,6 +1,7 @@
"""Code to handle a Pulse Hub."""
from __future__ import annotations
import asyncio
from typing import Optional
import aiopulse
@ -17,7 +18,7 @@ class PulseHub:
"""Initialize the system."""
self.config_entry = config_entry
self.hass = hass
self.api: Optional[aiopulse.Hub] = None
self.api: aiopulse.Hub | None = None
self.tasks = []
self.current_rollers = {}
self.cleanup_callbacks = []

View file

@ -1,5 +1,5 @@
"""Provides device automations for Alarm control panel."""
from typing import List, Optional
from __future__ import annotations
import voluptuous as vol
@ -41,7 +41,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 Alarm control panel devices."""
registry = await entity_registry.async_get_registry(hass)
actions = []
@ -109,7 +109,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 Alarm control panel."""
from typing import Dict, List
from __future__ import annotations
import voluptuous as vol
@ -58,7 +58,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 Alarm control panel devices."""
registry = await entity_registry.async_get_registry(hass)
conditions = []

View file

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

View file

@ -1,7 +1,9 @@
"""Reproduce an Alarm control panel 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,
@ -39,8 +41,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)
@ -83,8 +85,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 Alarm control panel states."""
await asyncio.gather(

View file

@ -1,7 +1,9 @@
"""Reproduce an Alert state."""
from __future__ import annotations
import asyncio
import logging
from typing import Any, Dict, Iterable, Optional
from typing import Any, Iterable
from homeassistant.const import (
ATTR_ENTITY_ID,
@ -24,8 +26,8 @@ async def _async_reproduce_state(
hass: HomeAssistantType,
state: State,
*,
context: Optional[Context] = None,
reproduce_options: Optional[Dict[str, Any]] = None,
context: Context | None = None,
reproduce_options: dict[str, Any] | None = None,
) -> None:
"""Reproduce a single state."""
cur_state = hass.states.get(state.entity_id)
@ -61,8 +63,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 Alert states."""
# Reproduce states in parallel.

View file

@ -1,6 +1,7 @@
"""Alexa capabilities."""
from __future__ import annotations
import logging
from typing import List, Optional
from homeassistant.components import (
cover,
@ -72,7 +73,7 @@ class AlexaCapability:
supported_locales = {"en-US"}
def __init__(self, entity: State, instance: Optional[str] = None):
def __init__(self, entity: State, instance: str | None = None):
"""Initialize an Alexa capability."""
self.entity = entity
self.instance = instance
@ -82,7 +83,7 @@ class AlexaCapability:
raise NotImplementedError
@staticmethod
def properties_supported() -> List[dict]:
def properties_supported() -> list[dict]:
"""Return what properties this entity supports."""
return []

View file

@ -1,6 +1,8 @@
"""Alexa entity adapters."""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, List
from typing import TYPE_CHECKING
from homeassistant.components import (
alarm_control_panel,
@ -300,7 +302,7 @@ class AlexaEntity:
Raises _UnsupportedInterface.
"""
def interfaces(self) -> List[AlexaCapability]:
def interfaces(self) -> list[AlexaCapability]:
"""Return a list of supported interfaces.
Used for discovery. The list should contain AlexaInterface instances.
@ -353,7 +355,7 @@ class AlexaEntity:
@callback
def async_get_entities(hass, config) -> List[AlexaEntity]:
def async_get_entities(hass, config) -> list[AlexaEntity]:
"""Return all entities that are supported by Alexa."""
entities = []
for state in hass.states.async_all():

View file

@ -1,8 +1,9 @@
"""Alexa state report code."""
from __future__ import annotations
import asyncio
import json
import logging
from typing import Optional
import aiohttp
import async_timeout
@ -45,8 +46,8 @@ async def async_enable_proactive_mode(hass, smart_home_config):
async def async_entity_state_listener(
changed_entity: str,
old_state: Optional[State],
new_state: Optional[State],
old_state: State | None,
new_state: State | None,
):
if not hass.is_running:
return

View file

@ -1,9 +1,10 @@
"""Support for Almond."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
import time
from typing import Optional
from aiohttp import ClientError, ClientSession
import async_timeout
@ -281,7 +282,7 @@ class AlmondAgent(conversation.AbstractConversationAgent):
return True
async def async_process(
self, text: str, context: Context, conversation_id: Optional[str] = None
self, text: str, context: Context, conversation_id: str | None = None
) -> intent.IntentResponse:
"""Process a sentence."""
response = await self.api.async_converse_text(text, conversation_id)

View file

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

View file

@ -1,5 +1,5 @@
"""Support for Arris TG2492LG router."""
from typing import List
from __future__ import annotations
from arris_tg2492lg import ConnectBox, Device
import voluptuous as vol
@ -36,7 +36,7 @@ class ArrisDeviceScanner(DeviceScanner):
def __init__(self, connect_box: ConnectBox):
"""Initialize the scanner."""
self.connect_box = connect_box
self.last_results: List[Device] = []
self.last_results: list[Device] = []
def scan_devices(self):
"""Scan for new devices and return a list with found device IDs."""

View file

@ -1,5 +1,5 @@
"""Support for ASUSWRT routers."""
from typing import Dict
from __future__ import annotations
from homeassistant.components.device_tracker import SOURCE_TYPE_ROUTER
from homeassistant.components.device_tracker.config_entry import ScannerEntity
@ -103,12 +103,12 @@ class AsusWrtDevice(ScannerEntity):
return self._icon
@property
def extra_state_attributes(self) -> Dict[str, any]:
def extra_state_attributes(self) -> dict[str, any]:
"""Return the attributes."""
return self._attrs
@property
def device_info(self) -> Dict[str, any]:
def device_info(self) -> dict[str, any]:
"""Return the device information."""
return {
"connections": {(CONNECTION_NETWORK_MAC, self._mac)},

View file

@ -1,7 +1,9 @@
"""Represent the AsusWrt router."""
from __future__ import annotations
from datetime import datetime, timedelta
import logging
from typing import Any, Dict, Optional
from typing import Any
from aioasuswrt.asuswrt import AsusWrt
@ -74,7 +76,7 @@ class AsusWrtSensorDataHandler:
async def _get_bytes(self):
"""Fetch byte information from the router."""
ret_dict: Dict[str, Any] = {}
ret_dict: dict[str, Any] = {}
try:
datas = await self._api.async_get_bytes_total()
except OSError as exc:
@ -87,7 +89,7 @@ class AsusWrtSensorDataHandler:
async def _get_rates(self):
"""Fetch rates information from the router."""
ret_dict: Dict[str, Any] = {}
ret_dict: dict[str, Any] = {}
try:
rates = await self._api.async_get_current_transfer_rates()
except OSError as exc:
@ -194,12 +196,12 @@ class AsusWrtRouter:
self._protocol = entry.data[CONF_PROTOCOL]
self._host = entry.data[CONF_HOST]
self._devices: Dict[str, Any] = {}
self._devices: dict[str, Any] = {}
self._connected_devices = 0
self._connect_error = False
self._sensors_data_handler: AsusWrtSensorDataHandler = None
self._sensors_coordinator: Dict[str, Any] = {}
self._sensors_coordinator: dict[str, Any] = {}
self._on_close = []
@ -245,7 +247,7 @@ class AsusWrtRouter:
async_track_time_interval(self.hass, self.update_all, SCAN_INTERVAL)
)
async def update_all(self, now: Optional[datetime] = None) -> None:
async def update_all(self, now: datetime | None = None) -> None:
"""Update all AsusWrt platforms."""
await self.update_devices()
@ -353,7 +355,7 @@ class AsusWrtRouter:
"""Add a function to call when router is closed."""
self._on_close.append(func)
def update_options(self, new_options: Dict) -> bool:
def update_options(self, new_options: dict) -> bool:
"""Update router options."""
req_reload = False
for name, new_opt in new_options.items():
@ -367,7 +369,7 @@ class AsusWrtRouter:
return req_reload
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return the device information."""
return {
"identifiers": {(DOMAIN, "AsusWRT")},
@ -392,12 +394,12 @@ class AsusWrtRouter:
return self._host
@property
def devices(self) -> Dict[str, Any]:
def devices(self) -> dict[str, Any]:
"""Return devices."""
return self._devices
@property
def sensors_coordinator(self) -> Dict[str, Any]:
def sensors_coordinator(self) -> dict[str, Any]:
"""Return sensors coordinators."""
return self._sensors_coordinator
@ -407,7 +409,7 @@ class AsusWrtRouter:
return self._api
def get_api(conf: Dict, options: Optional[Dict] = None) -> AsusWrt:
def get_api(conf: dict, options: dict | None = None) -> AsusWrt:
"""Get the AsusWrt API."""
opt = options or {}

View file

@ -1,7 +1,8 @@
"""Asuswrt status sensors."""
from __future__ import annotations
import logging
from numbers import Number
from typing import Dict
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import DATA_GIGABYTES, DATA_RATE_MEGABITS_PER_SECOND
@ -104,7 +105,7 @@ class AsusWrtSensor(CoordinatorEntity):
coordinator: DataUpdateCoordinator,
router: AsusWrtRouter,
sensor_type: str,
sensor: Dict[str, any],
sensor: dict[str, any],
) -> None:
"""Initialize a AsusWrt sensor."""
super().__init__(coordinator)
@ -159,11 +160,11 @@ class AsusWrtSensor(CoordinatorEntity):
return self._device_class
@property
def extra_state_attributes(self) -> Dict[str, any]:
def extra_state_attributes(self) -> dict[str, any]:
"""Return the attributes."""
return {"hostname": self._router.host}
@property
def device_info(self) -> Dict[str, any]:
def device_info(self) -> dict[str, any]:
"""Return the device information."""
return self._router.device_info

View file

@ -1,5 +1,5 @@
"""Initialization of ATAG One climate platform."""
from typing import List, Optional
from __future__ import annotations
from homeassistant.components.climate import ClimateEntity
from homeassistant.components.climate.const import (
@ -43,46 +43,46 @@ class AtagThermostat(AtagEntity, ClimateEntity):
return SUPPORT_FLAGS
@property
def hvac_mode(self) -> Optional[str]:
def hvac_mode(self) -> str | None:
"""Return hvac operation ie. heat, cool mode."""
if self.coordinator.data.climate.hvac_mode in HVAC_MODES:
return self.coordinator.data.climate.hvac_mode
return None
@property
def hvac_modes(self) -> List[str]:
def hvac_modes(self) -> list[str]:
"""Return the list of available hvac operation modes."""
return HVAC_MODES
@property
def hvac_action(self) -> Optional[str]:
def hvac_action(self) -> str | None:
"""Return the current running hvac operation."""
is_active = self.coordinator.data.climate.status
return CURRENT_HVAC_HEAT if is_active else CURRENT_HVAC_IDLE
@property
def temperature_unit(self) -> Optional[str]:
def temperature_unit(self) -> str | None:
"""Return the unit of measurement."""
return self.coordinator.data.climate.temp_unit
@property
def current_temperature(self) -> Optional[float]:
def current_temperature(self) -> float | None:
"""Return the current temperature."""
return self.coordinator.data.climate.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.climate.target_temperature
@property
def preset_mode(self) -> Optional[str]:
def preset_mode(self) -> str | None:
"""Return the current preset mode, e.g., auto, manual, fireplace, extend, etc."""
preset = self.coordinator.data.climate.preset_mode
return PRESET_INVERTED.get(preset)
@property
def preset_modes(self) -> Optional[List[str]]:
def preset_modes(self) -> list[str] | None:
"""Return a list of available preset modes."""
return list(PRESET_MAP.keys())

View file

@ -114,8 +114,9 @@ Result will be a long-lived access token:
}
"""
from __future__ import annotations
from datetime import timedelta
from typing import Union
import uuid
from aiohttp import web
@ -183,7 +184,7 @@ RESULT_TYPE_USER = "user"
@bind_hass
def create_auth_code(
hass, client_id: str, credential_or_user: Union[Credentials, User]
hass, client_id: str, credential_or_user: Credentials | User
) -> str:
"""Create an authorization code to fetch tokens."""
return hass.data[DOMAIN](client_id, credential_or_user)

View file

@ -1,6 +1,8 @@
"""Allow to set up simple automation rules via the config file."""
from __future__ import annotations
import logging
from typing import Any, Awaitable, Callable, Dict, List, Optional, Set, Union, cast
from typing import Any, Awaitable, Callable, Dict, cast
import voluptuous as vol
from voluptuous.humanize import humanize_error
@ -109,7 +111,7 @@ def is_on(hass, entity_id):
@callback
def automations_with_entity(hass: HomeAssistant, entity_id: str) -> List[str]:
def automations_with_entity(hass: HomeAssistant, entity_id: str) -> list[str]:
"""Return all automations that reference the entity."""
if DOMAIN not in hass.data:
return []
@ -124,7 +126,7 @@ def automations_with_entity(hass: HomeAssistant, entity_id: str) -> List[str]:
@callback
def entities_in_automation(hass: HomeAssistant, entity_id: str) -> List[str]:
def entities_in_automation(hass: HomeAssistant, entity_id: str) -> list[str]:
"""Return all entities in a scene."""
if DOMAIN not in hass.data:
return []
@ -140,7 +142,7 @@ def entities_in_automation(hass: HomeAssistant, entity_id: str) -> List[str]:
@callback
def automations_with_device(hass: HomeAssistant, device_id: str) -> List[str]:
def automations_with_device(hass: HomeAssistant, device_id: str) -> list[str]:
"""Return all automations that reference the device."""
if DOMAIN not in hass.data:
return []
@ -155,7 +157,7 @@ def automations_with_device(hass: HomeAssistant, device_id: str) -> List[str]:
@callback
def devices_in_automation(hass: HomeAssistant, entity_id: str) -> List[str]:
def devices_in_automation(hass: HomeAssistant, entity_id: str) -> list[str]:
"""Return all devices in a scene."""
if DOMAIN not in hass.data:
return []
@ -249,8 +251,8 @@ class AutomationEntity(ToggleEntity, RestoreEntity):
self.action_script.change_listener = self.async_write_ha_state
self._initial_state = initial_state
self._is_enabled = False
self._referenced_entities: Optional[Set[str]] = None
self._referenced_devices: Optional[Set[str]] = None
self._referenced_entities: set[str] | None = None
self._referenced_devices: set[str] | None = None
self._logger = LOGGER
self._variables: ScriptVariables = variables
self._trigger_variables: ScriptVariables = trigger_variables
@ -509,7 +511,7 @@ class AutomationEntity(ToggleEntity, RestoreEntity):
async def _async_attach_triggers(
self, home_assistant_start: bool
) -> Optional[Callable[[], None]]:
) -> Callable[[], None] | None:
"""Set up the triggers."""
def log_cb(level, msg, **kwargs):
@ -539,7 +541,7 @@ class AutomationEntity(ToggleEntity, RestoreEntity):
async def _async_process_config(
hass: HomeAssistant,
config: Dict[str, Any],
config: dict[str, Any],
component: EntityComponent,
) -> bool:
"""Process config and add automations.
@ -550,7 +552,7 @@ async def _async_process_config(
blueprints_used = False
for config_key in extract_domain_configs(config, DOMAIN):
conf: List[Union[Dict[str, Any], blueprint.BlueprintInputs]] = config[ # type: ignore
conf: list[dict[str, Any] | blueprint.BlueprintInputs] = config[ # type: ignore
config_key
]
@ -680,7 +682,7 @@ async def _async_process_if(hass, name, config, p_config):
@callback
def _trigger_extract_device(trigger_conf: dict) -> Optional[str]:
def _trigger_extract_device(trigger_conf: dict) -> str | None:
"""Extract devices from a trigger config."""
if trigger_conf[CONF_PLATFORM] != "device":
return None
@ -689,7 +691,7 @@ def _trigger_extract_device(trigger_conf: dict) -> Optional[str]:
@callback
def _trigger_extract_entities(trigger_conf: dict) -> List[str]:
def _trigger_extract_entities(trigger_conf: dict) -> list[str]:
"""Extract entities from a trigger config."""
if trigger_conf[CONF_PLATFORM] in ("state", "numeric_state"):
return trigger_conf[CONF_ENTITY_ID]

View file

@ -1,7 +1,9 @@
"""Reproduce an Automation state."""
from __future__ import annotations
import asyncio
import logging
from typing import Any, Dict, Iterable, Optional
from typing import Any, Iterable
from homeassistant.const import (
ATTR_ENTITY_ID,
@ -24,8 +26,8 @@ async def _async_reproduce_state(
hass: HomeAssistantType,
state: State,
*,
context: Optional[Context] = None,
reproduce_options: Optional[Dict[str, Any]] = None,
context: Context | None = None,
reproduce_options: dict[str, Any] | None = None,
) -> None:
"""Reproduce a single state."""
cur_state = hass.states.get(state.entity_id)
@ -60,8 +62,8 @@ async def async_reproduce_states(
hass: HomeAssistantType,
states: Iterable[State],
*,
context: Optional[Context] = None,
reproduce_options: Optional[Dict[str, Any]] = None,
context: Context | None = None,
reproduce_options: dict[str, Any] | None = None,
) -> None:
"""Reproduce Automation states."""
await asyncio.gather(

View file

@ -1,11 +1,13 @@
"""Trace support for automation."""
from __future__ import annotations
from collections import OrderedDict
from contextlib import contextmanager
import datetime as dt
from datetime import timedelta
from itertools import count
import logging
from typing import Any, Awaitable, Callable, Deque, Dict, Optional
from typing import Any, Awaitable, Callable, Deque
from homeassistant.core import Context, HomeAssistant, callback
from homeassistant.helpers.json import JSONEncoder as HAJSONEncoder
@ -30,28 +32,28 @@ class AutomationTrace:
def __init__(
self,
unique_id: Optional[str],
config: Dict[str, Any],
unique_id: str | None,
config: dict[str, Any],
context: Context,
):
"""Container for automation trace."""
self._action_trace: Optional[Dict[str, Deque[TraceElement]]] = None
self._condition_trace: Optional[Dict[str, Deque[TraceElement]]] = None
self._config: Dict[str, Any] = config
self._action_trace: dict[str, Deque[TraceElement]] | None = None
self._condition_trace: dict[str, Deque[TraceElement]] | None = None
self._config: dict[str, Any] = config
self.context: Context = context
self._error: Optional[Exception] = None
self._error: Exception | None = None
self._state: str = "running"
self.run_id: str = str(next(self._run_ids))
self._timestamp_finish: Optional[dt.datetime] = None
self._timestamp_finish: dt.datetime | None = None
self._timestamp_start: dt.datetime = dt_util.utcnow()
self._unique_id: Optional[str] = unique_id
self._variables: Optional[Dict[str, Any]] = None
self._unique_id: str | None = unique_id
self._variables: dict[str, Any] | None = None
def set_action_trace(self, trace: Dict[str, Deque[TraceElement]]) -> None:
def set_action_trace(self, trace: dict[str, Deque[TraceElement]]) -> None:
"""Set action trace."""
self._action_trace = trace
def set_condition_trace(self, trace: Dict[str, Deque[TraceElement]]) -> None:
def set_condition_trace(self, trace: dict[str, Deque[TraceElement]]) -> None:
"""Set condition trace."""
self._condition_trace = trace
@ -59,7 +61,7 @@ class AutomationTrace:
"""Set error."""
self._error = ex
def set_variables(self, variables: Dict[str, Any]) -> None:
def set_variables(self, variables: dict[str, Any]) -> None:
"""Set variables."""
self._variables = variables
@ -68,7 +70,7 @@ class AutomationTrace:
self._timestamp_finish = dt_util.utcnow()
self._state = "stopped"
def as_dict(self) -> Dict[str, Any]:
def as_dict(self) -> dict[str, Any]:
"""Return dictionary version of this AutomationTrace."""
result = self.as_short_dict()
@ -96,7 +98,7 @@ class AutomationTrace:
result["error"] = str(self._error)
return result
def as_short_dict(self) -> Dict[str, Any]:
def as_short_dict(self) -> dict[str, Any]:
"""Return a brief dictionary version of this AutomationTrace."""
last_action = None

View file

@ -1,7 +1,8 @@
"""The awair component."""
from __future__ import annotations
from asyncio import gather
from typing import Any, Optional
from typing import Any
from async_timeout import timeout
from python_awair import Awair
@ -70,7 +71,7 @@ class AwairDataUpdateCoordinator(DataUpdateCoordinator):
super().__init__(hass, LOGGER, name=DOMAIN, update_interval=UPDATE_INTERVAL)
async def _async_update_data(self) -> Optional[Any]:
async def _async_update_data(self) -> Any | None:
"""Update data via Awair client library."""
with timeout(API_TIMEOUT):
try:

View file

@ -1,6 +1,5 @@
"""Config flow for Awair."""
from typing import Optional
from __future__ import annotations
from python_awair import Awair
from python_awair.exceptions import AuthError, AwairError
@ -36,7 +35,7 @@ class AwairFlowHandler(ConfigFlow, domain=DOMAIN):
data={CONF_ACCESS_TOKEN: conf[CONF_ACCESS_TOKEN]},
)
async def async_step_user(self, user_input: Optional[dict] = None):
async def async_step_user(self, user_input: dict | None = None):
"""Handle a flow initialized by the user."""
errors = {}
@ -61,7 +60,7 @@ class AwairFlowHandler(ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def async_step_reauth(self, user_input: Optional[dict] = None):
async def async_step_reauth(self, user_input: dict | None = None):
"""Handle re-auth if token invalid."""
errors = {}

View file

@ -1,6 +1,7 @@
"""Support for Awair sensors."""
from __future__ import annotations
from typing import Callable, List, Optional
from typing import Callable
from python_awair.devices import AwairDevice
import voluptuous as vol
@ -55,13 +56,13 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
async def async_setup_entry(
hass: HomeAssistantType,
config_entry: ConfigType,
async_add_entities: Callable[[List[Entity], bool], None],
async_add_entities: Callable[[list[Entity], bool], None],
):
"""Set up Awair sensor entity based on a config entry."""
coordinator = hass.data[DOMAIN][config_entry.entry_id]
sensors = []
data: List[AwairResult] = coordinator.data.values()
data: list[AwairResult] = coordinator.data.values()
for result in data:
if result.air_data:
sensors.append(AwairSensor(API_SCORE, result.device, coordinator))
@ -228,9 +229,9 @@ class AwairSensor(CoordinatorEntity):
return info
@property
def _air_data(self) -> Optional[AwairResult]:
def _air_data(self) -> AwairResult | None:
"""Return the latest data for our device, or None."""
result: Optional[AwairResult] = self.coordinator.data.get(self._device.uuid)
result: AwairResult | None = self.coordinator.data.get(self._device.uuid)
if result:
return result.air_data

View file

@ -1,6 +1,8 @@
"""Support for Azure DevOps."""
from __future__ import annotations
import logging
from typing import Any, Dict
from typing import Any
from aioazuredevops.client import DevOpsClient
import aiohttp
@ -114,7 +116,7 @@ class AzureDevOpsDeviceEntity(AzureDevOpsEntity):
"""Defines a Azure DevOps device entity."""
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return device information about this Azure DevOps instance."""
return {
"identifiers": {

View file

@ -1,7 +1,8 @@
"""Support for Azure DevOps sensors."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import List
from aioazuredevops.builds import DevOpsBuild
from aioazuredevops.client import DevOpsClient
@ -39,7 +40,7 @@ async def async_setup_entry(
sensors = []
try:
builds: List[DevOpsBuild] = await client.get_builds(
builds: list[DevOpsBuild] = await client.get_builds(
organization, project, BUILDS_QUERY
)
except aiohttp.ClientError as exception:

View file

@ -1,9 +1,11 @@
"""Support for Azure Event Hubs."""
from __future__ import annotations
import asyncio
import json
import logging
import time
from typing import Any, Dict
from typing import Any
from azure.eventhub import EventData
from azure.eventhub.aio import EventHubProducerClient, EventHubSharedKeyCredential
@ -95,7 +97,7 @@ class AzureEventHub:
def __init__(
self,
hass: HomeAssistant,
client_args: Dict[str, Any],
client_args: dict[str, Any],
conn_str_client: bool,
entities_filter: vol.Schema,
send_interval: int,

View file

@ -1,8 +1,9 @@
"""Support for French FAI Bouygues Bbox routers."""
from __future__ import annotations
from collections import namedtuple
from datetime import timedelta
import logging
from typing import List
import pybbox
import voluptuous as vol
@ -47,7 +48,7 @@ class BboxDeviceScanner(DeviceScanner):
self.host = config[CONF_HOST]
"""Initialize the scanner."""
self.last_results: List[Device] = []
self.last_results: list[Device] = []
self.success_init = self._update_info()
_LOGGER.info("Scanner initialized")

View file

@ -1,5 +1,5 @@
"""Implement device conditions for binary sensor."""
from typing import Dict, List
from __future__ import annotations
import voluptuous as vol
@ -205,9 +205,9 @@ CONDITION_SCHEMA = cv.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."""
conditions: List[Dict[str, str]] = []
conditions: list[dict[str, str]] = []
entity_registry = await async_get_registry(hass)
entries = [
entry

View file

@ -1,5 +1,7 @@
"""Helper to test significant Binary Sensor state changes."""
from typing import Any, Optional
from __future__ import annotations
from typing import Any
from homeassistant.core import HomeAssistant, callback
@ -12,7 +14,7 @@ def async_check_significant_change(
new_state: str,
new_attrs: dict,
**kwargs: Any,
) -> Optional[bool]:
) -> bool | None:
"""Test if state significantly changed."""
if old_state != new_state:
return True

View file

@ -1,8 +1,9 @@
"""Import logic for blueprint."""
from __future__ import annotations
from dataclasses import dataclass
import html
import re
from typing import Optional
import voluptuous as vol
import yarl
@ -93,7 +94,7 @@ def _get_community_post_import_url(url: str) -> str:
def _extract_blueprint_from_community_topic(
url: str,
topic: dict,
) -> Optional[ImportedBlueprint]:
) -> ImportedBlueprint | None:
"""Extract a blueprint from a community post JSON.
Async friendly.
@ -136,7 +137,7 @@ def _extract_blueprint_from_community_topic(
async def fetch_blueprint_from_community_post(
hass: HomeAssistant, url: str
) -> Optional[ImportedBlueprint]:
) -> ImportedBlueprint | None:
"""Get blueprints from a community post url.
Method can raise aiohttp client exceptions, vol.Invalid.

View file

@ -1,9 +1,11 @@
"""Blueprint models."""
from __future__ import annotations
import asyncio
import logging
import pathlib
import shutil
from typing import Any, Dict, List, Optional, Union
from typing import Any
from awesomeversion import AwesomeVersion
import voluptuous as vol
@ -49,8 +51,8 @@ class Blueprint:
self,
data: dict,
*,
path: Optional[str] = None,
expected_domain: Optional[str] = None,
path: str | None = None,
expected_domain: str | None = None,
) -> None:
"""Initialize a blueprint."""
try:
@ -95,7 +97,7 @@ class Blueprint:
"""Return blueprint metadata."""
return self.data[CONF_BLUEPRINT]
def update_metadata(self, *, source_url: Optional[str] = None) -> None:
def update_metadata(self, *, source_url: str | None = None) -> None:
"""Update metadata."""
if source_url is not None:
self.data[CONF_BLUEPRINT][CONF_SOURCE_URL] = source_url
@ -105,7 +107,7 @@ class Blueprint:
return yaml.dump(self.data)
@callback
def validate(self) -> Optional[List[str]]:
def validate(self) -> list[str] | None:
"""Test if the Home Assistant installation supports this blueprint.
Return list of errors if not valid.
@ -126,7 +128,7 @@ class BlueprintInputs:
"""Inputs for a blueprint."""
def __init__(
self, blueprint: Blueprint, config_with_inputs: Dict[str, Any]
self, blueprint: Blueprint, config_with_inputs: dict[str, Any]
) -> None:
"""Instantiate a blueprint inputs object."""
self.blueprint = blueprint
@ -218,7 +220,7 @@ class DomainBlueprints:
blueprint_data, expected_domain=self.domain, path=blueprint_path
)
def _load_blueprints(self) -> Dict[str, Union[Blueprint, BlueprintException]]:
def _load_blueprints(self) -> dict[str, Blueprint | BlueprintException]:
"""Load all the blueprints."""
blueprint_folder = pathlib.Path(
self.hass.config.path(BLUEPRINT_FOLDER, self.domain)
@ -243,7 +245,7 @@ class DomainBlueprints:
async def async_get_blueprints(
self,
) -> Dict[str, Union[Blueprint, BlueprintException]]:
) -> dict[str, Blueprint | BlueprintException]:
"""Get all the blueprints."""
async with self._load_lock:
return await self.hass.async_add_executor_job(self._load_blueprints)

View file

@ -1,5 +1,5 @@
"""Websocket API for blueprint."""
from typing import Dict, Optional
from __future__ import annotations
import async_timeout
import voluptuous as vol
@ -33,7 +33,7 @@ def async_setup(hass: HomeAssistant):
)
async def ws_list_blueprints(hass, connection, msg):
"""List available blueprints."""
domain_blueprints: Optional[Dict[str, models.DomainBlueprints]] = hass.data.get(
domain_blueprints: dict[str, models.DomainBlueprints] | None = hass.data.get(
DOMAIN, {}
)
results = {}
@ -102,7 +102,7 @@ async def ws_save_blueprint(hass, connection, msg):
path = msg["path"]
domain = msg["domain"]
domain_blueprints: Optional[Dict[str, models.DomainBlueprints]] = hass.data.get(
domain_blueprints: dict[str, models.DomainBlueprints] | None = hass.data.get(
DOMAIN, {}
)
@ -149,7 +149,7 @@ async def ws_delete_blueprint(hass, connection, msg):
path = msg["path"]
domain = msg["domain"]
domain_blueprints: Optional[Dict[str, models.DomainBlueprints]] = hass.data.get(
domain_blueprints: dict[str, models.DomainBlueprints] | None = hass.data.get(
DOMAIN, {}
)

View file

@ -1,7 +1,8 @@
"""Tracking for bluetooth devices."""
from __future__ import annotations
import asyncio
import logging
from typing import List, Optional, Set, Tuple
import bluetooth # pylint: disable=import-error
from bt_proximity import BluetoothRSSI
@ -50,7 +51,7 @@ def is_bluetooth_device(device) -> bool:
return device.mac and device.mac[:3].upper() == BT_PREFIX
def discover_devices(device_id: int) -> List[Tuple[str, str]]:
def discover_devices(device_id: int) -> list[tuple[str, str]]:
"""Discover Bluetooth devices."""
result = bluetooth.discover_devices(
duration=8,
@ -79,7 +80,7 @@ async def see_device(
)
async def get_tracking_devices(hass: HomeAssistantType) -> Tuple[Set[str], Set[str]]:
async def get_tracking_devices(hass: HomeAssistantType) -> tuple[set[str], set[str]]:
"""
Load all known devices.
@ -90,17 +91,17 @@ async def get_tracking_devices(hass: HomeAssistantType) -> Tuple[Set[str], Set[s
devices = await async_load_config(yaml_path, hass, 0)
bluetooth_devices = [device for device in devices if is_bluetooth_device(device)]
devices_to_track: Set[str] = {
devices_to_track: set[str] = {
device.mac[3:] for device in bluetooth_devices if device.track
}
devices_to_not_track: Set[str] = {
devices_to_not_track: set[str] = {
device.mac[3:] for device in bluetooth_devices if not device.track
}
return devices_to_track, devices_to_not_track
def lookup_name(mac: str) -> Optional[str]:
def lookup_name(mac: str) -> str | None:
"""Lookup a Bluetooth device name."""
_LOGGER.debug("Scanning %s", mac)
return bluetooth.lookup_name(mac, timeout=5)

View file

@ -1,6 +1,8 @@
"""Config flow for Bond integration."""
from __future__ import annotations
import logging
from typing import Any, Dict, Optional, Tuple
from typing import Any
from aiohttp import ClientConnectionError, ClientResponseError
from bond_api import Bond
@ -28,7 +30,7 @@ DISCOVERY_SCHEMA = vol.Schema({vol.Required(CONF_ACCESS_TOKEN): str})
TOKEN_SCHEMA = vol.Schema({})
async def _validate_input(data: Dict[str, Any]) -> Tuple[str, str]:
async def _validate_input(data: dict[str, Any]) -> tuple[str, str]:
"""Validate the user input allows us to connect."""
bond = Bond(data[CONF_HOST], data[CONF_ACCESS_TOKEN])
@ -60,7 +62,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
def __init__(self) -> None:
"""Initialize config flow."""
self._discovered: Dict[str, str] = {}
self._discovered: dict[str, str] = {}
async def _async_try_automatic_configure(self) -> None:
"""Try to auto configure the device.
@ -83,7 +85,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
_, hub_name = await _validate_input(self._discovered)
self._discovered[CONF_NAME] = hub_name
async def async_step_zeroconf(self, discovery_info: DiscoveryInfoType) -> Dict[str, Any]: # type: ignore
async def async_step_zeroconf(self, discovery_info: DiscoveryInfoType) -> dict[str, Any]: # type: ignore
"""Handle a flow initialized by zeroconf discovery."""
name: str = discovery_info[CONF_NAME]
host: str = discovery_info[CONF_HOST]
@ -106,8 +108,8 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return await self.async_step_confirm()
async def async_step_confirm(
self, user_input: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, user_input: dict[str, Any] | None = None
) -> dict[str, Any]:
"""Handle confirmation flow for discovered bond hub."""
errors = {}
if user_input is not None:
@ -147,8 +149,8 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
)
async def async_step_user(
self, user_input: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, user_input: dict[str, Any] | None = None
) -> dict[str, Any]:
"""Handle a flow initialized by the user."""
errors = {}
if user_input is not None:

View file

@ -1,5 +1,7 @@
"""Support for Bond covers."""
from typing import Any, Callable, List, Optional
from __future__ import annotations
from typing import Any, Callable
from bond_api import Action, BPUPSubscriptions, DeviceType
@ -16,14 +18,14 @@ from .utils import BondDevice, BondHub
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 Bond cover devices."""
data = hass.data[DOMAIN][entry.entry_id]
hub: BondHub = data[HUB]
bpup_subs: BPUPSubscriptions = data[BPUP_SUBS]
covers: List[Entity] = [
covers: list[Entity] = [
BondCover(hub, device, bpup_subs)
for device in hub.devices
if device.type == DeviceType.MOTORIZED_SHADES
@ -41,19 +43,19 @@ class BondCover(BondEntity, CoverEntity):
"""Create HA entity representing Bond cover."""
super().__init__(hub, device, bpup_subs)
self._closed: Optional[bool] = None
self._closed: bool | None = None
def _apply_state(self, state: dict) -> None:
cover_open = state.get("open")
self._closed = True if cover_open == 0 else False if cover_open == 1 else None
@property
def device_class(self) -> Optional[str]:
def device_class(self) -> str | None:
"""Get device class."""
return DEVICE_CLASS_SHADE
@property
def is_closed(self) -> Optional[bool]:
def is_closed(self) -> bool | None:
"""Return if the cover is closed or not."""
return self._closed

View file

@ -1,9 +1,11 @@
"""An abstract class common to all Bond entities."""
from __future__ import annotations
from abc import abstractmethod
from asyncio import Lock, TimeoutError as AsyncIOTimeoutError
from datetime import timedelta
import logging
from typing import Any, Dict, Optional
from typing import Any
from aiohttp import ClientError
from bond_api import BPUPSubscriptions
@ -29,7 +31,7 @@ class BondEntity(Entity):
hub: BondHub,
device: BondDevice,
bpup_subs: BPUPSubscriptions,
sub_device: Optional[str] = None,
sub_device: str | None = None,
):
"""Initialize entity with API and device info."""
self._hub = hub
@ -38,11 +40,11 @@ class BondEntity(Entity):
self._sub_device = sub_device
self._available = True
self._bpup_subs = bpup_subs
self._update_lock: Optional[Lock] = None
self._update_lock: Lock | None = None
self._initialized = False
@property
def unique_id(self) -> Optional[str]:
def unique_id(self) -> str | None:
"""Get unique ID for the entity."""
hub_id = self._hub.bond_id
device_id = self._device_id
@ -50,7 +52,7 @@ class BondEntity(Entity):
return f"{hub_id}_{device_id}{sub_device_id}"
@property
def name(self) -> Optional[str]:
def name(self) -> str | None:
"""Get entity name."""
if self._sub_device:
sub_device_name = self._sub_device.replace("_", " ").title()
@ -63,7 +65,7 @@ class BondEntity(Entity):
return False
@property
def device_info(self) -> Optional[Dict[str, Any]]:
def device_info(self) -> dict[str, Any] | None:
"""Get a an HA device representing this Bond controlled device."""
device_info = {
ATTR_NAME: self.name,

View file

@ -1,7 +1,9 @@
"""Support for Bond fans."""
from __future__ import annotations
import logging
import math
from typing import Any, Callable, List, Optional, Tuple
from typing import Any, Callable
from bond_api import Action, BPUPSubscriptions, DeviceType, Direction
@ -31,14 +33,14 @@ _LOGGER = logging.getLogger(__name__)
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 Bond fan devices."""
data = hass.data[DOMAIN][entry.entry_id]
hub: BondHub = data[HUB]
bpup_subs: BPUPSubscriptions = data[BPUP_SUBS]
fans: List[Entity] = [
fans: list[Entity] = [
BondFan(hub, device, bpup_subs)
for device in hub.devices
if DeviceType.is_fan(device.type)
@ -54,9 +56,9 @@ class BondFan(BondEntity, FanEntity):
"""Create HA entity representing Bond fan."""
super().__init__(hub, device, bpup_subs)
self._power: Optional[bool] = None
self._speed: Optional[int] = None
self._direction: Optional[int] = None
self._power: bool | None = None
self._speed: int | None = None
self._direction: int | None = None
def _apply_state(self, state: dict) -> None:
self._power = state.get("power")
@ -75,7 +77,7 @@ class BondFan(BondEntity, FanEntity):
return features
@property
def _speed_range(self) -> Tuple[int, int]:
def _speed_range(self) -> tuple[int, int]:
"""Return the range of speeds."""
return (1, self._device.props.get("max_speed", 3))
@ -92,7 +94,7 @@ class BondFan(BondEntity, FanEntity):
return int_states_in_range(self._speed_range)
@property
def current_direction(self) -> Optional[str]:
def current_direction(self) -> str | None:
"""Return fan rotation direction."""
direction = None
if self._direction == Direction.FORWARD:
@ -125,9 +127,9 @@ class BondFan(BondEntity, FanEntity):
async def async_turn_on(
self,
speed: Optional[str] = None,
percentage: Optional[int] = None,
preset_mode: Optional[str] = None,
speed: str | None = None,
percentage: int | None = None,
preset_mode: str | None = None,
**kwargs: Any,
) -> None:
"""Turn on the fan."""

View file

@ -1,6 +1,8 @@
"""Support for Bond lights."""
from __future__ import annotations
import logging
from typing import Any, Callable, List, Optional
from typing import Any, Callable
from bond_api import Action, BPUPSubscriptions, DeviceType
@ -24,14 +26,14 @@ _LOGGER = logging.getLogger(__name__)
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 Bond light devices."""
data = hass.data[DOMAIN][entry.entry_id]
hub: BondHub = data[HUB]
bpup_subs: BPUPSubscriptions = data[BPUP_SUBS]
fan_lights: List[Entity] = [
fan_lights: list[Entity] = [
BondLight(hub, device, bpup_subs)
for device in hub.devices
if DeviceType.is_fan(device.type)
@ -39,31 +41,31 @@ async def async_setup_entry(
and not (device.supports_up_light() and device.supports_down_light())
]
fan_up_lights: List[Entity] = [
fan_up_lights: list[Entity] = [
BondUpLight(hub, device, bpup_subs, "up_light")
for device in hub.devices
if DeviceType.is_fan(device.type) and device.supports_up_light()
]
fan_down_lights: List[Entity] = [
fan_down_lights: list[Entity] = [
BondDownLight(hub, device, bpup_subs, "down_light")
for device in hub.devices
if DeviceType.is_fan(device.type) and device.supports_down_light()
]
fireplaces: List[Entity] = [
fireplaces: list[Entity] = [
BondFireplace(hub, device, bpup_subs)
for device in hub.devices
if DeviceType.is_fireplace(device.type)
]
fp_lights: List[Entity] = [
fp_lights: list[Entity] = [
BondLight(hub, device, bpup_subs, "light")
for device in hub.devices
if DeviceType.is_fireplace(device.type) and device.supports_light()
]
lights: List[Entity] = [
lights: list[Entity] = [
BondLight(hub, device, bpup_subs)
for device in hub.devices
if DeviceType.is_light(device.type)
@ -83,11 +85,11 @@ class BondBaseLight(BondEntity, LightEntity):
hub: BondHub,
device: BondDevice,
bpup_subs: BPUPSubscriptions,
sub_device: Optional[str] = None,
sub_device: str | None = None,
):
"""Create HA entity representing Bond light."""
super().__init__(hub, device, bpup_subs, sub_device)
self._light: Optional[int] = None
self._light: int | None = None
@property
def is_on(self) -> bool:
@ -108,11 +110,11 @@ class BondLight(BondBaseLight, BondEntity, LightEntity):
hub: BondHub,
device: BondDevice,
bpup_subs: BPUPSubscriptions,
sub_device: Optional[str] = None,
sub_device: str | None = None,
):
"""Create HA entity representing Bond light."""
super().__init__(hub, device, bpup_subs, sub_device)
self._brightness: Optional[int] = None
self._brightness: int | None = None
def _apply_state(self, state: dict) -> None:
self._light = state.get("light")
@ -126,7 +128,7 @@ class BondLight(BondBaseLight, BondEntity, LightEntity):
return 0
@property
def brightness(self) -> Optional[int]:
def brightness(self) -> int | None:
"""Return the brightness of this light between 1..255."""
brightness_value = (
round(self._brightness * 255 / 100) if self._brightness else None
@ -194,9 +196,9 @@ class BondFireplace(BondEntity, LightEntity):
"""Create HA entity representing Bond fireplace."""
super().__init__(hub, device, bpup_subs)
self._power: Optional[bool] = None
self._power: bool | None = None
# Bond flame level, 0-100
self._flame: Optional[int] = None
self._flame: int | None = None
def _apply_state(self, state: dict) -> None:
self._power = state.get("power")
@ -230,11 +232,11 @@ class BondFireplace(BondEntity, LightEntity):
await self._hub.bond.action(self._device.device_id, Action.turn_off())
@property
def brightness(self) -> Optional[int]:
def brightness(self) -> int | None:
"""Return the flame of this fireplace converted to HA brightness between 0..255."""
return round(self._flame * 255 / 100) if self._flame else None
@property
def icon(self) -> Optional[str]:
def icon(self) -> str | None:
"""Show fireplace icon for the entity."""
return "mdi:fireplace" if self._power == 1 else "mdi:fireplace-off"

View file

@ -1,5 +1,7 @@
"""Support for Bond generic devices."""
from typing import Any, Callable, List, Optional
from __future__ import annotations
from typing import Any, Callable
from bond_api import Action, BPUPSubscriptions, DeviceType
@ -16,14 +18,14 @@ from .utils import BondDevice, BondHub
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 Bond generic devices."""
data = hass.data[DOMAIN][entry.entry_id]
hub: BondHub = data[HUB]
bpup_subs: BPUPSubscriptions = data[BPUP_SUBS]
switches: List[Entity] = [
switches: list[Entity] = [
BondSwitch(hub, device, bpup_subs)
for device in hub.devices
if DeviceType.is_generic(device.type)
@ -39,7 +41,7 @@ class BondSwitch(BondEntity, SwitchEntity):
"""Create HA entity representing Bond generic device (switch)."""
super().__init__(hub, device, bpup_subs)
self._power: Optional[bool] = None
self._power: bool | None = None
def _apply_state(self, state: dict) -> None:
self._power = state.get("power")

View file

@ -1,7 +1,9 @@
"""Reusable utilities for the Bond component."""
from __future__ import annotations
import asyncio
import logging
from typing import Any, Dict, List, Optional, Set, cast
from typing import Any, cast
from aiohttp import ClientResponseError
from bond_api import Action, Bond
@ -15,7 +17,7 @@ class BondDevice:
"""Helper device class to hold ID and attributes together."""
def __init__(
self, device_id: str, attrs: Dict[str, Any], props: Dict[str, Any]
self, device_id: str, attrs: dict[str, Any], props: dict[str, Any]
) -> None:
"""Create a helper device from ID and attributes returned by API."""
self.device_id = device_id
@ -41,17 +43,17 @@ class BondDevice:
return cast(str, self._attrs["type"])
@property
def location(self) -> Optional[str]:
def location(self) -> str | None:
"""Get the location of this device."""
return self._attrs.get("location")
@property
def template(self) -> Optional[str]:
def template(self) -> str | None:
"""Return this model template."""
return self._attrs.get("template")
@property
def branding_profile(self) -> Optional[str]:
def branding_profile(self) -> str | None:
"""Return this branding profile."""
return self.props.get("branding_profile")
@ -60,9 +62,9 @@ class BondDevice:
"""Check if Trust State is turned on."""
return self.props.get("trust_state", False)
def _has_any_action(self, actions: Set[str]) -> bool:
def _has_any_action(self, actions: set[str]) -> bool:
"""Check to see if the device supports any of the actions."""
supported_actions: List[str] = self._attrs["actions"]
supported_actions: list[str] = self._attrs["actions"]
for action in supported_actions:
if action in actions:
return True
@ -101,11 +103,11 @@ class BondHub:
def __init__(self, bond: Bond):
"""Initialize Bond Hub."""
self.bond: Bond = bond
self._bridge: Dict[str, Any] = {}
self._version: Dict[str, Any] = {}
self._devices: List[BondDevice] = []
self._bridge: dict[str, Any] = {}
self._version: dict[str, Any] = {}
self._devices: list[BondDevice] = []
async def setup(self, max_devices: Optional[int] = None) -> None:
async def setup(self, max_devices: int | None = None) -> None:
"""Read hub version information."""
self._version = await self.bond.version()
_LOGGER.debug("Bond reported the following version info: %s", self._version)
@ -131,18 +133,18 @@ class BondHub:
_LOGGER.debug("Bond reported the following bridge info: %s", self._bridge)
@property
def bond_id(self) -> Optional[str]:
def bond_id(self) -> str | None:
"""Return unique Bond ID for this hub."""
# Old firmwares are missing the bondid
return self._version.get("bondid")
@property
def target(self) -> Optional[str]:
def target(self) -> str | None:
"""Return this hub target."""
return self._version.get("target")
@property
def model(self) -> Optional[str]:
def model(self) -> str | None:
"""Return this hub model."""
return self._version.get("model")
@ -159,19 +161,19 @@ class BondHub:
return cast(str, self._bridge["name"])
@property
def location(self) -> Optional[str]:
def location(self) -> str | None:
"""Get the location of this bridge."""
if not self.is_bridge and self._devices:
return self._devices[0].location
return self._bridge.get("location")
@property
def fw_ver(self) -> Optional[str]:
def fw_ver(self) -> str | None:
"""Return this hub firmware version."""
return self._version.get("fw_ver")
@property
def devices(self) -> List[BondDevice]:
def devices(self) -> list[BondDevice]:
"""Return a list of all devices controlled by this hub."""
return self._devices

View file

@ -1,7 +1,9 @@
"""BSBLAN platform to control a compatible Climate Device."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any, Callable, Dict, List, Optional
from typing import Any, Callable
from bsblan import BSBLan, BSBLanError, Info, State
@ -74,7 +76,7 @@ BSBLAN_TO_HA_PRESET = {
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 BSBLan device based on a config entry."""
bsblan: BSBLan = hass.data[DOMAIN][entry.entry_id][DATA_BSBLAN_CLIENT]
@ -92,10 +94,10 @@ class BSBLanClimate(ClimateEntity):
info: Info,
):
"""Initialize BSBLan climate device."""
self._current_temperature: Optional[float] = None
self._current_temperature: float | None = None
self._available = True
self._hvac_mode: Optional[str] = None
self._target_temperature: Optional[float] = None
self._hvac_mode: str | None = None
self._target_temperature: float | None = None
self._temperature_unit = None
self._preset_mode = None
self._store_hvac_mode = None
@ -229,7 +231,7 @@ class BSBLanClimate(ClimateEntity):
self._temperature_unit = state.current_temperature.unit
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return device information about this BSBLan device."""
return {
ATTR_IDENTIFIERS: {(DOMAIN, self._info.device_identification)},

View file

@ -1,6 +1,8 @@
"""Config flow for BSB-Lan integration."""
from __future__ import annotations
import logging
from typing import Any, Dict, Optional
from typing import Any
from bsblan import BSBLan, BSBLanError, Info
import voluptuous as vol
@ -26,8 +28,8 @@ class BSBLanFlowHandler(ConfigFlow, domain=DOMAIN):
CONNECTION_CLASS = CONN_CLASS_LOCAL_POLL
async def async_step_user(
self, user_input: Optional[ConfigType] = None
) -> Dict[str, Any]:
self, user_input: ConfigType | None = None
) -> dict[str, Any]:
"""Handle a flow initiated by the user."""
if user_input is None:
return self._show_setup_form()
@ -59,7 +61,7 @@ class BSBLanFlowHandler(ConfigFlow, domain=DOMAIN):
},
)
def _show_setup_form(self, errors: Optional[Dict] = None) -> Dict[str, Any]:
def _show_setup_form(self, errors: dict | None = None) -> dict[str, Any]:
"""Show the setup form to the user."""
return self.async_show_form(
step_id="user",
@ -78,9 +80,9 @@ class BSBLanFlowHandler(ConfigFlow, domain=DOMAIN):
async def _get_bsblan_info(
self,
host: str,
username: Optional[str],
password: Optional[str],
passkey: Optional[str],
username: str | None,
password: str | None,
passkey: str | None,
port: int,
) -> Info:
"""Get device information from an BSBLan device."""

View file

@ -1,8 +1,9 @@
"""Provide animated GIF loops of Buienradar imagery."""
from __future__ import annotations
import asyncio
from datetime import datetime, timedelta
import logging
from typing import Optional
import aiohttp
import voluptuous as vol
@ -85,13 +86,13 @@ class BuienradarCam(Camera):
# invariant: this condition is private to and owned by this instance.
self._condition = asyncio.Condition()
self._last_image: Optional[bytes] = None
self._last_image: bytes | None = None
# value of the last seen last modified header
self._last_modified: Optional[str] = None
self._last_modified: str | None = None
# loading status
self._loading = False
# deadline for image refresh - self.delta after last successful load
self._deadline: Optional[datetime] = None
self._deadline: datetime | None = None
self._unique_id = f"{self._dimension}_{self._country}"
@ -140,7 +141,7 @@ class BuienradarCam(Camera):
_LOGGER.error("Failed to fetch image, %s", type(err))
return False
async def async_camera_image(self) -> Optional[bytes]:
async def async_camera_image(self) -> bytes | None:
"""
Return a still image response from the camera.