Update typing 06 (#48039)
This commit is contained in:
parent
7c0734bdd5
commit
91df3fa904
40 changed files with 241 additions and 211 deletions
|
@ -1,5 +1,5 @@
|
|||
"""Allows to configuration ecoal (esterownik.pl) pumps as switches."""
|
||||
from typing import Optional
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.components.switch import SwitchEntity
|
||||
|
||||
|
@ -40,7 +40,7 @@ class EcoalSwitch(SwitchEntity):
|
|||
self._state = None
|
||||
|
||||
@property
|
||||
def name(self) -> Optional[str]:
|
||||
def name(self) -> str | None:
|
||||
"""Return the name of the switch."""
|
||||
return self._name
|
||||
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
"""Support for Ecobee Thermostats."""
|
||||
from __future__ import annotations
|
||||
|
||||
import collections
|
||||
from typing import Optional
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
|
@ -406,7 +407,7 @@ class Thermostat(ClimateEntity):
|
|||
)
|
||||
|
||||
@property
|
||||
def target_humidity(self) -> Optional[int]:
|
||||
def target_humidity(self) -> int | None:
|
||||
"""Return the desired humidity set point."""
|
||||
if self.has_humidifier_control:
|
||||
return self.thermostat["runtime"]["desiredHumidity"]
|
||||
|
@ -484,7 +485,7 @@ class Thermostat(ClimateEntity):
|
|||
return self._operation_list
|
||||
|
||||
@property
|
||||
def current_humidity(self) -> Optional[int]:
|
||||
def current_humidity(self) -> int | None:
|
||||
"""Return the current humidity."""
|
||||
return self.thermostat["runtime"]["actualHumidity"]
|
||||
|
||||
|
|
|
@ -1,9 +1,11 @@
|
|||
"""Support for esphome devices."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import functools
|
||||
import logging
|
||||
import math
|
||||
from typing import Any, Callable, Dict, List, Optional
|
||||
from typing import Any, Callable
|
||||
|
||||
from aioesphomeapi import (
|
||||
APIClient,
|
||||
|
@ -155,7 +157,7 @@ async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool
|
|||
await cli.send_home_assistant_state(entity_id, new_state.state)
|
||||
|
||||
async def _send_home_assistant_state(
|
||||
entity_id: str, new_state: Optional[State]
|
||||
entity_id: str, new_state: State | None
|
||||
) -> None:
|
||||
"""Forward Home Assistant states to ESPHome."""
|
||||
await cli.send_home_assistant_state(entity_id, new_state.state)
|
||||
|
@ -384,7 +386,7 @@ async def _register_service(
|
|||
|
||||
|
||||
async def _setup_services(
|
||||
hass: HomeAssistantType, entry_data: RuntimeEntryData, services: List[UserService]
|
||||
hass: HomeAssistantType, entry_data: RuntimeEntryData, services: list[UserService]
|
||||
):
|
||||
old_services = entry_data.services.copy()
|
||||
to_unregister = []
|
||||
|
@ -461,7 +463,7 @@ async def platform_async_setup_entry(
|
|||
entry_data.state[component_key] = {}
|
||||
|
||||
@callback
|
||||
def async_list_entities(infos: List[EntityInfo]):
|
||||
def async_list_entities(infos: list[EntityInfo]):
|
||||
"""Update entities of this platform when entities are listed."""
|
||||
old_infos = entry_data.info[component_key]
|
||||
new_infos = {}
|
||||
|
@ -535,7 +537,7 @@ def esphome_state_property(func):
|
|||
class EsphomeEnumMapper:
|
||||
"""Helper class to convert between hass and esphome enum values."""
|
||||
|
||||
def __init__(self, func: Callable[[], Dict[int, str]]):
|
||||
def __init__(self, func: Callable[[], dict[int, str]]):
|
||||
"""Construct a EsphomeEnumMapper."""
|
||||
self._func = func
|
||||
|
||||
|
@ -549,7 +551,7 @@ class EsphomeEnumMapper:
|
|||
return inverse[value]
|
||||
|
||||
|
||||
def esphome_map_enum(func: Callable[[], Dict[int, str]]):
|
||||
def esphome_map_enum(func: Callable[[], dict[int, str]]):
|
||||
"""Map esphome int enum values to hass string constants.
|
||||
|
||||
This class has to be used as a decorator. This ensures the aioesphomeapi
|
||||
|
@ -621,7 +623,7 @@ class EsphomeBaseEntity(Entity):
|
|||
return self._entry_data.client
|
||||
|
||||
@property
|
||||
def _state(self) -> Optional[EntityState]:
|
||||
def _state(self) -> EntityState | None:
|
||||
try:
|
||||
return self._entry_data.state[self._component_key][self._key]
|
||||
except KeyError:
|
||||
|
@ -640,14 +642,14 @@ class EsphomeBaseEntity(Entity):
|
|||
return self._entry_data.available
|
||||
|
||||
@property
|
||||
def unique_id(self) -> Optional[str]:
|
||||
def unique_id(self) -> str | None:
|
||||
"""Return a unique id identifying the entity."""
|
||||
if not self._static_info.unique_id:
|
||||
return None
|
||||
return self._static_info.unique_id
|
||||
|
||||
@property
|
||||
def device_info(self) -> Dict[str, Any]:
|
||||
def device_info(self) -> dict[str, Any]:
|
||||
"""Return device registry information for this entity."""
|
||||
return {
|
||||
"connections": {(dr.CONNECTION_NETWORK_MAC, self._device_info.mac_address)}
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
"""Support for ESPHome binary sensors."""
|
||||
from typing import Optional
|
||||
from __future__ import annotations
|
||||
|
||||
from aioesphomeapi import BinarySensorInfo, BinarySensorState
|
||||
|
||||
|
@ -29,11 +29,11 @@ class EsphomeBinarySensor(EsphomeEntity, BinarySensorEntity):
|
|||
return super()._static_info
|
||||
|
||||
@property
|
||||
def _state(self) -> Optional[BinarySensorState]:
|
||||
def _state(self) -> BinarySensorState | None:
|
||||
return super()._state
|
||||
|
||||
@property
|
||||
def is_on(self) -> Optional[bool]:
|
||||
def is_on(self) -> bool | None:
|
||||
"""Return true if the binary sensor is on."""
|
||||
if self._static_info.is_status_binary_sensor:
|
||||
# Status binary sensors indicated connected state.
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
"""Support for ESPHome cameras."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from typing import Optional
|
||||
|
||||
from aioesphomeapi import CameraInfo, CameraState
|
||||
|
||||
|
@ -42,7 +43,7 @@ class EsphomeCamera(Camera, EsphomeBaseEntity):
|
|||
return super()._static_info
|
||||
|
||||
@property
|
||||
def _state(self) -> Optional[CameraState]:
|
||||
def _state(self) -> CameraState | None:
|
||||
return super()._state
|
||||
|
||||
async def async_added_to_hass(self) -> None:
|
||||
|
@ -67,7 +68,7 @@ class EsphomeCamera(Camera, EsphomeBaseEntity):
|
|||
async with self._image_cond:
|
||||
self._image_cond.notify_all()
|
||||
|
||||
async def async_camera_image(self) -> Optional[bytes]:
|
||||
async def async_camera_image(self) -> bytes | None:
|
||||
"""Return single camera image bytes."""
|
||||
if not self.available:
|
||||
return None
|
||||
|
@ -78,7 +79,7 @@ class EsphomeCamera(Camera, EsphomeBaseEntity):
|
|||
return None
|
||||
return self._state.image[:]
|
||||
|
||||
async def _async_camera_stream_image(self) -> Optional[bytes]:
|
||||
async def _async_camera_stream_image(self) -> bytes | None:
|
||||
"""Return a single camera image in a stream."""
|
||||
if not self.available:
|
||||
return None
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
"""Support for ESPHome climate devices."""
|
||||
from typing import List, Optional
|
||||
from __future__ import annotations
|
||||
|
||||
from aioesphomeapi import (
|
||||
ClimateAction,
|
||||
|
@ -134,7 +134,7 @@ class EsphomeClimateEntity(EsphomeEntity, ClimateEntity):
|
|||
return super()._static_info
|
||||
|
||||
@property
|
||||
def _state(self) -> Optional[ClimateState]:
|
||||
def _state(self) -> ClimateState | None:
|
||||
return super()._state
|
||||
|
||||
@property
|
||||
|
@ -153,7 +153,7 @@ class EsphomeClimateEntity(EsphomeEntity, ClimateEntity):
|
|||
return TEMP_CELSIUS
|
||||
|
||||
@property
|
||||
def hvac_modes(self) -> List[str]:
|
||||
def hvac_modes(self) -> list[str]:
|
||||
"""Return the list of available operation modes."""
|
||||
return [
|
||||
_climate_modes.from_esphome(mode)
|
||||
|
@ -217,12 +217,12 @@ class EsphomeClimateEntity(EsphomeEntity, ClimateEntity):
|
|||
# pylint: disable=invalid-overridden-method
|
||||
|
||||
@esphome_state_property
|
||||
def hvac_mode(self) -> Optional[str]:
|
||||
def hvac_mode(self) -> str | None:
|
||||
"""Return current operation ie. heat, cool, idle."""
|
||||
return _climate_modes.from_esphome(self._state.mode)
|
||||
|
||||
@esphome_state_property
|
||||
def hvac_action(self) -> Optional[str]:
|
||||
def hvac_action(self) -> str | None:
|
||||
"""Return current action."""
|
||||
# HA has no support feature field for hvac_action
|
||||
if not self._static_info.supports_action:
|
||||
|
@ -245,22 +245,22 @@ class EsphomeClimateEntity(EsphomeEntity, ClimateEntity):
|
|||
return _swing_modes.from_esphome(self._state.swing_mode)
|
||||
|
||||
@esphome_state_property
|
||||
def current_temperature(self) -> Optional[float]:
|
||||
def current_temperature(self) -> float | None:
|
||||
"""Return the current temperature."""
|
||||
return self._state.current_temperature
|
||||
|
||||
@esphome_state_property
|
||||
def target_temperature(self) -> Optional[float]:
|
||||
def target_temperature(self) -> float | None:
|
||||
"""Return the temperature we try to reach."""
|
||||
return self._state.target_temperature
|
||||
|
||||
@esphome_state_property
|
||||
def target_temperature_low(self) -> Optional[float]:
|
||||
def target_temperature_low(self) -> float | None:
|
||||
"""Return the lowbound target temperature we try to reach."""
|
||||
return self._state.target_temperature_low
|
||||
|
||||
@esphome_state_property
|
||||
def target_temperature_high(self) -> Optional[float]:
|
||||
def target_temperature_high(self) -> float | None:
|
||||
"""Return the highbound target temperature we try to reach."""
|
||||
return self._state.target_temperature_high
|
||||
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
"""Config flow to configure esphome component."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import OrderedDict
|
||||
from typing import Optional
|
||||
|
||||
from aioesphomeapi import APIClient, APIConnectionError
|
||||
import voluptuous as vol
|
||||
|
@ -23,12 +24,12 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
|
|||
|
||||
def __init__(self):
|
||||
"""Initialize flow."""
|
||||
self._host: Optional[str] = None
|
||||
self._port: Optional[int] = None
|
||||
self._password: Optional[str] = None
|
||||
self._host: str | None = None
|
||||
self._port: int | None = None
|
||||
self._password: str | None = None
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: Optional[ConfigType] = None, error: Optional[str] = None
|
||||
self, user_input: ConfigType | None = None, error: str | None = None
|
||||
): # pylint: disable=arguments-differ
|
||||
"""Handle a flow initialized by the user."""
|
||||
if user_input is not None:
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
"""Support for ESPHome covers."""
|
||||
from typing import Optional
|
||||
from __future__ import annotations
|
||||
|
||||
from aioesphomeapi import CoverInfo, CoverOperation, CoverState
|
||||
|
||||
|
@ -64,14 +64,14 @@ class EsphomeCover(EsphomeEntity, CoverEntity):
|
|||
return self._static_info.assumed_state
|
||||
|
||||
@property
|
||||
def _state(self) -> Optional[CoverState]:
|
||||
def _state(self) -> CoverState | None:
|
||||
return super()._state
|
||||
|
||||
# https://github.com/PyCQA/pylint/issues/3150 for all @esphome_state_property
|
||||
# pylint: disable=invalid-overridden-method
|
||||
|
||||
@esphome_state_property
|
||||
def is_closed(self) -> Optional[bool]:
|
||||
def is_closed(self) -> bool | None:
|
||||
"""Return if the cover is closed or not."""
|
||||
# Check closed state with api version due to a protocol change
|
||||
return self._state.is_closed(self._client.api_version)
|
||||
|
@ -87,14 +87,14 @@ class EsphomeCover(EsphomeEntity, CoverEntity):
|
|||
return self._state.current_operation == CoverOperation.IS_CLOSING
|
||||
|
||||
@esphome_state_property
|
||||
def current_cover_position(self) -> Optional[int]:
|
||||
def current_cover_position(self) -> int | None:
|
||||
"""Return current position of cover. 0 is closed, 100 is open."""
|
||||
if not self._static_info.supports_position:
|
||||
return None
|
||||
return round(self._state.position * 100.0)
|
||||
|
||||
@esphome_state_property
|
||||
def current_cover_tilt_position(self) -> Optional[float]:
|
||||
def current_cover_tilt_position(self) -> float | None:
|
||||
"""Return current position of cover tilt. 0 is closed, 100 is open."""
|
||||
if not self._static_info.supports_tilt:
|
||||
return None
|
||||
|
|
|
@ -1,6 +1,8 @@
|
|||
"""Runtime entry data for ESPHome stored in hass.data."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Set, Tuple
|
||||
from typing import TYPE_CHECKING, Any, Callable
|
||||
|
||||
from aioesphomeapi import (
|
||||
COMPONENT_TYPE_TO_INFO,
|
||||
|
@ -52,22 +54,22 @@ class RuntimeEntryData:
|
|||
entry_id: str = attr.ib()
|
||||
client: "APIClient" = attr.ib()
|
||||
store: Store = attr.ib()
|
||||
reconnect_task: Optional[asyncio.Task] = attr.ib(default=None)
|
||||
state: Dict[str, Dict[str, Any]] = attr.ib(factory=dict)
|
||||
info: Dict[str, Dict[str, Any]] = attr.ib(factory=dict)
|
||||
reconnect_task: asyncio.Task | None = attr.ib(default=None)
|
||||
state: dict[str, dict[str, Any]] = attr.ib(factory=dict)
|
||||
info: dict[str, dict[str, Any]] = attr.ib(factory=dict)
|
||||
|
||||
# A second list of EntityInfo objects
|
||||
# This is necessary for when an entity is being removed. HA requires
|
||||
# some static info to be accessible during removal (unique_id, maybe others)
|
||||
# If an entity can't find anything in the info array, it will look for info here.
|
||||
old_info: Dict[str, Dict[str, Any]] = attr.ib(factory=dict)
|
||||
old_info: dict[str, dict[str, Any]] = attr.ib(factory=dict)
|
||||
|
||||
services: Dict[int, "UserService"] = attr.ib(factory=dict)
|
||||
services: dict[int, "UserService"] = attr.ib(factory=dict)
|
||||
available: bool = attr.ib(default=False)
|
||||
device_info: Optional[DeviceInfo] = attr.ib(default=None)
|
||||
cleanup_callbacks: List[Callable[[], None]] = attr.ib(factory=list)
|
||||
disconnect_callbacks: List[Callable[[], None]] = attr.ib(factory=list)
|
||||
loaded_platforms: Set[str] = attr.ib(factory=set)
|
||||
device_info: DeviceInfo | None = attr.ib(default=None)
|
||||
cleanup_callbacks: list[Callable[[], None]] = attr.ib(factory=list)
|
||||
disconnect_callbacks: list[Callable[[], None]] = attr.ib(factory=list)
|
||||
loaded_platforms: set[str] = attr.ib(factory=set)
|
||||
platform_load_lock: asyncio.Lock = attr.ib(factory=asyncio.Lock)
|
||||
|
||||
@callback
|
||||
|
@ -87,7 +89,7 @@ class RuntimeEntryData:
|
|||
async_dispatcher_send(hass, signal)
|
||||
|
||||
async def _ensure_platforms_loaded(
|
||||
self, hass: HomeAssistantType, entry: ConfigEntry, platforms: Set[str]
|
||||
self, hass: HomeAssistantType, entry: ConfigEntry, platforms: set[str]
|
||||
):
|
||||
async with self.platform_load_lock:
|
||||
needed = platforms - self.loaded_platforms
|
||||
|
@ -101,7 +103,7 @@ class RuntimeEntryData:
|
|||
self.loaded_platforms |= needed
|
||||
|
||||
async def async_update_static_infos(
|
||||
self, hass: HomeAssistantType, entry: ConfigEntry, infos: List[EntityInfo]
|
||||
self, hass: HomeAssistantType, entry: ConfigEntry, infos: list[EntityInfo]
|
||||
) -> None:
|
||||
"""Distribute an update of static infos to all platforms."""
|
||||
# First, load all platforms
|
||||
|
@ -129,7 +131,7 @@ class RuntimeEntryData:
|
|||
signal = f"esphome_{self.entry_id}_on_device_update"
|
||||
async_dispatcher_send(hass, signal)
|
||||
|
||||
async def async_load_from_store(self) -> Tuple[List[EntityInfo], List[UserService]]:
|
||||
async def async_load_from_store(self) -> tuple[list[EntityInfo], list[UserService]]:
|
||||
"""Load the retained data from store and return de-serialized data."""
|
||||
restored = await self.store.async_load()
|
||||
if restored is None:
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
"""Support for ESPHome fans."""
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
from typing import Optional
|
||||
|
||||
from aioesphomeapi import FanDirection, FanInfo, FanSpeed, FanState
|
||||
|
||||
|
@ -62,7 +63,7 @@ class EsphomeFan(EsphomeEntity, FanEntity):
|
|||
return super()._static_info
|
||||
|
||||
@property
|
||||
def _state(self) -> Optional[FanState]:
|
||||
def _state(self) -> FanState | None:
|
||||
return super()._state
|
||||
|
||||
@property
|
||||
|
@ -93,9 +94,9 @@ class EsphomeFan(EsphomeEntity, FanEntity):
|
|||
|
||||
async def async_turn_on(
|
||||
self,
|
||||
speed: Optional[str] = None,
|
||||
percentage: Optional[int] = None,
|
||||
preset_mode: Optional[str] = None,
|
||||
speed: str | None = None,
|
||||
percentage: int | None = None,
|
||||
preset_mode: str | None = None,
|
||||
**kwargs,
|
||||
) -> None:
|
||||
"""Turn on the fan."""
|
||||
|
@ -121,12 +122,12 @@ class EsphomeFan(EsphomeEntity, FanEntity):
|
|||
# pylint: disable=invalid-overridden-method
|
||||
|
||||
@esphome_state_property
|
||||
def is_on(self) -> Optional[bool]:
|
||||
def is_on(self) -> bool | None:
|
||||
"""Return true if the entity is on."""
|
||||
return self._state.state
|
||||
|
||||
@esphome_state_property
|
||||
def percentage(self) -> Optional[int]:
|
||||
def percentage(self) -> int | None:
|
||||
"""Return the current speed percentage."""
|
||||
if not self._static_info.supports_speed:
|
||||
return None
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
"""Support for ESPHome lights."""
|
||||
from typing import List, Optional, Tuple
|
||||
from __future__ import annotations
|
||||
|
||||
from aioesphomeapi import LightInfo, LightState
|
||||
|
||||
|
@ -54,14 +54,14 @@ class EsphomeLight(EsphomeEntity, LightEntity):
|
|||
return super()._static_info
|
||||
|
||||
@property
|
||||
def _state(self) -> Optional[LightState]:
|
||||
def _state(self) -> LightState | None:
|
||||
return super()._state
|
||||
|
||||
# https://github.com/PyCQA/pylint/issues/3150 for all @esphome_state_property
|
||||
# pylint: disable=invalid-overridden-method
|
||||
|
||||
@esphome_state_property
|
||||
def is_on(self) -> Optional[bool]:
|
||||
def is_on(self) -> bool | None:
|
||||
"""Return true if the switch is on."""
|
||||
return self._state.state
|
||||
|
||||
|
@ -96,29 +96,29 @@ class EsphomeLight(EsphomeEntity, LightEntity):
|
|||
await self._client.light_command(**data)
|
||||
|
||||
@esphome_state_property
|
||||
def brightness(self) -> Optional[int]:
|
||||
def brightness(self) -> int | None:
|
||||
"""Return the brightness of this light between 0..255."""
|
||||
return round(self._state.brightness * 255)
|
||||
|
||||
@esphome_state_property
|
||||
def hs_color(self) -> Optional[Tuple[float, float]]:
|
||||
def hs_color(self) -> tuple[float, float] | None:
|
||||
"""Return the hue and saturation color value [float, float]."""
|
||||
return color_util.color_RGB_to_hs(
|
||||
self._state.red * 255, self._state.green * 255, self._state.blue * 255
|
||||
)
|
||||
|
||||
@esphome_state_property
|
||||
def color_temp(self) -> Optional[float]:
|
||||
def color_temp(self) -> float | None:
|
||||
"""Return the CT color value in mireds."""
|
||||
return self._state.color_temperature
|
||||
|
||||
@esphome_state_property
|
||||
def white_value(self) -> Optional[int]:
|
||||
def white_value(self) -> int | None:
|
||||
"""Return the white value of this light between 0..255."""
|
||||
return round(self._state.white * 255)
|
||||
|
||||
@esphome_state_property
|
||||
def effect(self) -> Optional[str]:
|
||||
def effect(self) -> str | None:
|
||||
"""Return the current effect."""
|
||||
return self._state.effect
|
||||
|
||||
|
@ -140,7 +140,7 @@ class EsphomeLight(EsphomeEntity, LightEntity):
|
|||
return flags
|
||||
|
||||
@property
|
||||
def effect_list(self) -> List[str]:
|
||||
def effect_list(self) -> list[str]:
|
||||
"""Return the list of supported effects."""
|
||||
return self._static_info.effects
|
||||
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
"""Support for esphome sensors."""
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
from typing import Optional
|
||||
|
||||
from aioesphomeapi import SensorInfo, SensorState, TextSensorInfo, TextSensorState
|
||||
import voluptuous as vol
|
||||
|
@ -49,7 +50,7 @@ class EsphomeSensor(EsphomeEntity):
|
|||
return super()._static_info
|
||||
|
||||
@property
|
||||
def _state(self) -> Optional[SensorState]:
|
||||
def _state(self) -> SensorState | None:
|
||||
return super()._state
|
||||
|
||||
@property
|
||||
|
@ -65,7 +66,7 @@ class EsphomeSensor(EsphomeEntity):
|
|||
return self._static_info.force_update
|
||||
|
||||
@esphome_state_property
|
||||
def state(self) -> Optional[str]:
|
||||
def state(self) -> str | None:
|
||||
"""Return the state of the entity."""
|
||||
if math.isnan(self._state.state):
|
||||
return None
|
||||
|
@ -96,7 +97,7 @@ class EsphomeTextSensor(EsphomeEntity):
|
|||
return super()._static_info
|
||||
|
||||
@property
|
||||
def _state(self) -> Optional[TextSensorState]:
|
||||
def _state(self) -> TextSensorState | None:
|
||||
return super()._state
|
||||
|
||||
@property
|
||||
|
@ -105,7 +106,7 @@ class EsphomeTextSensor(EsphomeEntity):
|
|||
return self._static_info.icon
|
||||
|
||||
@esphome_state_property
|
||||
def state(self) -> Optional[str]:
|
||||
def state(self) -> str | None:
|
||||
"""Return the state of the entity."""
|
||||
if self._state.missing_state:
|
||||
return None
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
"""Support for ESPHome switches."""
|
||||
from typing import Optional
|
||||
from __future__ import annotations
|
||||
|
||||
from aioesphomeapi import SwitchInfo, SwitchState
|
||||
|
||||
|
@ -33,7 +33,7 @@ class EsphomeSwitch(EsphomeEntity, SwitchEntity):
|
|||
return super()._static_info
|
||||
|
||||
@property
|
||||
def _state(self) -> Optional[SwitchState]:
|
||||
def _state(self) -> SwitchState | None:
|
||||
return super()._state
|
||||
|
||||
@property
|
||||
|
@ -49,7 +49,7 @@ class EsphomeSwitch(EsphomeEntity, SwitchEntity):
|
|||
# https://github.com/PyCQA/pylint/issues/3150 for @esphome_state_property
|
||||
# pylint: disable=invalid-overridden-method
|
||||
@esphome_state_property
|
||||
def is_on(self) -> Optional[bool]:
|
||||
def is_on(self) -> bool | None:
|
||||
"""Return true if the switch is on."""
|
||||
return self._state.state
|
||||
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
"""Support for Essent API."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
from typing import Optional
|
||||
|
||||
from pyessent import PyEssent
|
||||
import voluptuous as vol
|
||||
|
@ -94,7 +95,7 @@ class EssentMeter(Entity):
|
|||
self._unit = unit
|
||||
|
||||
@property
|
||||
def unique_id(self) -> Optional[str]:
|
||||
def unique_id(self) -> str | None:
|
||||
"""Return a unique ID."""
|
||||
return f"{self._meter}-{self._type}-{self._tariff}"
|
||||
|
||||
|
|
|
@ -1,7 +1,8 @@
|
|||
"""Support for EverLights lights."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from typing import Tuple
|
||||
|
||||
import pyeverlights
|
||||
import voluptuous as vol
|
||||
|
@ -38,7 +39,7 @@ def color_rgb_to_int(red: int, green: int, blue: int) -> int:
|
|||
return red * 256 * 256 + green * 256 + blue
|
||||
|
||||
|
||||
def color_int_to_rgb(value: int) -> Tuple[int, int, int]:
|
||||
def color_int_to_rgb(value: int) -> tuple[int, int, int]:
|
||||
"""Return an RGB tuple from an integer."""
|
||||
return (value >> 16, (value >> 8) & 0xFF, value & 0xFF)
|
||||
|
||||
|
|
|
@ -2,10 +2,12 @@
|
|||
|
||||
Such systems include evohome, Round Thermostat, and others.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime as dt, timedelta
|
||||
import logging
|
||||
import re
|
||||
from typing import Any, Dict, Optional, Tuple
|
||||
from typing import Any
|
||||
|
||||
import aiohttp.client_exceptions
|
||||
import evohomeasync
|
||||
|
@ -114,7 +116,7 @@ def convert_until(status_dict: dict, until_key: str) -> None:
|
|||
status_dict[until_key] = dt_util.as_local(dt_utc_naive).isoformat()
|
||||
|
||||
|
||||
def convert_dict(dictionary: Dict[str, Any]) -> Dict[str, Any]:
|
||||
def convert_dict(dictionary: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Recursively convert a dict's keys to snake_case."""
|
||||
|
||||
def convert_key(key: str) -> str:
|
||||
|
@ -176,7 +178,7 @@ def _handle_exception(err) -> bool:
|
|||
async def async_setup(hass: HomeAssistantType, config: ConfigType) -> bool:
|
||||
"""Create a (EMEA/EU-based) Honeywell TCC system."""
|
||||
|
||||
async def load_auth_tokens(store) -> Tuple[Dict, Optional[Dict]]:
|
||||
async def load_auth_tokens(store) -> tuple[dict, dict | None]:
|
||||
app_storage = await store.async_load()
|
||||
tokens = dict(app_storage if app_storage else {})
|
||||
|
||||
|
@ -435,7 +437,7 @@ class EvoBroker:
|
|||
async def _update_v1_api_temps(self, *args, **kwargs) -> None:
|
||||
"""Get the latest high-precision temperatures of the default Location."""
|
||||
|
||||
def get_session_id(client_v1) -> Optional[str]:
|
||||
def get_session_id(client_v1) -> str | None:
|
||||
user_data = client_v1.user_data if client_v1 else None
|
||||
return user_data.get("sessionId") if user_data else None
|
||||
|
||||
|
@ -520,7 +522,7 @@ class EvoDevice(Entity):
|
|||
self._supported_features = None
|
||||
self._device_state_attrs = {}
|
||||
|
||||
async def async_refresh(self, payload: Optional[dict] = None) -> None:
|
||||
async def async_refresh(self, payload: dict | None = None) -> None:
|
||||
"""Process any signals."""
|
||||
if payload is None:
|
||||
self.async_schedule_update_ha_state(force_refresh=True)
|
||||
|
@ -546,7 +548,7 @@ class EvoDevice(Entity):
|
|||
return False
|
||||
|
||||
@property
|
||||
def unique_id(self) -> Optional[str]:
|
||||
def unique_id(self) -> str | None:
|
||||
"""Return a unique ID."""
|
||||
return self._unique_id
|
||||
|
||||
|
@ -556,7 +558,7 @@ class EvoDevice(Entity):
|
|||
return self._name
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> Dict[str, Any]:
|
||||
def extra_state_attributes(self) -> dict[str, Any]:
|
||||
"""Return the evohome-specific state attributes."""
|
||||
status = self._device_state_attrs
|
||||
if "systemModeStatus" in status:
|
||||
|
@ -606,7 +608,7 @@ class EvoChild(EvoDevice):
|
|||
self._setpoints = {}
|
||||
|
||||
@property
|
||||
def current_temperature(self) -> Optional[float]:
|
||||
def current_temperature(self) -> float | None:
|
||||
"""Return the current temperature of a Zone."""
|
||||
if (
|
||||
self._evo_broker.temps
|
||||
|
@ -618,7 +620,7 @@ class EvoChild(EvoDevice):
|
|||
return self._evo_device.temperatureStatus["temperature"]
|
||||
|
||||
@property
|
||||
def setpoints(self) -> Dict[str, Any]:
|
||||
def setpoints(self) -> dict[str, Any]:
|
||||
"""Return the current/next setpoints from the schedule.
|
||||
|
||||
Only Zones & DHW controllers (but not the TCS) can have schedules.
|
||||
|
|
|
@ -1,7 +1,8 @@
|
|||
"""Support for Climate devices of (EMEA/EU-based) Honeywell TCC systems."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime as dt
|
||||
import logging
|
||||
from typing import List, Optional
|
||||
|
||||
from homeassistant.components.climate import ClimateEntity
|
||||
from homeassistant.components.climate.const import (
|
||||
|
@ -129,12 +130,12 @@ class EvoClimateEntity(EvoDevice, ClimateEntity):
|
|||
self._preset_modes = None
|
||||
|
||||
@property
|
||||
def hvac_modes(self) -> List[str]:
|
||||
def hvac_modes(self) -> list[str]:
|
||||
"""Return a list of available hvac operation modes."""
|
||||
return list(HA_HVAC_TO_TCS)
|
||||
|
||||
@property
|
||||
def preset_modes(self) -> Optional[List[str]]:
|
||||
def preset_modes(self) -> list[str] | None:
|
||||
"""Return a list of available preset modes."""
|
||||
return self._preset_modes
|
||||
|
||||
|
@ -203,7 +204,7 @@ class EvoZone(EvoChild, EvoClimateEntity):
|
|||
return self._evo_device.setpointStatus["targetHeatTemperature"]
|
||||
|
||||
@property
|
||||
def preset_mode(self) -> Optional[str]:
|
||||
def preset_mode(self) -> str | None:
|
||||
"""Return the current preset mode, e.g., home, away, temp."""
|
||||
if self._evo_tcs.systemModeStatus["mode"] in [EVO_AWAY, EVO_HEATOFF]:
|
||||
return TCS_PRESET_TO_HA.get(self._evo_tcs.systemModeStatus["mode"])
|
||||
|
@ -268,7 +269,7 @@ class EvoZone(EvoChild, EvoClimateEntity):
|
|||
self._evo_device.cancel_temp_override()
|
||||
)
|
||||
|
||||
async def async_set_preset_mode(self, preset_mode: Optional[str]) -> None:
|
||||
async def async_set_preset_mode(self, preset_mode: str | None) -> None:
|
||||
"""Set the preset mode; if None, then revert to following the schedule."""
|
||||
evo_preset_mode = HA_PRESET_TO_EVO.get(preset_mode, EVO_FOLLOW)
|
||||
|
||||
|
@ -347,7 +348,7 @@ class EvoController(EvoClimateEntity):
|
|||
|
||||
await self._set_tcs_mode(mode, until=until)
|
||||
|
||||
async def _set_tcs_mode(self, mode: str, until: Optional[dt] = None) -> None:
|
||||
async def _set_tcs_mode(self, mode: str, until: dt | None = None) -> None:
|
||||
"""Set a Controller to any of its native EVO_* operating modes."""
|
||||
until = dt_util.as_utc(until) if until else None
|
||||
await self._evo_broker.call_client_api(
|
||||
|
@ -361,7 +362,7 @@ class EvoController(EvoClimateEntity):
|
|||
return HVAC_MODE_OFF if tcs_mode == EVO_HEATOFF else HVAC_MODE_HEAT
|
||||
|
||||
@property
|
||||
def current_temperature(self) -> Optional[float]:
|
||||
def current_temperature(self) -> float | None:
|
||||
"""Return the average current temperature of the heating Zones.
|
||||
|
||||
Controllers do not have a current temp, but one is expected by HA.
|
||||
|
@ -374,7 +375,7 @@ class EvoController(EvoClimateEntity):
|
|||
return round(sum(temps) / len(temps), 1) if temps else None
|
||||
|
||||
@property
|
||||
def preset_mode(self) -> Optional[str]:
|
||||
def preset_mode(self) -> str | None:
|
||||
"""Return the current preset mode, e.g., home, away, temp."""
|
||||
return TCS_PRESET_TO_HA.get(self._evo_tcs.systemModeStatus["mode"])
|
||||
|
||||
|
@ -396,7 +397,7 @@ class EvoController(EvoClimateEntity):
|
|||
"""Set an operating mode for a Controller."""
|
||||
await self._set_tcs_mode(HA_HVAC_TO_TCS.get(hvac_mode))
|
||||
|
||||
async def async_set_preset_mode(self, preset_mode: Optional[str]) -> None:
|
||||
async def async_set_preset_mode(self, preset_mode: str | None) -> None:
|
||||
"""Set the preset mode; if None, then revert to 'Auto' mode."""
|
||||
await self._set_tcs_mode(HA_PRESET_TO_TCS.get(preset_mode, EVO_AUTO))
|
||||
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
"""Support for WaterHeater devices of (EMEA/EU) Honeywell TCC systems."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import List
|
||||
|
||||
from homeassistant.components.water_heater import (
|
||||
SUPPORT_AWAY_MODE,
|
||||
|
@ -70,7 +71,7 @@ class EvoDHW(EvoChild, WaterHeaterEntity):
|
|||
return EVO_STATE_TO_HA[self._evo_device.stateStatus["state"]]
|
||||
|
||||
@property
|
||||
def operation_list(self) -> List[str]:
|
||||
def operation_list(self) -> list[str]:
|
||||
"""Return the list of available operations."""
|
||||
return list(HA_STATE_TO_EVO)
|
||||
|
||||
|
|
|
@ -1,9 +1,10 @@
|
|||
"""Provides functionality to interact with fans."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
import functools as ft
|
||||
import logging
|
||||
import math
|
||||
from typing import List, Optional
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
|
@ -274,16 +275,16 @@ class FanEntity(ToggleEntity):
|
|||
else:
|
||||
await self.async_set_speed(self.percentage_to_speed(percentage))
|
||||
|
||||
async def async_increase_speed(self, percentage_step: Optional[int] = None) -> None:
|
||||
async def async_increase_speed(self, percentage_step: int | None = None) -> None:
|
||||
"""Increase the speed of the fan."""
|
||||
await self._async_adjust_speed(1, percentage_step)
|
||||
|
||||
async def async_decrease_speed(self, percentage_step: Optional[int] = None) -> None:
|
||||
async def async_decrease_speed(self, percentage_step: int | None = None) -> None:
|
||||
"""Decrease the speed of the fan."""
|
||||
await self._async_adjust_speed(-1, percentage_step)
|
||||
|
||||
async def _async_adjust_speed(
|
||||
self, modifier: int, percentage_step: Optional[int]
|
||||
self, modifier: int, percentage_step: int | None
|
||||
) -> None:
|
||||
"""Increase or decrease the speed of the fan."""
|
||||
current_percentage = self.percentage or 0
|
||||
|
@ -338,9 +339,9 @@ class FanEntity(ToggleEntity):
|
|||
# pylint: disable=arguments-differ
|
||||
def turn_on(
|
||||
self,
|
||||
speed: Optional[str] = None,
|
||||
percentage: Optional[int] = None,
|
||||
preset_mode: Optional[str] = None,
|
||||
speed: str | None = None,
|
||||
percentage: int | None = None,
|
||||
preset_mode: str | None = None,
|
||||
**kwargs,
|
||||
) -> None:
|
||||
"""Turn on the fan."""
|
||||
|
@ -348,9 +349,9 @@ class FanEntity(ToggleEntity):
|
|||
|
||||
async def async_turn_on_compat(
|
||||
self,
|
||||
speed: Optional[str] = None,
|
||||
percentage: Optional[int] = None,
|
||||
preset_mode: Optional[str] = None,
|
||||
speed: str | None = None,
|
||||
percentage: int | None = None,
|
||||
preset_mode: str | None = None,
|
||||
**kwargs,
|
||||
) -> None:
|
||||
"""Turn on the fan.
|
||||
|
@ -387,9 +388,9 @@ class FanEntity(ToggleEntity):
|
|||
# pylint: disable=arguments-differ
|
||||
async def async_turn_on(
|
||||
self,
|
||||
speed: Optional[str] = None,
|
||||
percentage: Optional[int] = None,
|
||||
preset_mode: Optional[str] = None,
|
||||
speed: str | None = None,
|
||||
percentage: int | None = None,
|
||||
preset_mode: str | None = None,
|
||||
**kwargs,
|
||||
) -> None:
|
||||
"""Turn on the fan."""
|
||||
|
@ -441,7 +442,7 @@ class FanEntity(ToggleEntity):
|
|||
)
|
||||
|
||||
@property
|
||||
def speed(self) -> Optional[str]:
|
||||
def speed(self) -> str | None:
|
||||
"""Return the current speed."""
|
||||
if self._implemented_preset_mode:
|
||||
preset_mode = self.preset_mode
|
||||
|
@ -455,7 +456,7 @@ class FanEntity(ToggleEntity):
|
|||
return None
|
||||
|
||||
@property
|
||||
def percentage(self) -> Optional[int]:
|
||||
def percentage(self) -> int | None:
|
||||
"""Return the current speed as a percentage."""
|
||||
if not self._implemented_preset_mode:
|
||||
if self.speed in self.preset_modes:
|
||||
|
@ -488,7 +489,7 @@ class FanEntity(ToggleEntity):
|
|||
return speeds
|
||||
|
||||
@property
|
||||
def current_direction(self) -> Optional[str]:
|
||||
def current_direction(self) -> str | None:
|
||||
"""Return the current direction of the fan."""
|
||||
return None
|
||||
|
||||
|
@ -616,7 +617,7 @@ class FanEntity(ToggleEntity):
|
|||
return 0
|
||||
|
||||
@property
|
||||
def preset_mode(self) -> Optional[str]:
|
||||
def preset_mode(self) -> str | None:
|
||||
"""Return the current preset mode, e.g., auto, smart, interval, favorite.
|
||||
|
||||
Requires SUPPORT_SET_SPEED.
|
||||
|
@ -627,7 +628,7 @@ class FanEntity(ToggleEntity):
|
|||
return None
|
||||
|
||||
@property
|
||||
def preset_modes(self) -> Optional[List[str]]:
|
||||
def preset_modes(self) -> list[str] | None:
|
||||
"""Return a list of available preset modes.
|
||||
|
||||
Requires SUPPORT_SET_SPEED.
|
||||
|
@ -635,7 +636,7 @@ class FanEntity(ToggleEntity):
|
|||
return preset_modes_from_speed_list(self.speed_list)
|
||||
|
||||
|
||||
def speed_list_without_preset_modes(speed_list: List):
|
||||
def speed_list_without_preset_modes(speed_list: list):
|
||||
"""Filter out non-speeds from the speed list.
|
||||
|
||||
The goal is to get the speeds in a list from lowest to
|
||||
|
@ -659,7 +660,7 @@ def speed_list_without_preset_modes(speed_list: List):
|
|||
return [speed for speed in speed_list if speed.lower() not in _NOT_SPEEDS_FILTER]
|
||||
|
||||
|
||||
def preset_modes_from_speed_list(speed_list: List):
|
||||
def preset_modes_from_speed_list(speed_list: list):
|
||||
"""Filter out non-preset modes from the speed list.
|
||||
|
||||
The goal is to return only preset modes.
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
"""Provides device automations for Fan."""
|
||||
from typing import List, Optional
|
||||
from __future__ import annotations
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
|
@ -28,7 +28,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 Fan devices."""
|
||||
registry = await entity_registry.async_get_registry(hass)
|
||||
actions = []
|
||||
|
@ -59,7 +59,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)
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
"""Provide the device automations for Fan."""
|
||||
from typing import Dict, List
|
||||
from __future__ import annotations
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
|
@ -32,7 +32,7 @@ CONDITION_SCHEMA = DEVICE_CONDITION_BASE_SCHEMA.extend(
|
|||
|
||||
async def async_get_conditions(
|
||||
hass: HomeAssistant, device_id: str
|
||||
) -> List[Dict[str, str]]:
|
||||
) -> list[dict[str, str]]:
|
||||
"""List device conditions for Fan devices."""
|
||||
registry = await entity_registry.async_get_registry(hass)
|
||||
conditions = []
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
"""Provides device automations for Fan."""
|
||||
from typing import List
|
||||
from __future__ import annotations
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
|
@ -31,7 +31,7 @@ TRIGGER_SCHEMA = TRIGGER_BASE_SCHEMA.extend(
|
|||
)
|
||||
|
||||
|
||||
async def async_get_triggers(hass: HomeAssistant, device_id: str) -> List[dict]:
|
||||
async def async_get_triggers(hass: HomeAssistant, device_id: str) -> list[dict]:
|
||||
"""List device triggers for Fan devices."""
|
||||
registry = await entity_registry.async_get_registry(hass)
|
||||
triggers = []
|
||||
|
|
|
@ -1,8 +1,10 @@
|
|||
"""Reproduce an Fan state."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from types import MappingProxyType
|
||||
from typing import Any, Dict, Iterable, Optional
|
||||
from typing import Any, Iterable
|
||||
|
||||
from homeassistant.const import (
|
||||
ATTR_ENTITY_ID,
|
||||
|
@ -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)
|
||||
|
@ -98,8 +100,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 Fan states."""
|
||||
await asyncio.gather(
|
||||
|
|
|
@ -1,7 +1,8 @@
|
|||
"""Support for FFmpeg."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import re
|
||||
from typing import Optional
|
||||
|
||||
from haffmpeg.tools import IMAGE_JPEG, FFVersion, ImageFrame
|
||||
import voluptuous as vol
|
||||
|
@ -93,7 +94,7 @@ async def async_get_image(
|
|||
hass: HomeAssistantType,
|
||||
input_source: str,
|
||||
output_format: str = IMAGE_JPEG,
|
||||
extra_cmd: Optional[str] = None,
|
||||
extra_cmd: str | None = None,
|
||||
):
|
||||
"""Get an image from a frame of an RTSP stream."""
|
||||
manager = hass.data[DATA_FFMPEG]
|
||||
|
|
|
@ -1,7 +1,8 @@
|
|||
"""Support for the Fibaro devices."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import defaultdict
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
from fiblary3.client.v4.client import Client as FibaroClient, StateHandler
|
||||
import voluptuous as vol
|
||||
|
@ -496,7 +497,7 @@ class FibaroDevice(Entity):
|
|||
return self.fibaro_device.unique_id_str
|
||||
|
||||
@property
|
||||
def name(self) -> Optional[str]:
|
||||
def name(self) -> str | None:
|
||||
"""Return the name of the device."""
|
||||
return self._name
|
||||
|
||||
|
|
|
@ -1,4 +1,6 @@
|
|||
"""Allows the creation of a sensor that filters state property."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import Counter, deque
|
||||
from copy import copy
|
||||
from datetime import timedelta
|
||||
|
@ -6,7 +8,6 @@ from functools import partial
|
|||
import logging
|
||||
from numbers import Number
|
||||
import statistics
|
||||
from typing import Optional
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
|
@ -394,8 +395,8 @@ class Filter:
|
|||
self,
|
||||
name,
|
||||
window_size: int = 1,
|
||||
precision: Optional[int] = None,
|
||||
entity: Optional[str] = None,
|
||||
precision: int | None = None,
|
||||
entity: str | None = None,
|
||||
):
|
||||
"""Initialize common attributes.
|
||||
|
||||
|
@ -463,9 +464,9 @@ class RangeFilter(Filter):
|
|||
def __init__(
|
||||
self,
|
||||
entity,
|
||||
precision: Optional[int] = DEFAULT_PRECISION,
|
||||
lower_bound: Optional[float] = None,
|
||||
upper_bound: Optional[float] = None,
|
||||
precision: int | None = DEFAULT_PRECISION,
|
||||
lower_bound: float | None = None,
|
||||
upper_bound: float | None = None,
|
||||
):
|
||||
"""Initialize Filter.
|
||||
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
"""Entity for Firmata devices."""
|
||||
from typing import Type
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
|
||||
|
@ -32,7 +32,7 @@ class FirmataPinEntity(FirmataEntity):
|
|||
|
||||
def __init__(
|
||||
self,
|
||||
api: Type[FirmataBoardPin],
|
||||
api: type[FirmataBoardPin],
|
||||
config_entry: ConfigEntry,
|
||||
name: str,
|
||||
pin: FirmataPinType,
|
||||
|
|
|
@ -1,7 +1,7 @@
|
|||
"""Support for Firmata light output."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Type
|
||||
|
||||
from homeassistant.components.light import (
|
||||
ATTR_BRIGHTNESS,
|
||||
|
@ -55,7 +55,7 @@ class FirmataLight(FirmataPinEntity, LightEntity):
|
|||
|
||||
def __init__(
|
||||
self,
|
||||
api: Type[FirmataBoardPin],
|
||||
api: type[FirmataBoardPin],
|
||||
config_entry: ConfigEntry,
|
||||
name: str,
|
||||
pin: FirmataPinType,
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
"""Platform for Flexit AC units with CI66 Modbus adapter."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import List
|
||||
|
||||
from pyflexit.pyflexit import pyflexit
|
||||
import voluptuous as vol
|
||||
|
@ -135,7 +136,7 @@ class Flexit(ClimateEntity):
|
|||
return self._current_operation
|
||||
|
||||
@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.
|
||||
|
|
|
@ -1,6 +1,5 @@
|
|||
"""Support for Flo Water Monitor binary sensors."""
|
||||
|
||||
from typing import List
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.components.binary_sensor import (
|
||||
DEVICE_CLASS_PROBLEM,
|
||||
|
@ -14,7 +13,7 @@ from .entity import FloEntity
|
|||
|
||||
async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
"""Set up the Flo sensors from config entry."""
|
||||
devices: List[FloDeviceDataUpdateCoordinator] = hass.data[FLO_DOMAIN][
|
||||
devices: list[FloDeviceDataUpdateCoordinator] = hass.data[FLO_DOMAIN][
|
||||
config_entry.entry_id
|
||||
]["devices"]
|
||||
entities = []
|
||||
|
|
|
@ -1,7 +1,9 @@
|
|||
"""Flo device object."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Any, Dict, Optional
|
||||
from typing import Any
|
||||
|
||||
from aioflo.api import API
|
||||
from aioflo.errors import RequestError
|
||||
|
@ -26,8 +28,8 @@ class FloDeviceDataUpdateCoordinator(DataUpdateCoordinator):
|
|||
self._flo_location_id: str = location_id
|
||||
self._flo_device_id: str = device_id
|
||||
self._manufacturer: str = "Flo by Moen"
|
||||
self._device_information: Optional[Dict[str, Any]] = None
|
||||
self._water_usage: Optional[Dict[str, Any]] = None
|
||||
self._device_information: dict[str, Any] | None = None
|
||||
self._water_usage: dict[str, Any] | None = None
|
||||
super().__init__(
|
||||
hass,
|
||||
LOGGER,
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
"""Base entity class for Flo entities."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Dict
|
||||
from typing import Any
|
||||
|
||||
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
|
||||
from homeassistant.helpers.entity import Entity
|
||||
|
@ -36,7 +37,7 @@ class FloEntity(Entity):
|
|||
return self._unique_id
|
||||
|
||||
@property
|
||||
def device_info(self) -> Dict[str, Any]:
|
||||
def device_info(self) -> dict[str, Any]:
|
||||
"""Return a device description for device registry."""
|
||||
return {
|
||||
"identifiers": {(FLO_DOMAIN, self._device.id)},
|
||||
|
|
|
@ -1,6 +1,5 @@
|
|||
"""Support for Flo Water Monitor sensors."""
|
||||
|
||||
from typing import List, Optional
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.const import (
|
||||
DEVICE_CLASS_BATTERY,
|
||||
|
@ -31,7 +30,7 @@ NAME_BATTERY = "Battery"
|
|||
|
||||
async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
"""Set up the Flo sensors from config entry."""
|
||||
devices: List[FloDeviceDataUpdateCoordinator] = hass.data[FLO_DOMAIN][
|
||||
devices: list[FloDeviceDataUpdateCoordinator] = hass.data[FLO_DOMAIN][
|
||||
config_entry.entry_id
|
||||
]["devices"]
|
||||
entities = []
|
||||
|
@ -71,7 +70,7 @@ class FloDailyUsageSensor(FloEntity):
|
|||
return WATER_ICON
|
||||
|
||||
@property
|
||||
def state(self) -> Optional[float]:
|
||||
def state(self) -> float | None:
|
||||
"""Return the current daily usage."""
|
||||
if self._device.consumption_today is None:
|
||||
return None
|
||||
|
@ -92,7 +91,7 @@ class FloSystemModeSensor(FloEntity):
|
|||
self._state: str = None
|
||||
|
||||
@property
|
||||
def state(self) -> Optional[str]:
|
||||
def state(self) -> str | None:
|
||||
"""Return the current system mode."""
|
||||
if not self._device.current_system_mode:
|
||||
return None
|
||||
|
@ -113,7 +112,7 @@ class FloCurrentFlowRateSensor(FloEntity):
|
|||
return GAUGE_ICON
|
||||
|
||||
@property
|
||||
def state(self) -> Optional[float]:
|
||||
def state(self) -> float | None:
|
||||
"""Return the current flow rate."""
|
||||
if self._device.current_flow_rate is None:
|
||||
return None
|
||||
|
@ -134,7 +133,7 @@ class FloTemperatureSensor(FloEntity):
|
|||
self._state: float = None
|
||||
|
||||
@property
|
||||
def state(self) -> Optional[float]:
|
||||
def state(self) -> float | None:
|
||||
"""Return the current temperature."""
|
||||
if self._device.temperature is None:
|
||||
return None
|
||||
|
@ -146,7 +145,7 @@ class FloTemperatureSensor(FloEntity):
|
|||
return TEMP_FAHRENHEIT
|
||||
|
||||
@property
|
||||
def device_class(self) -> Optional[str]:
|
||||
def device_class(self) -> str | None:
|
||||
"""Return the device class for this sensor."""
|
||||
return DEVICE_CLASS_TEMPERATURE
|
||||
|
||||
|
@ -160,7 +159,7 @@ class FloHumiditySensor(FloEntity):
|
|||
self._state: float = None
|
||||
|
||||
@property
|
||||
def state(self) -> Optional[float]:
|
||||
def state(self) -> float | None:
|
||||
"""Return the current humidity."""
|
||||
if self._device.humidity is None:
|
||||
return None
|
||||
|
@ -172,7 +171,7 @@ class FloHumiditySensor(FloEntity):
|
|||
return PERCENTAGE
|
||||
|
||||
@property
|
||||
def device_class(self) -> Optional[str]:
|
||||
def device_class(self) -> str | None:
|
||||
"""Return the device class for this sensor."""
|
||||
return DEVICE_CLASS_HUMIDITY
|
||||
|
||||
|
@ -186,7 +185,7 @@ class FloPressureSensor(FloEntity):
|
|||
self._state: float = None
|
||||
|
||||
@property
|
||||
def state(self) -> Optional[float]:
|
||||
def state(self) -> float | None:
|
||||
"""Return the current water pressure."""
|
||||
if self._device.current_psi is None:
|
||||
return None
|
||||
|
@ -198,7 +197,7 @@ class FloPressureSensor(FloEntity):
|
|||
return PRESSURE_PSI
|
||||
|
||||
@property
|
||||
def device_class(self) -> Optional[str]:
|
||||
def device_class(self) -> str | None:
|
||||
"""Return the device class for this sensor."""
|
||||
return DEVICE_CLASS_PRESSURE
|
||||
|
||||
|
@ -212,7 +211,7 @@ class FloBatterySensor(FloEntity):
|
|||
self._state: float = None
|
||||
|
||||
@property
|
||||
def state(self) -> Optional[float]:
|
||||
def state(self) -> float | None:
|
||||
"""Return the current battery level."""
|
||||
return self._device.battery_level
|
||||
|
||||
|
@ -222,6 +221,6 @@ class FloBatterySensor(FloEntity):
|
|||
return PERCENTAGE
|
||||
|
||||
@property
|
||||
def device_class(self) -> Optional[str]:
|
||||
def device_class(self) -> str | None:
|
||||
"""Return the device class for this sensor."""
|
||||
return DEVICE_CLASS_BATTERY
|
||||
|
|
|
@ -1,6 +1,5 @@
|
|||
"""Switch representing the shutoff valve for the Flo by Moen integration."""
|
||||
|
||||
from typing import List
|
||||
from __future__ import annotations
|
||||
|
||||
from aioflo.location import SLEEP_MINUTE_OPTIONS, SYSTEM_MODE_HOME, SYSTEM_REVERT_MODES
|
||||
import voluptuous as vol
|
||||
|
@ -23,7 +22,7 @@ SERVICE_RUN_HEALTH_TEST = "run_health_test"
|
|||
|
||||
async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
"""Set up the Flo switches from config entry."""
|
||||
devices: List[FloDeviceDataUpdateCoordinator] = hass.data[FLO_DOMAIN][
|
||||
devices: list[FloDeviceDataUpdateCoordinator] = hass.data[FLO_DOMAIN][
|
||||
config_entry.entry_id
|
||||
]["devices"]
|
||||
entities = []
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
"""Support for Freebox devices (Freebox v6 and Freebox mini 4K)."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Dict
|
||||
|
||||
from homeassistant.components.device_tracker import SOURCE_TYPE_ROUTER
|
||||
from homeassistant.components.device_tracker.config_entry import ScannerEntity
|
||||
|
@ -52,7 +53,7 @@ def add_entities(router, async_add_entities, tracked):
|
|||
class FreeboxDevice(ScannerEntity):
|
||||
"""Representation of a Freebox device."""
|
||||
|
||||
def __init__(self, router: FreeboxRouter, device: Dict[str, any]) -> None:
|
||||
def __init__(self, router: FreeboxRouter, device: dict[str, any]) -> None:
|
||||
"""Initialize a Freebox device."""
|
||||
self._router = router
|
||||
self._name = device["primary_name"].strip() or DEFAULT_DEVICE_NAME
|
||||
|
@ -105,12 +106,12 @@ class FreeboxDevice(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)},
|
||||
|
|
|
@ -1,9 +1,11 @@
|
|||
"""Represent the Freebox router and its devices and sensors."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
from typing import Any
|
||||
|
||||
from freebox_api import Freepybox
|
||||
from freebox_api.api.wifi import Wifi
|
||||
|
@ -60,11 +62,11 @@ class FreeboxRouter:
|
|||
self._sw_v = None
|
||||
self._attrs = {}
|
||||
|
||||
self.devices: Dict[str, Dict[str, Any]] = {}
|
||||
self.disks: Dict[int, Dict[str, Any]] = {}
|
||||
self.sensors_temperature: Dict[str, int] = {}
|
||||
self.sensors_connection: Dict[str, float] = {}
|
||||
self.call_list: List[Dict[str, Any]] = []
|
||||
self.devices: dict[str, dict[str, Any]] = {}
|
||||
self.disks: dict[int, dict[str, Any]] = {}
|
||||
self.sensors_temperature: dict[str, int] = {}
|
||||
self.sensors_connection: dict[str, float] = {}
|
||||
self.call_list: list[dict[str, Any]] = []
|
||||
|
||||
self._unsub_dispatcher = None
|
||||
self.listeners = []
|
||||
|
@ -91,7 +93,7 @@ class FreeboxRouter:
|
|||
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 Freebox platforms."""
|
||||
await self.update_device_trackers()
|
||||
await self.update_sensors()
|
||||
|
@ -99,7 +101,7 @@ class FreeboxRouter:
|
|||
async def update_device_trackers(self) -> None:
|
||||
"""Update Freebox devices."""
|
||||
new_device = False
|
||||
fbx_devices: [Dict[str, Any]] = await self._api.lan.get_hosts_list()
|
||||
fbx_devices: [dict[str, Any]] = await self._api.lan.get_hosts_list()
|
||||
|
||||
# Adds the Freebox itself
|
||||
fbx_devices.append(
|
||||
|
@ -129,7 +131,7 @@ class FreeboxRouter:
|
|||
async def update_sensors(self) -> None:
|
||||
"""Update Freebox sensors."""
|
||||
# System sensors
|
||||
syst_datas: Dict[str, Any] = await self._api.system.get_config()
|
||||
syst_datas: dict[str, Any] = await self._api.system.get_config()
|
||||
|
||||
# According to the doc `syst_datas["sensors"]` is temperature sensors in celsius degree.
|
||||
# Name and id of sensors may vary under Freebox devices.
|
||||
|
@ -137,7 +139,7 @@ class FreeboxRouter:
|
|||
self.sensors_temperature[sensor["name"]] = sensor["value"]
|
||||
|
||||
# Connection sensors
|
||||
connection_datas: Dict[str, Any] = await self._api.connection.get_status()
|
||||
connection_datas: dict[str, Any] = await self._api.connection.get_status()
|
||||
for sensor_key in CONNECTION_SENSORS:
|
||||
self.sensors_connection[sensor_key] = connection_datas[sensor_key]
|
||||
|
||||
|
@ -161,7 +163,7 @@ class FreeboxRouter:
|
|||
async def _update_disks_sensors(self) -> None:
|
||||
"""Update Freebox disks."""
|
||||
# None at first request
|
||||
fbx_disks: [Dict[str, Any]] = await self._api.storage.get_disks() or []
|
||||
fbx_disks: [dict[str, Any]] = await self._api.storage.get_disks() or []
|
||||
|
||||
for fbx_disk in fbx_disks:
|
||||
self.disks[fbx_disk["id"]] = fbx_disk
|
||||
|
@ -178,7 +180,7 @@ class FreeboxRouter:
|
|||
self._api = None
|
||||
|
||||
@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)},
|
||||
|
@ -204,7 +206,7 @@ class FreeboxRouter:
|
|||
return f"{DOMAIN}-{self._host}-sensor-update"
|
||||
|
||||
@property
|
||||
def sensors(self) -> Dict[str, Any]:
|
||||
def sensors(self) -> dict[str, Any]:
|
||||
"""Return sensors."""
|
||||
return {**self.sensors_temperature, **self.sensors_connection}
|
||||
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
"""Support for Freebox devices (Freebox v6 and Freebox mini 4K)."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Dict
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import DATA_RATE_KILOBYTES_PER_SECOND
|
||||
|
@ -77,7 +78,7 @@ class FreeboxSensor(Entity):
|
|||
"""Representation of a Freebox sensor."""
|
||||
|
||||
def __init__(
|
||||
self, router: FreeboxRouter, sensor_type: str, sensor: Dict[str, any]
|
||||
self, router: FreeboxRouter, sensor_type: str, sensor: dict[str, any]
|
||||
) -> None:
|
||||
"""Initialize a Freebox sensor."""
|
||||
self._state = None
|
||||
|
@ -129,7 +130,7 @@ class FreeboxSensor(Entity):
|
|||
return self._device_class
|
||||
|
||||
@property
|
||||
def device_info(self) -> Dict[str, any]:
|
||||
def device_info(self) -> dict[str, any]:
|
||||
"""Return the device information."""
|
||||
return self._router.device_info
|
||||
|
||||
|
@ -160,7 +161,7 @@ class FreeboxCallSensor(FreeboxSensor):
|
|||
"""Representation of a Freebox call sensor."""
|
||||
|
||||
def __init__(
|
||||
self, router: FreeboxRouter, sensor_type: str, sensor: Dict[str, any]
|
||||
self, router: FreeboxRouter, sensor_type: str, sensor: dict[str, any]
|
||||
) -> None:
|
||||
"""Initialize a Freebox call sensor."""
|
||||
super().__init__(router, sensor_type, sensor)
|
||||
|
@ -180,7 +181,7 @@ class FreeboxCallSensor(FreeboxSensor):
|
|||
self._state = len(self._call_list_for_type)
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> Dict[str, any]:
|
||||
def extra_state_attributes(self) -> dict[str, any]:
|
||||
"""Return device specific state attributes."""
|
||||
return {
|
||||
dt_util.utc_from_timestamp(call["datetime"]).isoformat(): call["name"]
|
||||
|
@ -194,10 +195,10 @@ class FreeboxDiskSensor(FreeboxSensor):
|
|||
def __init__(
|
||||
self,
|
||||
router: FreeboxRouter,
|
||||
disk: Dict[str, any],
|
||||
partition: Dict[str, any],
|
||||
disk: dict[str, any],
|
||||
partition: dict[str, any],
|
||||
sensor_type: str,
|
||||
sensor: Dict[str, any],
|
||||
sensor: dict[str, any],
|
||||
) -> None:
|
||||
"""Initialize a Freebox disk sensor."""
|
||||
super().__init__(router, sensor_type, sensor)
|
||||
|
@ -207,7 +208,7 @@ class FreeboxDiskSensor(FreeboxSensor):
|
|||
self._unique_id = f"{self._router.mac} {sensor_type} {self._disk['id']} {self._partition['id']}"
|
||||
|
||||
@property
|
||||
def device_info(self) -> Dict[str, any]:
|
||||
def device_info(self) -> dict[str, any]:
|
||||
"""Return the device information."""
|
||||
return {
|
||||
"identifiers": {(DOMAIN, self._disk["id"])},
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
"""Support for Freebox Delta, Revolution and Mini 4K."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Dict
|
||||
|
||||
from freebox_api.exceptions import InsufficientPermissionsError
|
||||
|
||||
|
@ -48,7 +49,7 @@ class FreeboxWifiSwitch(SwitchEntity):
|
|||
return self._state
|
||||
|
||||
@property
|
||||
def device_info(self) -> Dict[str, any]:
|
||||
def device_info(self) -> dict[str, any]:
|
||||
"""Return the device information."""
|
||||
return self._router.device_info
|
||||
|
||||
|
|
|
@ -1,8 +1,9 @@
|
|||
"""Support for Fronius devices."""
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from typing import Dict
|
||||
|
||||
from pyfronius import Fronius
|
||||
import voluptuous as vol
|
||||
|
@ -195,7 +196,7 @@ class FroniusAdapter:
|
|||
for sensor in self._registered_sensors:
|
||||
sensor.async_schedule_update_ha_state(True)
|
||||
|
||||
async def _update(self) -> Dict:
|
||||
async def _update(self) -> dict:
|
||||
"""Return values of interest."""
|
||||
|
||||
async def register(self, sensor):
|
||||
|
|
|
@ -1,10 +1,12 @@
|
|||
"""Handle the frontend for Home Assistant."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import mimetypes
|
||||
import os
|
||||
import pathlib
|
||||
from typing import Any, Dict, Optional, Set, Tuple
|
||||
from typing import Any
|
||||
|
||||
from aiohttp import hdrs, web, web_urldispatcher
|
||||
import jinja2
|
||||
|
@ -119,19 +121,19 @@ class Panel:
|
|||
"""Abstract class for panels."""
|
||||
|
||||
# Name of the webcomponent
|
||||
component_name: Optional[str] = None
|
||||
component_name: str | None = None
|
||||
|
||||
# Icon to show in the sidebar
|
||||
sidebar_icon: Optional[str] = None
|
||||
sidebar_icon: str | None = None
|
||||
|
||||
# Title to show in the sidebar
|
||||
sidebar_title: Optional[str] = None
|
||||
sidebar_title: str | None = None
|
||||
|
||||
# Url to show the panel in the frontend
|
||||
frontend_url_path: Optional[str] = None
|
||||
frontend_url_path: str | None = None
|
||||
|
||||
# Config to pass to the webcomponent
|
||||
config: Optional[Dict[str, Any]] = None
|
||||
config: dict[str, Any] | None = None
|
||||
|
||||
# If the panel should only be visible to admins
|
||||
require_admin = False
|
||||
|
@ -443,7 +445,7 @@ class IndexView(web_urldispatcher.AbstractResource):
|
|||
|
||||
async def resolve(
|
||||
self, request: web.Request
|
||||
) -> Tuple[Optional[web_urldispatcher.UrlMappingMatchInfo], Set[str]]:
|
||||
) -> tuple[web_urldispatcher.UrlMappingMatchInfo | None, set[str]]:
|
||||
"""Resolve resource.
|
||||
|
||||
Return (UrlMappingMatchInfo, allowed_methods) pair.
|
||||
|
|
Loading…
Add table
Reference in a new issue