github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Bump devolo_plc_api to 1.0.0 (#85235) 

* Bump devolo_plc_api to 1.0.0

* Fix pylint
This commit is contained in:
Guido Schmitz 2023-01-06 00:49:59 +01:00 committed by GitHub
parent 57f792d88f
commit 21fbe07218
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 127 additions and 123 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -2,11 +2,12 @@
from __future__ import annotations
import logging
from typing import Any
import async_timeout
from devolo_plc_api.device import Device
from devolo_plc_api import Device
from devolo_plc_api.device_api import ConnectedStationInfo, NeighborAPInfo
from devolo_plc_api.exceptions.device import DeviceNotFound, DeviceUnavailable
from devolo_plc_api.plcnet_api import LogicalNetwork
from homeassistant.components import zeroconf
from homeassistant.config_entries import ConfigEntry
@ -48,27 +49,30 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
f"Unable to connect to {entry.data[CONF_IP_ADDRESS]}"
) from err
async def async_update_connected_plc_devices() -> dict[str, Any]:
async def async_update_connected_plc_devices() -> LogicalNetwork:
"""Fetch data from API endpoint."""
assert device.plcnet
try:
async with async_timeout.timeout(10):
return await device.plcnet.async_get_network_overview() # type: ignore[no-any-return, union-attr]
return await device.plcnet.async_get_network_overview()
except DeviceUnavailable as err:
raise UpdateFailed(err) from err
async def async_update_wifi_connected_station() -> dict[str, Any]:
async def async_update_wifi_connected_station() -> list[ConnectedStationInfo]:
"""Fetch data from API endpoint."""
assert device.device
try:
async with async_timeout.timeout(10):
return await device.device.async_get_wifi_connected_station() # type: ignore[no-any-return, union-attr]
return await device.device.async_get_wifi_connected_station()
except DeviceUnavailable as err:
raise UpdateFailed(err) from err
async def async_update_wifi_neighbor_access_points() -> dict[str, Any]:
async def async_update_wifi_neighbor_access_points() -> list[NeighborAPInfo]:
"""Fetch data from API endpoint."""
assert device.device
try:
async with async_timeout.timeout(30):
return await device.device.async_get_wifi_neighbor_access_points() # type: ignore[no-any-return, union-attr]
return await device.device.async_get_wifi_neighbor_access_points()
except DeviceUnavailable as err:
raise UpdateFailed(err) from err

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

@ -4,7 +4,7 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from devolo_plc_api.device import Device
from devolo_plc_api import Device
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
@ -24,9 +24,9 @@ from .entity import DevoloEntity
def _is_connected_to_router(entity: DevoloBinarySensorEntity) -> bool:
"""Check, if device is attached to the router."""
return all(
device["attached_to_router"]
for device in entity.coordinator.data["network"]["devices"]
if device["mac_address"] == entity.device.mac
device.attached_to_router
for device in entity.coordinator.data.devices
if device.mac_address == entity.device.mac
)

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

