Bump python-typing-update to v0.3.3 (#48992)
* Bump python-typing-update to 0.3.3 * Changes after updatepull/49006/head
parent
a0a8638a2d
commit
1a38d2089d
|
@ -69,7 +69,7 @@ repos:
|
|||
- id: prettier
|
||||
stages: [manual]
|
||||
- repo: https://github.com/cdce8p/python-typing-update
|
||||
rev: v0.3.2
|
||||
rev: v0.3.3
|
||||
hooks:
|
||||
# Run `python-typing-update` hook manually from time to time
|
||||
# to update python typing syntax.
|
||||
|
|
|
@ -1,6 +1,8 @@
|
|||
"""Config flow to configure Denon AVR receivers using their HTTP interface."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any, Dict, Optional
|
||||
from typing import Any
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import denonavr
|
||||
|
@ -44,7 +46,7 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
|
|||
"""Init object."""
|
||||
self.config_entry = config_entry
|
||||
|
||||
async def async_step_init(self, user_input: Optional[Dict[str, Any]] = None):
|
||||
async def async_step_init(self, user_input: dict[str, Any] | None = None):
|
||||
"""Manage the options."""
|
||||
if user_input is not None:
|
||||
return self.async_create_entry(title="", data=user_input)
|
||||
|
@ -96,7 +98,7 @@ class DenonAvrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
|||
"""Get the options flow."""
|
||||
return OptionsFlowHandler(config_entry)
|
||||
|
||||
async def async_step_user(self, user_input: Optional[Dict[str, Any]] = None):
|
||||
async def async_step_user(self, user_input: dict[str, Any] | None = None):
|
||||
"""Handle a flow initialized by the user."""
|
||||
errors = {}
|
||||
if user_input is not None:
|
||||
|
@ -123,8 +125,8 @@ class DenonAvrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
|||
)
|
||||
|
||||
async def async_step_select(
|
||||
self, user_input: Optional[Dict[str, Any]] = None
|
||||
) -> Dict[str, Any]:
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> dict[str, Any]:
|
||||
"""Handle multiple receivers found."""
|
||||
errors = {}
|
||||
if user_input is not None:
|
||||
|
@ -144,8 +146,8 @@ class DenonAvrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
|||
)
|
||||
|
||||
async def async_step_confirm(
|
||||
self, user_input: Optional[Dict[str, Any]] = None
|
||||
) -> Dict[str, Any]:
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> dict[str, Any]:
|
||||
"""Allow the user to confirm adding the device."""
|
||||
if user_input is not None:
|
||||
return await self.async_step_connect()
|
||||
|
@ -154,8 +156,8 @@ class DenonAvrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
|||
return self.async_show_form(step_id="confirm")
|
||||
|
||||
async def async_step_connect(
|
||||
self, user_input: Optional[Dict[str, Any]] = None
|
||||
) -> Dict[str, Any]:
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> dict[str, Any]:
|
||||
"""Connect to the receiver."""
|
||||
connect_denonavr = ConnectDenonAVR(
|
||||
self.host,
|
||||
|
@ -204,7 +206,7 @@ class DenonAvrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
|||
},
|
||||
)
|
||||
|
||||
async def async_step_ssdp(self, discovery_info: Dict[str, Any]) -> Dict[str, Any]:
|
||||
async def async_step_ssdp(self, discovery_info: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Handle a discovered Denon AVR.
|
||||
|
||||
This flow is triggered by the SSDP component. It will check if the
|
||||
|
|
|
@ -1,6 +1,8 @@
|
|||
"""Code to handle a DenonAVR receiver."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Callable, Optional
|
||||
from typing import Callable
|
||||
|
||||
from denonavr import DenonAVR
|
||||
|
||||
|
@ -18,7 +20,7 @@ class ConnectDenonAVR:
|
|||
zone2: bool,
|
||||
zone3: bool,
|
||||
async_client_getter: Callable,
|
||||
entry_state: Optional[str] = None,
|
||||
entry_state: str | None = None,
|
||||
):
|
||||
"""Initialize the class."""
|
||||
self._async_client_getter = async_client_getter
|
||||
|
@ -35,7 +37,7 @@ class ConnectDenonAVR:
|
|||
self._zones["Zone3"] = None
|
||||
|
||||
@property
|
||||
def receiver(self) -> Optional[DenonAVR]:
|
||||
def receiver(self) -> DenonAVR | None:
|
||||
"""Return the class containing all connections to the receiver."""
|
||||
return self._receiver
|
||||
|
||||
|
|
|
@ -1,9 +1,10 @@
|
|||
"""Code to handle the Plenticore API."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from collections import defaultdict
|
||||
from datetime import datetime, timedelta
|
||||
import logging
|
||||
from typing import Dict, Union
|
||||
|
||||
from aiohttp.client_exceptions import ClientError
|
||||
from kostal.plenticore import PlenticoreApiClient, PlenticoreAuthenticationException
|
||||
|
@ -151,7 +152,7 @@ class PlenticoreUpdateCoordinator(DataUpdateCoordinator):
|
|||
class ProcessDataUpdateCoordinator(PlenticoreUpdateCoordinator):
|
||||
"""Implementation of PlenticoreUpdateCoordinator for process data."""
|
||||
|
||||
async def _async_update_data(self) -> Dict[str, Dict[str, str]]:
|
||||
async def _async_update_data(self) -> dict[str, dict[str, str]]:
|
||||
client = self._plenticore.client
|
||||
|
||||
if not self._fetch or client is None:
|
||||
|
@ -172,7 +173,7 @@ class ProcessDataUpdateCoordinator(PlenticoreUpdateCoordinator):
|
|||
class SettingDataUpdateCoordinator(PlenticoreUpdateCoordinator):
|
||||
"""Implementation of PlenticoreUpdateCoordinator for settings data."""
|
||||
|
||||
async def _async_update_data(self) -> Dict[str, Dict[str, str]]:
|
||||
async def _async_update_data(self) -> dict[str, dict[str, str]]:
|
||||
client = self._plenticore.client
|
||||
|
||||
if not self._fetch or client is None:
|
||||
|
@ -223,7 +224,7 @@ class PlenticoreDataFormatter:
|
|||
return getattr(cls, name)
|
||||
|
||||
@staticmethod
|
||||
def format_round(state: str) -> Union[int, str]:
|
||||
def format_round(state: str) -> int | str:
|
||||
"""Return the given state value as rounded integer."""
|
||||
try:
|
||||
return round(float(state))
|
||||
|
@ -231,7 +232,7 @@ class PlenticoreDataFormatter:
|
|||
return state
|
||||
|
||||
@staticmethod
|
||||
def format_energy(state: str) -> Union[float, str]:
|
||||
def format_energy(state: str) -> float | str:
|
||||
"""Return the given state value as energy value, scaled to kWh."""
|
||||
try:
|
||||
return round(float(state) / 1000, 1)
|
||||
|
|
|
@ -1,7 +1,9 @@
|
|||
"""Platform for Kostal Plenticore sensors."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from typing import Any, Callable, Dict, Optional
|
||||
from typing import Any, Callable
|
||||
|
||||
from homeassistant.components.sensor import SensorEntity
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
|
@ -109,9 +111,9 @@ class PlenticoreDataSensor(CoordinatorEntity, SensorEntity):
|
|||
module_id: str,
|
||||
data_id: str,
|
||||
sensor_name: str,
|
||||
sensor_data: Dict[str, Any],
|
||||
sensor_data: dict[str, Any],
|
||||
formatter: Callable[[str], Any],
|
||||
device_info: Dict[str, Any],
|
||||
device_info: dict[str, Any],
|
||||
):
|
||||
"""Create a new Sensor Entity for Plenticore process data."""
|
||||
super().__init__(coordinator)
|
||||
|
@ -147,7 +149,7 @@ class PlenticoreDataSensor(CoordinatorEntity, SensorEntity):
|
|||
await super().async_will_remove_from_hass()
|
||||
|
||||
@property
|
||||
def device_info(self) -> Dict[str, Any]:
|
||||
def device_info(self) -> dict[str, Any]:
|
||||
"""Return the device info."""
|
||||
return self._device_info
|
||||
|
||||
|
@ -162,17 +164,17 @@ class PlenticoreDataSensor(CoordinatorEntity, SensorEntity):
|
|||
return f"{self.platform_name} {self._sensor_name}"
|
||||
|
||||
@property
|
||||
def unit_of_measurement(self) -> Optional[str]:
|
||||
def unit_of_measurement(self) -> str | None:
|
||||
"""Return the unit of this Sensor Entity or None."""
|
||||
return self._sensor_data.get(ATTR_UNIT_OF_MEASUREMENT)
|
||||
|
||||
@property
|
||||
def icon(self) -> Optional[str]:
|
||||
def icon(self) -> str | None:
|
||||
"""Return the icon name of this Sensor Entity or None."""
|
||||
return self._sensor_data.get(ATTR_ICON)
|
||||
|
||||
@property
|
||||
def device_class(self) -> Optional[str]:
|
||||
def device_class(self) -> str | None:
|
||||
"""Return the class of this device, from component DEVICE_CLASSES."""
|
||||
return self._sensor_data.get(ATTR_DEVICE_CLASS)
|
||||
|
||||
|
@ -182,7 +184,7 @@ class PlenticoreDataSensor(CoordinatorEntity, SensorEntity):
|
|||
return self._sensor_data.get(ATTR_ENABLED_DEFAULT, False)
|
||||
|
||||
@property
|
||||
def state(self) -> Optional[Any]:
|
||||
def state(self) -> Any | None:
|
||||
"""Return the state of the sensor."""
|
||||
if self.coordinator.data is None:
|
||||
# None is translated to STATE_UNKNOWN
|
||||
|
|
|
@ -1,5 +1,7 @@
|
|||
"""Support for Modbus."""
|
||||
from typing import Any, Union
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
|
@ -95,7 +97,7 @@ from .modbus import modbus_setup
|
|||
BASE_SCHEMA = vol.Schema({vol.Optional(CONF_NAME, default=DEFAULT_HUB): cv.string})
|
||||
|
||||
|
||||
def number(value: Any) -> Union[int, float]:
|
||||
def number(value: Any) -> int | float:
|
||||
"""Coerce a value to number without losing precision."""
|
||||
if isinstance(value, int):
|
||||
return value
|
||||
|
|
|
@ -1,6 +1,6 @@
|
|||
"""Axis conftest."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Optional
|
||||
from unittest.mock import patch
|
||||
|
||||
from axis.rtsp import (
|
||||
|
@ -34,7 +34,7 @@ def mock_axis_rtspclient():
|
|||
|
||||
rtsp_client_mock.return_value.stop = stop_stream
|
||||
|
||||
def make_rtsp_call(data: Optional[dict] = None, state: str = ""):
|
||||
def make_rtsp_call(data: dict | None = None, state: str = ""):
|
||||
"""Generate a RTSP call."""
|
||||
axis_streammanager_session_callback = rtsp_client_mock.call_args[0][4]
|
||||
|
||||
|
|
|
@ -1,7 +1,9 @@
|
|||
"""Tests for Climacell weather entity."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
import logging
|
||||
from typing import Any, Dict
|
||||
from typing import Any
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
@ -58,7 +60,7 @@ async def _enable_entity(hass: HomeAssistantType, entity_name: str) -> None:
|
|||
assert updated_entry.disabled is False
|
||||
|
||||
|
||||
async def _setup(hass: HomeAssistantType, config: Dict[str, Any]) -> State:
|
||||
async def _setup(hass: HomeAssistantType, config: dict[str, Any]) -> State:
|
||||
"""Set up entry and return entity state."""
|
||||
with patch(
|
||||
"homeassistant.util.dt.utcnow",
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
"""Tests for 1-Wire integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, List, Tuple
|
||||
from typing import Any
|
||||
from unittest.mock import patch
|
||||
|
||||
from pyownet.protocol import ProtocolError
|
||||
|
@ -129,8 +130,8 @@ def setup_owproxy_mock_devices(owproxy, domain, device_ids) -> None:
|
|||
|
||||
|
||||
def setup_sysbus_mock_devices(
|
||||
domain: str, device_ids: List[str]
|
||||
) -> Tuple[List[str], List[Any]]:
|
||||
domain: str, device_ids: list[str]
|
||||
) -> tuple[list[str], list[Any]]:
|
||||
"""Set up mock for sysbus."""
|
||||
glob_result = []
|
||||
read_side_effect = []
|
||||
|
|
Loading…
Reference in New Issue