jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 03:07:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add own coordinator to devolo_home_network (#128159)

Browse Source
This commit is contained in:
Guido Schmitz 2024-10-29 13:23:28 +01:00 committed by GitHub
parent 478bf643bf
commit 5ae2f3d081
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 86 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
homeassistant/components/devolo_home_network/__init__.py
View File

@ -2,6 +2,7 @@
from __future__ import annotations
from asyncio import Semaphore
from dataclasses import dataclass
import logging
from typing import Any
@ -32,7 +33,7 @@ from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.httpx_client import get_async_client
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.helpers.update_coordinator import UpdateFailed
from .const import (
CONNECTED_PLC_DEVICES,
@ -47,6 +48,7 @@ from .const import (
SWITCH_GUEST_WIFI,
SWITCH_LEDS,
)
from .coordinator import DevoloDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
@ -58,7 +60,7 @@ class DevoloHomeNetworkData:
"""The devolo Home Network data."""
device: Device
coordinators: dict[str, DataUpdateCoordinator[Any]]
coordinators: dict[str, DevoloDataUpdateCoordinator[Any]]
async def async_setup_entry(
@ -68,6 +70,7 @@ async def async_setup_entry(
zeroconf_instance = await zeroconf.async_get_async_instance(hass)
async_client = get_async_client(hass)
device_registry = dr.async_get(hass)
semaphore = Semaphore(1)
try:
device = Device(
@ -163,58 +166,65 @@ async def async_setup_entry(
"""Disconnect from device."""
await device.async_disconnect()
coordinators: dict[str, DataUpdateCoordinator[Any]] = {}
coordinators: dict[str, DevoloDataUpdateCoordinator[Any]] = {}
if device.plcnet:
coordinators[CONNECTED_PLC_DEVICES] = DataUpdateCoordinator(
coordinators[CONNECTED_PLC_DEVICES] = DevoloDataUpdateCoordinator(
hass,
_LOGGER,
name=CONNECTED_PLC_DEVICES,
semaphore=semaphore,
update_method=async_update_connected_plc_devices,
update_interval=LONG_UPDATE_INTERVAL,
)
if device.device and "led" in device.device.features:
coordinators[SWITCH_LEDS] = DataUpdateCoordinator(
coordinators[SWITCH_LEDS] = DevoloDataUpdateCoordinator(
hass,
_LOGGER,
name=SWITCH_LEDS,
semaphore=semaphore,
update_method=async_update_led_status,
update_interval=SHORT_UPDATE_INTERVAL,
)
if device.device and "restart" in device.device.features:
coordinators[LAST_RESTART] = DataUpdateCoordinator(
coordinators[LAST_RESTART] = DevoloDataUpdateCoordinator(
hass,
_LOGGER,
name=LAST_RESTART,
semaphore=semaphore,
update_method=async_update_last_restart,
update_interval=SHORT_UPDATE_INTERVAL,
)
if device.device and "update" in device.device.features:
coordinators[REGULAR_FIRMWARE] = DataUpdateCoordinator(
coordinators[REGULAR_FIRMWARE] = DevoloDataUpdateCoordinator(
hass,
_LOGGER,
name=REGULAR_FIRMWARE,
semaphore=semaphore,
update_method=async_update_firmware_available,
update_interval=FIRMWARE_UPDATE_INTERVAL,
)
if device.device and "wifi1" in device.device.features:
coordinators[CONNECTED_WIFI_CLIENTS] = DataUpdateCoordinator(
coordinators[CONNECTED_WIFI_CLIENTS] = DevoloDataUpdateCoordinator(
hass,
_LOGGER,
name=CONNECTED_WIFI_CLIENTS,
semaphore=semaphore,
update_method=async_update_wifi_connected_station,
update_interval=SHORT_UPDATE_INTERVAL,
)
coordinators[NEIGHBORING_WIFI_NETWORKS] = DataUpdateCoordinator(
coordinators[NEIGHBORING_WIFI_NETWORKS] = DevoloDataUpdateCoordinator(
hass,
_LOGGER,
name=NEIGHBORING_WIFI_NETWORKS,
semaphore=semaphore,
update_method=async_update_wifi_neighbor_access_points,
update_interval=LONG_UPDATE_INTERVAL,
)
coordinators[SWITCH_GUEST_WIFI] = DataUpdateCoordinator(
coordinators[SWITCH_GUEST_WIFI] = DevoloDataUpdateCoordinator(
hass,
_LOGGER,
name=SWITCH_GUEST_WIFI,
semaphore=semaphore,
update_method=async_update_guest_wifi_status,
update_interval=SHORT_UPDATE_INTERVAL,
)

6
homeassistant/components/devolo_home_network/binary_sensor.py
View File

@ -15,13 +15,13 @@ from homeassistant.components.binary_sensor import (
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import DevoloHomeNetworkConfigEntry
from .const import CONNECTED_PLC_DEVICES, CONNECTED_TO_ROUTER
from .coordinator import DevoloDataUpdateCoordinator
from .entity import DevoloCoordinatorEntity
PARALLEL_UPDATES = 1
PARALLEL_UPDATES = 0
def _is_connected_to_router(entity: DevoloBinarySensorEntity) -> bool:
@ -78,7 +78,7 @@ class DevoloBinarySensorEntity(
def __init__(
self,
entry: DevoloHomeNetworkConfigEntry,
coordinator: DataUpdateCoordinator[LogicalNetwork],
coordinator: DevoloDataUpdateCoordinator[LogicalNetwork],
description: DevoloBinarySensorEntityDescription,
) -> None:
"""Initialize entity."""

2
homeassistant/components/devolo_home_network/button.py
View File

@ -22,7 +22,7 @@ from . import DevoloHomeNetworkConfigEntry
from .const import DOMAIN, IDENTIFY, PAIRING, RESTART, START_WPS
from .entity import DevoloEntity
PARALLEL_UPDATES = 1
PARALLEL_UPDATES = 0
@dataclass(frozen=True, kw_only=True)

38
homeassistant/components/devolo_home_network/coordinator.py Normal file
View File

@ -0,0 +1,38 @@
"""Base coordinator."""
from asyncio import Semaphore
from collections.abc import Awaitable, Callable
from datetime import timedelta
from logging import Logger
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
class DevoloDataUpdateCoordinator[_DataT](DataUpdateCoordinator[_DataT]):
"""Class to manage fetching data from devolo Home Network devices."""
def __init__(
self,
hass: HomeAssistant,
logger: Logger,
*,
name: str,
semaphore: Semaphore,
update_interval: timedelta,
update_method: Callable[[], Awaitable[_DataT]],
) -> None:
"""Initialize global data updater."""
super().__init__(
hass,
logger,
name=name,
update_interval=update_interval,
update_method=update_method,
)
self._semaphore = semaphore
async def _async_update_data(self) -> _DataT:
"""Fetch the latest data from the source."""
async with self._semaphore:
return await super()._async_update_data()

15
homeassistant/components/devolo_home_network/device_tracker.py
View File

@ -13,15 +13,13 @@ from homeassistant.const import STATE_UNKNOWN, UnitOfFrequency
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import DevoloHomeNetworkConfigEntry
from .const import CONNECTED_WIFI_CLIENTS, DOMAIN, WIFI_APTYPE, WIFI_BANDS
from .coordinator import DevoloDataUpdateCoordinator
PARALLEL_UPDATES = 1
PARALLEL_UPDATES = 0
async def async_setup_entry(
@ -31,7 +29,7 @@ async def async_setup_entry(
) -> None:
"""Get all devices and sensors and setup them via config entry."""
device = entry.runtime_data.device
coordinators: dict[str, DataUpdateCoordinator[list[ConnectedStationInfo]]] = (
coordinators: dict[str, DevoloDataUpdateCoordinator[list[ConnectedStationInfo]]] = (
entry.runtime_data.coordinators
)
registry = er.async_get(hass)
@ -84,13 +82,14 @@ async def async_setup_entry(
class DevoloScannerEntity(
CoordinatorEntity[DataUpdateCoordinator[list[ConnectedStationInfo]]], ScannerEntity
CoordinatorEntity[DevoloDataUpdateCoordinator[list[ConnectedStationInfo]]],
ScannerEntity,
):
"""Representation of a devolo device tracker."""
def __init__(
self,
coordinator: DataUpdateCoordinator[list[ConnectedStationInfo]],
coordinator: DevoloDataUpdateCoordinator[list[ConnectedStationInfo]],
device: Device,
mac: str,
) -> None:

10
homeassistant/components/devolo_home_network/entity.py
View File

@ -12,13 +12,11 @@ from devolo_plc_api.plcnet_api import DataRate, LogicalNetwork
from homeassistant.const import ATTR_CONNECTIONS
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import DevoloHomeNetworkConfigEntry
from .const import DOMAIN
from .coordinator import DevoloDataUpdateCoordinator
type _DataType = (
LogicalNetwork
@ -64,14 +62,14 @@ class DevoloEntity(Entity):
class DevoloCoordinatorEntity[_DataT: _DataType](
CoordinatorEntity[DataUpdateCoordinator[_DataT]], DevoloEntity
CoordinatorEntity[DevoloDataUpdateCoordinator[_DataT]], DevoloEntity
):
"""Representation of a coordinated devolo home network device."""
def __init__(
self,
entry: DevoloHomeNetworkConfigEntry,
coordinator: DataUpdateCoordinator[_DataT],
coordinator: DevoloDataUpdateCoordinator[_DataT],
) -> None:
"""Initialize a devolo home network device."""
super().__init__(coordinator)

6
homeassistant/components/devolo_home_network/image.py
View File

@ -13,14 +13,14 @@ from homeassistant.components.image import ImageEntity, ImageEntityDescription
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
import homeassistant.util.dt as dt_util
from . import DevoloHomeNetworkConfigEntry
from .const import IMAGE_GUEST_WIFI, SWITCH_GUEST_WIFI
from .coordinator import DevoloDataUpdateCoordinator
from .entity import DevoloCoordinatorEntity
PARALLEL_UPDATES = 1
PARALLEL_UPDATES = 0
@dataclass(frozen=True, kw_only=True)
@ -66,7 +66,7 @@ class DevoloImageEntity(DevoloCoordinatorEntity[WifiGuestAccessGet], ImageEntity
def __init__(
self,
entry: DevoloHomeNetworkConfigEntry,
coordinator: DataUpdateCoordinator[WifiGuestAccessGet],
coordinator: DevoloDataUpdateCoordinator[WifiGuestAccessGet],
description: DevoloImageEntityDescription,
) -> None:
"""Initialize entity."""

8
homeassistant/components/devolo_home_network/sensor.py
View File

@ -20,7 +20,6 @@ from homeassistant.components.sensor import (
from homeassistant.const import EntityCategory, UnitOfDataRate
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from homeassistant.util.dt import utcnow
from . import DevoloHomeNetworkConfigEntry
@ -32,9 +31,10 @@ from .const import (
PLC_RX_RATE,
PLC_TX_RATE,
)
from .coordinator import DevoloDataUpdateCoordinator
from .entity import DevoloCoordinatorEntity
PARALLEL_UPDATES = 1
PARALLEL_UPDATES = 0
def _last_restart(runtime: int) -> datetime:
@ -198,7 +198,7 @@ class BaseDevoloSensorEntity[
def __init__(
self,
entry: DevoloHomeNetworkConfigEntry,
coordinator: DataUpdateCoordinator[_CoordinatorDataT],
coordinator: DevoloDataUpdateCoordinator[_CoordinatorDataT],
description: DevoloSensorEntityDescription[_ValueDataT, _SensorDataT],
) -> None:
"""Initialize entity."""
@ -231,7 +231,7 @@ class DevoloPlcDataRateSensorEntity(
def __init__(
self,
entry: DevoloHomeNetworkConfigEntry,
coordinator: DataUpdateCoordinator[LogicalNetwork],
coordinator: DevoloDataUpdateCoordinator[LogicalNetwork],
description: DevoloSensorEntityDescription[DataRate, float],
peer: str,
) -> None:

6
homeassistant/components/devolo_home_network/switch.py
View File

@ -15,13 +15,13 @@ from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import DevoloHomeNetworkConfigEntry
from .const import DOMAIN, SWITCH_GUEST_WIFI, SWITCH_LEDS
from .coordinator import DevoloDataUpdateCoordinator
from .entity import DevoloCoordinatorEntity
PARALLEL_UPDATES = 1
PARALLEL_UPDATES = 0
type _DataType = WifiGuestAccessGet | bool
@ -91,7 +91,7 @@ class DevoloSwitchEntity[_DataT: _DataType](
def __init__(
self,
entry: DevoloHomeNetworkConfigEntry,
coordinator: DataUpdateCoordinator[_DataT],
coordinator: DevoloDataUpdateCoordinator[_DataT],
description: DevoloSwitchEntityDescription[_DataT],
) -> None:
"""Initialize entity."""

6
homeassistant/components/devolo_home_network/update.py
View File

@ -20,13 +20,13 @@ from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import DevoloHomeNetworkConfigEntry
from .const import DOMAIN, REGULAR_FIRMWARE
from .coordinator import DevoloDataUpdateCoordinator
from .entity import DevoloCoordinatorEntity
PARALLEL_UPDATES = 1
PARALLEL_UPDATES = 0
@dataclass(frozen=True, kw_only=True)
@ -79,7 +79,7 @@ class DevoloUpdateEntity(DevoloCoordinatorEntity, UpdateEntity):
def __init__(
self,
entry: DevoloHomeNetworkConfigEntry,
coordinator: DataUpdateCoordinator,
coordinator: DevoloDataUpdateCoordinator,
description: DevoloUpdateEntityDescription,
) -> None:
"""Initialize entity."""