@ -40,7 +40,7 @@ async def validate_input(
return {
SERIAL_NUMBER: str(device.serial_number),
TITLE: device.hostname.split(".")[0],
TITLE: device.hostname.split(".", maxsplit=1)[0],
}

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

@ -2,12 +2,18 @@
from datetime import timedelta
from devolo_plc_api.device_api import (
WIFI_BAND_2G,
WIFI_BAND_5G,
WIFI_VAP_GUEST_AP,
WIFI_VAP_MAIN_AP,
)
from homeassistant.const import Platform
DOMAIN = "devolo_home_network"
PLATFORMS = [Platform.BINARY_SENSOR, Platform.DEVICE_TRACKER, Platform.SENSOR]
MAC_ADDRESS = "mac_address"
PRODUCT = "product"
SERIAL_NUMBER = "serial_number"
TITLE = "title"
@ -16,16 +22,15 @@ LONG_UPDATE_INTERVAL = timedelta(minutes=5)
SHORT_UPDATE_INTERVAL = timedelta(seconds=15)
CONNECTED_PLC_DEVICES = "connected_plc_devices"
CONNECTED_STATIONS = "connected_stations"
CONNECTED_TO_ROUTER = "connected_to_router"
CONNECTED_WIFI_CLIENTS = "connected_wifi_clients"
NEIGHBORING_WIFI_NETWORKS = "neighboring_wifi_networks"
WIFI_APTYPE = {
"WIFI_VAP_MAIN_AP": "Main",
"WIFI_VAP_GUEST_AP": "Guest",
WIFI_VAP_MAIN_AP: "Main",
WIFI_VAP_GUEST_AP: "Guest",
}
WIFI_BANDS = {
"WIFI_BAND_2G": 2.4,
"WIFI_BAND_5G": 5,
WIFI_BAND_2G: 2.4,
WIFI_BAND_5G: 5,
}

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

@ -1,9 +1,8 @@
"""Platform for device tracker integration."""
from __future__ import annotations
from typing import Any
from devolo_plc_api.device import Device
from devolo_plc_api.device_api import ConnectedStationInfo
from homeassistant.components.device_tracker import (
DOMAIN as DEVICE_TRACKER_DOMAIN,
@ -20,14 +19,7 @@ from homeassistant.helpers.update_coordinator import (
DataUpdateCoordinator,
)
from .const import (
CONNECTED_STATIONS,
CONNECTED_WIFI_CLIENTS,
DOMAIN,
MAC_ADDRESS,
WIFI_APTYPE,
WIFI_BANDS,
)
from .const import CONNECTED_WIFI_CLIENTS, DOMAIN, WIFI_APTYPE, WIFI_BANDS
async def async_setup_entry(
@ -35,9 +27,9 @@ async def async_setup_entry(
) -> None:
"""Get all devices and sensors and setup them via config entry."""
device: Device = hass.data[DOMAIN][entry.entry_id]["device"]
coordinators: dict[str, DataUpdateCoordinator] = hass.data[DOMAIN][entry.entry_id][
"coordinators"
]
coordinators: dict[
str, DataUpdateCoordinator[list[ConnectedStationInfo]]
] = hass.data[DOMAIN][entry.entry_id]["coordinators"]
registry = entity_registry.async_get(hass)
tracked = set()
@ -45,16 +37,16 @@ async def async_setup_entry(
def new_device_callback() -> None:
"""Add new devices if needed."""
new_entities = []
for station in coordinators[CONNECTED_WIFI_CLIENTS].data[CONNECTED_STATIONS]:
if station[MAC_ADDRESS] in tracked:
for station in coordinators[CONNECTED_WIFI_CLIENTS].data:
if station.mac_address in tracked:
continue
new_entities.append(
DevoloScannerEntity(
coordinators[CONNECTED_WIFI_CLIENTS], device, station[MAC_ADDRESS]
coordinators[CONNECTED_WIFI_CLIENTS], device, station.mac_address
)
)
tracked.add(station[MAC_ADDRESS])
tracked.add(station.mac_address)
async_add_entities(new_entities)
@callback
@ -90,7 +82,9 @@ async def async_setup_entry(
)
class DevoloScannerEntity(CoordinatorEntity, ScannerEntity):
class DevoloScannerEntity(
CoordinatorEntity[DataUpdateCoordinator[list[ConnectedStationInfo]]], ScannerEntity
):
"""Representation of a devolo device tracker."""
def __init__(
@ -105,22 +99,22 @@ class DevoloScannerEntity(CoordinatorEntity, ScannerEntity):
def extra_state_attributes(self) -> dict[str, str]:
"""Return the attributes."""
attrs: dict[str, str] = {}
if not self.coordinator.data[CONNECTED_STATIONS]:
if not self.coordinator.data:
return {}
station: dict[str, Any] = next(
station = next(
(
station
for station in self.coordinator.data[CONNECTED_STATIONS]
if station[MAC_ADDRESS] == self.mac_address
for station in self.coordinator.data
if station.mac_address == self.mac_address
),
{},
None,
)
if station:
attrs["wifi"] = WIFI_APTYPE.get(station["vap_type"], STATE_UNKNOWN)
attrs["wifi"] = WIFI_APTYPE.get(station.vap_type, STATE_UNKNOWN)
attrs["band"] = (
f"{WIFI_BANDS.get(station['band'])} {UnitOfFrequency.GIGAHERTZ}"
if WIFI_BANDS.get(station["band"])
f"{WIFI_BANDS.get(station.band)} {UnitOfFrequency.GIGAHERTZ}"
if WIFI_BANDS.get(station.band)
else STATE_UNKNOWN
)
return attrs
@ -137,8 +131,8 @@ class DevoloScannerEntity(CoordinatorEntity, ScannerEntity):
"""Return true if the device is connected to the network."""
return any(
station
for station in self.coordinator.data[CONNECTED_STATIONS]
if station[MAC_ADDRESS] == self.mac_address
for station in self.coordinator.data
if station.mac_address == self.mac_address
)
@property

2
homeassistant/components/devolo_home_network/manifest.json
View File

@ -4,7 +4,7 @@
"integration_type": "device",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/devolo_home_network",
"requirements": ["devolo-plc-api==0.9.0"],
"requirements": ["devolo-plc-api==1.0.0"],
"zeroconf": [
{ "type": "_dvl-deviceapi._tcp.local.", "properties": { "MT": "*" } }
],

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

@ -31,7 +31,7 @@ from .entity import DevoloEntity
class DevoloSensorRequiredKeysMixin:
"""Mixin for required keys."""
value_func: Callable[[dict[str, Any]], int]
value_func: Callable[[Any], int]
@dataclass
@ -49,7 +49,7 @@ SENSOR_TYPES: dict[str, DevoloSensorEntityDescription] = {
icon="mdi:lan",
name="Connected PLC devices",
value_func=lambda data: len(
{device["mac_address_from"] for device in data["network"]["data_rates"]}
{device.mac_address_from for device in data.data_rates}
),
),
CONNECTED_WIFI_CLIENTS: DevoloSensorEntityDescription(
@ -58,7 +58,7 @@ SENSOR_TYPES: dict[str, DevoloSensorEntityDescription] = {
icon="mdi:wifi",
name="Connected Wifi clients",
state_class=SensorStateClass.MEASUREMENT,
value_func=lambda data: len(data["connected_stations"]),
value_func=len,
),
NEIGHBORING_WIFI_NETWORKS: DevoloSensorEntityDescription(
key=NEIGHBORING_WIFI_NETWORKS,
@ -66,7 +66,7 @@ SENSOR_TYPES: dict[str, DevoloSensorEntityDescription] = {
entity_registry_enabled_default=False,
icon="mdi:wifi-marker",
name="Neighboring Wifi networks",
value_func=lambda data: len(data["neighbor_aps"]),
value_func=len,
),
}

2
requirements_all.txt
View File

@ -591,7 +591,7 @@ denonavr==0.10.12
devolo-home-control-api==0.18.2
# homeassistant.components.devolo_home_network
devolo-plc-api==0.9.0
devolo-plc-api==1.0.0
# homeassistant.components.directv
directv==0.4.0

2
requirements_test_all.txt
View File

@ -465,7 +465,7 @@ denonavr==0.10.12
devolo-home-control-api==0.18.2
# homeassistant.components.devolo_home_network
devolo-plc-api==0.9.0
devolo-plc-api==1.0.0
# homeassistant.components.directv
directv==0.4.0

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

@ -1,26 +1,31 @@
"""Constants used for mocking data."""
from homeassistant.components import zeroconf
from devolo_plc_api.device_api import (
WIFI_BAND_2G,
WIFI_BAND_5G,
WIFI_VAP_MAIN_AP,
ConnectedStationInfo,
NeighborAPInfo,
)
from devolo_plc_api.plcnet_api import LogicalNetwork
from homeassistant.components.zeroconf import ZeroconfServiceInfo
IP = "1.1.1.1"
CONNECTED_STATIONS = {
"connected_stations": [
{
"mac_address": "AA:BB:CC:DD:EE:FF",
"vap_type": "WIFI_VAP_MAIN_AP",
"band": "WIFI_BAND_5G",
"rx_rate": 87800,
"tx_rate": 87800,
}
],
}
CONNECTED_STATIONS = [
ConnectedStationInfo(
mac_address="AA:BB:CC:DD:EE:FF",
vap_type=WIFI_VAP_MAIN_AP,
band=WIFI_BAND_5G,
rx_rate=87800,
tx_rate=87800,
)
]
NO_CONNECTED_STATIONS = {
"connected_stations": [],
}
NO_CONNECTED_STATIONS = []
DISCOVERY_INFO = zeroconf.ZeroconfServiceInfo(
DISCOVERY_INFO = ZeroconfServiceInfo(
host=IP,
addresses=[IP],
port=14791,
@ -41,7 +46,7 @@ DISCOVERY_INFO = zeroconf.ZeroconfServiceInfo(
},
)
DISCOVERY_INFO_WRONG_DEVICE = zeroconf.ZeroconfServiceInfo(
DISCOVERY_INFO_WRONG_DEVICE = ZeroconfServiceInfo(
host="mock_host",
addresses=["mock_host"],
hostname="mock_hostname",
@ -51,46 +56,42 @@ DISCOVERY_INFO_WRONG_DEVICE = zeroconf.ZeroconfServiceInfo(
type="mock_type",
)
NEIGHBOR_ACCESS_POINTS = {
"neighbor_aps": [
NEIGHBOR_ACCESS_POINTS = [
NeighborAPInfo(
mac_address="AA:BB:CC:DD:EE:FF",
ssid="wifi",
band=WIFI_BAND_2G,
channel=1,
signal=-73,
signal_bars=1,
)
]
PLCNET = LogicalNetwork(
devices=[
{
"mac_address": "AA:BB:CC:DD:EE:FF",
"ssid": "wifi",
"band": "WIFI_BAND_2G",
"channel": 1,
"signal": -73,
"signal_bars": 1,
"attached_to_router": False,
}
]
}
],
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,
},
],
)
PLCNET = {
"network": {
"devices": [
{
"mac_address": "AA:BB:CC:DD:EE:FF",
"attached_to_router": False,
}
],
"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,
},
],
}
}
PLCNET_ATTACHED = {
"network": {
"devices": [
{
"mac_address": "AA:BB:CC:DD:EE:FF",
"attached_to_router": True,
}
],
"data_rates": [],
}
}
PLCNET_ATTACHED = LogicalNetwork(
devices=[
{
"mac_address": "AA:BB:CC:DD:EE:FF",
"attached_to_router": True,
}
],
data_rates=[],
)

10
tests/components/devolo_home_network/mock.py
View File

@ -1,8 +1,6 @@
"""Mock of a devolo Home Network device."""
from __future__ import annotations
import dataclasses
from typing import Any
from unittest.mock import AsyncMock
from devolo_plc_api.device import Device
@ -27,12 +25,10 @@ class MockDevice(Device):
def __init__(
self,
ip: str,
plcnetapi: dict[str, Any] | None = None,
deviceapi: dict[str, Any] | None = None,
zeroconf_instance: AsyncZeroconf | Zeroconf | None = None,
) -> None:
"""Bring mock in a well defined state."""
super().__init__(ip, plcnetapi, deviceapi, zeroconf_instance)
super().__init__(ip, zeroconf_instance)
self.reset()
async def async_connect(
@ -46,12 +42,12 @@ class MockDevice(Device):
def reset(self):
"""Reset mock to starting point."""
self.async_disconnect = AsyncMock()
self.device = DeviceApi(IP, None, dataclasses.asdict(DISCOVERY_INFO))
self.device = DeviceApi(IP, None, DISCOVERY_INFO)
self.device.async_get_wifi_connected_station = AsyncMock(
return_value=CONNECTED_STATIONS
)
self.device.async_get_wifi_neighbor_access_points = AsyncMock(
return_value=NEIGHBOR_ACCESS_POINTS
)
self.plcnet = PlcNetApi(IP, None, dataclasses.asdict(DISCOVERY_INFO))
self.plcnet = PlcNetApi(IP, None, DISCOVERY_INFO)
self.plcnet.async_get_network_overview = AsyncMock(return_value=PLCNET)

18
tests/components/devolo_home_network/test_device_tracker.py
View File

@ -11,10 +11,10 @@ from homeassistant.components.devolo_home_network.const import (
WIFI_BANDS,
)
from homeassistant.const import (
FREQUENCY_GIGAHERTZ,
STATE_HOME,
STATE_NOT_HOME,
STATE_UNAVAILABLE,
UnitOfFrequency,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry
@ -26,13 +26,15 @@ from .mock import MockDevice
from tests.common import async_fire_time_changed
STATION = CONNECTED_STATIONS["connected_stations"][0]
STATION = CONNECTED_STATIONS[0]
SERIAL = DISCOVERY_INFO.properties["SN"]
async def test_device_tracker(hass: HomeAssistant, mock_device: MockDevice):
"""Test device tracker states."""
state_key = f"{PLATFORM}.{DOMAIN}_{SERIAL}_{STATION['mac_address'].lower().replace(':', '_')}"
state_key = (
f"{PLATFORM}.{DOMAIN}_{SERIAL}_{STATION.mac_address.lower().replace(':', '_')}"
)
entry = configure_integration(hass)
er = entity_registry.async_get(hass)
await hass.config_entries.async_setup(entry.entry_id)
@ -49,10 +51,10 @@ async def test_device_tracker(hass: HomeAssistant, mock_device: MockDevice):
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_HOME
assert state.attributes["wifi"] == WIFI_APTYPE[STATION["vap_type"]]
assert state.attributes["wifi"] == WIFI_APTYPE[STATION.vap_type]
assert (
state.attributes["band"]
== f"{WIFI_BANDS[STATION['band']]} {FREQUENCY_GIGAHERTZ}"
== f"{WIFI_BANDS[STATION.band]} {UnitOfFrequency.GIGAHERTZ}"
)
# Emulate state change
@ -82,13 +84,15 @@ async def test_device_tracker(hass: HomeAssistant, mock_device: MockDevice):
async def test_restoring_clients(hass: HomeAssistant, mock_device: MockDevice):
"""Test restoring existing device_tracker entities."""
state_key = f"{PLATFORM}.{DOMAIN}_{SERIAL}_{STATION['mac_address'].lower().replace(':', '_')}"
state_key = (
f"{PLATFORM}.{DOMAIN}_{SERIAL}_{STATION.mac_address.lower().replace(':', '_')}"
)
entry = configure_integration(hass)
er = entity_registry.async_get(hass)
er.async_get_or_create(
PLATFORM,
DOMAIN,
f"{SERIAL}_{STATION['mac_address']}",
f"{SERIAL}_{STATION.mac_address}",
config_entry=entry,
)