Update typing 12 (#48073)

This commit is contained in:
Marc Mueller 2021-03-18 14:31:38 +01:00 committed by GitHub
parent 6cd6ad6904
commit a3cd1854f6
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
75 changed files with 399 additions and 312 deletions

View file

@ -1,7 +1,8 @@
"""Support for Queensland Bushfire Alert Feeds."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Optional
from georss_qld_bushfire_alert_client import QldBushfireAlertFeedManager
import voluptuous as vol
@ -209,22 +210,22 @@ class QldBushfireLocationEvent(GeolocationEvent):
return SOURCE
@property
def name(self) -> Optional[str]:
def name(self) -> str | None:
"""Return the name of the entity."""
return self._name
@property
def distance(self) -> Optional[float]:
def distance(self) -> float | None:
"""Return distance value of this external event."""
return self._distance
@property
def latitude(self) -> Optional[float]:
def latitude(self) -> float | None:
"""Return latitude value of this external event."""
return self._latitude
@property
def longitude(self) -> Optional[float]:
def longitude(self) -> float | None:
"""Return longitude value of this external event."""
return self._longitude

View file

@ -1,6 +1,7 @@
"""Adapter to wrap the rachiopy api for home assistant."""
from __future__ import annotations
import logging
from typing import Optional
import voluptuous as vol
@ -239,7 +240,7 @@ class RachioIro:
# Only enabled zones
return [z for z in self._zones if z[KEY_ENABLED]]
def get_zone(self, zone_id) -> Optional[dict]:
def get_zone(self, zone_id) -> dict | None:
"""Return the zone with the given ID."""
for zone in self.list_zones(include_disabled=True):
if zone[KEY_ID] == zone_id:

View file

@ -1,7 +1,8 @@
"""The ReCollect Waste integration."""
from __future__ import annotations
import asyncio
from datetime import date, timedelta
from typing import List
from aiorecollect.client import Client, PickupEvent
from aiorecollect.errors import RecollectError
@ -35,7 +36,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
entry.data[CONF_PLACE_ID], entry.data[CONF_SERVICE_ID], session=session
)
async def async_get_pickup_events() -> List[PickupEvent]:
async def async_get_pickup_events() -> list[PickupEvent]:
"""Get the next pickup."""
try:
return await client.async_get_pickup_events(

View file

@ -1,5 +1,5 @@
"""Config flow for ReCollect Waste integration."""
from typing import Optional
from __future__ import annotations
from aiorecollect.client import Client
from aiorecollect.errors import RecollectError
@ -83,7 +83,7 @@ class RecollectWasteOptionsFlowHandler(config_entries.OptionsFlow):
"""Initialize."""
self._entry = entry
async def async_step_init(self, user_input: Optional[dict] = None):
async def async_step_init(self, user_input: dict | None = None):
"""Manage the options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)

View file

@ -1,5 +1,7 @@
"""Support for ReCollect Waste sensors."""
from typing import Callable, List
from __future__ import annotations
from typing import Callable
from aiorecollect.client import PickupType
import voluptuous as vol
@ -36,8 +38,8 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
@callback
def async_get_pickup_type_names(
entry: ConfigEntry, pickup_types: List[PickupType]
) -> List[str]:
entry: ConfigEntry, pickup_types: list[PickupType]
) -> list[str]:
"""Return proper pickup type names from their associated objects."""
return [
t.friendly_name

View file

@ -1,4 +1,6 @@
"""Support for recording details."""
from __future__ import annotations
import asyncio
import concurrent.futures
from datetime import datetime
@ -7,7 +9,7 @@ import queue
import sqlite3
import threading
import time
from typing import Any, Callable, List, NamedTuple, Optional
from typing import Any, Callable, NamedTuple
from sqlalchemy import create_engine, event as sqlalchemy_event, exc, select
from sqlalchemy.orm import scoped_session, sessionmaker
@ -125,7 +127,7 @@ CONFIG_SCHEMA = vol.Schema(
)
def run_information(hass, point_in_time: Optional[datetime] = None):
def run_information(hass, point_in_time: datetime | None = None):
"""Return information about current run.
There is also the run that covers point_in_time.
@ -138,7 +140,7 @@ def run_information(hass, point_in_time: Optional[datetime] = None):
return run_information_with_session(session, point_in_time)
def run_information_from_instance(hass, point_in_time: Optional[datetime] = None):
def run_information_from_instance(hass, point_in_time: datetime | None = None):
"""Return information about current run from the existing instance.
Does not query the database for older runs.
@ -149,7 +151,7 @@ def run_information_from_instance(hass, point_in_time: Optional[datetime] = None
return ins.run_info
def run_information_with_session(session, point_in_time: Optional[datetime] = None):
def run_information_with_session(session, point_in_time: datetime | None = None):
"""Return information about current run from the database."""
recorder_runs = RecorderRuns
@ -249,7 +251,7 @@ class Recorder(threading.Thread):
db_max_retries: int,
db_retry_wait: int,
entity_filter: Callable[[str], bool],
exclude_t: List[str],
exclude_t: list[str],
db_integrity_check: bool,
) -> None:
"""Initialize the recorder."""

View file

@ -1,8 +1,10 @@
"""Support to interface with universal remote control devices."""
from __future__ import annotations
from datetime import timedelta
import functools as ft
import logging
from typing import Any, Dict, Iterable, List, Optional, cast
from typing import Any, Iterable, cast
import voluptuous as vol
@ -147,17 +149,17 @@ class RemoteEntity(ToggleEntity):
return 0
@property
def current_activity(self) -> Optional[str]:
def current_activity(self) -> str | None:
"""Active activity."""
return None
@property
def activity_list(self) -> Optional[List[str]]:
def activity_list(self) -> list[str] | None:
"""List of available activities."""
return None
@property
def state_attributes(self) -> Optional[Dict[str, Any]]:
def state_attributes(self) -> dict[str, Any] | None:
"""Return optional state attributes."""
if not self.supported_features & SUPPORT_ACTIVITY:
return None

View file

@ -1,5 +1,5 @@
"""Provides device actions for remotes."""
from typing import List
from __future__ import annotations
import voluptuous as vol
@ -25,6 +25,6 @@ async def async_call_action_from_config(
)
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."""
return await toggle_entity.async_get_actions(hass, device_id, DOMAIN)

View file

@ -1,5 +1,5 @@
"""Provides device conditions for remotes."""
from typing import Dict, List
from __future__ import annotations
import voluptuous as vol
@ -28,7 +28,7 @@ def async_condition_from_config(
async def async_get_conditions(
hass: HomeAssistant, device_id: str
) -> List[Dict[str, str]]:
) -> list[dict[str, str]]:
"""List device conditions."""
return await toggle_entity.async_get_conditions(hass, device_id, DOMAIN)

View file

@ -1,5 +1,5 @@
"""Provides device triggers for remotes."""
from typing import List
from __future__ import annotations
import voluptuous as vol
@ -28,7 +28,7 @@ async def async_attach_trigger(
)
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."""
return await toggle_entity.async_get_triggers(hass, device_id, DOMAIN)

View file

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

View file

@ -1,10 +1,11 @@
"""Support for Ring Doorbell/Chimes."""
from __future__ import annotations
import asyncio
from datetime import timedelta
from functools import partial
import logging
from pathlib import Path
from typing import Optional
from oauthlib.oauth2 import AccessDeniedError
import requests
@ -187,7 +188,7 @@ class GlobalDataUpdater:
self._unsub_interval()
self._unsub_interval = None
async def async_refresh_all(self, _now: Optional[int] = None) -> None:
async def async_refresh_all(self, _now: int | None = None) -> None:
"""Time to update."""
if not self.listeners:
return

View file

@ -1,8 +1,10 @@
"""Support for Roku."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
from typing import Any, Dict
from typing import Any
from rokuecp import Roku, RokuConnectionError, RokuError
from rokuecp.models import Device
@ -38,7 +40,7 @@ SCAN_INTERVAL = timedelta(seconds=15)
_LOGGER = logging.getLogger(__name__)
async def async_setup(hass: HomeAssistantType, config: Dict) -> bool:
async def async_setup(hass: HomeAssistantType, config: dict) -> bool:
"""Set up the Roku integration."""
hass.data.setdefault(DOMAIN, {})
return True
@ -151,7 +153,7 @@ class RokuEntity(CoordinatorEntity):
return self._name
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return device information about this Roku device."""
if self._device_id is None:
return None

View file

@ -1,6 +1,8 @@
"""Config flow for Roku."""
from __future__ import annotations
import logging
from typing import Any, Dict, Optional
from typing import Any
from urllib.parse import urlparse
from rokuecp import Roku, RokuError
@ -27,7 +29,7 @@ ERROR_UNKNOWN = "unknown"
_LOGGER = logging.getLogger(__name__)
async def validate_input(hass: HomeAssistantType, data: Dict) -> Dict:
async def validate_input(hass: HomeAssistantType, data: dict) -> dict:
"""Validate the user input allows us to connect.
Data has the keys from DATA_SCHEMA with values provided by the user.
@ -53,7 +55,7 @@ class RokuConfigFlow(ConfigFlow, domain=DOMAIN):
self.discovery_info = {}
@callback
def _show_form(self, errors: Optional[Dict] = None) -> Dict[str, Any]:
def _show_form(self, errors: dict | None = None) -> dict[str, Any]:
"""Show the form to the user."""
return self.async_show_form(
step_id="user",
@ -61,9 +63,7 @@ class RokuConfigFlow(ConfigFlow, domain=DOMAIN):
errors=errors or {},
)
async def async_step_user(
self, user_input: Optional[Dict] = None
) -> Dict[str, Any]:
async def async_step_user(self, user_input: dict | None = None) -> dict[str, Any]:
"""Handle a flow initialized by the user."""
if not user_input:
return self._show_form()
@ -115,8 +115,8 @@ class RokuConfigFlow(ConfigFlow, domain=DOMAIN):
return await self.async_step_discovery_confirm()
async def async_step_ssdp(
self, discovery_info: Optional[Dict] = None
) -> Dict[str, Any]:
self, discovery_info: dict | None = None
) -> dict[str, Any]:
"""Handle a flow initialized by discovery."""
host = urlparse(discovery_info[ATTR_SSDP_LOCATION]).hostname
name = discovery_info[ATTR_UPNP_FRIENDLY_NAME]
@ -141,8 +141,8 @@ class RokuConfigFlow(ConfigFlow, domain=DOMAIN):
return await self.async_step_discovery_confirm()
async def async_step_discovery_confirm(
self, user_input: Optional[Dict] = None
) -> Dict[str, Any]:
self, user_input: dict | None = None
) -> dict[str, Any]:
"""Handle user-confirmation of discovered device."""
if user_input is None:
return self.async_show_form(

View file

@ -1,6 +1,7 @@
"""Support for the Roku media player."""
from __future__ import annotations
import logging
from typing import List, Optional
import voluptuous as vol
@ -100,7 +101,7 @@ class RokuMediaPlayer(RokuEntity, MediaPlayerEntity):
return self._unique_id
@property
def device_class(self) -> Optional[str]:
def device_class(self) -> str | None:
"""Return the class of this device."""
if self.coordinator.data.info.device_type == "tv":
return DEVICE_CLASS_TV
@ -230,7 +231,7 @@ class RokuMediaPlayer(RokuEntity, MediaPlayerEntity):
return None
@property
def source_list(self) -> List:
def source_list(self) -> list:
"""List of available input sources."""
return ["Home"] + sorted(app.name for app in self.coordinator.data.apps)

View file

@ -1,5 +1,7 @@
"""Support for the Roku remote."""
from typing import Callable, List
from __future__ import annotations
from typing import Callable
from homeassistant.components.remote import ATTR_NUM_REPEATS, RemoteEntity
from homeassistant.config_entries import ConfigEntry
@ -12,7 +14,7 @@ from .const import DOMAIN
async def async_setup_entry(
hass: HomeAssistantType,
entry: ConfigEntry,
async_add_entities: Callable[[List, bool], None],
async_add_entities: Callable[[list, bool], None],
) -> bool:
"""Load Roku remote based on a config entry."""
coordinator = hass.data[DOMAIN][entry.entry_id]
@ -56,7 +58,7 @@ class RokuRemote(RokuEntity, RemoteEntity):
await self.coordinator.async_request_refresh()
@roku_exception_handler
async def async_send_command(self, command: List, **kwargs) -> None:
async def async_send_command(self, command: list, **kwargs) -> None:
"""Send a command to one device."""
num_repeats = kwargs[ATTR_NUM_REPEATS]

View file

@ -1,7 +1,8 @@
"""Update the IP addresses of your Route53 DNS records."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import List
import boto3
import requests
@ -77,7 +78,7 @@ def _update_route53(
aws_secret_access_key: str,
zone: str,
domain: str,
records: List[str],
records: list[str],
ttl: int,
):
_LOGGER.debug("Starting update for zone %s", zone)

View file

@ -1,5 +1,7 @@
"""Config flow for Raspberry Pi Power Supply Checker."""
from typing import Any, Dict, Optional
from __future__ import annotations
from typing import Any
from rpi_bad_power import new_under_voltage
@ -31,8 +33,8 @@ class RPiPowerFlow(DiscoveryFlowHandler, domain=DOMAIN):
)
async def async_step_onboarding(
self, data: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, data: dict[str, Any] | None = None
) -> dict[str, Any]:
"""Handle a flow initialized by onboarding."""
has_devices = await self._discovery_function(self.hass)

