Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add PLC PHY rates as sensor to devolo Home Network (#87039) 

* Add plc phyrate sensors

* Fix mypy

* Add tests

* Use suggested_display_precision

* Adapt to recent development

* Remove accidentally added constant

* Fix tests

* Fix pylint

* Use PHY rate instead of phyrate

* Adapt tests

* Hopefully fix mypy

* Hopefully fix mypy

* Use LogicalNetwork

* Apply mypy fixes
pull/106037/head
Guido Schmitz 2023-12-18 21:11:06 +01:00 committed by GitHub
parent b96d2cadac
commit b35afccdb7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 340 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -25,6 +25,8 @@ IDENTIFY = "identify"
IMAGE_GUEST_WIFI = "image_guest_wifi"
NEIGHBORING_WIFI_NETWORKS = "neighboring_wifi_networks"
PAIRING = "pairing"
PLC_RX_RATE = "plc_rx_rate"
PLC_TX_RATE = "plc_tx_rate"
REGULAR_FIRMWARE = "regular_firmware"
RESTART = "restart"
START_WPS = "start_wps"

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

@ -9,7 +9,7 @@ from devolo_plc_api.device_api import (
NeighborAPInfo,
WifiGuestAccessGet,
)
from devolo_plc_api.plcnet_api import LogicalNetwork
from devolo_plc_api.plcnet_api import DataRate, LogicalNetwork
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo
@ -25,6 +25,7 @@ _DataT = TypeVar(
"_DataT",
bound=(
LogicalNetwork
| DataRate
| list[ConnectedStationInfo]
| list[NeighborAPInfo]
| WifiGuestAccessGet

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

@ -3,19 +3,21 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from enum import StrEnum
from typing import Any, Generic, TypeVar
from devolo_plc_api.device import Device
from devolo_plc_api.device_api import ConnectedStationInfo, NeighborAPInfo
from devolo_plc_api.plcnet_api import LogicalNetwork
from devolo_plc_api.plcnet_api import REMOTE, DataRate, LogicalNetwork
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory
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
@ -25,25 +27,38 @@ from .const import (
CONNECTED_WIFI_CLIENTS,
DOMAIN,
NEIGHBORING_WIFI_NETWORKS,
PLC_RX_RATE,
PLC_TX_RATE,
)
from .entity import DevoloCoordinatorEntity
_DataT = TypeVar(
"_DataT",
bound=LogicalNetwork | list[ConnectedStationInfo] | list[NeighborAPInfo],
_CoordinatorDataT = TypeVar(
"_CoordinatorDataT",
bound=LogicalNetwork | DataRate | list[ConnectedStationInfo] | list[NeighborAPInfo],
)
_ValueDataT = TypeVar(
"_ValueDataT",
bound=LogicalNetwork | DataRate | list[ConnectedStationInfo] | list[NeighborAPInfo],
)
class DataRateDirection(StrEnum):
"""Direction of data transfer."""
RX = "rx_rate"
TX = "tx_rate"
@dataclass
class DevoloSensorRequiredKeysMixin(Generic[_DataT]):
class DevoloSensorRequiredKeysMixin(Generic[_CoordinatorDataT]):
"""Mixin for required keys."""
value_func: Callable[[_DataT], int]
value_func: Callable[[_CoordinatorDataT], float]
@dataclass
class DevoloSensorEntityDescription(
SensorEntityDescription, DevoloSensorRequiredKeysMixin[_DataT]
SensorEntityDescription, DevoloSensorRequiredKeysMixin[_CoordinatorDataT]
):
"""Describes devolo sensor entity."""
@ -71,6 +86,24 @@ SENSOR_TYPES: dict[str, DevoloSensorEntityDescription[Any]] = {
icon="mdi:wifi-marker",
value_func=len,
),
PLC_RX_RATE: DevoloSensorEntityDescription[DataRate](
key=PLC_RX_RATE,
entity_category=EntityCategory.DIAGNOSTIC,
name="PLC downlink PHY rate",
device_class=SensorDeviceClass.DATA_RATE,
native_unit_of_measurement=UnitOfDataRate.MEGABITS_PER_SECOND,
value_func=lambda data: getattr(data, DataRateDirection.RX, 0),
suggested_display_precision=0,
),
PLC_TX_RATE: DevoloSensorEntityDescription[DataRate](
key=PLC_TX_RATE,
entity_category=EntityCategory.DIAGNOSTIC,
name="PLC uplink PHY rate",
device_class=SensorDeviceClass.DATA_RATE,
native_unit_of_measurement=UnitOfDataRate.MEGABITS_PER_SECOND,
value_func=lambda data: getattr(data, DataRateDirection.TX, 0),
suggested_display_precision=0,
),
}
@ -83,7 +116,7 @@ async def async_setup_entry(
entry.entry_id
]["coordinators"]
entities: list[DevoloSensorEntity[Any]] = []
entities: list[BaseDevoloSensorEntity[Any, Any]] = []
if device.plcnet:
entities.append(
DevoloSensorEntity(
@ -93,6 +126,29 @@ async def async_setup_entry(
device,
)
)
network = await device.plcnet.async_get_network_overview()
peers = [
peer.mac_address for peer in network.devices if peer.topology == REMOTE
]
for peer in peers:
entities.append(
DevoloPlcDataRateSensorEntity(
entry,
coordinators[CONNECTED_PLC_DEVICES],
SENSOR_TYPES[PLC_TX_RATE],
device,
peer,
)
)
entities.append(
DevoloPlcDataRateSensorEntity(
entry,
coordinators[CONNECTED_PLC_DEVICES],
SENSOR_TYPES[PLC_RX_RATE],
device,
peer,
)
)
if device.device and "wifi1" in device.device.features:
entities.append(
DevoloSensorEntity(
@ -113,23 +169,70 @@ async def async_setup_entry(
async_add_entities(entities)
class DevoloSensorEntity(DevoloCoordinatorEntity[_DataT], SensorEntity):
class BaseDevoloSensorEntity(
Generic[_CoordinatorDataT, _ValueDataT],
DevoloCoordinatorEntity[_CoordinatorDataT],
SensorEntity,
):
"""Representation of a devolo sensor."""
entity_description: DevoloSensorEntityDescription[_DataT]
def __init__(
self,
entry: ConfigEntry,
coordinator: DataUpdateCoordinator[_DataT],
description: DevoloSensorEntityDescription[_DataT],
coordinator: DataUpdateCoordinator[_CoordinatorDataT],
description: DevoloSensorEntityDescription[_ValueDataT],
device: Device,
) -> None:
"""Initialize entity."""
self.entity_description = description
super().__init__(entry, coordinator, device)
class DevoloSensorEntity(BaseDevoloSensorEntity[_CoordinatorDataT, _CoordinatorDataT]):
"""Representation of a generic devolo sensor."""
entity_description: DevoloSensorEntityDescription[_CoordinatorDataT]
@property
def native_value(self) -> int:
def native_value(self) -> float:
"""State of the sensor."""
return self.entity_description.value_func(self.coordinator.data)
class DevoloPlcDataRateSensorEntity(BaseDevoloSensorEntity[LogicalNetwork, DataRate]):
"""Representation of a devolo PLC data rate sensor."""
entity_description: DevoloSensorEntityDescription[DataRate]
def __init__(
self,
entry: ConfigEntry,
coordinator: DataUpdateCoordinator[LogicalNetwork],
description: DevoloSensorEntityDescription[DataRate],
device: Device,
peer: str,
) -> None:
"""Initialize entity."""
super().__init__(entry, coordinator, description, device)
self._peer = peer
peer_device = next(
device
for device in self.coordinator.data.devices
if device.mac_address == peer
)
self._attr_unique_id = f"{self._attr_unique_id}_{peer}"
self._attr_name = f"{description.name} ({peer_device.user_device_name})"
self._attr_entity_registry_enabled_default = peer_device.attached_to_router
@property
def native_value(self) -> float:
"""State of the sensor."""
return self.entity_description.value_func(
next(
data_rate
for data_rate in self.coordinator.data.data_rates
if data_rate.mac_address_from == self.device.mac
and data_rate.mac_address_to == self._peer
)
)

6
homeassistant/components/devolo_home_network/strings.json
View File

@ -62,6 +62,12 @@
},
"neighboring_wifi_networks": {
"name": "Neighboring Wifi networks"
},
"plc_rx_rate": {
"name": "PLC downlink PHY rate"
},
"plc_tx_rate": {
"name": "PLC uplink PHY rate"
}
},
"switch": {

43
tests/components/devolo_home_network/const.py
View File

@ -12,7 +12,7 @@ from devolo_plc_api.device_api import (
UpdateFirmwareCheck,
WifiGuestAccessGet,
)
from devolo_plc_api.plcnet_api import LogicalNetwork
from devolo_plc_api.plcnet_api import LOCAL, REMOTE, LogicalNetwork
from homeassistant.components.zeroconf import ZeroconfServiceInfo
@ -117,14 +117,34 @@ PLCNET = LogicalNetwork(
{
"mac_address": "AA:BB:CC:DD:EE:FF",
"attached_to_router": False,
}
"topology": LOCAL,
"user_device_name": "test1",
},
{
"mac_address": "11:22:33:44:55:66",
"attached_to_router": True,
"topology": REMOTE,
"user_device_name": "test2",
},
{
"mac_address": "12:34:56:78:9A:BC",
"attached_to_router": False,
"topology": REMOTE,
"user_device_name": "test3",
},
],
data_rates=[
{
"mac_address_from": "AA:BB:CC:DD:EE:FF",
"mac_address_to": "11:22:33:44:55:66",
"rx_rate": 0.0,
"tx_rate": 0.0,
"rx_rate": 100.0,
"tx_rate": 100.0,
},
{
"mac_address_from": "AA:BB:CC:DD:EE:FF",
"mac_address_to": "12:34:56:78:9A:BC",
"rx_rate": 150.0,
"tx_rate": 150.0,
},
],
)
@ -136,5 +156,18 @@ PLCNET_ATTACHED = LogicalNetwork(
"attached_to_router": True,
}
],
data_rates=[],
data_rates=[
{
"mac_address_from": "AA:BB:CC:DD:EE:FF",
"mac_address_to": "11:22:33:44:55:66",
"rx_rate": 100.0,
"tx_rate": 100.0,
},
{
"mac_address_from": "AA:BB:CC:DD:EE:FF",
"mac_address_to": "12:34:56:78:9A:BC",
"rx_rate": 150.0,
"tx_rate": 150.0,
},
],
)

96
tests/components/devolo_home_network/snapshots/test_sensor.ambr
View File

@ -134,3 +134,99 @@
'unit_of_measurement': None,
})
# ---
# name: test_update_plc_phyrates
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'data_rate',
'friendly_name': 'Mock Title PLC downlink PHY rate (test2)',
'unit_of_measurement': <UnitOfDataRate.MEGABITS_PER_SECOND: 'Mbit/s'>,
}),
'context': <ANY>,
'entity_id': 'sensor.mock_title_plc_downlink_phy_rate_test2',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': '100.0',
})
# ---
# name: test_update_plc_phyrates.1
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.mock_title_plc_downlink_phy_rate_test2',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 0,
}),
}),
'original_device_class': <SensorDeviceClass.DATA_RATE: 'data_rate'>,
'original_icon': None,
'original_name': 'PLC downlink PHY rate (test2)',
'platform': 'devolo_home_network',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'plc_rx_rate',
'unique_id': '1234567890_plc_rx_rate_11:22:33:44:55:66',
'unit_of_measurement': <UnitOfDataRate.MEGABITS_PER_SECOND: 'Mbit/s'>,
})
# ---
# name: test_update_plc_phyrates.2
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'data_rate',
'friendly_name': 'Mock Title PLC downlink PHY rate (test2)',
'unit_of_measurement': <UnitOfDataRate.MEGABITS_PER_SECOND: 'Mbit/s'>,
}),
'context': <ANY>,
'entity_id': 'sensor.mock_title_plc_downlink_phy_rate_test2',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': '100.0',
})
# ---
# name: test_update_plc_phyrates.3
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.mock_title_plc_downlink_phy_rate_test2',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 0,
}),
}),
'original_device_class': <SensorDeviceClass.DATA_RATE: 'data_rate'>,
'original_icon': None,
'original_name': 'PLC downlink PHY rate (test2)',
'platform': 'devolo_home_network',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'plc_rx_rate',
'unique_id': '1234567890_plc_rx_rate_11:22:33:44:55:66',
'unit_of_measurement': <UnitOfDataRate.MEGABITS_PER_SECOND: 'Mbit/s'>,
})
# ---

