Update typing 09 (#48059)
parent
2ab640aaef
commit
283b4abe67
|
@ -1,8 +1,10 @@
|
|||
"""Component to embed Aqualink devices."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from functools import wraps
|
||||
import logging
|
||||
from typing import Any, Dict
|
||||
from typing import Any
|
||||
|
||||
import aiohttp.client_exceptions
|
||||
from iaqualink import (
|
||||
|
@ -234,7 +236,7 @@ class AqualinkEntity(Entity):
|
|||
return self.dev.system.online
|
||||
|
||||
@property
|
||||
def device_info(self) -> Dict[str, Any]:
|
||||
def device_info(self) -> dict[str, Any]:
|
||||
"""Return the device info."""
|
||||
return {
|
||||
"identifiers": {(DOMAIN, self.unique_id)},
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
"""Support for Aqualink Thermostats."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import List, Optional
|
||||
|
||||
from iaqualink import AqualinkHeater, AqualinkPump, AqualinkSensor, AqualinkState
|
||||
from iaqualink.const import (
|
||||
|
@ -53,7 +54,7 @@ class HassAqualinkThermostat(AqualinkEntity, ClimateEntity):
|
|||
return SUPPORT_TARGET_TEMPERATURE
|
||||
|
||||
@property
|
||||
def hvac_modes(self) -> List[str]:
|
||||
def hvac_modes(self) -> list[str]:
|
||||
"""Return the list of supported HVAC modes."""
|
||||
return CLIMATE_SUPPORTED_MODES
|
||||
|
||||
|
@ -119,7 +120,7 @@ class HassAqualinkThermostat(AqualinkEntity, ClimateEntity):
|
|||
return self.dev.system.devices[sensor]
|
||||
|
||||
@property
|
||||
def current_temperature(self) -> Optional[float]:
|
||||
def current_temperature(self) -> float | None:
|
||||
"""Return the current temperature."""
|
||||
if self.sensor.state != "":
|
||||
return float(self.sensor.state)
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
"""Config flow to configure zone component."""
|
||||
from typing import Optional
|
||||
from __future__ import annotations
|
||||
|
||||
from iaqualink import AqualinkClient, AqualinkLoginException
|
||||
import voluptuous as vol
|
||||
|
@ -18,7 +18,7 @@ class AqualinkFlowHandler(config_entries.ConfigFlow):
|
|||
VERSION = 1
|
||||
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
|
||||
|
||||
async def async_step_user(self, user_input: Optional[ConfigType] = None):
|
||||
async def async_step_user(self, user_input: ConfigType | None = None):
|
||||
"""Handle a flow start."""
|
||||
# Supporting a single account.
|
||||
entries = self.hass.config_entries.async_entries(DOMAIN)
|
||||
|
@ -46,6 +46,6 @@ class AqualinkFlowHandler(config_entries.ConfigFlow):
|
|||
errors=errors,
|
||||
)
|
||||
|
||||
async def async_step_import(self, user_input: Optional[ConfigType] = None):
|
||||
async def async_step_import(self, user_input: ConfigType | None = None):
|
||||
"""Occurs when an entry is setup through config."""
|
||||
return await self.async_step_user(user_input)
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
"""Support for Aqualink temperature sensors."""
|
||||
from typing import Optional
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.components.sensor import DOMAIN
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
|
@ -31,7 +31,7 @@ class HassAqualinkSensor(AqualinkEntity):
|
|||
return self.dev.label
|
||||
|
||||
@property
|
||||
def unit_of_measurement(self) -> Optional[str]:
|
||||
def unit_of_measurement(self) -> str | None:
|
||||
"""Return the measurement unit for the sensor."""
|
||||
if self.dev.name.endswith("_temp"):
|
||||
if self.dev.system.temp_unit == "F":
|
||||
|
@ -40,7 +40,7 @@ class HassAqualinkSensor(AqualinkEntity):
|
|||
return None
|
||||
|
||||
@property
|
||||
def state(self) -> Optional[str]:
|
||||
def state(self) -> str | None:
|
||||
"""Return the state of the sensor."""
|
||||
if self.dev.state == "":
|
||||
return None
|
||||
|
@ -52,7 +52,7 @@ class HassAqualinkSensor(AqualinkEntity):
|
|||
return state
|
||||
|
||||
@property
|
||||
def device_class(self) -> Optional[str]:
|
||||
def device_class(self) -> str | None:
|
||||
"""Return the class of the sensor."""
|
||||
if self.dev.name.endswith("_temp"):
|
||||
return DEVICE_CLASS_TEMPERATURE
|
||||
|
|
|
@ -1,8 +1,9 @@
|
|||
"""iCloud account."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
import operator
|
||||
from typing import Dict, Optional
|
||||
|
||||
from pyicloud import PyiCloudService
|
||||
from pyicloud.exceptions import (
|
||||
|
@ -95,7 +96,7 @@ class IcloudAccount:
|
|||
|
||||
self._icloud_dir = icloud_dir
|
||||
|
||||
self.api: Optional[PyiCloudService] = None
|
||||
self.api: PyiCloudService | None = None
|
||||
self._owner_fullname = None
|
||||
self._family_members_fullname = {}
|
||||
self._devices = {}
|
||||
|
@ -345,7 +346,7 @@ class IcloudAccount:
|
|||
return self._owner_fullname
|
||||
|
||||
@property
|
||||
def family_members_fullname(self) -> Dict[str, str]:
|
||||
def family_members_fullname(self) -> dict[str, str]:
|
||||
"""Return the account family members fullname."""
|
||||
return self._family_members_fullname
|
||||
|
||||
|
@ -355,7 +356,7 @@ class IcloudAccount:
|
|||
return self._fetch_interval
|
||||
|
||||
@property
|
||||
def devices(self) -> Dict[str, any]:
|
||||
def devices(self) -> dict[str, any]:
|
||||
"""Return the account devices."""
|
||||
return self._devices
|
||||
|
||||
|
@ -496,11 +497,11 @@ class IcloudDevice:
|
|||
return self._battery_status
|
||||
|
||||
@property
|
||||
def location(self) -> Dict[str, any]:
|
||||
def location(self) -> dict[str, any]:
|
||||
"""Return the Apple device location."""
|
||||
return self._location
|
||||
|
||||
@property
|
||||
def state_attributes(self) -> Dict[str, any]:
|
||||
def state_attributes(self) -> dict[str, any]:
|
||||
"""Return the attributes."""
|
||||
return self._attrs
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
"""Support for tracking for iCloud devices."""
|
||||
from typing import Dict
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.components.device_tracker import SOURCE_TYPE_GPS
|
||||
from homeassistant.components.device_tracker.config_entry import TrackerEntity
|
||||
|
@ -108,12 +108,12 @@ class IcloudTrackerEntity(TrackerEntity):
|
|||
return icon_for_icloud_device(self._device)
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> Dict[str, any]:
|
||||
def extra_state_attributes(self) -> dict[str, any]:
|
||||
"""Return the device state attributes."""
|
||||
return self._device.state_attributes
|
||||
|
||||
@property
|
||||
def device_info(self) -> Dict[str, any]:
|
||||
def device_info(self) -> dict[str, any]:
|
||||
"""Return the device information."""
|
||||
return {
|
||||
"identifiers": {(DOMAIN, self._device.unique_id)},
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
"""Support for iCloud sensors."""
|
||||
from typing import Dict
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import DEVICE_CLASS_BATTERY, PERCENTAGE
|
||||
|
@ -91,12 +91,12 @@ class IcloudDeviceBatterySensor(Entity):
|
|||
)
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> Dict[str, any]:
|
||||
def extra_state_attributes(self) -> dict[str, any]:
|
||||
"""Return default attributes for the iCloud device entity."""
|
||||
return self._device.state_attributes
|
||||
|
||||
@property
|
||||
def device_info(self) -> Dict[str, any]:
|
||||
def device_info(self) -> dict[str, any]:
|
||||
"""Return the device information."""
|
||||
return {
|
||||
"identifiers": {(DOMAIN, self._device.unique_id)},
|
||||
|
|
|
@ -1,7 +1,8 @@
|
|||
"""Support for IGN Sismologia (Earthquakes) Feeds."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
from georss_ign_sismologia_client import IgnSismologiaFeedManager
|
||||
import voluptuous as vol
|
||||
|
@ -207,7 +208,7 @@ class IgnSismologiaLocationEvent(GeolocationEvent):
|
|||
return SOURCE
|
||||
|
||||
@property
|
||||
def name(self) -> Optional[str]:
|
||||
def name(self) -> str | None:
|
||||
"""Return the name of the entity."""
|
||||
if self._magnitude and self._region:
|
||||
return f"M {self._magnitude:.1f} - {self._region}"
|
||||
|
@ -218,17 +219,17 @@ class IgnSismologiaLocationEvent(GeolocationEvent):
|
|||
return self._title
|
||||
|
||||
@property
|
||||
def distance(self) -> Optional[float]:
|
||||
def distance(self) -> float | None:
|
||||
"""Return distance value of this external event."""
|
||||
return self._distance
|
||||
|
||||
@property
|
||||
def latitude(self) -> Optional[float]:
|
||||
def latitude(self) -> float | None:
|
||||
"""Return latitude value of this external event."""
|
||||
return self._latitude
|
||||
|
||||
@property
|
||||
def longitude(self) -> Optional[float]:
|
||||
def longitude(self) -> float | None:
|
||||
"""Return longitude value of this external event."""
|
||||
return self._longitude
|
||||
|
||||
|
|
|
@ -1,10 +1,11 @@
|
|||
"""The Picture integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import pathlib
|
||||
import secrets
|
||||
import shutil
|
||||
import typing
|
||||
|
||||
from PIL import Image, ImageOps, UnidentifiedImageError
|
||||
from aiohttp import hdrs, web
|
||||
|
@ -69,7 +70,7 @@ class ImageStorageCollection(collection.StorageCollection):
|
|||
self.async_add_listener(self._change_listener)
|
||||
self.image_dir = image_dir
|
||||
|
||||
async def _process_create_data(self, data: typing.Dict) -> typing.Dict:
|
||||
async def _process_create_data(self, data: dict) -> dict:
|
||||
"""Validate the config is valid."""
|
||||
data = self.CREATE_SCHEMA(dict(data))
|
||||
uploaded_file: FileField = data["file"]
|
||||
|
@ -117,11 +118,11 @@ class ImageStorageCollection(collection.StorageCollection):
|
|||
return media_file.stat().st_size
|
||||
|
||||
@callback
|
||||
def _get_suggested_id(self, info: typing.Dict) -> str:
|
||||
def _get_suggested_id(self, info: dict) -> str:
|
||||
"""Suggest an ID based on the config."""
|
||||
return info[CONF_ID]
|
||||
|
||||
async def _update_data(self, data: dict, update_data: typing.Dict) -> typing.Dict:
|
||||
async def _update_data(self, data: dict, update_data: dict) -> dict:
|
||||
"""Return a new updated data object."""
|
||||
return {**data, **self.UPDATE_SCHEMA(update_data)}
|
||||
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
"""Support for an Intergas boiler via an InComfort/Intouch Lan2RF gateway."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
from aiohttp import ClientResponseError
|
||||
from incomfortclient import Gateway as InComfortGateway
|
||||
|
@ -68,12 +69,12 @@ class IncomfortEntity(Entity):
|
|||
self._unique_id = self._name = None
|
||||
|
||||
@property
|
||||
def unique_id(self) -> Optional[str]:
|
||||
def unique_id(self) -> str | None:
|
||||
"""Return a unique ID."""
|
||||
return self._unique_id
|
||||
|
||||
@property
|
||||
def name(self) -> Optional[str]:
|
||||
def name(self) -> str | None:
|
||||
"""Return the name of the sensor."""
|
||||
return self._name
|
||||
|
||||
|
|
|
@ -1,5 +1,7 @@
|
|||
"""Support for an Intergas heater via an InComfort/InTouch Lan2RF gateway."""
|
||||
from typing import Any, Dict, Optional
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from homeassistant.components.binary_sensor import (
|
||||
DOMAIN as BINARY_SENSOR_DOMAIN,
|
||||
|
@ -40,6 +42,6 @@ class IncomfortFailed(IncomfortChild, BinarySensorEntity):
|
|||
return self._heater.status["is_failed"]
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> Optional[Dict[str, Any]]:
|
||||
def extra_state_attributes(self) -> dict[str, Any] | None:
|
||||
"""Return the device state attributes."""
|
||||
return {"fault_code": self._heater.status["fault_code"]}
|
||||
|
|
|
@ -1,5 +1,7 @@
|
|||
"""Support for an Intergas boiler via an InComfort/InTouch Lan2RF gateway."""
|
||||
from typing import Any, Dict, List, Optional
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from homeassistant.components.climate import DOMAIN as CLIMATE_DOMAIN, ClimateEntity
|
||||
from homeassistant.components.climate.const import (
|
||||
|
@ -39,7 +41,7 @@ class InComfortClimate(IncomfortChild, ClimateEntity):
|
|||
self._room = room
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> Dict[str, Any]:
|
||||
def extra_state_attributes(self) -> dict[str, Any]:
|
||||
"""Return the device state attributes."""
|
||||
return {"status": self._room.status}
|
||||
|
||||
|
@ -54,17 +56,17 @@ class InComfortClimate(IncomfortChild, ClimateEntity):
|
|||
return HVAC_MODE_HEAT
|
||||
|
||||
@property
|
||||
def hvac_modes(self) -> List[str]:
|
||||
def hvac_modes(self) -> list[str]:
|
||||
"""Return the list of available hvac operation modes."""
|
||||
return [HVAC_MODE_HEAT]
|
||||
|
||||
@property
|
||||
def current_temperature(self) -> Optional[float]:
|
||||
def current_temperature(self) -> float | None:
|
||||
"""Return the current temperature."""
|
||||
return self._room.room_temp
|
||||
|
||||
@property
|
||||
def target_temperature(self) -> Optional[float]:
|
||||
def target_temperature(self) -> float | None:
|
||||
"""Return the temperature we try to reach."""
|
||||
return self._room.setpoint
|
||||
|
||||
|
|
|
@ -1,5 +1,7 @@
|
|||
"""Support for an Intergas heater via an InComfort/InTouch Lan2RF gateway."""
|
||||
from typing import Any, Dict, Optional
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
|
||||
from homeassistant.const import (
|
||||
|
@ -57,17 +59,17 @@ class IncomfortSensor(IncomfortChild):
|
|||
self._unit_of_measurement = None
|
||||
|
||||
@property
|
||||
def state(self) -> Optional[str]:
|
||||
def state(self) -> str | None:
|
||||
"""Return the state of the sensor."""
|
||||
return self._heater.status[self._state_attr]
|
||||
|
||||
@property
|
||||
def device_class(self) -> Optional[str]:
|
||||
def device_class(self) -> str | None:
|
||||
"""Return the device class of the sensor."""
|
||||
return self._device_class
|
||||
|
||||
@property
|
||||
def unit_of_measurement(self) -> Optional[str]:
|
||||
def unit_of_measurement(self) -> str | None:
|
||||
"""Return the unit of measurement of the sensor."""
|
||||
return self._unit_of_measurement
|
||||
|
||||
|
@ -95,6 +97,6 @@ class IncomfortTemperature(IncomfortSensor):
|
|||
self._unit_of_measurement = TEMP_CELSIUS
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> Optional[Dict[str, Any]]:
|
||||
def extra_state_attributes(self) -> dict[str, Any] | None:
|
||||
"""Return the device state attributes."""
|
||||
return {self._attr: self._heater.status[self._attr]}
|
||||
|
|
|
@ -1,7 +1,9 @@
|
|||
"""Support for an Intergas boiler via an InComfort/Intouch Lan2RF gateway."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import Any, Dict
|
||||
from typing import Any
|
||||
|
||||
from aiohttp import ClientResponseError
|
||||
|
||||
|
@ -50,7 +52,7 @@ class IncomfortWaterHeater(IncomfortEntity, WaterHeaterEntity):
|
|||
return "mdi:thermometer-lines"
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> Dict[str, Any]:
|
||||
def extra_state_attributes(self) -> dict[str, Any]:
|
||||
"""Return the device state attributes."""
|
||||
return {k: v for k, v in self._heater.status.items() if k in HEATER_ATTRS}
|
||||
|
||||
|
|
|
@ -1,11 +1,13 @@
|
|||
"""Support for sending data to an Influx database."""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
import logging
|
||||
import math
|
||||
import queue
|
||||
import threading
|
||||
import time
|
||||
from typing import Any, Callable, Dict, List
|
||||
from typing import Any, Callable
|
||||
|
||||
from influxdb import InfluxDBClient, exceptions
|
||||
from influxdb_client import InfluxDBClient as InfluxDBClientV2
|
||||
|
@ -100,7 +102,7 @@ from .const import (
|
|||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def create_influx_url(conf: Dict) -> Dict:
|
||||
def create_influx_url(conf: dict) -> dict:
|
||||
"""Build URL used from config inputs and default when necessary."""
|
||||
if conf[CONF_API_VERSION] == API_VERSION_2:
|
||||
if CONF_SSL not in conf:
|
||||
|
@ -125,7 +127,7 @@ def create_influx_url(conf: Dict) -> Dict:
|
|||
return conf
|
||||
|
||||
|
||||
def validate_version_specific_config(conf: Dict) -> Dict:
|
||||
def validate_version_specific_config(conf: dict) -> dict:
|
||||
"""Ensure correct config fields are provided based on API version used."""
|
||||
if conf[CONF_API_VERSION] == API_VERSION_2:
|
||||
if CONF_TOKEN not in conf:
|
||||
|
@ -193,7 +195,7 @@ CONFIG_SCHEMA = vol.Schema(
|
|||
)
|
||||
|
||||
|
||||
def _generate_event_to_json(conf: Dict) -> Callable[[Dict], str]:
|
||||
def _generate_event_to_json(conf: dict) -> Callable[[dict], str]:
|
||||
"""Build event to json converter and add to config."""
|
||||
entity_filter = convert_include_exclude_filter(conf)
|
||||
tags = conf.get(CONF_TAGS)
|
||||
|
@ -208,7 +210,7 @@ def _generate_event_to_json(conf: Dict) -> Callable[[Dict], str]:
|
|||
conf[CONF_COMPONENT_CONFIG_GLOB],
|
||||
)
|
||||
|
||||
def event_to_json(event: Dict) -> str:
|
||||
def event_to_json(event: dict) -> str:
|
||||
"""Convert event into json in format Influx expects."""
|
||||
state = event.data.get(EVENT_NEW_STATE)
|
||||
if (
|
||||
|
@ -319,9 +321,9 @@ def _generate_event_to_json(conf: Dict) -> Callable[[Dict], str]:
|
|||
class InfluxClient:
|
||||
"""An InfluxDB client wrapper for V1 or V2."""
|
||||
|
||||
data_repositories: List[str]
|
||||
data_repositories: list[str]
|
||||
write: Callable[[str], None]
|
||||
query: Callable[[str, str], List[Any]]
|
||||
query: Callable[[str, str], list[Any]]
|
||||
close: Callable[[], None]
|
||||
|
||||
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
"""InfluxDB component which allows you to get data from an Influx database."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Dict
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
|
@ -67,7 +68,7 @@ def _merge_connection_config_into_query(conf, query):
|
|||
query[key] = conf[key]
|
||||
|
||||
|
||||
def validate_query_format_for_version(conf: Dict) -> Dict:
|
||||
def validate_query_format_for_version(conf: dict) -> dict:
|
||||
"""Ensure queries are provided in correct format based on API version."""
|
||||
if conf[CONF_API_VERSION] == API_VERSION_2:
|
||||
if CONF_QUERIES_FLUX not in conf:
|
||||
|
|
|
@ -2,7 +2,6 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import typing
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
|
@ -62,16 +61,16 @@ class InputBooleanStorageCollection(collection.StorageCollection):
|
|||
CREATE_SCHEMA = vol.Schema(CREATE_FIELDS)
|
||||
UPDATE_SCHEMA = vol.Schema(UPDATE_FIELDS)
|
||||
|
||||
async def _process_create_data(self, data: typing.Dict) -> typing.Dict:
|
||||
async def _process_create_data(self, data: dict) -> dict:
|
||||
"""Validate the config is valid."""
|
||||
return self.CREATE_SCHEMA(data)
|
||||
|
||||
@callback
|
||||
def _get_suggested_id(self, info: typing.Dict) -> str:
|
||||
def _get_suggested_id(self, info: dict) -> str:
|
||||
"""Suggest an ID based on the config."""
|
||||
return info[CONF_NAME]
|
||||
|
||||
async def _update_data(self, data: dict, update_data: typing.Dict) -> typing.Dict:
|
||||
async def _update_data(self, data: dict, update_data: dict) -> dict:
|
||||
"""Return a new updated data object."""
|
||||
update_data = self.UPDATE_SCHEMA(update_data)
|
||||
return {**data, **update_data}
|
||||
|
@ -145,14 +144,14 @@ async def async_setup(hass: HomeAssistantType, config: ConfigType) -> bool:
|
|||
class InputBoolean(ToggleEntity, RestoreEntity):
|
||||
"""Representation of a boolean input."""
|
||||
|
||||
def __init__(self, config: typing.Optional[dict]):
|
||||
def __init__(self, config: dict | None):
|
||||
"""Initialize a boolean input."""
|
||||
self._config = config
|
||||
self.editable = True
|
||||
self._state = config.get(CONF_INITIAL)
|
||||
|
||||
@classmethod
|
||||
def from_yaml(cls, config: typing.Dict) -> InputBoolean:
|
||||
def from_yaml(cls, config: dict) -> InputBoolean:
|
||||
"""Return entity instance initialized from yaml storage."""
|
||||
input_bool = cls(config)
|
||||
input_bool.entity_id = f"{DOMAIN}.{config[CONF_ID]}"
|
||||
|
@ -209,7 +208,7 @@ class InputBoolean(ToggleEntity, RestoreEntity):
|
|||
self._state = False
|
||||
self.async_write_ha_state()
|
||||
|
||||
async def async_update_config(self, config: typing.Dict) -> None:
|
||||
async def async_update_config(self, config: dict) -> None:
|
||||
"""Handle when the config is updated."""
|
||||
self._config = config
|
||||
self.async_write_ha_state()
|
||||
|
|
|
@ -1,7 +1,9 @@
|
|||
"""Reproduce an input boolean 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,
|
||||
|
@ -22,8 +24,8 @@ async def _async_reproduce_states(
|
|||
hass: HomeAssistantType,
|
||||
state: State,
|
||||
*,
|
||||
context: Optional[Context] = None,
|
||||
reproduce_options: Optional[Dict[str, Any]] = None,
|
||||
context: Context | None = None,
|
||||
reproduce_options: dict[str, Any] | None = None,
|
||||
) -> None:
|
||||
"""Reproduce input boolean states."""
|
||||
cur_state = hass.states.get(state.entity_id)
|
||||
|
@ -56,8 +58,8 @@ async def async_reproduce_states(
|
|||
hass: HomeAssistantType,
|
||||
states: Iterable[State],
|
||||
*,
|
||||
context: Optional[Context] = None,
|
||||
reproduce_options: Optional[Dict[str, Any]] = None,
|
||||
context: Context | None = None,
|
||||
reproduce_options: dict[str, Any] | None = None,
|
||||
) -> None:
|
||||
"""Reproduce component states."""
|
||||
await asyncio.gather(
|
||||
|
|
|
@ -3,7 +3,6 @@ from __future__ import annotations
|
|||
|
||||
import datetime as py_datetime
|
||||
import logging
|
||||
import typing
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
|
@ -178,16 +177,16 @@ class DateTimeStorageCollection(collection.StorageCollection):
|
|||
CREATE_SCHEMA = vol.Schema(vol.All(CREATE_FIELDS, has_date_or_time))
|
||||
UPDATE_SCHEMA = vol.Schema(UPDATE_FIELDS)
|
||||
|
||||
async def _process_create_data(self, data: typing.Dict) -> typing.Dict:
|
||||
async def _process_create_data(self, data: dict) -> dict:
|
||||
"""Validate the config is valid."""
|
||||
return self.CREATE_SCHEMA(data)
|
||||
|
||||
@callback
|
||||
def _get_suggested_id(self, info: typing.Dict) -> str:
|
||||
def _get_suggested_id(self, info: dict) -> str:
|
||||
"""Suggest an ID based on the config."""
|
||||
return info[CONF_NAME]
|
||||
|
||||
async def _update_data(self, data: dict, update_data: typing.Dict) -> typing.Dict:
|
||||
async def _update_data(self, data: dict, update_data: dict) -> dict:
|
||||
"""Return a new updated data object."""
|
||||
update_data = self.UPDATE_SCHEMA(update_data)
|
||||
return has_date_or_time({**data, **update_data})
|
||||
|
@ -196,7 +195,7 @@ class DateTimeStorageCollection(collection.StorageCollection):
|
|||
class InputDatetime(RestoreEntity):
|
||||
"""Representation of a datetime input."""
|
||||
|
||||
def __init__(self, config: typing.Dict) -> None:
|
||||
def __init__(self, config: dict) -> None:
|
||||
"""Initialize a select input."""
|
||||
self._config = config
|
||||
self.editable = True
|
||||
|
@ -230,7 +229,7 @@ class InputDatetime(RestoreEntity):
|
|||
)
|
||||
|
||||
@classmethod
|
||||
def from_yaml(cls, config: typing.Dict) -> InputDatetime:
|
||||
def from_yaml(cls, config: dict) -> InputDatetime:
|
||||
"""Return entity instance initialized from yaml storage."""
|
||||
input_dt = cls(config)
|
||||
input_dt.entity_id = f"{DOMAIN}.{config[CONF_ID]}"
|
||||
|
@ -360,7 +359,7 @@ class InputDatetime(RestoreEntity):
|
|||
return attrs
|
||||
|
||||
@property
|
||||
def unique_id(self) -> typing.Optional[str]:
|
||||
def unique_id(self) -> str | None:
|
||||
"""Return unique id of the entity."""
|
||||
return self._config[CONF_ID]
|
||||
|
||||
|
@ -394,7 +393,7 @@ class InputDatetime(RestoreEntity):
|
|||
)
|
||||
self.async_write_ha_state()
|
||||
|
||||
async def async_update_config(self, config: typing.Dict) -> None:
|
||||
async def async_update_config(self, config: dict) -> None:
|
||||
"""Handle when the config is updated."""
|
||||
self._config = config
|
||||
self.async_write_ha_state()
|
||||
|
|
|
@ -1,7 +1,9 @@
|
|||
"""Reproduce an Input datetime state."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import Any, Dict, Iterable, Optional
|
||||
from typing import Any, Iterable
|
||||
|
||||
from homeassistant.const import ATTR_ENTITY_ID
|
||||
from homeassistant.core import Context, State
|
||||
|
@ -35,8 +37,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)
|
||||
|
@ -80,8 +82,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 Input datetime states."""
|
||||
await asyncio.gather(
|
||||
|
|
|
@ -2,7 +2,6 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import typing
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
|
@ -179,16 +178,16 @@ class NumberStorageCollection(collection.StorageCollection):
|
|||
CREATE_SCHEMA = vol.Schema(vol.All(CREATE_FIELDS, _cv_input_number))
|
||||
UPDATE_SCHEMA = vol.Schema(UPDATE_FIELDS)
|
||||
|
||||
async def _process_create_data(self, data: typing.Dict) -> typing.Dict:
|
||||
async def _process_create_data(self, data: dict) -> dict:
|
||||
"""Validate the config is valid."""
|
||||
return self.CREATE_SCHEMA(data)
|
||||
|
||||
@callback
|
||||
def _get_suggested_id(self, info: typing.Dict) -> str:
|
||||
def _get_suggested_id(self, info: dict) -> str:
|
||||
"""Suggest an ID based on the config."""
|
||||
return info[CONF_NAME]
|
||||
|
||||
async def _update_data(self, data: dict, update_data: typing.Dict) -> typing.Dict:
|
||||
async def _update_data(self, data: dict, update_data: dict) -> dict:
|
||||
"""Return a new updated data object."""
|
||||
update_data = self.UPDATE_SCHEMA(update_data)
|
||||
return _cv_input_number({**data, **update_data})
|
||||
|
@ -197,14 +196,14 @@ class NumberStorageCollection(collection.StorageCollection):
|
|||
class InputNumber(RestoreEntity):
|
||||
"""Representation of a slider."""
|
||||
|
||||
def __init__(self, config: typing.Dict):
|
||||
def __init__(self, config: dict):
|
||||
"""Initialize an input number."""
|
||||
self._config = config
|
||||
self.editable = True
|
||||
self._current_value = config.get(CONF_INITIAL)
|
||||
|
||||
@classmethod
|
||||
def from_yaml(cls, config: typing.Dict) -> InputNumber:
|
||||
def from_yaml(cls, config: dict) -> InputNumber:
|
||||
"""Return entity instance initialized from yaml storage."""
|
||||
input_num = cls(config)
|
||||
input_num.entity_id = f"{DOMAIN}.{config[CONF_ID]}"
|
||||
|
@ -252,7 +251,7 @@ class InputNumber(RestoreEntity):
|
|||
return self._config.get(CONF_UNIT_OF_MEASUREMENT)
|
||||
|
||||
@property
|
||||
def unique_id(self) -> typing.Optional[str]:
|
||||
def unique_id(self) -> str | None:
|
||||
"""Return unique id of the entity."""
|
||||
return self._config[CONF_ID]
|
||||
|
||||
|
@ -303,7 +302,7 @@ class InputNumber(RestoreEntity):
|
|||
"""Decrement value."""
|
||||
await self.async_set_value(max(self._current_value - self._step, self._minimum))
|
||||
|
||||
async def async_update_config(self, config: typing.Dict) -> None:
|
||||
async def async_update_config(self, config: dict) -> None:
|
||||
"""Handle when the config is updated."""
|
||||
self._config = config
|
||||
# just in case min/max values changed
|
||||
|
|
|
@ -1,7 +1,9 @@
|
|||
"""Reproduce an Input number state."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import Any, Dict, Iterable, Optional
|
||||
from typing import Any, Iterable
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
|
@ -18,8 +20,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)
|
||||
|
@ -56,8 +58,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 Input number states."""
|
||||
# Reproduce states in parallel.
|
||||
|
|
|
@ -2,7 +2,6 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import typing
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
|
@ -184,16 +183,16 @@ class InputSelectStorageCollection(collection.StorageCollection):
|
|||
CREATE_SCHEMA = vol.Schema(vol.All(CREATE_FIELDS, _cv_input_select))
|
||||
UPDATE_SCHEMA = vol.Schema(UPDATE_FIELDS)
|
||||
|
||||
async def _process_create_data(self, data: typing.Dict) -> typing.Dict:
|
||||
async def _process_create_data(self, data: dict) -> dict:
|
||||
"""Validate the config is valid."""
|
||||
return self.CREATE_SCHEMA(data)
|
||||
|
||||
@callback
|
||||
def _get_suggested_id(self, info: typing.Dict) -> str:
|
||||
def _get_suggested_id(self, info: dict) -> str:
|
||||
"""Suggest an ID based on the config."""
|
||||
return info[CONF_NAME]
|
||||
|
||||
async def _update_data(self, data: dict, update_data: typing.Dict) -> typing.Dict:
|
||||
async def _update_data(self, data: dict, update_data: dict) -> dict:
|
||||
"""Return a new updated data object."""
|
||||
update_data = self.UPDATE_SCHEMA(update_data)
|
||||
return _cv_input_select({**data, **update_data})
|
||||
|
@ -202,14 +201,14 @@ class InputSelectStorageCollection(collection.StorageCollection):
|
|||
class InputSelect(RestoreEntity):
|
||||
"""Representation of a select input."""
|
||||
|
||||
def __init__(self, config: typing.Dict):
|
||||
def __init__(self, config: dict):
|
||||
"""Initialize a select input."""
|
||||
self._config = config
|
||||
self.editable = True
|
||||
self._current_option = config.get(CONF_INITIAL)
|
||||
|
||||
@classmethod
|
||||
def from_yaml(cls, config: typing.Dict) -> InputSelect:
|
||||
def from_yaml(cls, config: dict) -> InputSelect:
|
||||
"""Return entity instance initialized from yaml storage."""
|
||||
input_select = cls(config)
|
||||
input_select.entity_id = f"{DOMAIN}.{config[CONF_ID]}"
|
||||
|
@ -244,7 +243,7 @@ class InputSelect(RestoreEntity):
|
|||
return self._config.get(CONF_ICON)
|
||||
|
||||
@property
|
||||
def _options(self) -> typing.List[str]:
|
||||
def _options(self) -> list[str]:
|
||||
"""Return a list of selection options."""
|
||||
return self._config[CONF_OPTIONS]
|
||||
|
||||
|
@ -259,7 +258,7 @@ class InputSelect(RestoreEntity):
|
|||
return {ATTR_OPTIONS: self._config[ATTR_OPTIONS], ATTR_EDITABLE: self.editable}
|
||||
|
||||
@property
|
||||
def unique_id(self) -> typing.Optional[str]:
|
||||
def unique_id(self) -> str | None:
|
||||
"""Return unique id for the entity."""
|
||||
return self._config[CONF_ID]
|
||||
|
||||
|
@ -315,7 +314,7 @@ class InputSelect(RestoreEntity):
|
|||
self._config[CONF_OPTIONS] = options
|
||||
self.async_write_ha_state()
|
||||
|
||||
async def async_update_config(self, config: typing.Dict) -> None:
|
||||
async def async_update_config(self, config: dict) -> None:
|
||||
"""Handle when the config is updated."""
|
||||
self._config = config
|
||||
self.async_write_ha_state()
|
||||
|
|
|
@ -1,8 +1,10 @@
|
|||
"""Reproduce an Input select 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
|
||||
from homeassistant.core import Context, State
|
||||
|
@ -25,8 +27,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)
|
||||
|
@ -71,8 +73,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 Input select states."""
|
||||
# Reproduce states in parallel.
|
||||
|
|
|
@ -2,7 +2,6 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import typing
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
|
@ -173,16 +172,16 @@ class InputTextStorageCollection(collection.StorageCollection):
|
|||
CREATE_SCHEMA = vol.Schema(vol.All(CREATE_FIELDS, _cv_input_text))
|
||||
UPDATE_SCHEMA = vol.Schema(UPDATE_FIELDS)
|
||||
|
||||
async def _process_create_data(self, data: typing.Dict) -> typing.Dict:
|
||||
async def _process_create_data(self, data: dict) -> dict:
|
||||
"""Validate the config is valid."""
|
||||
return self.CREATE_SCHEMA(data)
|
||||
|
||||
@callback
|
||||
def _get_suggested_id(self, info: typing.Dict) -> str:
|
||||
def _get_suggested_id(self, info: dict) -> str:
|
||||
"""Suggest an ID based on the config."""
|
||||
return info[CONF_NAME]
|
||||
|
||||
async def _update_data(self, data: dict, update_data: typing.Dict) -> typing.Dict:
|
||||
async def _update_data(self, data: dict, update_data: dict) -> dict:
|
||||
"""Return a new updated data object."""
|
||||
update_data = self.UPDATE_SCHEMA(update_data)
|
||||
return _cv_input_text({**data, **update_data})
|
||||
|
@ -191,14 +190,14 @@ class InputTextStorageCollection(collection.StorageCollection):
|
|||
class InputText(RestoreEntity):
|
||||
"""Represent a text box."""
|
||||
|
||||
def __init__(self, config: typing.Dict):
|
||||
def __init__(self, config: dict):
|
||||
"""Initialize a text input."""
|
||||
self._config = config
|
||||
self.editable = True
|
||||
self._current_value = config.get(CONF_INITIAL)
|
||||
|
||||
@classmethod
|
||||
def from_yaml(cls, config: typing.Dict) -> InputText:
|
||||
def from_yaml(cls, config: dict) -> InputText:
|
||||
"""Return entity instance initialized from yaml storage."""
|
||||
input_text = cls(config)
|
||||
input_text.entity_id = f"{DOMAIN}.{config[CONF_ID]}"
|
||||
|
@ -241,7 +240,7 @@ class InputText(RestoreEntity):
|
|||
return self._config.get(CONF_UNIT_OF_MEASUREMENT)
|
||||
|
||||
@property
|
||||
def unique_id(self) -> typing.Optional[str]:
|
||||
def unique_id(self) -> str | None:
|
||||
"""Return unique id for the entity."""
|
||||
return self._config[CONF_ID]
|
||||
|
||||
|
@ -282,7 +281,7 @@ class InputText(RestoreEntity):
|
|||
self._current_value = value
|
||||
self.async_write_ha_state()
|
||||
|
||||
async def async_update_config(self, config: typing.Dict) -> None:
|
||||
async def async_update_config(self, config: dict) -> None:
|
||||
"""Handle when the config is updated."""
|
||||
self._config = config
|
||||
self.async_write_ha_state()
|
||||
|
|
|
@ -1,7 +1,9 @@
|
|||
"""Reproduce an Input text state."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import Any, Dict, Iterable, Optional
|
||||
from typing import Any, Iterable
|
||||
|
||||
from homeassistant.const import ATTR_ENTITY_ID
|
||||
from homeassistant.core import Context, State
|
||||
|
@ -16,8 +18,8 @@ async def _async_reproduce_state(
|
|||
hass: HomeAssistantType,
|
||||
state: State,
|
||||
*,
|
||||
context: Optional[Context] = None,
|
||||
reproduce_options: Optional[Dict[str, Any]] = None,
|
||||
context: Context | None = None,
|
||||
reproduce_options: dict[str, Any] | None = None,
|
||||
) -> None:
|
||||
"""Reproduce a single state."""
|
||||
cur_state = hass.states.get(state.entity_id)
|
||||
|
@ -44,8 +46,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 Input text states."""
|
||||
# Reproduce states in parallel.
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
"""Support for Insteon thermostat."""
|
||||
from typing import List, Optional
|
||||
from __future__ import annotations
|
||||
|
||||
from pyinsteon.constants import ThermostatMode
|
||||
from pyinsteon.operating_flag import CELSIUS
|
||||
|
@ -97,7 +97,7 @@ class InsteonClimateEntity(InsteonEntity, ClimateEntity):
|
|||
return TEMP_FAHRENHEIT
|
||||
|
||||
@property
|
||||
def current_humidity(self) -> Optional[int]:
|
||||
def current_humidity(self) -> int | None:
|
||||
"""Return the current humidity."""
|
||||
return self._insteon_device.groups[HUMIDITY].value
|
||||
|
||||
|
@ -107,17 +107,17 @@ class InsteonClimateEntity(InsteonEntity, ClimateEntity):
|
|||
return HVAC_MODES[self._insteon_device.groups[SYSTEM_MODE].value]
|
||||
|
||||
@property
|
||||
def hvac_modes(self) -> List[str]:
|
||||
def hvac_modes(self) -> list[str]:
|
||||
"""Return the list of available hvac operation modes."""
|
||||
return list(HVAC_MODES.values())
|
||||
|
||||
@property
|
||||
def current_temperature(self) -> Optional[float]:
|
||||
def current_temperature(self) -> float | None:
|
||||
"""Return the current temperature."""
|
||||
return self._insteon_device.groups[TEMPERATURE].value
|
||||
|
||||
@property
|
||||
def target_temperature(self) -> Optional[float]:
|
||||
def target_temperature(self) -> float | None:
|
||||
"""Return the temperature we try to reach."""
|
||||
if self._insteon_device.groups[SYSTEM_MODE].value == ThermostatMode.HEAT:
|
||||
return self._insteon_device.groups[HEAT_SET_POINT].value
|
||||
|
@ -126,31 +126,31 @@ class InsteonClimateEntity(InsteonEntity, ClimateEntity):
|
|||
return None
|
||||
|
||||
@property
|
||||
def target_temperature_high(self) -> Optional[float]:
|
||||
def target_temperature_high(self) -> float | None:
|
||||
"""Return the highbound target temperature we try to reach."""
|
||||
if self._insteon_device.groups[SYSTEM_MODE].value == ThermostatMode.AUTO:
|
||||
return self._insteon_device.groups[COOL_SET_POINT].value
|
||||
return None
|
||||
|
||||
@property
|
||||
def target_temperature_low(self) -> Optional[float]:
|
||||
def target_temperature_low(self) -> float | None:
|
||||
"""Return the lowbound target temperature we try to reach."""
|
||||
if self._insteon_device.groups[SYSTEM_MODE].value == ThermostatMode.AUTO:
|
||||
return self._insteon_device.groups[HEAT_SET_POINT].value
|
||||
return None
|
||||
|
||||
@property
|
||||
def fan_mode(self) -> Optional[str]:
|
||||
def fan_mode(self) -> str | None:
|
||||
"""Return the fan setting."""
|
||||
return FAN_MODES[self._insteon_device.groups[FAN_MODE].value]
|
||||
|
||||
@property
|
||||
def fan_modes(self) -> Optional[List[str]]:
|
||||
def fan_modes(self) -> list[str] | None:
|
||||
"""Return the list of available fan modes."""
|
||||
return list(FAN_MODES.values())
|
||||
|
||||
@property
|
||||
def target_humidity(self) -> Optional[int]:
|
||||
def target_humidity(self) -> int | None:
|
||||
"""Return the humidity we try to reach."""
|
||||
high = self._insteon_device.groups[HUMIDITY_HIGH].value
|
||||
low = self._insteon_device.groups[HUMIDITY_LOW].value
|
||||
|
@ -163,7 +163,7 @@ class InsteonClimateEntity(InsteonEntity, ClimateEntity):
|
|||
return 1
|
||||
|
||||
@property
|
||||
def hvac_action(self) -> Optional[str]:
|
||||
def hvac_action(self) -> str | None:
|
||||
"""Return the current running hvac operation if supported.
|
||||
|
||||
Need to be one of CURRENT_HVAC_*.
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
"""Schemas used by insteon component."""
|
||||
from __future__ import annotations
|
||||
|
||||
from binascii import Error as HexError, unhexlify
|
||||
from typing import Dict
|
||||
|
||||
from pyinsteon.address import Address
|
||||
from pyinsteon.constants import HC_LOOKUP
|
||||
|
@ -51,7 +52,7 @@ from .const import (
|
|||
)
|
||||
|
||||
|
||||
def set_default_port(schema: Dict) -> Dict:
|
||||
def set_default_port(schema: dict) -> dict:
|
||||
"""Set the default port based on the Hub version."""
|
||||
# If the ip_port is found do nothing
|
||||
# If it is not found the set the default
|
||||
|
|
|
@ -1,8 +1,10 @@
|
|||
"""The Internet Printing Protocol (IPP) integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from typing import Any, Dict
|
||||
from typing import Any
|
||||
|
||||
from pyipp import IPP, IPPError, Printer as IPPPrinter
|
||||
|
||||
|
@ -39,7 +41,7 @@ SCAN_INTERVAL = timedelta(seconds=60)
|
|||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_setup(hass: HomeAssistant, config: Dict) -> bool:
|
||||
async def async_setup(hass: HomeAssistant, config: dict) -> bool:
|
||||
"""Set up the IPP component."""
|
||||
hass.data.setdefault(DOMAIN, {})
|
||||
return True
|
||||
|
@ -166,7 +168,7 @@ class IPPEntity(CoordinatorEntity):
|
|||
return self._enabled_default
|
||||
|
||||
@property
|
||||
def device_info(self) -> Dict[str, Any]:
|
||||
def device_info(self) -> dict[str, Any]:
|
||||
"""Return device information about this IPP device."""
|
||||
if self._device_id is None:
|
||||
return None
|
||||
|
|
|
@ -1,6 +1,8 @@
|
|||
"""Config flow to configure the IPP integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any, Dict, Optional
|
||||
from typing import Any
|
||||
|
||||
from pyipp import (
|
||||
IPP,
|
||||
|
@ -30,7 +32,7 @@ from .const import DOMAIN # pylint: disable=unused-import
|
|||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def validate_input(hass: HomeAssistantType, data: dict) -> Dict[str, Any]:
|
||||
async def validate_input(hass: HomeAssistantType, data: dict) -> dict[str, Any]:
|
||||
"""Validate the user input allows us to connect.
|
||||
|
||||
Data has the keys from DATA_SCHEMA with values provided by the user.
|
||||
|
@ -61,8 +63,8 @@ class IPPFlowHandler(ConfigFlow, domain=DOMAIN):
|
|||
self.discovery_info = {}
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: Optional[ConfigType] = None
|
||||
) -> Dict[str, Any]:
|
||||
self, user_input: ConfigType | None = None
|
||||
) -> dict[str, Any]:
|
||||
"""Handle a flow initiated by the user."""
|
||||
if user_input is None:
|
||||
return self._show_setup_form()
|
||||
|
@ -98,7 +100,7 @@ class IPPFlowHandler(ConfigFlow, domain=DOMAIN):
|
|||
|
||||
return self.async_create_entry(title=user_input[CONF_HOST], data=user_input)
|
||||
|
||||
async def async_step_zeroconf(self, discovery_info: ConfigType) -> Dict[str, Any]:
|
||||
async def async_step_zeroconf(self, discovery_info: ConfigType) -> dict[str, Any]:
|
||||
"""Handle zeroconf discovery."""
|
||||
port = discovery_info[CONF_PORT]
|
||||
zctype = discovery_info["type"]
|
||||
|
@ -166,7 +168,7 @@ class IPPFlowHandler(ConfigFlow, domain=DOMAIN):
|
|||
|
||||
async def async_step_zeroconf_confirm(
|
||||
self, user_input: ConfigType = None
|
||||
) -> Dict[str, Any]:
|
||||
) -> dict[str, Any]:
|
||||
"""Handle a confirmation flow initiated by zeroconf."""
|
||||
if user_input is None:
|
||||
return self.async_show_form(
|
||||
|
@ -180,7 +182,7 @@ class IPPFlowHandler(ConfigFlow, domain=DOMAIN):
|
|||
data=self.discovery_info,
|
||||
)
|
||||
|
||||
def _show_setup_form(self, errors: Optional[Dict] = None) -> Dict[str, Any]:
|
||||
def _show_setup_form(self, errors: dict | None = None) -> dict[str, Any]:
|
||||
"""Show the setup form to the user."""
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
|
|
|
@ -1,6 +1,8 @@
|
|||
"""Support for IPP sensors."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
from typing import Any, Callable, Dict, List, Optional
|
||||
from typing import Any, Callable
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import ATTR_LOCATION, DEVICE_CLASS_TIMESTAMP, PERCENTAGE
|
||||
|
@ -26,7 +28,7 @@ from .const import (
|
|||
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 IPP sensor based on a config entry."""
|
||||
coordinator: IPPDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
|
||||
|
@ -63,7 +65,7 @@ class IPPSensor(IPPEntity):
|
|||
icon: str,
|
||||
key: str,
|
||||
name: str,
|
||||
unit_of_measurement: Optional[str] = None,
|
||||
unit_of_measurement: str | None = None,
|
||||
) -> None:
|
||||
"""Initialize IPP sensor."""
|
||||
self._unit_of_measurement = unit_of_measurement
|
||||
|
@ -117,7 +119,7 @@ class IPPMarkerSensor(IPPSensor):
|
|||
)
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> Optional[Dict[str, Any]]:
|
||||
def extra_state_attributes(self) -> dict[str, Any] | None:
|
||||
"""Return the state attributes of the entity."""
|
||||
return {
|
||||
ATTR_MARKER_HIGH_LEVEL: self.coordinator.data.markers[
|
||||
|
@ -132,7 +134,7 @@ class IPPMarkerSensor(IPPSensor):
|
|||
}
|
||||
|
||||
@property
|
||||
def state(self) -> Optional[int]:
|
||||
def state(self) -> int | None:
|
||||
"""Return the state of the sensor."""
|
||||
level = self.coordinator.data.markers[self.marker_index].level
|
||||
|
||||
|
@ -160,7 +162,7 @@ class IPPPrinterSensor(IPPSensor):
|
|||
)
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> Optional[Dict[str, Any]]:
|
||||
def extra_state_attributes(self) -> dict[str, Any] | None:
|
||||
"""Return the state attributes of the entity."""
|
||||
return {
|
||||
ATTR_INFO: self.coordinator.data.info.printer_info,
|
||||
|
@ -202,6 +204,6 @@ class IPPUptimeSensor(IPPSensor):
|
|||
return uptime.replace(microsecond=0).isoformat()
|
||||
|
||||
@property
|
||||
def device_class(self) -> Optional[str]:
|
||||
def device_class(self) -> str | None:
|
||||
"""Return the class of this sensor."""
|
||||
return DEVICE_CLASS_TIMESTAMP
|
||||
|
|
|
@ -1,7 +1,8 @@
|
|||
"""Support the ISY-994 controllers."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from functools import partial
|
||||
from typing import Optional
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from pyisy import ISY
|
||||
|
@ -67,7 +68,7 @@ CONFIG_SCHEMA = vol.Schema(
|
|||
|
||||
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||
"""Set up the isy994 integration from YAML."""
|
||||
isy_config: Optional[ConfigType] = config.get(DOMAIN)
|
||||
isy_config: ConfigType | None = config.get(DOMAIN)
|
||||
hass.data.setdefault(DOMAIN, {})
|
||||
|
||||
if not isy_config:
|
||||
|
|
|
@ -1,6 +1,8 @@
|
|||
"""Support for ISY994 binary sensors."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
from typing import Callable, Union
|
||||
from typing import Callable
|
||||
|
||||
from pyisy.constants import (
|
||||
CMD_OFF,
|
||||
|
@ -173,7 +175,7 @@ async def async_setup_entry(
|
|||
async_add_entities(devices)
|
||||
|
||||
|
||||
def _detect_device_type_and_class(node: Union[Group, Node]) -> (str, str):
|
||||
def _detect_device_type_and_class(node: Group | Node) -> (str, str):
|
||||
try:
|
||||
device_type = node.type
|
||||
except AttributeError:
|
||||
|
|
|
@ -1,5 +1,7 @@
|
|||
"""Support for Insteon Thermostats via ISY994 Platform."""
|
||||
from typing import Callable, List, Optional
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Callable
|
||||
|
||||
from pyisy.constants import (
|
||||
CMD_CLIMATE_FAN_SETTING,
|
||||
|
@ -114,7 +116,7 @@ class ISYThermostatEntity(ISYNodeEntity, ClimateEntity):
|
|||
return TEMP_FAHRENHEIT
|
||||
|
||||
@property
|
||||
def current_humidity(self) -> Optional[int]:
|
||||
def current_humidity(self) -> int | None:
|
||||
"""Return the current humidity."""
|
||||
humidity = self._node.aux_properties.get(PROP_HUMIDITY)
|
||||
if not humidity:
|
||||
|
@ -122,7 +124,7 @@ class ISYThermostatEntity(ISYNodeEntity, ClimateEntity):
|
|||
return int(humidity.value)
|
||||
|
||||
@property
|
||||
def hvac_mode(self) -> Optional[str]:
|
||||
def hvac_mode(self) -> str | None:
|
||||
"""Return hvac operation ie. heat, cool mode."""
|
||||
hvac_mode = self._node.aux_properties.get(CMD_CLIMATE_MODE)
|
||||
if not hvac_mode:
|
||||
|
@ -140,12 +142,12 @@ class ISYThermostatEntity(ISYNodeEntity, ClimateEntity):
|
|||
return UOM_TO_STATES[uom].get(hvac_mode.value)
|
||||
|
||||
@property
|
||||
def hvac_modes(self) -> List[str]:
|
||||
def hvac_modes(self) -> list[str]:
|
||||
"""Return the list of available hvac operation modes."""
|
||||
return ISY_HVAC_MODES
|
||||
|
||||
@property
|
||||
def hvac_action(self) -> Optional[str]:
|
||||
def hvac_action(self) -> str | None:
|
||||
"""Return the current running hvac operation if supported."""
|
||||
hvac_action = self._node.aux_properties.get(PROP_HEAT_COOL_STATE)
|
||||
if not hvac_action:
|
||||
|
@ -153,19 +155,19 @@ class ISYThermostatEntity(ISYNodeEntity, ClimateEntity):
|
|||
return UOM_TO_STATES[UOM_HVAC_ACTIONS].get(hvac_action.value)
|
||||
|
||||
@property
|
||||
def current_temperature(self) -> Optional[float]:
|
||||
def current_temperature(self) -> float | None:
|
||||
"""Return the current temperature."""
|
||||
return convert_isy_value_to_hass(
|
||||
self._node.status, self._uom, self._node.prec, 1
|
||||
)
|
||||
|
||||
@property
|
||||
def target_temperature_step(self) -> Optional[float]:
|
||||
def target_temperature_step(self) -> float | None:
|
||||
"""Return the supported step of target temperature."""
|
||||
return 1.0
|
||||
|
||||
@property
|
||||
def target_temperature(self) -> Optional[float]:
|
||||
def target_temperature(self) -> float | None:
|
||||
"""Return the temperature we try to reach."""
|
||||
if self.hvac_mode == HVAC_MODE_COOL:
|
||||
return self.target_temperature_high
|
||||
|
@ -174,7 +176,7 @@ class ISYThermostatEntity(ISYNodeEntity, ClimateEntity):
|
|||
return None
|
||||
|
||||
@property
|
||||
def target_temperature_high(self) -> Optional[float]:
|
||||
def target_temperature_high(self) -> float | None:
|
||||
"""Return the highbound target temperature we try to reach."""
|
||||
target = self._node.aux_properties.get(PROP_SETPOINT_COOL)
|
||||
if not target:
|
||||
|
@ -182,7 +184,7 @@ class ISYThermostatEntity(ISYNodeEntity, ClimateEntity):
|
|||
return convert_isy_value_to_hass(target.value, target.uom, target.prec, 1)
|
||||
|
||||
@property
|
||||
def target_temperature_low(self) -> Optional[float]:
|
||||
def target_temperature_low(self) -> float | None:
|
||||
"""Return the lowbound target temperature we try to reach."""
|
||||
target = self._node.aux_properties.get(PROP_SETPOINT_HEAT)
|
||||
if not target:
|
||||
|
|
|
@ -1,6 +1,8 @@
|
|||
"""Support for ISY994 fans."""
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
from typing import Callable, Optional
|
||||
from typing import Callable
|
||||
|
||||
from pyisy.constants import ISY_VALUE_UNKNOWN, PROTO_INSTEON
|
||||
|
||||
|
@ -43,7 +45,7 @@ class ISYFanEntity(ISYNodeEntity, FanEntity):
|
|||
"""Representation of an ISY994 fan device."""
|
||||
|
||||
@property
|
||||
def percentage(self) -> Optional[int]:
|
||||
def percentage(self) -> int | None:
|
||||
"""Return the current speed percentage."""
|
||||
if self._node.status == ISY_VALUE_UNKNOWN:
|
||||
return None
|
||||
|
@ -97,7 +99,7 @@ class ISYFanProgramEntity(ISYProgramEntity, FanEntity):
|
|||
"""Representation of an ISY994 fan program."""
|
||||
|
||||
@property
|
||||
def percentage(self) -> Optional[int]:
|
||||
def percentage(self) -> int | None:
|
||||
"""Return the current speed percentage."""
|
||||
if self._node.status == ISY_VALUE_UNKNOWN:
|
||||
return None
|
||||
|
|
|
@ -1,5 +1,7 @@
|
|||
"""Sorting helpers for ISY994 device classifications."""
|
||||
from typing import Any, List, Optional, Union
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from pyisy.constants import (
|
||||
ISY_VALUE_UNKNOWN,
|
||||
|
@ -56,7 +58,7 @@ BINARY_SENSOR_ISY_STATES = ["on", "off"]
|
|||
|
||||
|
||||
def _check_for_node_def(
|
||||
hass_isy_data: dict, node: Union[Group, Node], single_platform: str = None
|
||||
hass_isy_data: dict, node: Group | Node, single_platform: str = None
|
||||
) -> bool:
|
||||
"""Check if the node matches the node_def_id for any platforms.
|
||||
|
||||
|
@ -79,7 +81,7 @@ def _check_for_node_def(
|
|||
|
||||
|
||||
def _check_for_insteon_type(
|
||||
hass_isy_data: dict, node: Union[Group, Node], single_platform: str = None
|
||||
hass_isy_data: dict, node: Group | Node, single_platform: str = None
|
||||
) -> bool:
|
||||
"""Check if the node matches the Insteon type for any platforms.
|
||||
|
||||
|
@ -144,7 +146,7 @@ def _check_for_insteon_type(
|
|||
|
||||
|
||||
def _check_for_zwave_cat(
|
||||
hass_isy_data: dict, node: Union[Group, Node], single_platform: str = None
|
||||
hass_isy_data: dict, node: Group | Node, single_platform: str = None
|
||||
) -> bool:
|
||||
"""Check if the node matches the ISY Z-Wave Category for any platforms.
|
||||
|
||||
|
@ -174,7 +176,7 @@ def _check_for_zwave_cat(
|
|||
|
||||
def _check_for_uom_id(
|
||||
hass_isy_data: dict,
|
||||
node: Union[Group, Node],
|
||||
node: Group | Node,
|
||||
single_platform: str = None,
|
||||
uom_list: list = None,
|
||||
) -> bool:
|
||||
|
@ -209,7 +211,7 @@ def _check_for_uom_id(
|
|||
|
||||
def _check_for_states_in_uom(
|
||||
hass_isy_data: dict,
|
||||
node: Union[Group, Node],
|
||||
node: Group | Node,
|
||||
single_platform: str = None,
|
||||
states_list: list = None,
|
||||
) -> bool:
|
||||
|
@ -244,7 +246,7 @@ def _check_for_states_in_uom(
|
|||
return False
|
||||
|
||||
|
||||
def _is_sensor_a_binary_sensor(hass_isy_data: dict, node: Union[Group, Node]) -> bool:
|
||||
def _is_sensor_a_binary_sensor(hass_isy_data: dict, node: Group | Node) -> bool:
|
||||
"""Determine if the given sensor node should be a binary_sensor."""
|
||||
if _check_for_node_def(hass_isy_data, node, single_platform=BINARY_SENSOR):
|
||||
return True
|
||||
|
@ -364,7 +366,7 @@ def _categorize_variables(
|
|||
|
||||
|
||||
async def migrate_old_unique_ids(
|
||||
hass: HomeAssistantType, platform: str, devices: Optional[List[Any]]
|
||||
hass: HomeAssistantType, platform: str, devices: list[Any] | None
|
||||
) -> None:
|
||||
"""Migrate to new controller-specific unique ids."""
|
||||
registry = await async_get_registry(hass)
|
||||
|
@ -396,11 +398,11 @@ async def migrate_old_unique_ids(
|
|||
|
||||
|
||||
def convert_isy_value_to_hass(
|
||||
value: Union[int, float, None],
|
||||
value: int | float | None,
|
||||
uom: str,
|
||||
precision: Union[int, str],
|
||||
fallback_precision: Optional[int] = None,
|
||||
) -> Union[float, int]:
|
||||
precision: int | str,
|
||||
fallback_precision: int | None = None,
|
||||
) -> float | int:
|
||||
"""Fix ISY Reported Values.
|
||||
|
||||
ISY provides float values as an integer and precision component.
|
||||
|
|
|
@ -1,5 +1,7 @@
|
|||
"""Support for ISY994 lights."""
|
||||
from typing import Callable, Dict
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Callable
|
||||
|
||||
from pyisy.constants import ISY_VALUE_UNKNOWN
|
||||
|
||||
|
@ -98,7 +100,7 @@ class ISYLightEntity(ISYNodeEntity, LightEntity, RestoreEntity):
|
|||
_LOGGER.debug("Unable to turn on light")
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> Dict:
|
||||
def extra_state_attributes(self) -> dict:
|
||||
"""Return the light attributes."""
|
||||
attribs = super().extra_state_attributes
|
||||
attribs[ATTR_LAST_BRIGHTNESS] = self._last_brightness
|
||||
|
|
|
@ -1,5 +1,7 @@
|
|||
"""Support for ISY994 sensors."""
|
||||
from typing import Callable, Dict, Union
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Callable
|
||||
|
||||
from pyisy.constants import ISY_VALUE_UNKNOWN
|
||||
|
||||
|
@ -47,7 +49,7 @@ class ISYSensorEntity(ISYNodeEntity):
|
|||
"""Representation of an ISY994 sensor device."""
|
||||
|
||||
@property
|
||||
def raw_unit_of_measurement(self) -> Union[dict, str]:
|
||||
def raw_unit_of_measurement(self) -> dict | str:
|
||||
"""Get the raw unit of measurement for the ISY994 sensor device."""
|
||||
uom = self._node.uom
|
||||
|
||||
|
@ -117,7 +119,7 @@ class ISYSensorVariableEntity(ISYEntity):
|
|||
return convert_isy_value_to_hass(self._node.status, "", self._node.prec)
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> Dict:
|
||||
def extra_state_attributes(self) -> dict:
|
||||
"""Get the state attributes for the device."""
|
||||
return {
|
||||
"init_value": convert_isy_value_to_hass(
|
||||
|
|
|
@ -1,7 +1,7 @@
|
|||
"""Support for the iZone HVAC."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import List, Optional
|
||||
|
||||
from pizone import Controller, Zone
|
||||
import voluptuous as vol
|
||||
|
@ -324,7 +324,7 @@ class ControllerDevice(ClimateEntity):
|
|||
|
||||
@property
|
||||
@_return_on_connection_error([])
|
||||
def hvac_modes(self) -> List[str]:
|
||||
def hvac_modes(self) -> list[str]:
|
||||
"""Return the list of available operation modes."""
|
||||
if self._controller.free_air:
|
||||
return [HVAC_MODE_OFF, HVAC_MODE_FAN_ONLY]
|
||||
|
@ -346,7 +346,7 @@ class ControllerDevice(ClimateEntity):
|
|||
|
||||
@property
|
||||
@_return_on_connection_error()
|
||||
def current_temperature(self) -> Optional[float]:
|
||||
def current_temperature(self) -> float | None:
|
||||
"""Return the current temperature."""
|
||||
if self._controller.mode == Controller.Mode.FREE_AIR:
|
||||
return self._controller.temp_supply
|
||||
|
@ -364,7 +364,7 @@ class ControllerDevice(ClimateEntity):
|
|||
return zone.name
|
||||
|
||||
@property
|
||||
def control_zone_setpoint(self) -> Optional[float]:
|
||||
def control_zone_setpoint(self) -> float | None:
|
||||
"""Return the temperature setpoint of the zone that currently controls the AC unit (if target temp not set by controller)."""
|
||||
if self._supported_features & SUPPORT_TARGET_TEMPERATURE:
|
||||
return None
|
||||
|
@ -376,7 +376,7 @@ class ControllerDevice(ClimateEntity):
|
|||
|
||||
@property
|
||||
@_return_on_connection_error()
|
||||
def target_temperature(self) -> Optional[float]:
|
||||
def target_temperature(self) -> float | None:
|
||||
"""Return the temperature we try to reach (either from control zone or master unit)."""
|
||||
if self._supported_features & SUPPORT_TARGET_TEMPERATURE:
|
||||
return self._controller.temp_setpoint
|
||||
|
@ -388,17 +388,17 @@ class ControllerDevice(ClimateEntity):
|
|||
return self._controller.temp_supply
|
||||
|
||||
@property
|
||||
def target_temperature_step(self) -> Optional[float]:
|
||||
def target_temperature_step(self) -> float | None:
|
||||
"""Return the supported step of target temperature."""
|
||||
return 0.5
|
||||
|
||||
@property
|
||||
def fan_mode(self) -> Optional[str]:
|
||||
def fan_mode(self) -> str | None:
|
||||
"""Return the fan setting."""
|
||||
return _IZONE_FAN_TO_HA[self._controller.fan]
|
||||
|
||||
@property
|
||||
def fan_modes(self) -> Optional[List[str]]:
|
||||
def fan_modes(self) -> list[str] | None:
|
||||
"""Return the list of available fan modes."""
|
||||
return list(self._fan_to_pizone)
|
||||
|
||||
|
|
Loading…
Reference in New Issue