Improve kostal_plenticore generic typing (#84634)

This commit is contained in:
Marc Mueller 2022-12-27 21:23:22 +01:00 committed by GitHub
parent b8f83074ce
commit 1cad55f721
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 30 additions and 18 deletions

View File

@ -6,7 +6,7 @@ from collections import defaultdict
from collections.abc import Callable from collections.abc import Callable
from datetime import datetime, timedelta from datetime import datetime, timedelta
import logging import logging
from typing import Any from typing import Any, TypeVar, cast
from aiohttp.client_exceptions import ClientError from aiohttp.client_exceptions import ClientError
from kostal.plenticore import ( from kostal.plenticore import (
@ -26,6 +26,7 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN from .const import DOMAIN
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
_DataT = TypeVar("_DataT")
class Plenticore: class Plenticore:
@ -154,7 +155,7 @@ class DataUpdateCoordinatorMixin:
return True return True
class PlenticoreUpdateCoordinator(DataUpdateCoordinator): class PlenticoreUpdateCoordinator(DataUpdateCoordinator[_DataT]):
"""Base implementation of DataUpdateCoordinator for Plenticore data.""" """Base implementation of DataUpdateCoordinator for Plenticore data."""
def __init__( def __init__(
@ -192,7 +193,9 @@ class PlenticoreUpdateCoordinator(DataUpdateCoordinator):
self._fetch[module_id].remove(data_id) self._fetch[module_id].remove(data_id)
class ProcessDataUpdateCoordinator(PlenticoreUpdateCoordinator): class ProcessDataUpdateCoordinator(
PlenticoreUpdateCoordinator[dict[str, dict[str, str]]]
):
"""Implementation of PlenticoreUpdateCoordinator for process data.""" """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]]:
@ -214,7 +217,7 @@ class ProcessDataUpdateCoordinator(PlenticoreUpdateCoordinator):
class SettingDataUpdateCoordinator( class SettingDataUpdateCoordinator(
PlenticoreUpdateCoordinator, DataUpdateCoordinatorMixin PlenticoreUpdateCoordinator[dict[str, dict[str, str]]], DataUpdateCoordinatorMixin
): ):
"""Implementation of PlenticoreUpdateCoordinator for settings data.""" """Implementation of PlenticoreUpdateCoordinator for settings data."""
@ -230,7 +233,7 @@ class SettingDataUpdateCoordinator(
return fetched_data return fetched_data
class PlenticoreSelectUpdateCoordinator(DataUpdateCoordinator): class PlenticoreSelectUpdateCoordinator(DataUpdateCoordinator[_DataT]):
"""Base implementation of DataUpdateCoordinator for Plenticore data.""" """Base implementation of DataUpdateCoordinator for Plenticore data."""
def __init__( def __init__(
@ -249,10 +252,12 @@ class PlenticoreSelectUpdateCoordinator(DataUpdateCoordinator):
update_interval=update_inverval, update_interval=update_inverval,
) )
# data ids to poll # data ids to poll
self._fetch: dict[str, list[str]] = defaultdict(list) self._fetch: dict[str, list[str | list[str]]] = defaultdict(list)
self._plenticore = plenticore self._plenticore = plenticore
def start_fetch_data(self, module_id: str, data_id: str, all_options: str) -> None: def start_fetch_data(
self, module_id: str, data_id: str, all_options: list[str]
) -> None:
"""Start fetching the given data (module-id and entry-id).""" """Start fetching the given data (module-id and entry-id)."""
self._fetch[module_id].append(data_id) self._fetch[module_id].append(data_id)
self._fetch[module_id].append(all_options) self._fetch[module_id].append(all_options)
@ -264,14 +269,17 @@ class PlenticoreSelectUpdateCoordinator(DataUpdateCoordinator):
async_call_later(self.hass, 2, force_refresh) async_call_later(self.hass, 2, force_refresh)
def stop_fetch_data(self, module_id: str, data_id: str, all_options: str) -> None: def stop_fetch_data(
self, module_id: str, data_id: str, all_options: list[str]
) -> None:
"""Stop fetching the given data (module-id and entry-id).""" """Stop fetching the given data (module-id and entry-id)."""
self._fetch[module_id].remove(all_options) self._fetch[module_id].remove(all_options)
self._fetch[module_id].remove(data_id) self._fetch[module_id].remove(data_id)
class SelectDataUpdateCoordinator( class SelectDataUpdateCoordinator(
PlenticoreSelectUpdateCoordinator, DataUpdateCoordinatorMixin PlenticoreSelectUpdateCoordinator[dict[str, dict[str, str]]],
DataUpdateCoordinatorMixin,
): ):
"""Implementation of PlenticoreUpdateCoordinator for select data.""" """Implementation of PlenticoreUpdateCoordinator for select data."""
@ -287,11 +295,11 @@ class SelectDataUpdateCoordinator(
async def _async_get_current_option( async def _async_get_current_option(
self, self,
module_id: dict[str, list[str]], module_id: dict[str, list[str | list[str]]],
) -> dict[str, dict[str, str]]: ) -> dict[str, dict[str, str]]:
"""Get current option.""" """Get current option."""
for mid, pids in module_id.items(): for mid, pids in module_id.items():
all_options = pids[1] all_options = cast(list[str], pids[1])
for all_option in all_options: for all_option in all_options:
if all_option == "None" or not ( if all_option == "None" or not (
val := await self.async_read_data(mid, all_option) val := await self.async_read_data(mid, all_option)
@ -299,10 +307,10 @@ class SelectDataUpdateCoordinator(
continue continue
for option in val.values(): for option in val.values():
if option[all_option] == "1": if option[all_option] == "1":
fetched = {mid: {pids[0]: all_option}} fetched = {mid: {cast(str, pids[0]): all_option}}
return fetched return fetched
return {mid: {pids[0]: "None"}} return {mid: {cast(str, pids[0]): "None"}}
return {} return {}

View File

@ -1,7 +1,6 @@
"""Platform for Kostal Plenticore numbers.""" """Platform for Kostal Plenticore numbers."""
from __future__ import annotations from __future__ import annotations
from abc import ABC
from dataclasses import dataclass from dataclasses import dataclass
from datetime import timedelta from datetime import timedelta
import logging import logging
@ -130,11 +129,12 @@ async def async_setup_entry(
async_add_entities(entities) async_add_entities(entities)
class PlenticoreDataNumber(CoordinatorEntity, NumberEntity, ABC): class PlenticoreDataNumber(
CoordinatorEntity[SettingDataUpdateCoordinator], NumberEntity
):
"""Representation of a Kostal Plenticore Number entity.""" """Representation of a Kostal Plenticore Number entity."""
entity_description: PlenticoreNumberEntityDescription entity_description: PlenticoreNumberEntityDescription
coordinator: SettingDataUpdateCoordinator
def __init__( def __init__(
self, self,

View File

@ -87,7 +87,9 @@ async def async_setup_entry(
async_add_entities(entities) async_add_entities(entities)
class PlenticoreDataSelect(CoordinatorEntity, SelectEntity): class PlenticoreDataSelect(
CoordinatorEntity[SelectDataUpdateCoordinator], SelectEntity
):
"""Representation of a Plenticore Select.""" """Representation of a Plenticore Select."""
_attr_entity_category = EntityCategory.CONFIG _attr_entity_category = EntityCategory.CONFIG

View File

@ -725,7 +725,9 @@ async def async_setup_entry(
async_add_entities(entities) async_add_entities(entities)
class PlenticoreDataSensor(CoordinatorEntity, SensorEntity): class PlenticoreDataSensor(
CoordinatorEntity[ProcessDataUpdateCoordinator], SensorEntity
):
"""Representation of a Plenticore data Sensor.""" """Representation of a Plenticore data Sensor."""
entity_description: PlenticoreSensorEntityDescription entity_description: PlenticoreSensorEntityDescription