78
tests/components/devolo_home_network/test_sensor.py
View File

@ -17,6 +17,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import configure_integration
from .const import PLCNET
from .mock import MockDevice
from tests.common import async_fire_time_changed
@ -33,6 +34,30 @@ async def test_sensor_setup(hass: HomeAssistant) -> None:
assert hass.states.get(f"{DOMAIN}.{device_name}_connected_wifi_clients") is not None
assert hass.states.get(f"{DOMAIN}.{device_name}_connected_plc_devices") is None
assert hass.states.get(f"{DOMAIN}.{device_name}_neighboring_wifi_networks") is None
assert (
hass.states.get(
f"{DOMAIN}.{device_name}_plc_downlink_phy_rate_{PLCNET.devices[1].user_device_name}"
)
is not None
)
assert (
hass.states.get(
f"{DOMAIN}.{device_name}_plc_uplink_phy_rate_{PLCNET.devices[1].user_device_name}"
)
is not None
)
assert (
hass.states.get(
f"{DOMAIN}.{device_name}_plc_downlink_phyrate_{PLCNET.devices[2].user_device_name}"
)
is None
)
assert (
hass.states.get(
f"{DOMAIN}.{device_name}_plc_uplink_phyrate_{PLCNET.devices[2].user_device_name}"
)
is None
)
await hass.config_entries.async_unload(entry.entry_id)
@ -100,3 +125,56 @@ async def test_sensor(
assert state.state == "1"
await hass.config_entries.async_unload(entry.entry_id)
async def test_update_plc_phyrates(
hass: HomeAssistant,
mock_device: MockDevice,
entity_registry: er.EntityRegistry,
freezer: FrozenDateTimeFactory,
snapshot: SnapshotAssertion,
) -> None:
"""Test state change of plc_downlink_phyrate and plc_uplink_phyrate sensor devices."""
entry = configure_integration(hass)
device_name = entry.title.replace(" ", "_").lower()
state_key_downlink = f"{DOMAIN}.{device_name}_plc_downlink_phy_rate_{PLCNET.devices[1].user_device_name}"
state_key_uplink = f"{DOMAIN}.{device_name}_plc_uplink_phy_rate_{PLCNET.devices[1].user_device_name}"
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get(state_key_downlink) == snapshot
assert entity_registry.async_get(state_key_downlink) == snapshot
assert hass.states.get(state_key_downlink) == snapshot
assert entity_registry.async_get(state_key_downlink) == snapshot
# Emulate device failure
mock_device.plcnet.async_get_network_overview = AsyncMock(
side_effect=DeviceUnavailable
)
freezer.tick(LONG_UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
state = hass.states.get(state_key_downlink)
assert state is not None
assert state.state == STATE_UNAVAILABLE
state = hass.states.get(state_key_uplink)
assert state is not None
assert state.state == STATE_UNAVAILABLE
# Emulate state change
mock_device.reset()
freezer.tick(LONG_UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
state = hass.states.get(state_key_downlink)
assert state is not None
assert state.state == str(PLCNET.data_rates[0].rx_rate)
state = hass.states.get(state_key_uplink)
assert state is not None
assert state.state == str(PLCNET.data_rates[0].tx_rate)
await hass.config_entries.async_unload(entry.entry_id)