jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 05:07:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Update typing 08 (#48058)

Browse Source
This commit is contained in:
Marc Mueller 2021-03-18 09:25:40 +01:00 committed by GitHub
parent 9e1a6610dc
commit 5cdd945f44
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
48 changed files with 355 additions and 296 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
homeassistant/components/habitica/config_flow.py
View File

@ -1,6 +1,7 @@
"""Config flow for habitica integration."""
from __future__ import annotations
import logging
from typing import Dict
from aiohttp import ClientResponseError
from habitipy.aio import HabitipyAsync
@ -25,8 +26,8 @@ _LOGGER = logging.getLogger(__name__)
async def validate_input(
hass: core.HomeAssistant, data: Dict[str, str]
) -> Dict[str, str]:
hass: core.HomeAssistant, data: dict[str, str]
) -> dict[str, str]:
"""Validate the user input allows us to connect."""
websession = async_get_clientsession(hass)

14
homeassistant/components/hassio/__init__.py
View File

@ -1,9 +1,11 @@
"""Support for Hass.io."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
import os
from typing import Any, Dict, List, Optional
from typing import Any
import voluptuous as vol
@ -236,7 +238,7 @@ async def async_set_addon_options(
@bind_hass
async def async_get_addon_discovery_info(
hass: HomeAssistantType, slug: str
) -> Optional[dict]:
) -> dict | None:
"""Return discovery data for an add-on."""
hassio = hass.data[DOMAIN]
data = await hassio.retrieve_discovery_messages()
@ -545,7 +547,7 @@ async def async_unload_entry(
@callback
def async_register_addons_in_dev_reg(
entry_id: str, dev_reg: DeviceRegistry, addons: List[Dict[str, Any]]
entry_id: str, dev_reg: DeviceRegistry, addons: list[dict[str, Any]]
) -> None:
"""Register addons in the device registry."""
for addon in addons:
@ -564,7 +566,7 @@ def async_register_addons_in_dev_reg(
@callback
def async_register_os_in_dev_reg(
entry_id: str, dev_reg: DeviceRegistry, os_dict: Dict[str, Any]
entry_id: str, dev_reg: DeviceRegistry, os_dict: dict[str, Any]
) -> None:
"""Register OS in the device registry."""
params = {
@ -581,7 +583,7 @@ def async_register_os_in_dev_reg(
@callback
def async_remove_addons_from_dev_reg(
dev_reg: DeviceRegistry, addons: List[Dict[str, Any]]
dev_reg: DeviceRegistry, addons: list[dict[str, Any]]
) -> None:
"""Remove addons from the device registry."""
for addon_slug in addons:
@ -607,7 +609,7 @@ class HassioDataUpdateCoordinator(DataUpdateCoordinator):
self.dev_reg = dev_reg
self.is_hass_os = "hassos" in get_info(self.hass)
async def _async_update_data(self) -> Dict[str, Any]:
async def _async_update_data(self) -> dict[str, Any]:
"""Update data via library."""
new_data = {}
addon_data = get_supervisor_info(self.hass)

6
homeassistant/components/hassio/binary_sensor.py
View File

@ -1,5 +1,7 @@
"""Binary sensor platform for Hass.io addons."""
from typing import Callable, List
from __future__ import annotations
from typing import Callable
from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.config_entries import ConfigEntry
@ -14,7 +16,7 @@ from .entity import HassioAddonEntity, HassioOSEntity
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: Callable[[List[Entity], bool], None],
async_add_entities: Callable[[list[Entity], bool], None],
) -> None:
"""Binary sensor set up for Hass.io config entry."""
coordinator = hass.data[ADDONS_COORDINATOR]

14
homeassistant/components/hassio/entity.py
View File

@ -1,5 +1,7 @@
"""Base for Hass.io entities."""
from typing import Any, Dict
from __future__ import annotations
from typing import Any
from homeassistant.const import ATTR_NAME
from homeassistant.helpers.update_coordinator import CoordinatorEntity
@ -14,7 +16,7 @@ class HassioAddonEntity(CoordinatorEntity):
def __init__(
self,
coordinator: HassioDataUpdateCoordinator,
addon: Dict[str, Any],
addon: dict[str, Any],
attribute_name: str,
sensor_name: str,
) -> None:
@ -27,7 +29,7 @@ class HassioAddonEntity(CoordinatorEntity):
super().__init__(coordinator)
@property
def addon_info(self) -> Dict[str, Any]:
def addon_info(self) -> dict[str, Any]:
"""Return add-on info."""
return self.coordinator.data[self._data_key][self.addon_slug]
@ -47,7 +49,7 @@ class HassioAddonEntity(CoordinatorEntity):
return f"{self.addon_slug}_{self.attribute_name}"
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return device specific attributes."""
return {"identifiers": {(DOMAIN, self.addon_slug)}}
@ -68,7 +70,7 @@ class HassioOSEntity(CoordinatorEntity):
super().__init__(coordinator)
@property
def os_info(self) -> Dict[str, Any]:
def os_info(self) -> dict[str, Any]:
"""Return OS info."""
return self.coordinator.data[self._data_key]
@ -88,6 +90,6 @@ class HassioOSEntity(CoordinatorEntity):
return f"home_assistant_os_{self.attribute_name}"
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return device specific attributes."""
return {"identifiers": {(DOMAIN, "OS")}}

9
homeassistant/components/hassio/http.py
View File

@ -1,9 +1,10 @@
"""HTTP Support for Hass.io."""
from __future__ import annotations
import asyncio
import logging
import os
import re
from typing import Dict, Union
import aiohttp
from aiohttp import web
@ -57,7 +58,7 @@ class HassIOView(HomeAssistantView):
async def _handle(
self, request: web.Request, path: str
) -> Union[web.Response, web.StreamResponse]:
) -> web.Response | web.StreamResponse:
"""Route data to Hass.io."""
hass = request.app["hass"]
if _need_auth(hass, path) and not request[KEY_AUTHENTICATED]:
@ -71,7 +72,7 @@ class HassIOView(HomeAssistantView):
async def _command_proxy(
self, path: str, request: web.Request
) -> Union[web.Response, web.StreamResponse]:
) -> web.Response | web.StreamResponse:
"""Return a client request with proxy origin for Hass.io supervisor.
This method is a coroutine.
@ -131,7 +132,7 @@ class HassIOView(HomeAssistantView):
raise HTTPBadGateway()
def _init_header(request: web.Request) -> Dict[str, str]:
def _init_header(request: web.Request) -> dict[str, str]:
"""Create initial header."""
headers = {
X_HASSIO: os.environ.get("HASSIO_TOKEN", ""),

13
homeassistant/components/hassio/ingress.py
View File

@ -1,9 +1,10 @@
"""Hass.io Add-on ingress service."""
from __future__ import annotations
import asyncio
from ipaddress import ip_address
import logging
import os
from typing import Dict, Union
import aiohttp
from aiohttp import hdrs, web
@ -46,7 +47,7 @@ class HassIOIngress(HomeAssistantView):
async def _handle(
self, request: web.Request, token: str, path: str
) -> Union[web.Response, web.StreamResponse, web.WebSocketResponse]:
) -> web.Response | web.StreamResponse | web.WebSocketResponse:
"""Route data to Hass.io ingress service."""
try:
# Websocket
@ -114,7 +115,7 @@ class HassIOIngress(HomeAssistantView):
async def _handle_request(
self, request: web.Request, token: str, path: str
) -> Union[web.Response, web.StreamResponse]:
) -> web.Response | web.StreamResponse:
"""Ingress route for request."""
url = self._create_url(token, path)
data = await request.read()
@ -159,9 +160,7 @@ class HassIOIngress(HomeAssistantView):
return response
def _init_header(
request: web.Request, token: str
) -> Union[CIMultiDict, Dict[str, str]]:
def _init_header(request: web.Request, token: str) -> CIMultiDict | dict[str, str]:
"""Create initial header."""
headers = {}
@ -208,7 +207,7 @@ def _init_header(
return headers
def _response_header(response: aiohttp.ClientResponse) -> Dict[str, str]:
def _response_header(response: aiohttp.ClientResponse) -> dict[str, str]:
"""Create response header."""
headers = {}

6
homeassistant/components/hassio/sensor.py
View File

@ -1,5 +1,7 @@
"""Sensor platform for Hass.io addons."""
from typing import Callable, List
from __future__ import annotations
from typing import Callable
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
@ -13,7 +15,7 @@ from .entity import HassioAddonEntity, HassioOSEntity
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: Callable[[List[Entity], bool], None],
async_add_entities: Callable[[list[Entity], bool], None],
) -> None:
"""Sensor set up for Hass.io config entry."""
coordinator = hass.data[ADDONS_COORDINATOR]

5
homeassistant/components/heatmiser/climate.py
View File

@ -1,6 +1,7 @@
"""Support for the PRT Heatmiser themostats using the V3 protocol."""
from __future__ import annotations
import logging
from typing import List
from heatmiserV3 import connection, heatmiser
import voluptuous as vol
@ -103,7 +104,7 @@ class HeatmiserV3Thermostat(ClimateEntity):
return self._hvac_mode
@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.

5
homeassistant/components/heos/__init__.py
View File

@ -1,8 +1,9 @@
"""Denon HEOS Media Player."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
from typing import Dict
from pyheos import Heos, HeosError, const as heos_const
import voluptuous as vol
@ -191,7 +192,7 @@ class ControllerManager:
# Update players
self._hass.helpers.dispatcher.async_dispatcher_send(SIGNAL_HEOS_UPDATED)
def update_ids(self, mapped_ids: Dict[int, int]):
def update_ids(self, mapped_ids: dict[int, int]):
"""Update the IDs in the device and entity registry."""
# mapped_ids contains the mapped IDs (new:old)
for new_id, old_id in mapped_ids.items():

16
homeassistant/components/here_travel_time/sensor.py
View File

@ -1,7 +1,9 @@
"""Support for HERE travel time sensors."""
from __future__ import annotations
from datetime import datetime, timedelta
import logging
from typing import Callable, Dict, Optional, Union
from typing import Callable
import herepy
import voluptuous as vol
@ -143,9 +145,9 @@ PLATFORM_SCHEMA = vol.All(
async def async_setup_platform(
hass: HomeAssistant,
config: Dict[str, Union[str, bool]],
config: dict[str, str | bool],
async_add_entities: Callable,
discovery_info: Optional[DiscoveryInfoType] = None,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the HERE travel time platform."""
api_key = config[CONF_API_KEY]
@ -255,7 +257,7 @@ class HERETravelTimeSensor(Entity):
)
@property
def state(self) -> Optional[str]:
def state(self) -> str | None:
"""Return the state of the sensor."""
if self._here_data.traffic_mode:
if self._here_data.traffic_time is not None:
@ -273,7 +275,7 @@ class HERETravelTimeSensor(Entity):
@property
def extra_state_attributes(
self,
) -> Optional[Dict[str, Union[None, float, str, bool]]]:
) -> dict[str, None | float | str | bool] | None:
"""Return the state attributes."""
if self._here_data.base_time is None:
return None
@ -324,7 +326,7 @@ class HERETravelTimeSensor(Entity):
await self.hass.async_add_executor_job(self._here_data.update)
async def _get_location_from_entity(self, entity_id: str) -> Optional[str]:
async def _get_location_from_entity(self, entity_id: str) -> str | None:
"""Get the location from the entity state or attributes."""
entity = self.hass.states.get(entity_id)
@ -480,7 +482,7 @@ class HERETravelTimeData:
self.destination_name = waypoint[1]["mappedRoadName"]
@staticmethod
def _build_hass_attribution(source_attribution: Dict) -> Optional[str]:
def _build_hass_attribution(source_attribution: dict) -> str | None:
"""Build a hass frontend ready string out of the sourceAttribution."""
suppliers = source_attribution.get("supplier")
if suppliers is not None:

6
homeassistant/components/history/__init__.py
View File

@ -1,11 +1,13 @@
"""Provide pre-made queries on top of the recorder component."""
from __future__ import annotations
from collections import defaultdict
from datetime import datetime as dt, timedelta
from itertools import groupby
import json
import logging
import time
from typing import Iterable, Optional, cast
from typing import Iterable, cast
from aiohttp import web
from sqlalchemy import and_, bindparam, func, not_, or_
@ -462,7 +464,7 @@ class HistoryPeriodView(HomeAssistantView):
self.use_include_order = use_include_order
async def get(
self, request: web.Request, datetime: Optional[str] = None
self, request: web.Request, datetime: str | None = None
) -> web.Response:
"""Return history over a period of time."""
datetime_ = None

8
homeassistant/components/homeassistant/scene.py
View File

@ -1,7 +1,9 @@
"""Allow users to set and activate scenes."""
from __future__ import annotations
from collections import namedtuple
import logging
from typing import Any, List
from typing import Any
import voluptuous as vol
@ -118,7 +120,7 @@ _LOGGER = logging.getLogger(__name__)
@callback
def scenes_with_entity(hass: HomeAssistant, entity_id: str) -> List[str]:
def scenes_with_entity(hass: HomeAssistant, entity_id: str) -> list[str]:
"""Return all scenes that reference the entity."""
if DATA_PLATFORM not in hass.data:
return []
@ -133,7 +135,7 @@ def scenes_with_entity(hass: HomeAssistant, entity_id: str) -> List[str]:
@callback
def entities_in_scene(hass: HomeAssistant, entity_id: str) -> List[str]:
def entities_in_scene(hass: HomeAssistant, entity_id: str) -> list[str]:
"""Return all entities in a scene."""
if DATA_PLATFORM not in hass.data:
return []

10
homeassistant/components/homeassistant/triggers/state.py
View File

@ -1,7 +1,9 @@
"""Offer state listening automation rules."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any, Dict, Optional
from typing import Any
import voluptuous as vol
@ -79,7 +81,7 @@ async def async_attach_trigger(
template.attach(hass, time_delta)
match_all = from_state == MATCH_ALL and to_state == MATCH_ALL
unsub_track_same = {}
period: Dict[str, timedelta] = {}
period: dict[str, timedelta] = {}
match_from_state = process_state_match(from_state)
match_to_state = process_state_match(to_state)
attribute = config.get(CONF_ATTRIBUTE)
@ -93,8 +95,8 @@ async def async_attach_trigger(
def state_automation_listener(event: Event):
"""Listen for state changes and calls action."""
entity: str = event.data["entity_id"]
from_s: Optional[State] = event.data.get("old_state")
to_s: Optional[State] = event.data.get("new_state")
from_s: State | None = event.data.get("old_state")
to_s: State | None = event.data.get("new_state")
if from_s is None:
old_value = None

6
homeassistant/components/homekit_controller/__init__.py
View File

@ -1,5 +1,7 @@
"""Support for Homekit device discovery."""
from typing import Any, Dict
from __future__ import annotations
from typing import Any
import aiohomekit
from aiohomekit.model import Accessory
@ -77,7 +79,7 @@ class HomeKitEntity(Entity):
signal_remove()
self._signals.clear()
async def async_put_characteristics(self, characteristics: Dict[str, Any]):
async def async_put_characteristics(self, characteristics: dict[str, Any]):
"""
Write characteristics to the device.

4
homeassistant/components/homekit_controller/device_trigger.py
View File

