1
mirror of https://github.com/home-assistant/core synced 2024-08-02 23:40:32 +02:00

Improve WLED typing (#77200)

This commit is contained in:
Marc Mueller 2022-08-26 10:25:33 +02:00 committed by GitHub
parent f9a46cc79f
commit c438c26df3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 25 additions and 14 deletions

View File

@ -2,13 +2,12 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
from collections.abc import Callable
from wled import WLED, Device as WLEDDevice, WLEDConnectionClosed, WLEDError from wled import WLED, Device as WLEDDevice, WLEDConnectionClosed, WLEDError
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, EVENT_HOMEASSISTANT_STOP from homeassistant.const import CONF_HOST, EVENT_HOMEASSISTANT_STOP
from homeassistant.core import HomeAssistant, callback from homeassistant.core import CALLBACK_TYPE, Event, HomeAssistant, callback
from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
@ -38,7 +37,7 @@ class WLEDDataUpdateCoordinator(DataUpdateCoordinator[WLEDDevice]):
CONF_KEEP_MASTER_LIGHT, DEFAULT_KEEP_MASTER_LIGHT CONF_KEEP_MASTER_LIGHT, DEFAULT_KEEP_MASTER_LIGHT
) )
self.wled = WLED(entry.data[CONF_HOST], session=async_get_clientsession(hass)) self.wled = WLED(entry.data[CONF_HOST], session=async_get_clientsession(hass))
self.unsub: Callable | None = None self.unsub: CALLBACK_TYPE | None = None
super().__init__( super().__init__(
hass, hass,
@ -85,7 +84,7 @@ class WLEDDataUpdateCoordinator(DataUpdateCoordinator[WLEDDevice]):
self.unsub() self.unsub()
self.unsub = None self.unsub = None
async def close_websocket(_) -> None: async def close_websocket(_: Event) -> None:
"""Close WebSocket connection.""" """Close WebSocket connection."""
self.unsub = None self.unsub = None
await self.wled.disconnect() await self.wled.disconnect()

View File

@ -17,7 +17,7 @@ async def async_get_config_entry_diagnostics(
"""Return diagnostics for a config entry.""" """Return diagnostics for a config entry."""
coordinator: WLEDDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id] coordinator: WLEDDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
data = { data: dict[str, Any] = {
"info": async_redact_data(coordinator.data.info.__dict__, "wifi"), "info": async_redact_data(coordinator.data.info.__dict__, "wifi"),
"state": coordinator.data.state.__dict__, "state": coordinator.data.state.__dict__,
"effects": { "effects": {

View File

@ -1,18 +1,30 @@
"""Helpers for WLED.""" """Helpers for WLED."""
from __future__ import annotations
from collections.abc import Callable, Coroutine
from typing import Any, TypeVar
from typing_extensions import Concatenate, ParamSpec
from wled import WLEDConnectionError, WLEDError from wled import WLEDConnectionError, WLEDError
from homeassistant.exceptions import HomeAssistantError from homeassistant.exceptions import HomeAssistantError
from .models import WLEDEntity
def wled_exception_handler(func): _WLEDEntityT = TypeVar("_WLEDEntityT", bound=WLEDEntity)
_P = ParamSpec("_P")
def wled_exception_handler(
func: Callable[Concatenate[_WLEDEntityT, _P], Coroutine[Any, Any, Any]]
) -> Callable[Concatenate[_WLEDEntityT, _P], Coroutine[Any, Any, None]]:
"""Decorate WLED calls to handle WLED exceptions. """Decorate WLED calls to handle WLED exceptions.
A decorator that wraps the passed in function, catches WLED errors, A decorator that wraps the passed in function, catches WLED errors,
and handles the availability of the device in the data coordinator. and handles the availability of the device in the data coordinator.
""" """
async def handler(self, *args, **kwargs): async def handler(self: _WLEDEntityT, *args: _P.args, **kwargs: _P.kwargs) -> None:
try: try:
await func(self, *args, **kwargs) await func(self, *args, **kwargs)
self.coordinator.async_update_listeners() self.coordinator.async_update_listeners()

View File

@ -253,7 +253,7 @@ class WLEDSegmentLight(WLEDEntity, LightEntity):
def async_update_segments( def async_update_segments(
coordinator: WLEDDataUpdateCoordinator, coordinator: WLEDDataUpdateCoordinator,
current_ids: set[int], current_ids: set[int],
async_add_entities, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
"""Update segments.""" """Update segments."""
segment_ids = {light.segment_id for light in coordinator.data.state.segments} segment_ids = {light.segment_id for light in coordinator.data.state.segments}

View File

@ -115,12 +115,12 @@ class WLEDNumber(WLEDEntity, NumberEntity):
def async_update_segments( def async_update_segments(
coordinator: WLEDDataUpdateCoordinator, coordinator: WLEDDataUpdateCoordinator,
current_ids: set[int], current_ids: set[int],
async_add_entities, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
"""Update segments.""" """Update segments."""
segment_ids = {segment.segment_id for segment in coordinator.data.state.segments} segment_ids = {segment.segment_id for segment in coordinator.data.state.segments}
new_entities = [] new_entities: list[WLEDNumber] = []
# Process new segments, add them to Home Assistant # Process new segments, add them to Home Assistant
for segment_id in segment_ids - current_ids: for segment_id in segment_ids - current_ids:

View File

@ -183,12 +183,12 @@ class WLEDPaletteSelect(WLEDEntity, SelectEntity):
def async_update_segments( def async_update_segments(
coordinator: WLEDDataUpdateCoordinator, coordinator: WLEDDataUpdateCoordinator,
current_ids: set[int], current_ids: set[int],
async_add_entities, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
"""Update segments.""" """Update segments."""
segment_ids = {segment.segment_id for segment in coordinator.data.state.segments} segment_ids = {segment.segment_id for segment in coordinator.data.state.segments}
new_entities = [] new_entities: list[WLEDPaletteSelect] = []
# Process new segments, add them to Home Assistant # Process new segments, add them to Home Assistant
for segment_id in segment_ids - current_ids: for segment_id in segment_ids - current_ids:

View File

@ -203,12 +203,12 @@ class WLEDReverseSwitch(WLEDEntity, SwitchEntity):
def async_update_segments( def async_update_segments(
coordinator: WLEDDataUpdateCoordinator, coordinator: WLEDDataUpdateCoordinator,
current_ids: set[int], current_ids: set[int],
async_add_entities, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
"""Update segments.""" """Update segments."""
segment_ids = {segment.segment_id for segment in coordinator.data.state.segments} segment_ids = {segment.segment_id for segment in coordinator.data.state.segments}
new_entities = [] new_entities: list[WLEDReverseSwitch] = []
# Process new segments, add them to Home Assistant # Process new segments, add them to Home Assistant
for segment_id in segment_ids - current_ids: for segment_id in segment_ids - current_ids: