1
0
mirror of https://github.com/home-assistant/core synced 2024-07-08 20:17:01 +00:00

Add type hints to yeelight (#69213)

This commit is contained in:
epenet 2022-04-04 10:58:44 +02:00 committed by GitHub
parent 8ee9695e3b
commit 9231819532
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 57 additions and 57 deletions

View File

@ -8,7 +8,7 @@ import voluptuous as vol
from yeelight import BulbException
from yeelight.aio import AsyncBulb
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry, ConfigEntryNotReady
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
CONF_DEVICES,
CONF_HOST,
@ -18,6 +18,7 @@ from homeassistant.const import (
EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType

View File

@ -253,25 +253,25 @@ def _async_cmd(func):
except asyncio.TimeoutError as ex:
# The wifi likely dropped, so we want to retry once since
# python-yeelight will auto reconnect
exc_message = str(ex) or type(ex)
if attempts == 0:
continue
raise HomeAssistantError(
f"Timed out when calling {func.__name__} for bulb {self.device.name} at {self.device.host}: {exc_message}"
f"Timed out when calling {func.__name__} for bulb "
f"{self.device.name} at {self.device.host}: {str(ex) or type(ex)}"
) from ex
except OSError as ex:
# A network error happened, the bulb is likely offline now
self.device.async_mark_unavailable()
self.async_state_changed()
exc_message = str(ex) or type(ex)
raise HomeAssistantError(
f"Error when calling {func.__name__} for bulb {self.device.name} at {self.device.host}: {exc_message}"
f"Error when calling {func.__name__} for bulb "
f"{self.device.name} at {self.device.host}: {str(ex) or type(ex)}"
) from ex
except BulbException as ex:
# The bulb likely responded but had an error
exc_message = str(ex) or type(ex)
raise HomeAssistantError(
f"Error when calling {func.__name__} for bulb {self.device.name} at {self.device.host}: {exc_message}"
f"Error when calling {func.__name__} for bulb "
f"{self.device.name} at {self.device.host}: {str(ex) or type(ex)}"
) from ex
return _async_wrap
@ -413,8 +413,8 @@ def _async_setup_services(hass: HomeAssistant):
class YeelightGenericLight(YeelightEntity, LightEntity):
"""Representation of a Yeelight generic light."""
_attr_color_mode = COLOR_MODE_BRIGHTNESS
_attr_supported_color_modes = {COLOR_MODE_BRIGHTNESS}
_attr_color_mode: str | None = COLOR_MODE_BRIGHTNESS
_attr_supported_color_modes: set[str] | None = {COLOR_MODE_BRIGHTNESS}
_attr_should_poll = False
def __init__(self, device, entry, custom_effects=None):
@ -522,7 +522,7 @@ class YeelightGenericLight(YeelightEntity, LightEntity):
return self._light_type
@property
def hs_color(self) -> tuple:
def hs_color(self) -> tuple[int, int] | None:
"""Return the color property."""
hue = self._get_property("hue")
sat = self._get_property("sat")
@ -532,7 +532,7 @@ class YeelightGenericLight(YeelightEntity, LightEntity):
return (int(hue), int(sat))
@property
def rgb_color(self) -> tuple:
def rgb_color(self) -> tuple[int, int, int] | None:
"""Return the color property."""
if (rgb := self._get_property("rgb")) is None:
return None
@ -637,7 +637,11 @@ class YeelightGenericLight(YeelightEntity, LightEntity):
@_async_cmd
async def async_set_hs(self, hs_color, duration) -> None:
"""Set bulb's color."""
if not hs_color or COLOR_MODE_HS not in self.supported_color_modes:
if (
not hs_color
or not self.supported_color_modes
or COLOR_MODE_HS not in self.supported_color_modes
):
return
if (
not self.device.is_color_flow_enabled
@ -658,7 +662,11 @@ class YeelightGenericLight(YeelightEntity, LightEntity):
@_async_cmd
async def async_set_rgb(self, rgb, duration) -> None:
"""Set bulb's color."""
if not rgb or COLOR_MODE_RGB not in self.supported_color_modes:
if (
not rgb
or not self.supported_color_modes
or COLOR_MODE_RGB not in self.supported_color_modes
):
return
if (
not self.device.is_color_flow_enabled
@ -679,7 +687,11 @@ class YeelightGenericLight(YeelightEntity, LightEntity):
@_async_cmd
async def async_set_colortemp(self, colortemp, duration) -> None:
"""Set bulb's color temperature."""
if not colortemp or COLOR_MODE_COLOR_TEMP not in self.supported_color_modes:
if (
not colortemp
or not self.supported_color_modes
or COLOR_MODE_COLOR_TEMP not in self.supported_color_modes
):
return
temp_in_k = mired_to_kelvin(colortemp)
@ -708,7 +720,7 @@ class YeelightGenericLight(YeelightEntity, LightEntity):
"""Activate flash."""
if not flash:
return
if int(self._get_property("color_mode")) != 1:
if int(self._get_property("color_mode")) != 1 or not self.hs_color:
_LOGGER.error("Flash supported currently only in RGB mode")
return
@ -782,7 +794,7 @@ class YeelightGenericLight(YeelightEntity, LightEntity):
duration = int(self.config[CONF_TRANSITION]) # in ms
if ATTR_TRANSITION in kwargs: # passed kwarg overrides config
duration = int(kwargs.get(ATTR_TRANSITION) * 1000) # kwarg in s
duration = int(kwargs[ATTR_TRANSITION] * 1000) # kwarg in s
if not self.is_on:
await self._async_turn_on(duration)
@ -840,7 +852,7 @@ class YeelightGenericLight(YeelightEntity, LightEntity):
duration = int(self.config[CONF_TRANSITION]) # in ms
if ATTR_TRANSITION in kwargs: # passed kwarg overrides config
duration = int(kwargs.get(ATTR_TRANSITION) * 1000) # kwarg in s
duration = int(kwargs[ATTR_TRANSITION] * 1000) # kwarg in s
await self._async_turn_off(duration)
self._async_schedule_state_check(False)
@ -893,8 +905,8 @@ class YeelightColorLightSupport(YeelightGenericLight):
class YeelightWhiteTempLightSupport:
"""Representation of a White temp Yeelight light."""
_attr_color_mode = COLOR_MODE_COLOR_TEMP
_attr_supported_color_modes = {COLOR_MODE_COLOR_TEMP}
_attr_color_mode: str | None = COLOR_MODE_COLOR_TEMP
_attr_supported_color_modes: set[str] | None = {COLOR_MODE_COLOR_TEMP}
@property
def _predefined_effects(self):
@ -909,7 +921,7 @@ class YeelightNightLightSupport:
return PowerMode.NORMAL
class YeelightWithoutNightlightSwitchMixIn:
class YeelightWithoutNightlightSwitchMixIn(YeelightGenericLight):
"""A mix-in for yeelights without a nightlight switch."""
@property
@ -931,9 +943,7 @@ class YeelightWithoutNightlightSwitchMixIn:
class YeelightColorLightWithoutNightlightSwitch(
YeelightColorLightSupport,
YeelightWithoutNightlightSwitchMixIn,
YeelightGenericLight,
YeelightColorLightSupport, YeelightWithoutNightlightSwitchMixIn
):
"""Representation of a Color Yeelight light."""
@ -953,9 +963,7 @@ class YeelightColorLightWithNightlightSwitch(
class YeelightWhiteTempWithoutNightlightSwitch(
YeelightWhiteTempLightSupport,
YeelightWithoutNightlightSwitchMixIn,
YeelightGenericLight,
YeelightWhiteTempLightSupport, YeelightWithoutNightlightSwitchMixIn
):
"""White temp light, when nightlight switch is not set to light."""

View File

@ -2,8 +2,10 @@
from __future__ import annotations
import asyncio
from collections.abc import Awaitable, Callable, ValuesView
import contextlib
from ipaddress import IPv4Address, IPv6Address
from datetime import datetime
from ipaddress import IPv4Address
import logging
from urllib.parse import urlparse
@ -11,7 +13,7 @@ from async_upnp_client.search import SsdpHeaders, SsdpSearchListener
from homeassistant import config_entries
from homeassistant.components import network, ssdp
from homeassistant.core import HomeAssistant, callback
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
from homeassistant.helpers.event import async_call_later, async_track_time_interval
from .const import (
@ -34,7 +36,7 @@ class YeelightScanner:
@classmethod
@callback
def async_get(cls, hass: HomeAssistant):
def async_get(cls, hass: HomeAssistant) -> YeelightScanner:
"""Get scanner instance."""
if cls._scanner is None:
cls._scanner = cls(hass)
@ -43,14 +45,14 @@ class YeelightScanner:
def __init__(self, hass: HomeAssistant) -> None:
"""Initialize class."""
self._hass = hass
self._host_discovered_events = {}
self._unique_id_capabilities = {}
self._host_capabilities = {}
self._track_interval = None
self._listeners = []
self._connected_events = []
self._host_discovered_events: dict[str, list[asyncio.Event]] = {}
self._unique_id_capabilities: dict[str, SsdpHeaders] = {}
self._host_capabilities: dict[str, SsdpHeaders] = {}
self._track_interval: CALLBACK_TYPE | None = None
self._listeners: list[SsdpSearchListener] = []
self._connected_events: list[asyncio.Event] = []
async def async_setup(self):
async def async_setup(self) -> None:
"""Set up the scanner."""
if self._connected_events:
await self._async_wait_connected()
@ -59,10 +61,10 @@ class YeelightScanner:
for idx, source_ip in enumerate(await self._async_build_source_set()):
self._connected_events.append(asyncio.Event())
def _wrap_async_connected_idx(idx):
def _wrap_async_connected_idx(idx) -> Callable[[], Awaitable[None]]:
"""Create a function to capture the idx cell variable."""
async def _async_connected():
async def _async_connected() -> None:
self._connected_events[idx].set()
return _async_connected
@ -118,10 +120,10 @@ class YeelightScanner:
return {
source_ip
for source_ip in await network.async_get_enabled_source_ips(self._hass)
if not source_ip.is_loopback and not isinstance(source_ip, IPv6Address)
if isinstance(source_ip, IPv4Address) and not source_ip.is_loopback
}
async def async_discover(self):
async def async_discover(self) -> ValuesView[SsdpHeaders]:
"""Discover bulbs."""
_LOGGER.debug("Yeelight discover with interval %s", DISCOVERY_SEARCH_INTERVAL)
await self.async_setup()
@ -131,13 +133,13 @@ class YeelightScanner:
return self._unique_id_capabilities.values()
@callback
def async_scan(self, *_):
def async_scan(self, _: datetime | None = None) -> None:
"""Send discovery packets."""
_LOGGER.debug("Yeelight scanning")
for listener in self._listeners:
listener.async_search()
async def async_get_capabilities(self, host):
async def async_get_capabilities(self, host: str) -> SsdpHeaders | None:
"""Get capabilities via SSDP."""
if host in self._host_capabilities:
return self._host_capabilities[host]
@ -155,9 +157,9 @@ class YeelightScanner:
self._host_discovered_events[host].remove(host_event)
return self._host_capabilities.get(host)
def _async_discovered_by_ssdp(self, response):
def _async_discovered_by_ssdp(self, response: SsdpHeaders) -> None:
@callback
def _async_start_flow(*_):
def _async_start_flow(*_) -> None:
asyncio.create_task(
self._hass.config_entries.flow.async_init(
DOMAIN,
@ -175,11 +177,12 @@ class YeelightScanner:
# of another discovery
async_call_later(self._hass, 1, _async_start_flow)
async def _async_process_entry(self, headers: SsdpHeaders):
async def _async_process_entry(self, headers: SsdpHeaders) -> None:
"""Process a discovery."""
_LOGGER.debug("Discovered via SSDP: %s", headers)
unique_id = headers["id"]
host = urlparse(headers["location"]).hostname
assert host
current_entry = self._unique_id_capabilities.get(unique_id)
# Make sure we handle ip changes
if not current_entry or host != urlparse(current_entry["location"]).hostname:

View File

@ -3019,15 +3019,6 @@ ignore_errors = true
[mypy-homeassistant.components.xiaomi_miio.switch]
ignore_errors = true
[mypy-homeassistant.components.yeelight]
ignore_errors = true
[mypy-homeassistant.components.yeelight.light]
ignore_errors = true
[mypy-homeassistant.components.yeelight.scanner]
ignore_errors = true
[mypy-homeassistant.components.zha.alarm_control_panel]
ignore_errors = true

View File

@ -167,9 +167,6 @@ IGNORED_MODULES: Final[list[str]] = [
"homeassistant.components.xiaomi_miio.light",
"homeassistant.components.xiaomi_miio.sensor",
"homeassistant.components.xiaomi_miio.switch",
"homeassistant.components.yeelight",
"homeassistant.components.yeelight.light",
"homeassistant.components.yeelight.scanner",
"homeassistant.components.zha.alarm_control_panel",
"homeassistant.components.zha.api",
"homeassistant.components.zha.binary_sensor",