View file

@ -1,5 +1,5 @@
"""Support for Ruckus Unleashed devices."""
from typing import Optional
from __future__ import annotations
from homeassistant.components.device_tracker import SOURCE_TYPE_ROUTER
from homeassistant.components.device_tracker.config_entry import ScannerEntity
@ -115,7 +115,7 @@ class RuckusUnleashedDevice(CoordinatorEntity, ScannerEntity):
return SOURCE_TYPE_ROUTER
@property
def device_info(self) -> Optional[dict]:
def device_info(self) -> dict | None:
"""Return the device information."""
if self.is_connected:
return {

View file

@ -1,8 +1,10 @@
"""Allow users to set and activate scenes."""
from __future__ import annotations
import functools as ft
import importlib
import logging
from typing import Any, Optional
from typing import Any
import voluptuous as vol
@ -94,7 +96,7 @@ class Scene(Entity):
return False
@property
def state(self) -> Optional[str]:
def state(self) -> str | None:
"""Return the state of the scene."""
return STATE

View file

@ -1,7 +1,8 @@
"""Support for scripts."""
from __future__ import annotations
import asyncio
import logging
from typing import List
import voluptuous as vol
@ -89,7 +90,7 @@ def is_on(hass, entity_id):
@callback
def scripts_with_entity(hass: HomeAssistant, entity_id: str) -> List[str]:
def scripts_with_entity(hass: HomeAssistant, entity_id: str) -> list[str]:
"""Return all scripts that reference the entity."""
if DOMAIN not in hass.data:
return []
@ -104,7 +105,7 @@ def scripts_with_entity(hass: HomeAssistant, entity_id: str) -> List[str]:
@callback
def entities_in_script(hass: HomeAssistant, entity_id: str) -> List[str]:
def entities_in_script(hass: HomeAssistant, entity_id: str) -> list[str]:
"""Return all entities in script."""
if DOMAIN not in hass.data:
return []
@ -120,7 +121,7 @@ def entities_in_script(hass: HomeAssistant, entity_id: str) -> List[str]:
@callback
def scripts_with_device(hass: HomeAssistant, device_id: str) -> List[str]:
def scripts_with_device(hass: HomeAssistant, device_id: str) -> list[str]:
"""Return all scripts that reference the device."""
if DOMAIN not in hass.data:
return []
@ -135,7 +136,7 @@ def scripts_with_device(hass: HomeAssistant, device_id: str) -> List[str]:
@callback
def devices_in_script(hass: HomeAssistant, entity_id: str) -> List[str]:
def devices_in_script(hass: HomeAssistant, entity_id: str) -> list[str]:
"""Return all devices in script."""
if DOMAIN not in hass.data:
return []

View file

@ -1,5 +1,5 @@
"""Provides device conditions for sensors."""
from typing import Dict, List
from __future__ import annotations
import voluptuous as vol
@ -109,9 +109,9 @@ CONDITION_SCHEMA = vol.All(
async def async_get_conditions(
hass: HomeAssistant, device_id: str
) -> List[Dict[str, str]]:
) -> list[dict[str, str]]:
"""List device conditions."""
conditions: List[Dict[str, str]] = []
conditions: list[dict[str, str]] = []
entity_registry = await async_get_registry(hass)
entries = [
entry

View file

@ -1,5 +1,7 @@
"""Helper to test significant sensor state changes."""
from typing import Any, Optional, Union
from __future__ import annotations
from typing import Any
from homeassistant.const import (
ATTR_DEVICE_CLASS,
@ -19,7 +21,7 @@ def async_check_significant_change(
new_state: str,
new_attrs: dict,
**kwargs: Any,
) -> Optional[bool]:
) -> bool | None:
"""Test if state significantly changed."""
device_class = new_attrs.get(ATTR_DEVICE_CLASS)
@ -28,7 +30,7 @@ def async_check_significant_change(
if device_class == DEVICE_CLASS_TEMPERATURE:
if new_attrs.get(ATTR_UNIT_OF_MEASUREMENT) == TEMP_FAHRENHEIT:
change: Union[float, int] = 1
change: float | int = 1
else:
change = 0.5

View file

@ -1,6 +1,7 @@
"""The sentry integration."""
from __future__ import annotations
import re
from typing import Dict, Union
import sentry_sdk
from sentry_sdk.integrations.aiohttp import AioHttpIntegration
@ -126,8 +127,8 @@ def process_before_send(
options,
channel: str,
huuid: str,
system_info: Dict[str, Union[bool, str]],
custom_components: Dict[str, Integration],
system_info: dict[str, bool | str],
custom_components: dict[str, Integration],
event,
hint,
):

View file

@ -2,7 +2,7 @@
from __future__ import annotations
import logging
from typing import Any, Dict, Optional
from typing import Any
from sentry_sdk.utils import BadDsn, Dsn
import voluptuous as vol
@ -47,8 +47,8 @@ class SentryConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return SentryOptionsFlow(config_entry)
async def async_step_user(
self, user_input: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, user_input: dict[str, Any] | None = None
) -> dict[str, Any]:
"""Handle a user config flow."""
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
@ -78,8 +78,8 @@ class SentryOptionsFlow(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 Sentry options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)

View file

@ -1,7 +1,7 @@
"""Config flow for Shark IQ integration."""
from __future__ import annotations
import asyncio
from typing import Dict, Optional
import aiohttp
import async_timeout
@ -62,7 +62,7 @@ class SharkIqConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors["base"] = "unknown"
return info, errors
async def async_step_user(self, user_input: Optional[Dict] = None):
async def async_step_user(self, user_input: dict | None = None):
"""Handle the initial step."""
errors = {}
if user_input is not None:
@ -76,7 +76,7 @@ class SharkIqConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
step_id="user", data_schema=SHARKIQ_SCHEMA, errors=errors
)
async def async_step_reauth(self, user_input: Optional[dict] = None):
async def async_step_reauth(self, user_input: dict | None = None):
"""Handle re-auth if login is invalid."""
errors = {}

View file

@ -1,7 +1,7 @@
"""Data update coordinator for shark iq vacuums."""
from __future__ import annotations
import asyncio
from typing import Dict, List, Set
from async_timeout import timeout
from sharkiqpy import (
@ -27,11 +27,11 @@ class SharkIqUpdateCoordinator(DataUpdateCoordinator):
hass: HomeAssistant,
config_entry: ConfigEntry,
ayla_api: AylaApi,
shark_vacs: List[SharkIqVacuum],
shark_vacs: list[SharkIqVacuum],
) -> None:
"""Set up the SharkIqUpdateCoordinator class."""
self.ayla_api = ayla_api
self.shark_vacs: Dict[str, SharkIqVacuum] = {
self.shark_vacs: dict[str, SharkIqVacuum] = {
sharkiq.serial_number: sharkiq for sharkiq in shark_vacs
}
self._config_entry = config_entry
@ -40,7 +40,7 @@ class SharkIqUpdateCoordinator(DataUpdateCoordinator):
super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=UPDATE_INTERVAL)
@property
def online_dsns(self) -> Set[str]:
def online_dsns(self) -> set[str]:
"""Get the set of all online DSNs."""
return self._online_dsns

View file

@ -1,8 +1,8 @@
"""Shark IQ Wrapper."""
from __future__ import annotations
import logging
from typing import Dict, Iterable, Optional
from typing import Iterable
from sharkiqpy import OperatingModes, PowerModes, Properties, SharkIqVacuum
@ -118,7 +118,7 @@ class SharkVacuumEntity(CoordinatorEntity, StateVacuumEntity):
return self.sharkiq.oem_model_number
@property
def device_info(self) -> Dict:
def device_info(self) -> dict:
"""Device info dictionary."""
return {
"identifiers": {(DOMAIN, self.serial_number)},
@ -136,30 +136,30 @@ class SharkVacuumEntity(CoordinatorEntity, StateVacuumEntity):
return SUPPORT_SHARKIQ
@property
def is_docked(self) -> Optional[bool]:
def is_docked(self) -> bool | None:
"""Is vacuum docked."""
return self.sharkiq.get_property_value(Properties.DOCKED_STATUS)
@property
def error_code(self) -> Optional[int]:
def error_code(self) -> int | None:
"""Return the last observed error code (or None)."""
return self.sharkiq.error_code
@property
def error_message(self) -> Optional[str]:
def error_message(self) -> str | None:
"""Return the last observed error message (or None)."""
if not self.error_code:
return None
return self.sharkiq.error_text
@property
def operating_mode(self) -> Optional[str]:
def operating_mode(self) -> str | None:
"""Operating mode.."""
op_mode = self.sharkiq.get_property_value(Properties.OPERATING_MODE)
return OPERATING_STATE_MAP.get(op_mode)
@property
def recharging_to_resume(self) -> Optional[int]:
def recharging_to_resume(self) -> int | None:
"""Return True if vacuum set to recharge and resume cleaning."""
return self.sharkiq.get_property_value(Properties.RECHARGING_TO_RESUME)
@ -240,12 +240,12 @@ class SharkVacuumEntity(CoordinatorEntity, StateVacuumEntity):
# Various attributes we want to expose
@property
def recharge_resume(self) -> Optional[bool]:
def recharge_resume(self) -> bool | None:
"""Recharge and resume mode active."""
return self.sharkiq.get_property_value(Properties.RECHARGE_RESUME)
@property
def rssi(self) -> Optional[int]:
def rssi(self) -> int | None:
"""Get the WiFi RSSI."""
return self.sharkiq.get_property_value(Properties.RSSI)
@ -255,7 +255,7 @@ class SharkVacuumEntity(CoordinatorEntity, StateVacuumEntity):
return self.sharkiq.get_property_value(Properties.LOW_LIGHT_MISSION)
@property
def extra_state_attributes(self) -> Dict:
def extra_state_attributes(self) -> dict:
"""Return a dictionary of device state attributes specific to sharkiq."""
data = {
ATTR_ERROR_CODE: self.error_code,

View file

@ -1,5 +1,5 @@
"""Provides device triggers for Shelly."""
from typing import List
from __future__ import annotations
import voluptuous as vol
@ -60,7 +60,7 @@ async def async_validate_trigger_config(hass, config):
)
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 Shelly devices."""
triggers = []

View file

@ -1,7 +1,9 @@
"""Shelly entity helper."""
from __future__ import annotations
from dataclasses import dataclass
import logging
from typing import Any, Callable, Optional, Union
from typing import Any, Callable
import aioshelly
@ -142,15 +144,15 @@ class BlockAttributeDescription:
name: str
# Callable = lambda attr_info: unit
icon: Optional[str] = None
unit: Union[None, str, Callable[[dict], str]] = None
icon: str | None = None
unit: None | str | Callable[[dict], str] = None
value: Callable[[Any], Any] = lambda val: val
device_class: Optional[str] = None
device_class: str | None = None
default_enabled: bool = True
available: Optional[Callable[[aioshelly.Block], bool]] = None
available: Callable[[aioshelly.Block], bool] | None = None
# Callable (settings, block), return true if entity should be removed
removal_condition: Optional[Callable[[dict, aioshelly.Block], bool]] = None
extra_state_attributes: Optional[Callable[[aioshelly.Block], Optional[dict]]] = None
removal_condition: Callable[[dict, aioshelly.Block], bool] | None = None
extra_state_attributes: Callable[[aioshelly.Block], dict | None] | None = None
@dataclass
@ -158,12 +160,12 @@ class RestAttributeDescription:
"""Class to describe a REST sensor."""
name: str
icon: Optional[str] = None
unit: Optional[str] = None
icon: str | None = None
unit: str | None = None
value: Callable[[dict, Any], Any] = None
device_class: Optional[str] = None
device_class: str | None = None
default_enabled: bool = True
extra_state_attributes: Optional[Callable[[dict], Optional[dict]]] = None
extra_state_attributes: Callable[[dict], dict | None] | None = None
class ShellyBlockEntity(entity.Entity):
@ -385,7 +387,7 @@ class ShellySleepingBlockAttributeEntity(ShellyBlockAttributeEntity, RestoreEnti
block: aioshelly.Block,
attribute: str,
description: BlockAttributeDescription,
entry: Optional[ConfigEntry] = None,
entry: ConfigEntry | None = None,
) -> None:
"""Initialize the sleeping sensor."""
self.last_state = None

View file

@ -1,5 +1,5 @@
"""Light for Shelly."""
from typing import Optional, Tuple
from __future__ import annotations
from aioshelly import Block
@ -96,7 +96,7 @@ class ShellyLight(ShellyBlockEntity, LightEntity):
return self.block.output
@property
def mode(self) -> Optional[str]:
def mode(self) -> str | None:
"""Return the color mode of the light."""
if self.mode_result:
return self.mode_result["mode"]
@ -138,7 +138,7 @@ class ShellyLight(ShellyBlockEntity, LightEntity):
return int(white)
@property
def hs_color(self) -> Tuple[float, float]:
def hs_color(self) -> tuple[float, float]:
"""Return the hue and saturation color value of light."""
if self.mode == "white":
return color_RGB_to_hs(255, 255, 255)
@ -154,7 +154,7 @@ class ShellyLight(ShellyBlockEntity, LightEntity):
return color_RGB_to_hs(red, green, blue)
@property
def color_temp(self) -> Optional[int]:
def color_temp(self) -> int | None:
"""Return the CT color value in mireds."""
if self.mode == "color":
return None

View file

@ -1,8 +1,8 @@
"""Shelly helpers functions."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import List, Optional, Tuple
import aioshelly
@ -67,7 +67,7 @@ def get_number_of_channels(device: aioshelly.Device, block: aioshelly.Block) ->
def get_entity_name(
device: aioshelly.Device,
block: aioshelly.Block,
description: Optional[str] = None,
description: str | None = None,
) -> str:
"""Naming for switch and sensors."""
channel_name = get_device_channel_name(device, block)
@ -143,7 +143,7 @@ def get_device_uptime(status: dict, last_uptime: str) -> str:
def get_input_triggers(
device: aioshelly.Device, block: aioshelly.Block
) -> List[Tuple[str, str]]:
) -> list[tuple[str, str]]:
"""Return list of input triggers for block."""
if "inputEvent" not in block.sensor_ids or "inputEventCnt" not in block.sensor_ids:
return []

View file

@ -1,5 +1,7 @@
"""Support for binary sensors through the SmartThings cloud API."""
from typing import Optional, Sequence
from __future__ import annotations
from typing import Sequence
from pysmartthings import Attribute, Capability
@ -52,7 +54,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
async_add_entities(sensors)
def get_capabilities(capabilities: Sequence[str]) -> Optional[Sequence[str]]:
def get_capabilities(capabilities: Sequence[str]) -> Sequence[str] | None:
"""Return all capabilities supported if minimum required are present."""
return [
capability for capability in CAPABILITY_TO_ATTRIB if capability in capabilities

View file

@ -1,8 +1,10 @@
"""Support for climate devices through the SmartThings cloud API."""
from __future__ import annotations
import asyncio
from collections.abc import Iterable
import logging
from typing import Optional, Sequence
from typing import Sequence
from pysmartthings import Attribute, Capability
@ -103,7 +105,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
async_add_entities(entities, True)
def get_capabilities(capabilities: Sequence[str]) -> Optional[Sequence[str]]:
def get_capabilities(capabilities: Sequence[str]) -> Sequence[str] | None:
"""Return all capabilities supported if minimum required are present."""
supported = [
Capability.air_conditioner_mode,
@ -274,7 +276,7 @@ class SmartThingsThermostat(SmartThingsEntity, ClimateEntity):
return self._device.status.supported_thermostat_fan_modes
@property
def hvac_action(self) -> Optional[str]:
def hvac_action(self) -> str | None:
"""Return the current running hvac operation if supported."""
return OPERATING_STATE_TO_ACTION.get(
self._device.status.thermostat_operating_state

View file

@ -1,5 +1,7 @@
"""Support for covers through the SmartThings cloud API."""
from typing import Optional, Sequence
from __future__ import annotations
from typing import Sequence
from pysmartthings import Attribute, Capability
@ -46,7 +48,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
)
def get_capabilities(capabilities: Sequence[str]) -> Optional[Sequence[str]]:
def get_capabilities(capabilities: Sequence[str]) -> Sequence[str] | None:
"""Return all capabilities supported if minimum required are present."""
min_required = [
Capability.door_control,

View file

@ -1,6 +1,8 @@
"""Support for fans through the SmartThings cloud API."""
from __future__ import annotations
import math
from typing import Optional, Sequence
from typing import Sequence
from pysmartthings import Capability
@ -29,7 +31,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
)
def get_capabilities(capabilities: Sequence[str]) -> Optional[Sequence[str]]:
def get_capabilities(capabilities: Sequence[str]) -> Sequence[str] | None:
"""Return all capabilities supported if minimum required are present."""
supported = [Capability.switch, Capability.fan_speed]
# Must have switch and fan_speed

View file

@ -1,6 +1,8 @@
"""Support for lights through the SmartThings cloud API."""
from __future__ import annotations
import asyncio
from typing import Optional, Sequence
from typing import Sequence
from pysmartthings import Capability
@ -34,7 +36,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
)
def get_capabilities(capabilities: Sequence[str]) -> Optional[Sequence[str]]:
def get_capabilities(capabilities: Sequence[str]) -> Sequence[str] | None:
"""Return all capabilities supported if minimum required are present."""
supported = [
Capability.switch,

View file

@ -1,5 +1,7 @@
"""Support for locks through the SmartThings cloud API."""
from typing import Optional, Sequence
from __future__ import annotations
from typing import Sequence
from pysmartthings import Attribute, Capability
@ -31,7 +33,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
)
def get_capabilities(capabilities: Sequence[str]) -> Optional[Sequence[str]]:
def get_capabilities(capabilities: Sequence[str]) -> Sequence[str] | None:
"""Return all capabilities supported if minimum required are present."""
if Capability.lock in capabilities:
return [Capability.lock]

View file

@ -1,6 +1,8 @@
"""Support for sensors through the SmartThings cloud API."""
from __future__ import annotations
from collections import namedtuple
from typing import Optional, Sequence
from typing import Sequence
from pysmartthings import Attribute, Capability
@ -297,7 +299,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
async_add_entities(sensors)
def get_capabilities(capabilities: Sequence[str]) -> Optional[Sequence[str]]:
def get_capabilities(capabilities: Sequence[str]) -> Sequence[str] | None:
"""Return all capabilities supported if minimum required are present."""
return [
capability for capability in CAPABILITY_TO_SENSORS if capability in capabilities

View file

@ -1,5 +1,7 @@
"""Support for switches through the SmartThings cloud API."""
from typing import Optional, Sequence
from __future__ import annotations
from typing import Sequence
from pysmartthings import Attribute, Capability
@ -21,7 +23,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
)
def get_capabilities(capabilities: Sequence[str]) -> Optional[Sequence[str]]:
def get_capabilities(capabilities: Sequence[str]) -> Sequence[str] | None:
"""Return all capabilities supported if minimum required are present."""
# Must be able to be turned on/off.
if Capability.switch in capabilities:

View file

@ -1,8 +1,9 @@
"""Support for the Swedish weather institute weather service."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
from typing import Dict, List
import aiohttp
import async_timeout
@ -210,7 +211,7 @@ class SmhiWeather(WeatherEntity):
return "Swedish weather institute (SMHI)"
@property
def forecast(self) -> List:
def forecast(self) -> list:
"""Return the forecast."""
if self._forecasts is None or len(self._forecasts) < 2:
return None
@ -235,7 +236,7 @@ class SmhiWeather(WeatherEntity):
return data
@property
def extra_state_attributes(self) -> Dict:
def extra_state_attributes(self) -> dict:
"""Return SMHI specific attributes."""
if self.cloudiness:
return {ATTR_SMHI_CLOUDINESS: self.cloudiness}

View file

@ -1,6 +1,7 @@
"""API for Somfy bound to Home Assistant OAuth."""
from __future__ import annotations
from asyncio import run_coroutine_threadsafe
from typing import Dict, Union
from pymfy.api import somfy_api
@ -27,7 +28,7 @@ class ConfigEntrySomfyApi(somfy_api.SomfyApi):
def refresh_tokens(
self,
) -> Dict[str, Union[str, int]]:
) -> dict[str, str | int]:
"""Refresh and return new Somfy tokens using Home Assistant OAuth2 session."""
run_coroutine_threadsafe(
self.session.async_ensure_token_valid(), self.hass.loop

View file

@ -1,6 +1,5 @@
"""Support for Somfy Thermostat."""
from typing import List, Optional
from __future__ import annotations
from pymfy.api.devices.category import Category
from pymfy.api.devices.thermostat import (
@ -125,7 +124,7 @@ class SomfyClimate(SomfyEntity, ClimateEntity):
return HVAC_MODES_MAPPING.get(self._climate.get_hvac_state())
@property
def hvac_modes(self) -> List[str]:
def hvac_modes(self) -> list[str]:
"""Return the list of available hvac operation modes.
HEAT and COOL mode are exclusive. End user has to enable a mode manually within the Somfy application.
@ -144,13 +143,13 @@ class SomfyClimate(SomfyEntity, ClimateEntity):
)
@property
def preset_mode(self) -> Optional[str]:
def preset_mode(self) -> str | None:
"""Return the current preset mode."""
mode = self._climate.get_target_mode()
return PRESETS_MAPPING.get(mode)
@property
def preset_modes(self) -> Optional[List[str]]:
def preset_modes(self) -> list[str] | None:
"""Return a list of available preset modes."""
return list(PRESETS_MAPPING.values())

View file

@ -1,8 +1,10 @@
"""The Sonarr component."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
from typing import Any, Dict
from typing import Any
from sonarr import Sonarr, SonarrAccessRestricted, SonarrError
@ -40,7 +42,7 @@ SCAN_INTERVAL = timedelta(seconds=30)
_LOGGER = logging.getLogger(__name__)
async def async_setup(hass: HomeAssistantType, config: Dict) -> bool:
async def async_setup(hass: HomeAssistantType, config: dict) -> bool:
"""Set up the Sonarr component."""
hass.data.setdefault(DOMAIN, {})
return True
@ -164,7 +166,7 @@ class SonarrEntity(Entity):
return self._enabled_default
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return device information about the application."""
if self._device_id is None:
return None

View file

@ -1,6 +1,8 @@
"""Config flow for Sonarr."""
from __future__ import annotations
import logging
from typing import Any, Dict, Optional
from typing import Any
from sonarr import Sonarr, SonarrAccessRestricted, SonarrError
import voluptuous as vol
@ -33,7 +35,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.
@ -73,9 +75,7 @@ class SonarrConfigFlow(ConfigFlow, domain=DOMAIN):
"""Get the options flow for this handler."""
return SonarrOptionsFlowHandler(config_entry)
async def async_step_reauth(
self, data: Optional[ConfigType] = None
) -> Dict[str, Any]:
async def async_step_reauth(self, data: ConfigType | None = None) -> dict[str, Any]:
"""Handle configuration by re-auth."""
self._reauth = True
self._entry_data = dict(data)
@ -84,8 +84,8 @@ class SonarrConfigFlow(ConfigFlow, domain=DOMAIN):
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: Optional[ConfigType] = None
) -> Dict[str, Any]:
self, user_input: ConfigType | None = None
) -> dict[str, Any]:
"""Confirm reauth dialog."""
if user_input is None:
return self.async_show_form(
@ -98,8 +98,8 @@ class SonarrConfigFlow(ConfigFlow, domain=DOMAIN):
return await self.async_step_user()
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."""
errors = {}
@ -138,7 +138,7 @@ class SonarrConfigFlow(ConfigFlow, domain=DOMAIN):
async def _async_reauth_update_entry(
self, entry_id: str, data: dict
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Update existing config entry."""
entry = self.hass.config_entries.async_get_entry(entry_id)
self.hass.config_entries.async_update_entry(entry, data=data)
@ -146,7 +146,7 @@ class SonarrConfigFlow(ConfigFlow, domain=DOMAIN):
return self.async_abort(reason="reauth_successful")
def _get_user_data_schema(self) -> Dict[str, Any]:
def _get_user_data_schema(self) -> dict[str, Any]:
"""Get the data schema to display user form."""
if self._reauth:
return {vol.Required(CONF_API_KEY): str}
@ -174,7 +174,7 @@ class SonarrOptionsFlowHandler(OptionsFlow):
"""Initialize options flow."""
self.config_entry = config_entry
async def async_step_init(self, user_input: Optional[ConfigType] = None):
async def async_step_init(self, user_input: ConfigType | None = None):
"""Manage Sonarr options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)

View file

@ -1,7 +1,9 @@
"""Support for Sonarr sensors."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any, Callable, Dict, List, Optional
from typing import Any, Callable
from sonarr import Sonarr, SonarrConnectionError, SonarrError
@ -20,7 +22,7 @@ _LOGGER = logging.getLogger(__name__)
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 Sonarr sensors based on a config entry."""
options = entry.options
@ -75,7 +77,7 @@ class SonarrSensor(SonarrEntity):
icon: str,
key: str,
name: str,
unit_of_measurement: Optional[str] = None,
unit_of_measurement: str | None = None,
) -> None:
"""Initialize Sonarr sensor."""
self._unit_of_measurement = unit_of_measurement
@ -131,7 +133,7 @@ class SonarrCommandsSensor(SonarrSensor):
self._commands = await self.sonarr.commands()
@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."""
attrs = {}
@ -172,7 +174,7 @@ class SonarrDiskspaceSensor(SonarrSensor):
self._total_free = sum([disk.free for disk in self._disks])
@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."""
attrs = {}
@ -217,7 +219,7 @@ class SonarrQueueSensor(SonarrSensor):
self._queue = await self.sonarr.queue()
@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."""
attrs = {}
@ -258,7 +260,7 @@ class SonarrSeriesSensor(SonarrSensor):
self._items = await self.sonarr.series()
@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."""
attrs = {}
@ -301,7 +303,7 @@ class SonarrUpcomingSensor(SonarrSensor):
)
@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."""
attrs = {}
@ -323,7 +325,7 @@ class SonarrWantedSensor(SonarrSensor):
"""Initialize Sonarr Wanted sensor."""
self._max_items = max_items
self._results = None
self._total: Optional[int] = None
self._total: int | None = None
super().__init__(
sonarr=sonarr,
@ -342,7 +344,7 @@ class SonarrWantedSensor(SonarrSensor):
self._total = self._results.total
@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."""
attrs = {}
@ -354,6 +356,6 @@ class SonarrWantedSensor(SonarrSensor):
return attrs
@property
def state(self) -> Optional[int]:
def state(self) -> int | None:
"""Return the state of the sensor."""
return self._total

View file

@ -1,6 +1,7 @@
"""Config flow to configure songpal component."""
from __future__ import annotations
import logging
from typing import Optional
from urllib.parse import urlparse
from songpal import Device, SongpalException
@ -34,7 +35,7 @@ class SongpalConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
def __init__(self):
"""Initialize the flow."""
self.conf: Optional[SongpalConfig] = None
self.conf: SongpalConfig | None = None
async def async_step_user(self, user_input=None):
"""Handle a flow initiated by the user."""

View file

@ -1,6 +1,8 @@
"""Config flow for Spotify."""
from __future__ import annotations
import logging
from typing import Any, Dict, Optional
from typing import Any
from spotipy import Spotify
import voluptuous as vol
@ -24,7 +26,7 @@ class SpotifyFlowHandler(
def __init__(self) -> None:
"""Instantiate config flow."""
super().__init__()
self.entry: Optional[Dict[str, Any]] = None
self.entry: dict[str, Any] | None = None
@property
def logger(self) -> logging.Logger:
@ -32,11 +34,11 @@ class SpotifyFlowHandler(
return logging.getLogger(__name__)
@property
def extra_authorize_data(self) -> Dict[str, Any]:
def extra_authorize_data(self) -> dict[str, Any]:
"""Extra data that needs to be appended to the authorize url."""
return {"scope": ",".join(SPOTIFY_SCOPES)}
async def async_oauth_create_entry(self, data: Dict[str, Any]) -> Dict[str, Any]:
async def async_oauth_create_entry(self, data: dict[str, Any]) -> dict[str, Any]:
"""Create an entry for Spotify."""
spotify = Spotify(auth=data["token"]["access_token"])
@ -58,7 +60,7 @@ class SpotifyFlowHandler(
return self.async_create_entry(title=name, data=data)
async def async_step_reauth(self, entry: Dict[str, Any]) -> Dict[str, Any]:
async def async_step_reauth(self, entry: dict[str, Any]) -> dict[str, Any]:
"""Perform reauth upon migration of old entries."""
if entry:
self.entry = entry
@ -73,8 +75,8 @@ class SpotifyFlowHandler(
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, user_input: dict[str, Any] | None = None
) -> dict[str, Any]:
"""Confirm reauth dialog."""
if user_input is None:
return self.async_show_form(

View file

@ -1,9 +1,11 @@
"""Support for interacting with Spotify Connect."""
from __future__ import annotations
from asyncio import run_coroutine_threadsafe
import datetime as dt
from datetime import timedelta
import logging
from typing import Any, Callable, Dict, List, Optional
from typing import Any, Callable
import requests
from spotipy import Spotify, SpotifyException
@ -185,7 +187,7 @@ class UnknownMediaType(BrowseError):
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: Callable[[List[Entity], bool], None],
async_add_entities: Callable[[list[Entity], bool], None],
) -> None:
"""Set up Spotify based on a config entry."""
spotify = SpotifyMediaPlayer(
@ -237,9 +239,9 @@ class SpotifyMediaPlayer(MediaPlayerEntity):
SPOTIFY_SCOPES
)
self._currently_playing: Optional[dict] = {}
self._devices: Optional[List[dict]] = []
self._playlist: Optional[dict] = None
self._currently_playing: dict | None = {}
self._devices: list[dict] | None = []
self._playlist: dict | None = None
self._spotify: Spotify = None
self.player_available = False
@ -265,7 +267,7 @@ class SpotifyMediaPlayer(MediaPlayerEntity):
return self._id
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> dict[str, Any]:
"""Return device information about this entity."""
if self._me is not None:
model = self._me["product"]
@ -278,7 +280,7 @@ class SpotifyMediaPlayer(MediaPlayerEntity):
}
@property
def state(self) -> Optional[str]:
def state(self) -> str | None:
"""Return the playback state."""
if not self._currently_playing:
return STATE_IDLE
@ -287,44 +289,44 @@ class SpotifyMediaPlayer(MediaPlayerEntity):
return STATE_PAUSED
@property
def volume_level(self) -> Optional[float]:
def volume_level(self) -> float | None:
"""Return the device volume."""
return self._currently_playing.get("device", {}).get("volume_percent", 0) / 100
@property
def media_content_id(self) -> Optional[str]:
def media_content_id(self) -> str | None:
"""Return the media URL."""
item = self._currently_playing.get("item") or {}
return item.get("uri")
@property
def media_content_type(self) -> Optional[str]:
def media_content_type(self) -> str | None:
"""Return the media type."""
return MEDIA_TYPE_MUSIC
@property
def media_duration(self) -> Optional[int]:
def media_duration(self) -> int | None:
"""Duration of current playing media in seconds."""
if self._currently_playing.get("item") is None:
return None
return self._currently_playing["item"]["duration_ms"] / 1000
@property
def media_position(self) -> Optional[str]:
def media_position(self) -> str | None:
"""Position of current playing media in seconds."""
if not self._currently_playing:
return None
return self._currently_playing["progress_ms"] / 1000
@property
def media_position_updated_at(self) -> Optional[dt.datetime]:
def media_position_updated_at(self) -> dt.datetime | None:
"""When was the position of the current playing media valid."""
if not self._currently_playing:
return None
return utc_from_timestamp(self._currently_playing["timestamp"] / 1000)
@property
def media_image_url(self) -> Optional[str]:
def media_image_url(self) -> str | None:
"""Return the media image URL."""
if (
self._currently_playing.get("item") is None
@ -339,13 +341,13 @@ class SpotifyMediaPlayer(MediaPlayerEntity):
return False
@property
def media_title(self) -> Optional[str]:
def media_title(self) -> str | None:
"""Return the media title."""
item = self._currently_playing.get("item") or {}
return item.get("name")
@property
def media_artist(self) -> Optional[str]:
def media_artist(self) -> str | None:
"""Return the media artist."""
if self._currently_playing.get("item") is None:
return None
@ -354,14 +356,14 @@ class SpotifyMediaPlayer(MediaPlayerEntity):
)
@property
def media_album_name(self) -> Optional[str]:
def media_album_name(self) -> str | None:
"""Return the media album."""
if self._currently_playing.get("item") is None:
return None
return self._currently_playing["item"]["album"]["name"]
@property
def media_track(self) -> Optional[int]:
def media_track(self) -> int | None:
"""Track number of current playing media, music track only."""
item = self._currently_playing.get("item") or {}
return item.get("track_number")
@ -374,12 +376,12 @@ class SpotifyMediaPlayer(MediaPlayerEntity):
return self._playlist["name"]
@property
def source(self) -> Optional[str]:
def source(self) -> str | None:
"""Return the current playback device."""
return self._currently_playing.get("device", {}).get("name")
@property
def source_list(self) -> Optional[List[str]]:
def source_list(self) -> list[str] | None:
"""Return a list of source devices."""
if not self._devices:
return None
@ -391,7 +393,7 @@ class SpotifyMediaPlayer(MediaPlayerEntity):
return bool(self._currently_playing.get("shuffle_state"))
@property
def repeat(self) -> Optional[str]:
def repeat(self) -> str | None:
"""Return current repeat mode."""
repeat_state = self._currently_playing.get("repeat_state")
return REPEAT_MODE_MAPPING_TO_HA.get(repeat_state)

View file

@ -1,6 +1,8 @@
"""StarLine Account."""
from __future__ import annotations
from datetime import datetime, timedelta
from typing import Any, Callable, Dict, Optional
from typing import Any, Callable
from starline import StarlineApi, StarlineDevice
@ -29,8 +31,8 @@ class StarlineAccount:
self._config_entry: ConfigEntry = config_entry
self._update_interval: int = DEFAULT_SCAN_INTERVAL
self._update_obd_interval: int = DEFAULT_SCAN_OBD_INTERVAL
self._unsubscribe_auto_updater: Optional[Callable] = None
self._unsubscribe_auto_obd_updater: Optional[Callable] = None
self._unsubscribe_auto_updater: Callable | None = None
self._unsubscribe_auto_obd_updater: Callable | None = None
self._api: StarlineApi = StarlineApi(
config_entry.data[DATA_USER_ID], config_entry.data[DATA_SLNET_TOKEN]
)
@ -123,7 +125,7 @@ class StarlineAccount:
self._unsubscribe_auto_obd_updater = None
@staticmethod
def device_info(device: StarlineDevice) -> Dict[str, Any]:
def device_info(device: StarlineDevice) -> dict[str, Any]:
"""Device information for entities."""
return {
"identifiers": {(DOMAIN, device.device_id)},
@ -134,7 +136,7 @@ class StarlineAccount:
}
@staticmethod
def gps_attrs(device: StarlineDevice) -> Dict[str, Any]:
def gps_attrs(device: StarlineDevice) -> dict[str, Any]:
"""Attributes for device tracker."""
return {
"updated": datetime.utcfromtimestamp(device.position["ts"]).isoformat(),
@ -142,7 +144,7 @@ class StarlineAccount:
}
@staticmethod
def balance_attrs(device: StarlineDevice) -> Dict[str, Any]:
def balance_attrs(device: StarlineDevice) -> dict[str, Any]:
"""Attributes for balance sensor."""
return {
"operator": device.balance.get("operator"),
@ -151,7 +153,7 @@ class StarlineAccount:
}
@staticmethod
def gsm_attrs(device: StarlineDevice) -> Dict[str, Any]:
def gsm_attrs(device: StarlineDevice) -> dict[str, Any]:
"""Attributes for GSM sensor."""
return {
"raw": device.gsm_level,
@ -161,7 +163,7 @@ class StarlineAccount:
}
@staticmethod
def engine_attrs(device: StarlineDevice) -> Dict[str, Any]:
def engine_attrs(device: StarlineDevice) -> dict[str, Any]:
"""Attributes for engine switch."""
return {
"autostart": device.car_state.get("r_start"),
@ -169,6 +171,6 @@ class StarlineAccount:
}
@staticmethod
def errors_attrs(device: StarlineDevice) -> Dict[str, Any]:
def errors_attrs(device: StarlineDevice) -> dict[str, Any]:
"""Attributes for errors sensor."""
return {"errors": device.errors.get("errors")}

View file

@ -1,5 +1,5 @@
"""Config flow to configure StarLine component."""
from typing import Optional
from __future__ import annotations
from starline import StarlineAuth
import voluptuous as vol
@ -32,11 +32,11 @@ class StarlineFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
def __init__(self):
"""Initialize flow."""
self._app_id: Optional[str] = None
self._app_secret: Optional[str] = None
self._username: Optional[str] = None
self._password: Optional[str] = None
self._mfa_code: Optional[str] = None
self._app_id: str | None = None
self._app_secret: str | None = None
self._username: str | None = None
self._password: str | None = None
self._mfa_code: str | None = None
self._app_code = None
self._app_token = None

View file

@ -1,5 +1,7 @@
"""StarLine base entity."""
from typing import Callable, Optional
from __future__ import annotations
from typing import Callable
from homeassistant.helpers.entity import Entity
@ -17,7 +19,7 @@ class StarlineEntity(Entity):
self._device = device
self._key = key
self._name = name
self._unsubscribe_api: Optional[Callable] = None
self._unsubscribe_api: Callable | None = None
@property
def should_poll(self):

View file

@ -1,8 +1,10 @@
"""Provides core stream functionality."""
from __future__ import annotations
import asyncio
from collections import deque
import io
from typing import Any, Callable, List
from typing import Any, Callable
from aiohttp import web
import attr
@ -104,7 +106,7 @@ class StreamOutput:
return self._idle_timer.idle
@property
def segments(self) -> List[int]:
def segments(self) -> list[int]:
"""Return current sequence from segments."""
return [s.sequence for s in self._segments]

View file

@ -1,8 +1,10 @@
"""Provide functionality to record stream."""
from __future__ import annotations
import logging
import os
import threading
from typing import Deque, List
from typing import Deque
import av
@ -115,7 +117,7 @@ class RecorderOutput(StreamOutput):
"""Return provider name."""
return "recorder"
def prepend(self, segments: List[Segment]) -> None:
def prepend(self, segments: list[Segment]) -> None:
"""Prepend segments to existing list."""
self._segments.extendleft(reversed(segments))

View file

@ -1,8 +1,9 @@
"""Provide functionality to STT."""
from __future__ import annotations
from abc import ABC, abstractmethod
import asyncio
import logging
from typing import Dict, List, Optional
from aiohttp import StreamReader, web
from aiohttp.hdrs import istr
@ -96,44 +97,44 @@ class SpeechMetadata:
class SpeechResult:
"""Result of audio Speech."""
text: Optional[str] = attr.ib()
text: str | None = attr.ib()
result: SpeechResultState = attr.ib()
class Provider(ABC):
"""Represent a single STT provider."""
hass: Optional[HomeAssistantType] = None
name: Optional[str] = None
hass: HomeAssistantType | None = None
name: str | None = None
@property
@abstractmethod
def supported_languages(self) -> List[str]:
def supported_languages(self) -> list[str]:
"""Return a list of supported languages."""
@property
@abstractmethod
def supported_formats(self) -> List[AudioFormats]:
def supported_formats(self) -> list[AudioFormats]:
"""Return a list of supported formats."""
@property
@abstractmethod
def supported_codecs(self) -> List[AudioCodecs]:
def supported_codecs(self) -> list[AudioCodecs]:
"""Return a list of supported codecs."""
@property
@abstractmethod
def supported_bit_rates(self) -> List[AudioBitRates]:
def supported_bit_rates(self) -> list[AudioBitRates]:
"""Return a list of supported bit rates."""
@property
@abstractmethod
def supported_sample_rates(self) -> List[AudioSampleRates]:
def supported_sample_rates(self) -> list[AudioSampleRates]:
"""Return a list of supported sample rates."""
@property
@abstractmethod
def supported_channels(self) -> List[AudioChannels]:
def supported_channels(self) -> list[AudioChannels]:
"""Return a list of supported channels."""
@abstractmethod
@ -167,12 +168,12 @@ class SpeechToTextView(HomeAssistantView):
url = "/api/stt/{provider}"
name = "api:stt:provider"
def __init__(self, providers: Dict[str, Provider]) -> None:
def __init__(self, providers: dict[str, Provider]) -> None:
"""Initialize a tts view."""
self.providers = providers
@staticmethod
def _metadata_from_header(request: web.Request) -> Optional[SpeechMetadata]:
def _metadata_from_header(request: web.Request) -> SpeechMetadata | None:
"""Extract metadata from header.
X-Speech-Content: format=wav; codec=pcm; sample_rate=16000; bit_rate=16; channel=1; language=de_de

View file

@ -1,7 +1,8 @@
"""Support for Supla devices."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Optional
import async_timeout
from asyncpysupla import SuplaAPI
@ -180,7 +181,7 @@ class SuplaChannel(CoordinatorEntity):
)
@property
def name(self) -> Optional[str]:
def name(self) -> str | None:
"""Return the name of the device."""
return self.channel_data["caption"]

View file

@ -1,6 +1,8 @@
"""Support for Sure Petcare cat/pet flaps."""
from __future__ import annotations
import logging
from typing import Any, Dict, List
from typing import Any
from surepy import (
MESTART_RESOURCE,
@ -185,12 +187,12 @@ async def async_setup(hass, config) -> bool:
class SurePetcareAPI:
"""Define a generic Sure Petcare object."""
def __init__(self, hass, surepy: SurePetcare, ids: List[Dict[str, Any]]) -> None:
def __init__(self, hass, surepy: SurePetcare, ids: list[dict[str, Any]]) -> None:
"""Initialize the Sure Petcare object."""
self.hass = hass
self.surepy = surepy
self.ids = ids
self.states: Dict[str, Any] = {}
self.states: dict[str, Any] = {}
async def async_update(self, arg: Any = None) -> None:
"""Refresh Sure Petcare data."""

View file

@ -1,7 +1,9 @@
"""Support for Sure PetCare Flaps/Pets binary sensors."""
from __future__ import annotations
from datetime import datetime
import logging
from typing import Any, Dict, Optional
from typing import Any
from surepy import SureLocationID, SurepyProduct
@ -71,8 +73,8 @@ class SurePetcareBinarySensor(BinarySensorEntity):
self._device_class = device_class
self._spc: SurePetcareAPI = spc
self._spc_data: Dict[str, Any] = self._spc.states[self._sure_type].get(self._id)
self._state: Dict[str, Any] = {}
self._spc_data: dict[str, Any] = self._spc.states[self._sure_type].get(self._id)
self._state: dict[str, Any] = {}
# cover special case where a device has no name set
if "name" in self._spc_data:
@ -85,7 +87,7 @@ class SurePetcareBinarySensor(BinarySensorEntity):
self._async_unsub_dispatcher_connect = None
@property
def is_on(self) -> Optional[bool]:
def is_on(self) -> bool | None:
"""Return true if entity is on/unlocked."""
return bool(self._state)
@ -151,7 +153,7 @@ class Hub(SurePetcareBinarySensor):
return self.available
@property
def extra_state_attributes(self) -> Optional[Dict[str, Any]]:
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return the state attributes of the device."""
attributes = None
if self._state:
@ -179,7 +181,7 @@ class Pet(SurePetcareBinarySensor):
return False
@property
def extra_state_attributes(self) -> Optional[Dict[str, Any]]:
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return the state attributes of the device."""
attributes = None
if self._state:
@ -232,7 +234,7 @@ class DeviceConnectivity(SurePetcareBinarySensor):
return self.available
@property
def extra_state_attributes(self) -> Optional[Dict[str, Any]]:
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return the state attributes of the device."""
attributes = None
if self._state:

View file

@ -1,6 +1,8 @@
"""Support for Sure PetCare Flaps/Pets sensors."""
from __future__ import annotations
import logging
from typing import Any, Dict, Optional
from typing import Any
from surepy import SureLockStateID, SurepyProduct
@ -62,8 +64,8 @@ class SurePetcareSensor(Entity):
self._sure_type = sure_type
self._spc = spc
self._spc_data: Dict[str, Any] = self._spc.states[self._sure_type].get(self._id)
self._state: Dict[str, Any] = {}
self._spc_data: dict[str, Any] = self._spc.states[self._sure_type].get(self._id)
self._state: dict[str, Any] = {}
self._name = (
f"{self._sure_type.name.capitalize()} "
@ -120,12 +122,12 @@ class Flap(SurePetcareSensor):
"""Sure Petcare Flap."""
@property
def state(self) -> Optional[int]:
def state(self) -> int | None:
"""Return battery level in percent."""
return SureLockStateID(self._state["locking"]["mode"]).name.capitalize()
@property
def extra_state_attributes(self) -> Optional[Dict[str, Any]]:
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return the state attributes of the device."""
attributes = None
if self._state:
@ -143,9 +145,9 @@ class SureBattery(SurePetcareSensor):
return f"{self._name} Battery Level"
@property
def state(self) -> Optional[int]:
def state(self) -> int | None:
"""Return battery level in percent."""
battery_percent: Optional[int]
battery_percent: int | None
try:
per_battery_voltage = self._state["battery"] / 4
voltage_diff = per_battery_voltage - SURE_BATT_VOLTAGE_LOW
@ -166,7 +168,7 @@ class SureBattery(SurePetcareSensor):
return DEVICE_CLASS_BATTERY
@property
def extra_state_attributes(self) -> Optional[Dict[str, Any]]:
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return state attributes."""
attributes = None
if self._state:

View file

@ -1,5 +1,5 @@
"""Provides device actions for switches."""
from typing import List
from __future__ import annotations
import voluptuous as vol
@ -25,6 +25,6 @@ async def async_call_action_from_config(
)
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."""
return await toggle_entity.async_get_actions(hass, device_id, DOMAIN)

View file

@ -1,5 +1,5 @@
"""Provides device conditions for switches."""
from typing import Dict, List
from __future__ import annotations
import voluptuous as vol
@ -28,7 +28,7 @@ def async_condition_from_config(
async def async_get_conditions(
hass: HomeAssistant, device_id: str
) -> List[Dict[str, str]]:
) -> list[dict[str, str]]:
"""List device conditions."""
return await toggle_entity.async_get_conditions(hass, device_id, DOMAIN)

View file

@ -1,5 +1,5 @@
"""Provides device triggers for switches."""
from typing import List
from __future__ import annotations
import voluptuous as vol
@ -28,7 +28,7 @@ async def async_attach_trigger(
)
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."""
return await toggle_entity.async_get_triggers(hass, device_id, DOMAIN)

View file

@ -1,5 +1,7 @@
"""Light support for switch entities."""
from typing import Any, Callable, Optional, Sequence, cast
from __future__ import annotations
from typing import Any, Callable, Sequence, cast
import voluptuous as vol
@ -38,7 +40,7 @@ async def async_setup_platform(
hass: HomeAssistantType,
config: ConfigType,
async_add_entities: Callable[[Sequence[Entity]], None],
discovery_info: Optional[DiscoveryInfoType] = None,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Initialize Light Switch platform."""
@ -65,7 +67,7 @@ class LightSwitch(LightEntity):
self._name = name
self._switch_entity_id = switch_entity_id
self._unique_id = unique_id
self._switch_state: Optional[State] = None
self._switch_state: State | None = None
@property
def name(self) -> str:

View file

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

View file

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

View file

@ -1,5 +1,7 @@
"""Support for Switchbot."""
from typing import Any, Dict
from __future__ import annotations
from typing import Any
# pylint: disable=import-error
import switchbot
@ -86,6 +88,6 @@ class SwitchBot(SwitchEntity, RestoreEntity):
return self._name
@property
def extra_state_attributes(self) -> Dict[str, Any]:
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes."""
return {"last_run_success": self._last_run_success}

View file

@ -1,8 +1,9 @@
"""Home Assistant Switcher Component."""
from __future__ import annotations
from asyncio import QueueEmpty, TimeoutError as Asyncio_TimeoutError, wait_for
from datetime import datetime, timedelta
import logging
from typing import Dict, Optional
from aioswitcher.bridge import SwitcherV2Bridge
import voluptuous as vol
@ -45,7 +46,7 @@ CONFIG_SCHEMA = vol.Schema(
)
async def async_setup(hass: HomeAssistantType, config: Dict) -> bool:
async def async_setup(hass: HomeAssistantType, config: dict) -> bool:
"""Set up the switcher component."""
phone_id = config[DOMAIN][CONF_PHONE_ID]
device_id = config[DOMAIN][CONF_DEVICE_ID]
@ -72,7 +73,7 @@ async def async_setup(hass: HomeAssistantType, config: Dict) -> bool:
hass.async_create_task(async_load_platform(hass, SWITCH_DOMAIN, DOMAIN, {}, config))
@callback
def device_updates(timestamp: Optional[datetime]) -> None:
def device_updates(timestamp: datetime | None) -> None:
"""Use for updating the device data from the queue."""
if v2bridge.running:
try:

View file

@ -1,5 +1,7 @@
"""Home Assistant Switcher Component Switch platform."""
from typing import Callable, Dict
from __future__ import annotations
from typing import Callable
from aioswitcher.api import SwitcherV2Api
from aioswitcher.api.messages import SwitcherV2ControlResponseMSG
@ -52,9 +54,9 @@ SERVICE_TURN_ON_WITH_TIMER_SCHEMA = {
async def async_setup_platform(
hass: HomeAssistantType,
config: Dict,
config: dict,
async_add_entities: Callable,
discovery_info: Dict,
discovery_info: dict,
) -> None:
"""Set up the switcher platform for the switch component."""
if discovery_info is None:
@ -139,7 +141,7 @@ class SwitcherControl(SwitchEntity):
return self._device_data.power_consumption
@property
def extra_state_attributes(self) -> Dict:
def extra_state_attributes(self) -> dict:
"""Return the optional state attributes."""
attribs = {}
@ -173,11 +175,11 @@ class SwitcherControl(SwitchEntity):
self._state = self._device_data.state
self.async_write_ha_state()
async def async_turn_on(self, **kwargs: Dict) -> None:
async def async_turn_on(self, **kwargs: dict) -> None:
"""Turn the entity on."""
await self._control_device(True)
async def async_turn_off(self, **kwargs: Dict) -> None:
async def async_turn_off(self, **kwargs: dict) -> None:
"""Turn the entity off."""
await self._control_device(False)

View file

@ -1,7 +1,7 @@
"""The syncthru component."""
from __future__ import annotations
import logging
from typing import Set, Tuple
from pysyncthru import SyncThru
@ -65,12 +65,12 @@ async def async_unload_entry(hass: HomeAssistantType, entry: ConfigEntry) -> boo
return True
def device_identifiers(printer: SyncThru) -> Set[Tuple[str, str]]:
def device_identifiers(printer: SyncThru) -> set[tuple[str, str]]:
"""Get device identifiers for device registry."""
return {(DOMAIN, printer.serial_number())}
def device_connections(printer: SyncThru) -> Set[Tuple[str, str]]:
def device_connections(printer: SyncThru) -> set[tuple[str, str]]:
"""Get device connections for device registry."""
connections = set()
try:

View file

@ -1,8 +1,9 @@
"""The Synology DSM component."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
from typing import Dict
import async_timeout
from synology_dsm import SynologyDSM
@ -599,7 +600,7 @@ class SynologyDSMBaseEntity(CoordinatorEntity):
self,
api: SynoApi,
entity_type: str,
entity_info: Dict[str, str],
entity_info: dict[str, str],
coordinator: DataUpdateCoordinator,
):
"""Initialize the Synology DSM entity."""
@ -643,12 +644,12 @@ class SynologyDSMBaseEntity(CoordinatorEntity):
return self._class
@property
def extra_state_attributes(self) -> Dict[str, any]:
def extra_state_attributes(self) -> dict[str, any]:
"""Return the state attributes."""
return {ATTR_ATTRIBUTION: ATTRIBUTION}
@property
def device_info(self) -> Dict[str, any]:
def device_info(self) -> dict[str, any]:
"""Return the device information."""
return {
"identifiers": {(DOMAIN, self._api.information.serial)},
@ -676,7 +677,7 @@ class SynologyDSMDeviceEntity(SynologyDSMBaseEntity):
self,
api: SynoApi,
entity_type: str,
entity_info: Dict[str, str],
entity_info: dict[str, str],
coordinator: DataUpdateCoordinator,
device_id: str = None,
):
@ -718,7 +719,7 @@ class SynologyDSMDeviceEntity(SynologyDSMBaseEntity):
return bool(self._api.storage)
@property
def device_info(self) -> Dict[str, any]:
def device_info(self) -> dict[str, any]:
"""Return the device information."""
return {
"identifiers": {(DOMAIN, self._api.information.serial, self._device_id)},

View file

@ -1,5 +1,5 @@
"""Support for Synology DSM binary sensors."""
from typing import Dict
from __future__ import annotations
from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.config_entries import ConfigEntry
@ -71,7 +71,7 @@ class SynoDSMSecurityBinarySensor(SynologyDSMBaseEntity, BinarySensorEntity):
return bool(self._api.security)
@property
def extra_state_attributes(self) -> Dict[str, str]:
def extra_state_attributes(self) -> dict[str, str]:
"""Return security checks details."""
return self._api.security.status_by_check

View file

@ -1,6 +1,7 @@
"""Support for Synology DSM cameras."""
from __future__ import annotations
import logging
from typing import Dict
from synology_dsm.api.surveillance_station import SynoSurveillanceStation
from synology_dsm.exceptions import (
@ -79,7 +80,7 @@ class SynoDSMCamera(SynologyDSMBaseEntity, Camera):
return self.coordinator.data["cameras"][self._camera_id]
@property
def device_info(self) -> Dict[str, any]:
def device_info(self) -> dict[str, any]:
"""Return the device information."""
return {
"identifiers": {

View file

@ -1,6 +1,7 @@
"""Support for Synology DSM sensors."""
from __future__ import annotations
from datetime import timedelta
from typing import Dict
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
@ -145,7 +146,7 @@ class SynoDSMInfoSensor(SynologyDSMBaseEntity):
self,
api: SynoApi,
entity_type: str,
entity_info: Dict[str, str],
entity_info: dict[str, str],
coordinator: DataUpdateCoordinator,
):
"""Initialize the Synology SynoDSMInfoSensor entity."""

View file

@ -1,6 +1,7 @@
"""Support for Synology DSM switch."""
from __future__ import annotations
import logging
from typing import Dict
from synology_dsm.api.surveillance_station import SynoSurveillanceStation
@ -49,7 +50,7 @@ class SynoDSMSurveillanceHomeModeToggle(SynologyDSMBaseEntity, ToggleEntity):
self,
api: SynoApi,
entity_type: str,
entity_info: Dict[str, str],
entity_info: dict[str, str],
version: str,
coordinator: DataUpdateCoordinator,
):
@ -95,7 +96,7 @@ class SynoDSMSurveillanceHomeModeToggle(SynologyDSMBaseEntity, ToggleEntity):
return bool(self._api.surveillance_station)
@property
def device_info(self) -> Dict[str, any]:
def device_info(self) -> dict[str, any]:
"""Return the device information."""
return {
"identifiers": {

View file

@ -1,9 +1,11 @@
"""Support for System health ."""
from __future__ import annotations
import asyncio
import dataclasses
from datetime import datetime
import logging
from typing import Awaitable, Callable, Dict, Optional
from typing import Awaitable, Callable
import aiohttp
import async_timeout
@ -27,7 +29,7 @@ INFO_CALLBACK_TIMEOUT = 5
def async_register_info(
hass: HomeAssistant,
domain: str,
info_callback: Callable[[HomeAssistant], Dict],
info_callback: Callable[[HomeAssistant], dict],
):
"""Register an info callback.
@ -89,10 +91,10 @@ def _format_value(val):
@websocket_api.async_response
@websocket_api.websocket_command({vol.Required("type"): "system_health/info"})
async def handle_info(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: Dict
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
):
"""Handle an info request via a subscription."""
registrations: Dict[str, SystemHealthRegistration] = hass.data[DOMAIN]
registrations: dict[str, SystemHealthRegistration] = hass.data[DOMAIN]
data = {}
pending_info = {}
@ -187,14 +189,14 @@ class SystemHealthRegistration:
hass: HomeAssistant
domain: str
info_callback: Optional[Callable[[HomeAssistant], Awaitable[Dict]]] = None
manage_url: Optional[str] = None
info_callback: Callable[[HomeAssistant], Awaitable[dict]] | None = None
manage_url: str | None = None
@callback
def async_register_info(
self,
info_callback: Callable[[HomeAssistant], Awaitable[Dict]],
manage_url: Optional[str] = None,
info_callback: Callable[[HomeAssistant], Awaitable[dict]],
manage_url: str | None = None,
):
"""Register an info callback."""
self.info_callback = info_callback
@ -203,7 +205,7 @@ class SystemHealthRegistration:
async def async_check_can_reach_url(
hass: HomeAssistant, url: str, more_info: Optional[str] = None
hass: HomeAssistant, url: str, more_info: str | None = None
) -> str:
"""Test if the url can be reached."""
session = aiohttp_client.async_get_clientsession(hass)