@ -1,5 +1,5 @@
"""Provides device automations for homekit devices."""
from typing import List
from __future__ import annotations
from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.characteristics.const import InputEventValues
@ -226,7 +226,7 @@ def async_fire_triggers(conn, events):
source.fire(iid, ev)
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 homekit devices."""
if device_id not in hass.data.get(TRIGGERS, {}):

14
homeassistant/components/homekit_controller/humidifier.py
View File

@ -1,5 +1,5 @@
"""Support for HomeKit Controller humidifier."""
from typing import List, Optional
from __future__ import annotations
from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.services import ServicesTypes
@ -69,14 +69,14 @@ class HomeKitHumidifier(HomeKitEntity, HumidifierEntity):
await self.async_put_characteristics({CharacteristicsTypes.ACTIVE: False})
@property
def target_humidity(self) -> Optional[int]:
def target_humidity(self) -> int | None:
"""Return the humidity we try to reach."""
return self.service.value(
CharacteristicsTypes.RELATIVE_HUMIDITY_HUMIDIFIER_THRESHOLD
)
@property
def mode(self) -> Optional[str]:
def mode(self) -> str | None:
"""Return the current mode, e.g., home, auto, baby.
Requires SUPPORT_MODES.
@ -87,7 +87,7 @@ class HomeKitHumidifier(HomeKitEntity, HumidifierEntity):
return MODE_AUTO if mode == 1 else MODE_NORMAL
@property
def available_modes(self) -> Optional[List[str]]:
def available_modes(self) -> list[str] | None:
"""Return a list of available modes.
Requires SUPPORT_MODES.
@ -175,14 +175,14 @@ class HomeKitDehumidifier(HomeKitEntity, HumidifierEntity):
await self.async_put_characteristics({CharacteristicsTypes.ACTIVE: False})
@property
def target_humidity(self) -> Optional[int]:
def target_humidity(self) -> int | None:
"""Return the humidity we try to reach."""
return self.service.value(
CharacteristicsTypes.RELATIVE_HUMIDITY_DEHUMIDIFIER_THRESHOLD
)
@property
def mode(self) -> Optional[str]:
def mode(self) -> str | None:
"""Return the current mode, e.g., home, auto, baby.
Requires SUPPORT_MODES.
@ -193,7 +193,7 @@ class HomeKitDehumidifier(HomeKitEntity, HumidifierEntity):
return MODE_AUTO if mode == 1 else MODE_NORMAL
@property
def available_modes(self) -> Optional[List[str]]:
def available_modes(self) -> list[str] | None:
"""Return a list of available modes.
Requires SUPPORT_MODES.

6
homeassistant/components/homematicip_cloud/alarm_control_panel.py
View File

@ -1,6 +1,8 @@
"""Support for HomematicIP Cloud alarm control panel."""
from __future__ import annotations
import logging
from typing import Any, Dict
from typing import Any
from homematicip.functionalHomes import SecurityAndAlarmHome
@ -44,7 +46,7 @@ class HomematicipAlarmControlPanelEntity(AlarmControlPanelEntity):
_LOGGER.info("Setting up %s", self.name)
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return device specific attributes."""
return {
"identifiers": {(HMIPC_DOMAIN, f"ACP {self._home.id}")},

16
homeassistant/components/homematicip_cloud/binary_sensor.py
View File

@ -1,5 +1,7 @@
"""Support for HomematicIP Cloud binary sensor."""
from typing import Any, Dict
from __future__ import annotations
from typing import Any
from homematicip.aio.device import (
AsyncAccelerationSensor,
@ -166,7 +168,7 @@ class HomematicipCloudConnectionSensor(HomematicipGenericEntity, BinarySensorEnt
return name if not self._home.name else f"{self._home.name} {name}"
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return device specific attributes."""
# Adds a sensor to the existing HAP device
return {
@ -210,7 +212,7 @@ class HomematicipBaseActionSensor(HomematicipGenericEntity, BinarySensorEntity):
return self._device.accelerationSensorTriggered
@property
def extra_state_attributes(self) -> Dict[str, Any]:
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the acceleration sensor."""
state_attr = super().extra_state_attributes
@ -285,7 +287,7 @@ class HomematicipShutterContact(HomematicipMultiContactInterface, BinarySensorEn
return DEVICE_CLASS_DOOR
@property
def extra_state_attributes(self) -> Dict[str, Any]:
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the Shutter Contact."""
state_attr = super().extra_state_attributes
@ -412,7 +414,7 @@ class HomematicipSunshineSensor(HomematicipGenericEntity, BinarySensorEntity):
return self._device.sunshine
@property
def extra_state_attributes(self) -> Dict[str, Any]:
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the illuminance sensor."""
state_attr = super().extra_state_attributes
@ -482,7 +484,7 @@ class HomematicipSecurityZoneSensorGroup(HomematicipGenericEntity, BinarySensorE
return True
@property
def extra_state_attributes(self) -> Dict[str, Any]:
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the security zone group."""
state_attr = super().extra_state_attributes
@ -526,7 +528,7 @@ class HomematicipSecuritySensorGroup(
super().__init__(hap, device, post="Sensors")
@property
def extra_state_attributes(self) -> Dict[str, Any]:
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the security group."""
state_attr = super().extra_state_attributes

24
homeassistant/components/homematicip_cloud/climate.py
View File

@ -1,5 +1,7 @@
"""Support for HomematicIP Cloud climate devices."""
from typing import Any, Dict, List, Optional, Union
from __future__ import annotations
from typing import Any
from homematicip.aio.device import AsyncHeatingThermostat, AsyncHeatingThermostatCompact
from homematicip.aio.group import AsyncHeatingGroup
@ -71,7 +73,7 @@ class HomematicipHeatingGroup(HomematicipGenericEntity, ClimateEntity):
self._simple_heating = self._first_radiator_thermostat
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return device specific attributes."""
return {
"identifiers": {(HMIPC_DOMAIN, self._device.id)},
@ -121,7 +123,7 @@ class HomematicipHeatingGroup(HomematicipGenericEntity, ClimateEntity):
return HVAC_MODE_AUTO
@property
def hvac_modes(self) -> List[str]:
def hvac_modes(self) -> list[str]:
"""Return the list of available hvac operation modes."""
if self._disabled_by_cooling_mode and not self._has_switch:
return [HVAC_MODE_OFF]
@ -133,7 +135,7 @@ class HomematicipHeatingGroup(HomematicipGenericEntity, ClimateEntity):
)
@property
def hvac_action(self) -> Optional[str]:
def hvac_action(self) -> str | None:
"""
Return the current hvac_action.
@ -151,7 +153,7 @@ class HomematicipHeatingGroup(HomematicipGenericEntity, ClimateEntity):
return None
@property
def preset_mode(self) -> Optional[str]:
def preset_mode(self) -> str | None:
"""Return the current preset mode."""
if self._device.boostMode:
return PRESET_BOOST
@ -174,7 +176,7 @@ class HomematicipHeatingGroup(HomematicipGenericEntity, ClimateEntity):
)
@property
def preset_modes(self) -> List[str]:
def preset_modes(self) -> list[str]:
"""Return a list of available preset modes incl. hmip profiles."""
# Boost is only available if a radiator thermostat is in the room,
# and heat mode is enabled.
@ -237,7 +239,7 @@ class HomematicipHeatingGroup(HomematicipGenericEntity, ClimateEntity):
await self._device.set_active_profile(profile_idx)
@property
def extra_state_attributes(self) -> Dict[str, Any]:
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the access point."""
state_attr = super().extra_state_attributes
@ -259,7 +261,7 @@ class HomematicipHeatingGroup(HomematicipGenericEntity, ClimateEntity):
return self._home.get_functionalHome(IndoorClimateHome)
@property
def _device_profiles(self) -> List[str]:
def _device_profiles(self) -> list[str]:
"""Return the relevant profiles."""
return [
profile
@ -270,7 +272,7 @@ class HomematicipHeatingGroup(HomematicipGenericEntity, ClimateEntity):
]
@property
def _device_profile_names(self) -> List[str]:
def _device_profile_names(self) -> list[str]:
"""Return a collection of profile names."""
return [profile.name for profile in self._device_profiles]
@ -298,7 +300,7 @@ class HomematicipHeatingGroup(HomematicipGenericEntity, ClimateEntity):
)
@property
def _relevant_profile_group(self) -> List[str]:
def _relevant_profile_group(self) -> list[str]:
"""Return the relevant profile groups."""
if self._disabled_by_cooling_mode:
return []
@ -322,7 +324,7 @@ class HomematicipHeatingGroup(HomematicipGenericEntity, ClimateEntity):
@property
def _first_radiator_thermostat(
self,
) -> Optional[Union[AsyncHeatingThermostat, AsyncHeatingThermostatCompact]]:
) -> AsyncHeatingThermostat | AsyncHeatingThermostatCompact | None:
"""Return the first radiator thermostat from the hmip heating group."""
for device in self._device.devices:
if isinstance(

12
homeassistant/components/homematicip_cloud/config_flow.py
View File

@ -1,5 +1,7 @@
"""Config flow to configure the HomematicIP Cloud component."""
from typing import Any, Dict
from __future__ import annotations
from typing import Any
import voluptuous as vol
@ -27,11 +29,11 @@ class HomematicipCloudFlowHandler(config_entries.ConfigFlow):
"""Initialize HomematicIP Cloud config flow."""
self.auth = None
async def async_step_user(self, user_input=None) -> Dict[str, Any]:
async def async_step_user(self, user_input=None) -> dict[str, Any]:
"""Handle a flow initialized by the user."""
return await self.async_step_init(user_input)
async def async_step_init(self, user_input=None) -> Dict[str, Any]:
async def async_step_init(self, user_input=None) -> dict[str, Any]:
"""Handle a flow start."""
errors = {}
@ -62,7 +64,7 @@ class HomematicipCloudFlowHandler(config_entries.ConfigFlow):
errors=errors,
)
async def async_step_link(self, user_input=None) -> Dict[str, Any]:
async def async_step_link(self, user_input=None) -> dict[str, Any]:
"""Attempt to link with the HomematicIP Cloud access point."""
errors = {}
@ -84,7 +86,7 @@ class HomematicipCloudFlowHandler(config_entries.ConfigFlow):
return self.async_show_form(step_id="link", errors=errors)
async def async_step_import(self, import_info) -> Dict[str, Any]:
async def async_step_import(self, import_info) -> dict[str, Any]:
"""Import a new access point as a config entry."""
hapid = import_info[HMIPC_HAPID].replace("-", "").upper()
authtoken = import_info[HMIPC_AUTHTOKEN]

10
homeassistant/components/homematicip_cloud/cover.py
View File

@ -1,5 +1,5 @@
"""Support for HomematicIP Cloud cover devices."""
from typing import Optional
from __future__ import annotations
from homematicip.aio.device import (
AsyncBlindModule,
@ -95,7 +95,7 @@ class HomematicipBlindModule(HomematicipGenericEntity, CoverEntity):
)
@property
def is_closed(self) -> Optional[bool]:
def is_closed(self) -> bool | None:
"""Return if the cover is closed."""
if self._device.primaryShadingLevel is not None:
return self._device.primaryShadingLevel == HMIP_COVER_CLOSED
@ -168,7 +168,7 @@ class HomematicipMultiCoverShutter(HomematicipGenericEntity, CoverEntity):
await self._device.set_shutter_level(level, self._channel)
@property
def is_closed(self) -> Optional[bool]:
def is_closed(self) -> bool | None:
"""Return if the cover is closed."""
if self._device.functionalChannels[self._channel].shutterLevel is not None:
return (
@ -265,7 +265,7 @@ class HomematicipGarageDoorModule(HomematicipGenericEntity, CoverEntity):
return door_state_to_position.get(self._device.doorState)
@property
def is_closed(self) -> Optional[bool]:
def is_closed(self) -> bool | None:
"""Return if the cover is closed."""
return self._device.doorState == DoorState.CLOSED
@ -305,7 +305,7 @@ class HomematicipCoverShutterGroup(HomematicipGenericEntity, CoverEntity):
return None
@property
def is_closed(self) -> Optional[bool]:
def is_closed(self) -> bool | None:
"""Return if the cover is closed."""
if self._device.shutterLevel is not None:
return self._device.shutterLevel == HMIP_COVER_CLOSED

16
homeassistant/components/homematicip_cloud/generic_entity.py
View File

@ -1,6 +1,8 @@
"""Generic entity for the HomematicIP Cloud component."""
from __future__ import annotations
import logging
from typing import Any, Dict, Optional
from typing import Any
from homematicip.aio.device import AsyncDevice
from homematicip.aio.group import AsyncGroup
@ -74,9 +76,9 @@ class HomematicipGenericEntity(Entity):
self,
hap: HomematicipHAP,
device,
post: Optional[str] = None,
channel: Optional[int] = None,
is_multi_channel: Optional[bool] = False,
post: str | None = None,
channel: int | None = None,
is_multi_channel: bool | None = False,
) -> None:
"""Initialize the generic entity."""
self._hap = hap
@ -90,7 +92,7 @@ class HomematicipGenericEntity(Entity):
_LOGGER.info("Setting up %s (%s)", self.name, self._device.modelType)
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return device specific attributes."""
# Only physical devices should be HA devices.
if isinstance(self._device, AsyncDevice):
@ -223,7 +225,7 @@ class HomematicipGenericEntity(Entity):
return unique_id
@property
def icon(self) -> Optional[str]:
def icon(self) -> str | None:
"""Return the icon."""
for attr, icon in DEVICE_ATTRIBUTE_ICONS.items():
if getattr(self._device, attr, None):
@ -232,7 +234,7 @@ class HomematicipGenericEntity(Entity):
return None
@property
def extra_state_attributes(self) -> Dict[str, Any]:
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the generic entity."""
state_attr = {}

8
homeassistant/components/homematicip_cloud/light.py
View File

@ -1,5 +1,7 @@
"""Support for HomematicIP Cloud lights."""
from typing import Any, Dict
from __future__ import annotations
from typing import Any
from homematicip.aio.device import (
AsyncBrandDimmer,
@ -90,7 +92,7 @@ class HomematicipLightMeasuring(HomematicipLight):
"""Representation of the HomematicIP measuring light."""
@property
def extra_state_attributes(self) -> Dict[str, Any]:
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the light."""
state_attr = super().extra_state_attributes
@ -206,7 +208,7 @@ class HomematicipNotificationLight(HomematicipGenericEntity, LightEntity):
return self._color_switcher.get(simple_rgb_color, [0.0, 0.0])
@property
def extra_state_attributes(self) -> Dict[str, Any]:
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the notification light sensor."""
state_attr = super().extra_state_attributes

12
homeassistant/components/homematicip_cloud/sensor.py
View File

@ -1,5 +1,7 @@
"""Support for HomematicIP Cloud sensors."""
from typing import Any, Dict
from __future__ import annotations
from typing import Any
from homematicip.aio.device import (
AsyncBrandSwitchMeasuring,
@ -222,7 +224,7 @@ class HomematicipTemperatureSensor(HomematicipGenericEntity):
return TEMP_CELSIUS
@property
def extra_state_attributes(self) -> Dict[str, Any]:
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the windspeed sensor."""
state_attr = super().extra_state_attributes
@ -259,7 +261,7 @@ class HomematicipIlluminanceSensor(HomematicipGenericEntity):
return LIGHT_LUX
@property
def extra_state_attributes(self) -> Dict[str, Any]:
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the wind speed sensor."""
state_attr = super().extra_state_attributes
@ -312,7 +314,7 @@ class HomematicipWindspeedSensor(HomematicipGenericEntity):
return SPEED_KILOMETERS_PER_HOUR
@property
def extra_state_attributes(self) -> Dict[str, Any]:
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the wind speed sensor."""
state_attr = super().extra_state_attributes
@ -354,7 +356,7 @@ class HomematicipPassageDetectorDeltaCounter(HomematicipGenericEntity):
return self._device.leftRightCounterDelta
@property
def extra_state_attributes(self) -> Dict[str, Any]:
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the delta counter."""
state_attr = super().extra_state_attributes

5
homeassistant/components/homematicip_cloud/services.py
View File

@ -1,7 +1,8 @@
"""Support for HomematicIP Cloud devices."""
from __future__ import annotations
import logging
from pathlib import Path
from typing import Optional
from homematicip.aio.device import AsyncSwitchMeasuring
from homematicip.aio.group import AsyncHeatingGroup
@ -342,7 +343,7 @@ async def _async_reset_energy_counter(
await device.reset_energy_counter()
def _get_home(hass: HomeAssistantType, hapid: str) -> Optional[AsyncHome]:
def _get_home(hass: HomeAssistantType, hapid: str) -> AsyncHome | None:
"""Return a HmIP home."""
hap = hass.data[HMIPC_DOMAIN].get(hapid)
if hap:

6
homeassistant/components/homematicip_cloud/switch.py
View File

@ -1,5 +1,7 @@
"""Support for HomematicIP Cloud switches."""
from typing import Any, Dict
from __future__ import annotations
from typing import Any
from homematicip.aio.device import (
AsyncBrandSwitchMeasuring,
@ -141,7 +143,7 @@ class HomematicipGroupSwitch(HomematicipGenericEntity, SwitchEntity):
return True
@property
def extra_state_attributes(self) -> Dict[str, Any]:
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the switch-group."""
state_attr = super().extra_state_attributes

32
homeassistant/components/honeywell/climate.py
View File

@ -1,7 +1,9 @@
"""Support for Honeywell (US) Total Connect Comfort climate systems."""
from __future__ import annotations
import datetime
import logging
from typing import Any, Dict, List, Optional
from typing import Any
import requests
import somecomfort
@ -192,12 +194,12 @@ class HoneywellUSThermostat(ClimateEntity):
self._supported_features |= SUPPORT_FAN_MODE
@property
def name(self) -> Optional[str]:
def name(self) -> str | None:
"""Return the name of the honeywell, if any."""
return self._device.name
@property
def extra_state_attributes(self) -> Dict[str, Any]:
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the device specific state attributes."""
data = {}
data[ATTR_FAN_ACTION] = "running" if self._device.fan_running else "idle"
@ -235,7 +237,7 @@ class HoneywellUSThermostat(ClimateEntity):
return TEMP_CELSIUS if self._device.temperature_unit == "C" else TEMP_FAHRENHEIT
@property
def current_humidity(self) -> Optional[int]:
def current_humidity(self) -> int | None:
"""Return the current humidity."""
return self._device.current_humidity
@ -245,24 +247,24 @@ class HoneywellUSThermostat(ClimateEntity):
return HW_MODE_TO_HVAC_MODE[self._device.system_mode]
@property
def hvac_modes(self) -> List[str]:
def hvac_modes(self) -> list[str]:
"""Return the list of available hvac operation modes."""
return list(self._hvac_mode_map)
@property
def hvac_action(self) -> Optional[str]:
def hvac_action(self) -> str | None:
"""Return the current running hvac operation if supported."""
if self.hvac_mode == HVAC_MODE_OFF:
return None
return HW_MODE_TO_HA_HVAC_ACTION[self._device.equipment_output_status]
@property
def current_temperature(self) -> Optional[float]:
def current_temperature(self) -> float | None:
"""Return the current temperature."""
return self._device.current_temperature
@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._device.setpoint_cool
@ -271,41 +273,41 @@ class HoneywellUSThermostat(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.hvac_mode == HVAC_MODE_HEAT_COOL:
return self._device.setpoint_cool
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.hvac_mode == HVAC_MODE_HEAT_COOL:
return self._device.setpoint_heat
return 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 PRESET_AWAY if self._away else None
@property
def preset_modes(self) -> Optional[List[str]]:
def preset_modes(self) -> list[str] | None:
"""Return a list of available preset modes."""
return [PRESET_NONE, PRESET_AWAY]
@property
def is_aux_heat(self) -> Optional[str]:
def is_aux_heat(self) -> str | None:
"""Return true if aux heater."""
return self._device.system_mode == "emheat"
@property
def fan_mode(self) -> Optional[str]:
def fan_mode(self) -> str | None:
"""Return the fan setting."""
return HW_FAN_MODE_TO_HA[self._device.fan_mode]
@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_mode_map)

12
homeassistant/components/http/__init__.py
View File

@ -1,10 +1,12 @@
"""Support to serve the Home Assistant API as WSGI application."""
from __future__ import annotations
from contextvars import ContextVar
from ipaddress import ip_network
import logging
import os
import ssl
from typing import Dict, Optional, cast
from typing import Optional, cast
from aiohttp import web
from aiohttp.web_exceptions import HTTPMovedPermanently
@ -102,7 +104,7 @@ CONFIG_SCHEMA = vol.Schema({DOMAIN: HTTP_SCHEMA}, extra=vol.ALLOW_EXTRA)
@bind_hass
async def async_get_last_config(hass: HomeAssistant) -> Optional[dict]:
async def async_get_last_config(hass: HomeAssistant) -> dict | None:
"""Return the last known working config."""
store = storage.Store(hass, STORAGE_VERSION, STORAGE_KEY)
return cast(Optional[dict], await store.async_load())
@ -115,7 +117,7 @@ class ApiConfig:
self,
local_ip: str,
host: str,
port: Optional[int] = SERVER_PORT,
port: int | None = SERVER_PORT,
use_ssl: bool = False,
) -> None:
"""Initialize a new API config object."""
@ -379,7 +381,7 @@ class HomeAssistantHTTP:
async def start_http_server_and_save_config(
hass: HomeAssistant, conf: Dict, server: HomeAssistantHTTP
hass: HomeAssistant, conf: dict, server: HomeAssistantHTTP
) -> None:
"""Startup the http server and save the config."""
await server.start() # type: ignore
@ -395,6 +397,6 @@ async def start_http_server_and_save_config(
await store.async_save(conf)
current_request: ContextVar[Optional[web.Request]] = ContextVar(
current_request: ContextVar[web.Request | None] = ContextVar(
"current_request", default=None
)

9
homeassistant/components/http/ban.py
View File

@ -1,10 +1,11 @@
"""Ban logic for HTTP component."""
from __future__ import annotations
from collections import defaultdict
from datetime import datetime
from ipaddress import ip_address
import logging
from socket import gethostbyaddr, herror
from typing import List, Optional
from aiohttp.web import middleware
from aiohttp.web_exceptions import HTTPForbidden, HTTPUnauthorized
@ -178,15 +179,15 @@ async def process_success_login(request):
class IpBan:
"""Represents banned IP address."""
def __init__(self, ip_ban: str, banned_at: Optional[datetime] = None) -> None:
def __init__(self, ip_ban: str, banned_at: datetime | None = None) -> None:
"""Initialize IP Ban object."""
self.ip_address = ip_address(ip_ban)
self.banned_at = banned_at or dt_util.utcnow()
async def async_load_ip_bans_config(hass: HomeAssistant, path: str) -> List[IpBan]:
async def async_load_ip_bans_config(hass: HomeAssistant, path: str) -> list[IpBan]:
"""Load list of banned IPs from config file."""
ip_list: List[IpBan] = []
ip_list: list[IpBan] = []
try:
list_ = await hass.async_add_executor_job(load_yaml_config_file, path)

14
homeassistant/components/http/view.py
View File

@ -1,8 +1,10 @@
"""Support for views."""
from __future__ import annotations
import asyncio
import json
import logging
from typing import Any, Callable, List, Optional
from typing import Any, Callable
from aiohttp import web
from aiohttp.typedefs import LooseHeaders
@ -26,8 +28,8 @@ _LOGGER = logging.getLogger(__name__)
class HomeAssistantView:
"""Base view for all views."""
url: Optional[str] = None
extra_urls: List[str] = []
url: str | None = None
extra_urls: list[str] = []
# Views inheriting from this class can override this
requires_auth = True
cors_allowed = False
@ -45,7 +47,7 @@ class HomeAssistantView:
def json(
result: Any,
status_code: int = HTTP_OK,
headers: Optional[LooseHeaders] = None,
headers: LooseHeaders | None = None,
) -> web.Response:
"""Return a JSON response."""
try:
@ -66,8 +68,8 @@ class HomeAssistantView:
self,
message: str,
status_code: int = HTTP_OK,
message_code: Optional[str] = None,
headers: Optional[LooseHeaders] = None,
message_code: str | None = None,
headers: LooseHeaders | None = None,
) -> web.Response:
"""Return a JSON message response."""
data = {"message": message}

11
homeassistant/components/http/web_runner.py
View File

@ -1,7 +1,8 @@
"""HomeAssistant specific aiohttp Site."""
from __future__ import annotations
import asyncio
from ssl import SSLContext
from typing import List, Optional, Union
from aiohttp import web
from yarl import URL
@ -25,14 +26,14 @@ class HomeAssistantTCPSite(web.BaseSite):
def __init__(
self,
runner: "web.BaseRunner",
host: Union[None, str, List[str]],
host: None | str | list[str],
port: int,
*,
shutdown_timeout: float = 10.0,
ssl_context: Optional[SSLContext] = None,
ssl_context: SSLContext | None = None,
backlog: int = 128,
reuse_address: Optional[bool] = None,
reuse_port: Optional[bool] = None,
reuse_address: bool | None = None,
reuse_port: bool | None = None,
) -> None: # noqa: D107
super().__init__(
runner,

25
homeassistant/components/huawei_lte/__init__.py
View File

@ -1,4 +1,5 @@
"""Support for Huawei LTE routers."""
from __future__ import annotations
from collections import defaultdict
from datetime import timedelta
@ -6,7 +7,7 @@ from functools import partial
import ipaddress
import logging
import time
from typing import Any, Callable, Dict, List, Set, Tuple, cast
from typing import Any, Callable, cast
from urllib.parse import urlparse
import attr
@ -138,13 +139,13 @@ class Router:
mac: str = attr.ib()
signal_update: CALLBACK_TYPE = attr.ib()
data: Dict[str, Any] = attr.ib(init=False, factory=dict)
subscriptions: Dict[str, Set[str]] = attr.ib(
data: dict[str, Any] = attr.ib(init=False, factory=dict)
subscriptions: dict[str, set[str]] = attr.ib(
init=False,
factory=lambda: defaultdict(set, ((x, {"initial_scan"}) for x in ALL_KEYS)),
)
inflight_gets: Set[str] = attr.ib(init=False, factory=set)
unload_handlers: List[CALLBACK_TYPE] = attr.ib(init=False, factory=list)
inflight_gets: set[str] = attr.ib(init=False, factory=set)
unload_handlers: list[CALLBACK_TYPE] = attr.ib(init=False, factory=list)
client: Client
suspended = attr.ib(init=False, default=False)
notify_last_attempt: float = attr.ib(init=False, default=-1)
@ -167,7 +168,7 @@ class Router:
return DEFAULT_DEVICE_NAME
@property
def device_identifiers(self) -> Set[Tuple[str, str]]:
def device_identifiers(self) -> set[tuple[str, str]]:
"""Get router identifiers for device registry."""
try:
return {(DOMAIN, self.data[KEY_DEVICE_INFORMATION]["SerialNumber"])}
@ -175,7 +176,7 @@ class Router:
return set()
@property
def device_connections(self) -> Set[Tuple[str, str]]:
def device_connections(self) -> set[tuple[str, str]]:
"""Get router connections for device registry."""
return {(dr.CONNECTION_NETWORK_MAC, self.mac)} if self.mac else set()
@ -304,8 +305,8 @@ class HuaweiLteData:
hass_config: dict = attr.ib()
# Our YAML config, keyed by router URL
config: Dict[str, Dict[str, Any]] = attr.ib()
routers: Dict[str, Router] = attr.ib(init=False, factory=dict)
config: dict[str, dict[str, Any]] = attr.ib()
routers: dict[str, Router] = attr.ib(init=False, factory=dict)
async def async_setup_entry(hass: HomeAssistantType, config_entry: ConfigEntry) -> bool:
@ -484,7 +485,7 @@ async def async_setup(hass: HomeAssistantType, config: ConfigType) -> bool:
logging.getLogger("dicttoxml").setLevel(logging.WARNING)
# Arrange our YAML config to dict with normalized URLs as keys
domain_config: Dict[str, Dict[str, Any]] = {}
domain_config: dict[str, dict[str, Any]] = {}
if DOMAIN not in hass.data:
hass.data[DOMAIN] = HuaweiLteData(hass_config=config, config=domain_config)
for router_config in config.get(DOMAIN, []):
@ -588,7 +589,7 @@ class HuaweiLteBaseEntity(Entity):
router: Router = attr.ib()
_available: bool = attr.ib(init=False, default=True)
_unsub_handlers: List[Callable] = attr.ib(init=False, factory=list)
_unsub_handlers: list[Callable] = attr.ib(init=False, factory=list)
@property
def _entity_name(self) -> str:
@ -620,7 +621,7 @@ class HuaweiLteBaseEntity(Entity):
return False
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Get info for matching with parent router."""
return {
"identifiers": self.router.device_identifiers,

11
homeassistant/components/huawei_lte/binary_sensor.py
View File

@ -1,7 +1,8 @@
"""Support for Huawei LTE binary sensors."""
from __future__ import annotations
import logging
from typing import Any, Callable, Dict, List, Optional
from typing import Any, Callable
import attr
from huawei_lte_api.enums.cradle import ConnectionStatusEnum
@ -29,11 +30,11 @@ _LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistantType,
config_entry: ConfigEntry,
async_add_entities: Callable[[List[Entity], bool], None],
async_add_entities: Callable[[list[Entity], bool], None],
) -> None:
"""Set up from config entry."""
router = hass.data[DOMAIN].routers[config_entry.data[CONF_URL]]
entities: List[Entity] = []
entities: list[Entity] = []
if router.data.get(KEY_MONITORING_STATUS):
entities.append(HuaweiLteMobileConnectionBinarySensor(router))
@ -53,7 +54,7 @@ class HuaweiLteBaseBinarySensor(HuaweiLteBaseEntity, BinarySensorEntity):
key: str
item: str
_raw_state: Optional[str] = attr.ib(init=False, default=None)
_raw_state: str | None = attr.ib(init=False, default=None)
@property
def entity_registry_enabled_default(self) -> bool:
@ -142,7 +143,7 @@ class HuaweiLteMobileConnectionBinarySensor(HuaweiLteBaseBinarySensor):
return True
@property
def extra_state_attributes(self) -> Optional[Dict[str, Any]]:
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Get additional attributes related to connection status."""
attributes = {}
if self._raw_state in CONNECTION_STATE_ATTRIBUTES:

30
homeassistant/components/huawei_lte/config_flow.py
View File

@ -3,7 +3,7 @@ from __future__ import annotations
from collections import OrderedDict
import logging
from typing import Any, Dict, Optional
from typing import Any
from urllib.parse import urlparse
from huawei_lte_api.AuthorizedConnection import AuthorizedConnection
@ -55,9 +55,9 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
async def _async_show_user_form(
self,
user_input: Optional[Dict[str, Any]] = None,
errors: Optional[Dict[str, str]] = None,
) -> Dict[str, Any]:
user_input: dict[str, Any] | None = None,
errors: dict[str, str] | None = None,
) -> dict[str, Any]:
if user_input is None:
user_input = {}
return self.async_show_form(
@ -94,12 +94,12 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
)
async def async_step_import(
self, user_input: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, user_input: dict[str, Any] | None = None
) -> dict[str, Any]:
"""Handle import initiated config flow."""
return await self.async_step_user(user_input)
def _already_configured(self, user_input: Dict[str, Any]) -> bool:
def _already_configured(self, user_input: dict[str, Any]) -> bool:
"""See if we already have a router matching user input configured."""
existing_urls = {
url_normalize(entry.data[CONF_URL], default_scheme="http")
@ -108,8 +108,8 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return user_input[CONF_URL] in existing_urls
async def async_step_user(
self, user_input: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, user_input: dict[str, Any] | None = None
) -> dict[str, Any]:
"""Handle user initiated config flow."""
if user_input is None:
return await self._async_show_user_form()
@ -129,7 +129,7 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
if self._already_configured(user_input):
return self.async_abort(reason="already_configured")
conn: Optional[Connection] = None
conn: Connection | None = None
def logout() -> None:
if isinstance(conn, AuthorizedConnection):
@ -138,7 +138,7 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
except Exception: # pylint: disable=broad-except
_LOGGER.debug("Could not logout", exc_info=True)
def try_connect(user_input: Dict[str, Any]) -> Connection:
def try_connect(user_input: dict[str, Any]) -> Connection:
"""Try connecting with given credentials."""
username = user_input.get(CONF_USERNAME)
password = user_input.get(CONF_PASSWORD)
@ -222,8 +222,8 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_create_entry(title=title, data=user_input)
async def async_step_ssdp( # type: ignore # mypy says signature incompatible with supertype, but it's the same?
self, discovery_info: Dict[str, Any]
) -> Dict[str, Any]:
self, discovery_info: dict[str, Any]
) -> dict[str, Any]:
"""Handle SSDP initiated config flow."""
await self.async_set_unique_id(discovery_info[ssdp.ATTR_UPNP_UDN])
self._abort_if_unique_id_configured()
@ -263,8 +263,8 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
self.config_entry = config_entry
async def async_step_init(
self, user_input: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, user_input: dict[str, Any] | None = None
) -> dict[str, Any]:
"""Handle options flow."""
# Recipients are persisted as a list, but handled as comma separated string in UI

21
homeassistant/components/huawei_lte/device_tracker.py
View File

@ -1,8 +1,9 @@
"""Support for device tracking of Huawei LTE routers."""
from __future__ import annotations
import logging
import re
from typing import Any, Callable, Dict, List, Optional, Set, cast
from typing import Any, Callable, cast
import attr
from stringcase import snakecase
@ -31,7 +32,7 @@ _DEVICE_SCAN = f"{DEVICE_TRACKER_DOMAIN}/device_scan"
async def async_setup_entry(
hass: HomeAssistantType,
config_entry: ConfigEntry,
async_add_entities: Callable[[List[Entity], bool], None],
async_add_entities: Callable[[list[Entity], bool], None],
) -> None:
"""Set up from config entry."""
@ -46,9 +47,9 @@ async def async_setup_entry(
return
# Initialize already tracked entities
tracked: Set[str] = set()
tracked: set[str] = set()
registry = await entity_registry.async_get_registry(hass)
known_entities: List[Entity] = []
known_entities: list[Entity] = []
for entity in registry.entities.values():
if (
entity.domain == DEVICE_TRACKER_DOMAIN
@ -82,8 +83,8 @@ async def async_setup_entry(
def async_add_new_entities(
hass: HomeAssistantType,
router_url: str,
async_add_entities: Callable[[List[Entity], bool], None],
tracked: Set[str],
async_add_entities: Callable[[list[Entity], bool], None],
tracked: set[str],
) -> None:
"""Add new entities that are not already being tracked."""
router = hass.data[DOMAIN].routers[router_url]
@ -93,7 +94,7 @@ def async_add_new_entities(
_LOGGER.debug("%s[%s][%s] not in data", KEY_WLAN_HOST_LIST, "Hosts", "Host")
return
new_entities: List[Entity] = []
new_entities: list[Entity] = []
for host in (x for x in hosts if x.get("MacAddress")):
entity = HuaweiLteScannerEntity(router, host["MacAddress"])
if entity.unique_id in tracked:
@ -125,8 +126,8 @@ class HuaweiLteScannerEntity(HuaweiLteBaseEntity, ScannerEntity):
mac: str = attr.ib()
_is_connected: bool = attr.ib(init=False, default=False)
_hostname: Optional[str] = attr.ib(init=False, default=None)
_extra_state_attributes: Dict[str, Any] = attr.ib(init=False, factory=dict)
_hostname: str | None = attr.ib(init=False, default=None)
_extra_state_attributes: dict[str, Any] = attr.ib(init=False, factory=dict)
def __attrs_post_init__(self) -> None:
"""Initialize internal state."""
@ -151,7 +152,7 @@ class HuaweiLteScannerEntity(HuaweiLteBaseEntity, ScannerEntity):
return self._is_connected
@property
def extra_state_attributes(self) -> Dict[str, Any]:
def extra_state_attributes(self) -> dict[str, Any]:
"""Get additional attributes related to entity state."""
return self._extra_state_attributes

10
homeassistant/components/huawei_lte/notify.py
View File

@ -3,7 +3,7 @@ from __future__ import annotations
import logging
import time
from typing import Any, Dict, List, Optional
from typing import Any
import attr
from huawei_lte_api.exceptions import ResponseErrorException
@ -20,9 +20,9 @@ _LOGGER = logging.getLogger(__name__)
async def async_get_service(
hass: HomeAssistantType,
config: Dict[str, Any],
discovery_info: Optional[Dict[str, Any]] = None,
) -> Optional[HuaweiLteSmsNotificationService]:
config: dict[str, Any],
discovery_info: dict[str, Any] | None = None,
) -> HuaweiLteSmsNotificationService | None:
"""Get the notification service."""
if discovery_info is None:
return None
@ -38,7 +38,7 @@ class HuaweiLteSmsNotificationService(BaseNotificationService):
"""Huawei LTE router SMS notification service."""
router: Router = attr.ib()
default_targets: List[str] = attr.ib()
default_targets: list[str] = attr.ib()
def send_message(self, message: str = "", **kwargs: Any) -> None:
"""Send message to target numbers."""

33
homeassistant/components/huawei_lte/sensor.py
View File

@ -1,9 +1,10 @@
"""Support for Huawei LTE sensors."""
from __future__ import annotations
from bisect import bisect
import logging
import re
from typing import Callable, Dict, List, NamedTuple, Optional, Pattern, Tuple, Union
from typing import Callable, NamedTuple, Pattern
import attr
@ -45,17 +46,17 @@ _LOGGER = logging.getLogger(__name__)
class SensorMeta(NamedTuple):
"""Metadata for defining sensors."""
name: Optional[str] = None
device_class: Optional[str] = None
icon: Union[str, Callable[[StateType], str], None] = None
unit: Optional[str] = None
name: str | None = None
device_class: str | None = None
icon: str | Callable[[StateType], str] | None = None
unit: str | None = None
enabled_default: bool = False
include: Optional[Pattern[str]] = None
exclude: Optional[Pattern[str]] = None
formatter: Optional[Callable[[str], Tuple[StateType, Optional[str]]]] = None
include: Pattern[str] | None = None
exclude: Pattern[str] | None = None
formatter: Callable[[str], tuple[StateType, str | None]] | None = None
SENSOR_META: Dict[Union[str, Tuple[str, str]], SensorMeta] = {
SENSOR_META: dict[str | tuple[str, str], SensorMeta] = {
KEY_DEVICE_INFORMATION: SensorMeta(
include=re.compile(r"^WanIP.*Address$", re.IGNORECASE)
),
@ -329,11 +330,11 @@ SENSOR_META: Dict[Union[str, Tuple[str, str]], SensorMeta] = {
async def async_setup_entry(
hass: HomeAssistantType,
config_entry: ConfigEntry,
async_add_entities: Callable[[List[Entity], bool], None],
async_add_entities: Callable[[list[Entity], bool], None],
) -> None:
"""Set up from config entry."""
router = hass.data[DOMAIN].routers[config_entry.data[CONF_URL]]
sensors: List[Entity] = []
sensors: list[Entity] = []
for key in SENSOR_KEYS:
items = router.data.get(key)
if not items:
@ -354,7 +355,7 @@ async def async_setup_entry(
async_add_entities(sensors, True)
def format_default(value: StateType) -> Tuple[StateType, Optional[str]]:
def format_default(value: StateType) -> tuple[StateType, str | None]:
"""Format value."""
unit = None
if value is not None:
@ -380,7 +381,7 @@ class HuaweiLteSensor(HuaweiLteBaseEntity):
meta: SensorMeta = attr.ib()
_state: StateType = attr.ib(init=False, default=STATE_UNKNOWN)
_unit: Optional[str] = attr.ib(init=False)
_unit: str | None = attr.ib(init=False)
async def async_added_to_hass(self) -> None:
"""Subscribe to needed data on add."""
@ -406,17 +407,17 @@ class HuaweiLteSensor(HuaweiLteBaseEntity):
return self._state
@property
def device_class(self) -> Optional[str]:
def device_class(self) -> str | None:
"""Return sensor device class."""
return self.meta.device_class
@property
def unit_of_measurement(self) -> Optional[str]:
def unit_of_measurement(self) -> str | None:
"""Return sensor's unit of measurement."""
return self.meta.unit or self._unit
@property
def icon(self) -> Optional[str]:
def icon(self) -> str | None:
"""Return icon for sensor."""
icon = self.meta.icon
if callable(icon):

9
homeassistant/components/huawei_lte/switch.py
View File

@ -1,7 +1,8 @@
"""Support for Huawei LTE switches."""
from __future__ import annotations
import logging
from typing import Any, Callable, List, Optional
from typing import Any, Callable
import attr
@ -24,11 +25,11 @@ _LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistantType,
config_entry: ConfigEntry,
async_add_entities: Callable[[List[Entity], bool], None],
async_add_entities: Callable[[list[Entity], bool], None],
) -> None:
"""Set up from config entry."""
router = hass.data[DOMAIN].routers[config_entry.data[CONF_URL]]
switches: List[Entity] = []
switches: list[Entity] = []
if router.data.get(KEY_DIALUP_MOBILE_DATASWITCH):
switches.append(HuaweiLteMobileDataSwitch(router))
@ -42,7 +43,7 @@ class HuaweiLteBaseSwitch(HuaweiLteBaseEntity, SwitchEntity):
key: str
item: str
_raw_state: Optional[str] = attr.ib(init=False, default=None)
_raw_state: str | None = attr.ib(init=False, default=None)
def _turn(self, state: bool) -> None:
raise NotImplementedError

18
homeassistant/components/hue/config_flow.py
View File

@ -1,6 +1,8 @@
"""Config flow to configure Philips Hue."""
from __future__ import annotations
import asyncio
from typing import Any, Dict, Optional
from typing import Any
from urllib.parse import urlparse
import aiohue
@ -44,8 +46,8 @@ class HueFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
def __init__(self):
"""Initialize the Hue flow."""
self.bridge: Optional[aiohue.Bridge] = None
self.discovered_bridges: Optional[Dict[str, aiohue.Bridge]] = None
self.bridge: aiohue.Bridge | None = None
self.discovered_bridges: dict[str, aiohue.Bridge] | None = None
async def async_step_user(self, user_input=None):
"""Handle a flow initialized by the user."""
@ -53,7 +55,7 @@ class HueFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return await self.async_step_init(user_input)
@core.callback
def _async_get_bridge(self, host: str, bridge_id: Optional[str] = None):
def _async_get_bridge(self, host: str, bridge_id: str | None = None):
"""Return a bridge object."""
if bridge_id is not None:
bridge_id = normalize_bridge_id(bridge_id)
@ -114,8 +116,8 @@ class HueFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
)
async def async_step_manual(
self, user_input: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, user_input: dict[str, Any] | None = None
) -> dict[str, Any]:
"""Handle manual bridge setup."""
if user_input is None:
return self.async_show_form(
@ -249,8 +251,8 @@ class HueOptionsFlowHandler(config_entries.OptionsFlow):
self.config_entry = config_entry
async def async_step_init(
self, user_input: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, user_input: dict[str, Any] | None = None
) -> dict[str, Any]:
"""Manage Hue options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)

14
homeassistant/components/humidifier/__init__.py
View File

@ -1,7 +1,9 @@
"""Provides functionality to interact with humidifier devices."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any, Dict, List, Optional
from typing import Any
import voluptuous as vol
@ -100,7 +102,7 @@ class HumidifierEntity(ToggleEntity):
"""Representation of a humidifier device."""
@property
def capability_attributes(self) -> Dict[str, Any]:
def capability_attributes(self) -> dict[str, Any]:
"""Return capability attributes."""
supported_features = self.supported_features or 0
data = {
@ -114,7 +116,7 @@ class HumidifierEntity(ToggleEntity):
return data
@property
def state_attributes(self) -> Dict[str, Any]:
def state_attributes(self) -> dict[str, Any]:
"""Return the optional state attributes."""
supported_features = self.supported_features or 0
data = {}
@ -128,12 +130,12 @@ class HumidifierEntity(ToggleEntity):
return data
@property
def target_humidity(self) -> Optional[int]:
def target_humidity(self) -> int | None:
"""Return the humidity we try to reach."""
return None
@property
def mode(self) -> Optional[str]:
def mode(self) -> str | None:
"""Return the current mode, e.g., home, auto, baby.
Requires SUPPORT_MODES.
@ -141,7 +143,7 @@ class HumidifierEntity(ToggleEntity):
raise NotImplementedError
@property
def available_modes(self) -> Optional[List[str]]:
def available_modes(self) -> list[str] | None:
"""Return a list of available modes.
Requires SUPPORT_MODES.

6
homeassistant/components/humidifier/device_action.py
View File

@ -1,5 +1,5 @@
"""Provides device actions for Humidifier."""
from typing import List, Optional
from __future__ import annotations
import voluptuous as vol
@ -40,7 +40,7 @@ ONOFF_SCHEMA = toggle_entity.ACTION_SCHEMA.extend({vol.Required(CONF_DOMAIN): DO
ACTION_SCHEMA = vol.Any(SET_HUMIDITY_SCHEMA, SET_MODE_SCHEMA, ONOFF_SCHEMA)
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 Humidifier devices."""
registry = await entity_registry.async_get_registry(hass)
actions = await toggle_entity.async_get_actions(hass, device_id, DOMAIN)
@ -79,7 +79,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)

4
homeassistant/components/humidifier/device_condition.py
View File

@ -1,5 +1,5 @@
"""Provide the device automations for Humidifier."""
from typing import Dict, List
from __future__ import annotations
import voluptuous as vol
@ -38,7 +38,7 @@ CONDITION_SCHEMA = vol.Any(TOGGLE_CONDITION, MODE_CONDITION)
async def async_get_conditions(
hass: HomeAssistant, device_id: str
) -> List[Dict[str, str]]:
) -> list[dict[str, str]]:
"""List device conditions for Humidifier devices."""
registry = await entity_registry.async_get_registry(hass)
conditions = await toggle_entity.async_get_conditions(hass, device_id, DOMAIN)

4
homeassistant/components/humidifier/device_trigger.py
View File

@ -1,5 +1,5 @@
"""Provides device automations for Climate."""
from typing import List
from __future__ import annotations
import voluptuous as vol
@ -48,7 +48,7 @@ TOGGLE_TRIGGER_SCHEMA = toggle_entity.TRIGGER_SCHEMA.extend(
TRIGGER_SCHEMA = vol.Any(TARGET_TRIGGER_SCHEMA, TOGGLE_TRIGGER_SCHEMA)
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 Humidifier devices."""
registry = await entity_registry.async_get_registry(hass)
triggers = await toggle_entity.async_get_triggers(hass, device_id, DOMAIN)

12
homeassistant/components/humidifier/reproduce_state.py
View File

@ -1,7 +1,9 @@
"""Module that groups code required to handle state restore for component."""
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_MODE,
@ -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 component states."""
cur_state = hass.states.get(state.entity_id)
@ -82,8 +84,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(

15
homeassistant/components/hyperion/__init__.py
View File

@ -1,8 +1,9 @@
"""The Hyperion component."""
from __future__ import annotations
import asyncio
import logging
from typing import Any, Callable, Dict, List, Optional, Set, Tuple, cast
from typing import Any, Callable, cast
from awesomeversion import AwesomeVersion
from hyperion import client, const as hyperion_const
@ -70,7 +71,7 @@ def get_hyperion_unique_id(server_id: str, instance: int, name: str) -> str:
return f"{server_id}_{instance}_{name}"
def split_hyperion_unique_id(unique_id: str) -> Optional[Tuple[str, int, str]]:
def split_hyperion_unique_id(unique_id: str) -> tuple[str, int, str] | None:
"""Split a unique_id into a (server_id, instance, type) tuple."""
data = tuple(unique_id.split("_", 2))
if len(data) != 3:
@ -92,7 +93,7 @@ def create_hyperion_client(
async def async_create_connect_hyperion_client(
*args: Any,
**kwargs: Any,
) -> Optional[client.HyperionClient]:
) -> client.HyperionClient | None:
"""Create and connect a Hyperion Client."""
hyperion_client = create_hyperion_client(*args, **kwargs)
@ -207,17 +208,17 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
CONF_ON_UNLOAD: [],
}
async def async_instances_to_clients(response: Dict[str, Any]) -> None:
async def async_instances_to_clients(response: dict[str, Any]) -> None:
"""Convert instances to Hyperion clients."""
if not response or hyperion_const.KEY_DATA not in response:
return
await async_instances_to_clients_raw(response[hyperion_const.KEY_DATA])
async def async_instances_to_clients_raw(instances: List[Dict[str, Any]]) -> None:
async def async_instances_to_clients_raw(instances: list[dict[str, Any]]) -> None:
"""Convert instances to Hyperion clients."""
registry = await async_get_registry(hass)
running_instances: Set[int] = set()
stopped_instances: Set[int] = set()
running_instances: set[int] = set()
stopped_instances: set[int] = set()
existing_instances = hass.data[DOMAIN][config_entry.entry_id][
CONF_INSTANCE_CLIENTS
]

52
homeassistant/components/hyperion/config_flow.py
View File

@ -3,7 +3,7 @@ from __future__ import annotations
import asyncio
import logging
from typing import Any, Dict, Optional
from typing import Any
from urllib.parse import urlparse
from hyperion import client, const
@ -111,9 +111,9 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
def __init__(self) -> None:
"""Instantiate config flow."""
self._data: Dict[str, Any] = {}
self._request_token_task: Optional[asyncio.Task] = None
self._auth_id: Optional[str] = None
self._data: dict[str, Any] = {}
self._request_token_task: asyncio.Task | None = None
self._auth_id: str | None = None
self._require_confirm: bool = False
self._port_ui: int = const.DEFAULT_PORT_UI
@ -128,7 +128,7 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
async def _advance_to_auth_step_if_necessary(
self, hyperion_client: client.HyperionClient
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Determine if auth is required."""
auth_resp = await hyperion_client.async_is_auth_required()
@ -143,7 +143,7 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
async def async_step_reauth(
self,
config_data: ConfigType,
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Handle a reauthentication flow."""
self._data = dict(config_data)
async with self._create_client(raw_connection=True) as hyperion_client:
@ -152,8 +152,8 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
return await self._advance_to_auth_step_if_necessary(hyperion_client)
async def async_step_ssdp( # type: ignore[override]
self, discovery_info: Dict[str, Any]
) -> Dict[str, Any]:
self, discovery_info: dict[str, Any]
) -> dict[str, Any]:
"""Handle a flow initiated by SSDP."""
# Sample data provided by SSDP: {
# 'ssdp_location': 'http://192.168.0.1:8090/description.xml',
@ -223,8 +223,8 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
async def async_step_user(
self,
user_input: Optional[ConfigType] = None,
) -> Dict[str, Any]:
user_input: ConfigType | None = None,
) -> dict[str, Any]:
"""Handle a flow initiated by the user."""
errors = {}
if user_input:
@ -262,7 +262,7 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
async def _request_token_task_func(self, auth_id: str) -> None:
"""Send an async_request_token request."""
auth_resp: Optional[Dict[str, Any]] = None
auth_resp: dict[str, Any] | None = None
async with self._create_client(raw_connection=True) as hyperion_client:
if hyperion_client:
# The Hyperion-py client has a default timeout of 3 minutes on this request.
@ -283,7 +283,7 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
# used to open a URL, that the user already knows the address of).
return f"http://{self._data[CONF_HOST]}:{self._port_ui}"
async def _can_login(self) -> Optional[bool]:
async def _can_login(self) -> bool | None:
"""Verify login details."""
async with self._create_client(raw_connection=True) as hyperion_client:
if not hyperion_client:
@ -296,8 +296,8 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
async def async_step_auth(
self,
user_input: Optional[ConfigType] = None,
) -> Dict[str, Any]:
user_input: ConfigType | None = None,
) -> dict[str, Any]:
"""Handle the auth step of a flow."""
errors = {}
if user_input:
@ -325,8 +325,8 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
)
async def async_step_create_token(
self, user_input: Optional[ConfigType] = None
) -> Dict[str, Any]:
self, user_input: ConfigType | None = None
) -> dict[str, Any]:
"""Send a request for a new token."""
if user_input is None:
self._auth_id = client.generate_random_auth_id()
@ -351,8 +351,8 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
)
async def async_step_create_token_external(
self, auth_resp: Optional[ConfigType] = None
) -> Dict[str, Any]:
self, auth_resp: ConfigType | None = None
) -> dict[str, Any]:
"""Handle completion of the request for a new token."""
if auth_resp is not None and client.ResponseOK(auth_resp):
token = auth_resp.get(const.KEY_INFO, {}).get(const.KEY_TOKEN)
@ -364,8 +364,8 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
return self.async_external_step_done(next_step_id="create_token_fail")
async def async_step_create_token_success(
self, _: Optional[ConfigType] = None
) -> Dict[str, Any]:
self, _: ConfigType | None = None
) -> dict[str, Any]:
"""Create an entry after successful token creation."""
# Clean-up the request task.
await self._cancel_request_token_task()
@ -380,16 +380,16 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
return await self.async_step_confirm()
async def async_step_create_token_fail(
self, _: Optional[ConfigType] = None
) -> Dict[str, Any]:
self, _: ConfigType | None = None
) -> dict[str, Any]:
"""Show an error on the auth form."""
# Clean-up the request task.
await self._cancel_request_token_task()
return self.async_abort(reason="auth_new_token_not_granted_error")
async def async_step_confirm(
self, user_input: Optional[ConfigType] = None
) -> Dict[str, Any]:
self, user_input: ConfigType | None = None
) -> dict[str, Any]:
"""Get final confirmation before entry creation."""
if user_input is None and self._require_confirm:
return self.async_show_form(
@ -440,8 +440,8 @@ class HyperionOptionsFlow(OptionsFlow):
self._config_entry = config_entry
async def async_step_init(
self, user_input: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, user_input: dict[str, Any] | None = None
) -> dict[str, Any]:
"""Manage the options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)

46
homeassistant/components/hyperion/light.py
View File

@ -4,7 +4,7 @@ from __future__ import annotations
import functools
import logging
from types import MappingProxyType
from typing import Any, Callable, Dict, List, Mapping, Optional, Sequence, Tuple
from typing import Any, Callable, Mapping, Sequence
from hyperion import client, const
@ -63,7 +63,7 @@ DEFAULT_EFFECT = KEY_EFFECT_SOLID
DEFAULT_NAME = "Hyperion"
DEFAULT_PORT = const.DEFAULT_PORT_JSON
DEFAULT_HDMI_PRIORITY = 880
DEFAULT_EFFECT_LIST: List[str] = []
DEFAULT_EFFECT_LIST: list[str] = []
SUPPORT_HYPERION = SUPPORT_COLOR | SUPPORT_BRIGHTNESS | SUPPORT_EFFECT
@ -142,12 +142,12 @@ class HyperionBaseLight(LightEntity):
self._rgb_color: Sequence[int] = DEFAULT_COLOR
self._effect: str = KEY_EFFECT_SOLID
self._static_effect_list: List[str] = [KEY_EFFECT_SOLID]
self._static_effect_list: list[str] = [KEY_EFFECT_SOLID]
if self._support_external_effects:
self._static_effect_list += list(const.KEY_COMPONENTID_EXTERNAL_SOURCES)
self._effect_list: List[str] = self._static_effect_list[:]
self._effect_list: list[str] = self._static_effect_list[:]
self._client_callbacks: Mapping[str, Callable[[Dict[str, Any]], None]] = {
self._client_callbacks: Mapping[str, Callable[[dict[str, Any]], None]] = {
f"{const.KEY_ADJUSTMENT}-{const.KEY_UPDATE}": self._update_adjustment,
f"{const.KEY_COMPONENTS}-{const.KEY_UPDATE}": self._update_components,
f"{const.KEY_EFFECTS}-{const.KEY_UPDATE}": self._update_effect_list,
@ -176,7 +176,7 @@ class HyperionBaseLight(LightEntity):
return self._brightness
@property
def hs_color(self) -> Tuple[float, float]:
def hs_color(self) -> tuple[float, float]:
"""Return last color value set."""
return color_util.color_RGB_to_hs(*self._rgb_color)
@ -196,7 +196,7 @@ class HyperionBaseLight(LightEntity):
return self._effect
@property
def effect_list(self) -> List[str]:
def effect_list(self) -> list[str]:
"""Return the list of supported effects."""
return self._effect_list
@ -305,9 +305,9 @@ class HyperionBaseLight(LightEntity):
def _set_internal_state(
self,
brightness: Optional[int] = None,
rgb_color: Optional[Sequence[int]] = None,
effect: Optional[str] = None,
brightness: int | None = None,
rgb_color: Sequence[int] | None = None,
effect: str | None = None,
) -> None:
"""Set the internal state."""
if brightness is not None:
@ -318,12 +318,12 @@ class HyperionBaseLight(LightEntity):
self._effect = effect
@callback
def _update_components(self, _: Optional[Dict[str, Any]] = None) -> None:
def _update_components(self, _: dict[str, Any] | None = None) -> None:
"""Update Hyperion components."""
self.async_write_ha_state()
@callback
def _update_adjustment(self, _: Optional[Dict[str, Any]] = None) -> None:
def _update_adjustment(self, _: dict[str, Any] | None = None) -> None:
"""Update Hyperion adjustments."""
if self._client.adjustment:
brightness_pct = self._client.adjustment[0].get(
@ -337,7 +337,7 @@ class HyperionBaseLight(LightEntity):
self.async_write_ha_state()
@callback
def _update_priorities(self, _: Optional[Dict[str, Any]] = None) -> None:
def _update_priorities(self, _: dict[str, Any] | None = None) -> None:
"""Update Hyperion priorities."""
priority = self._get_priority_entry_that_dictates_state()
if priority and self._allow_priority_update(priority):
@ -361,11 +361,11 @@ class HyperionBaseLight(LightEntity):
self.async_write_ha_state()
@callback
def _update_effect_list(self, _: Optional[Dict[str, Any]] = None) -> None:
def _update_effect_list(self, _: dict[str, Any] | None = None) -> None:
"""Update Hyperion effects."""
if not self._client.effects:
return
effect_list: List[str] = []
effect_list: list[str] = []
for effect in self._client.effects or []:
if const.KEY_NAME in effect:
effect_list.append(effect[const.KEY_NAME])
@ -391,7 +391,7 @@ class HyperionBaseLight(LightEntity):
)
@callback
def _update_client(self, _: Optional[Dict[str, Any]] = None) -> None:
def _update_client(self, _: dict[str, Any] | None = None) -> None:
"""Update client connection state."""
self.async_write_ha_state()
@ -419,18 +419,18 @@ class HyperionBaseLight(LightEntity):
"""Whether or not to support setting external effects from the light entity."""
return True
def _get_priority_entry_that_dictates_state(self) -> Optional[Dict[str, Any]]:
def _get_priority_entry_that_dictates_state(self) -> dict[str, Any] | None:
"""Get the relevant Hyperion priority entry to consider."""
# Return the visible priority (whether or not it is the HA priority).
# Explicit type specifier to ensure this works when the underlying (typed)
# library is installed along with the tests. Casts would trigger a
# redundant-cast warning in this case.
priority: Optional[Dict[str, Any]] = self._client.visible_priority
priority: dict[str, Any] | None = self._client.visible_priority
return priority
# pylint: disable=no-self-use
def _allow_priority_update(self, priority: Optional[Dict[str, Any]] = None) -> bool:
def _allow_priority_update(self, priority: dict[str, Any] | None = None) -> bool:
"""Determine whether to allow a priority to update internal state."""
return True
@ -525,7 +525,7 @@ class HyperionPriorityLight(HyperionBaseLight):
"""Whether or not to support setting external effects from the light entity."""
return False
def _get_priority_entry_that_dictates_state(self) -> Optional[Dict[str, Any]]:
def _get_priority_entry_that_dictates_state(self) -> dict[str, Any] | None:
"""Get the relevant Hyperion priority entry to consider."""
# Return the active priority (if any) at the configured HA priority.
for candidate in self._client.priorities or []:
@ -537,12 +537,12 @@ class HyperionPriorityLight(HyperionBaseLight):
# Explicit type specifier to ensure this works when the underlying
# (typed) library is installed along with the tests. Casts would trigger
# a redundant-cast warning in this case.
output: Dict[str, Any] = candidate
output: dict[str, Any] = candidate
return output
return None
@classmethod
def _is_priority_entry_black(cls, priority: Optional[Dict[str, Any]]) -> bool:
def _is_priority_entry_black(cls, priority: dict[str, Any] | None) -> bool:
"""Determine if a given priority entry is the color black."""
if not priority:
return False
@ -552,7 +552,7 @@ class HyperionPriorityLight(HyperionBaseLight):
return True
return False
def _allow_priority_update(self, priority: Optional[Dict[str, Any]] = None) -> bool:
def _allow_priority_update(self, priority: dict[str, Any] | None = None) -> bool:
"""Determine whether to allow a Hyperion priority to update entity attributes."""
# Black is treated as 'off' (and Home Assistant does not support selecting black
# from the color selector). Do not set our internal attributes if the priority is

5
homeassistant/components/hyperion/switch.py
View File

@ -1,7 +1,8 @@
"""Switch platform for Hyperion."""
from __future__ import annotations
import functools
from typing import Any, Callable, Dict, Optional
from typing import Any, Callable
from hyperion import client
from hyperion.const import (
@ -187,7 +188,7 @@ class HyperionComponentSwitch(SwitchEntity):
await self._async_send_set_component(False)
@callback
def _update_components(self, _: Optional[Dict[str, Any]] = None) -> None:
def _update_components(self, _: dict[str, Any] | None = None) -> None:
"""Update Hyperion components."""
self.async_write_ha_state()