diff --git a/devolo_plc_api/device_api/__init__.py b/devolo_plc_api/device_api/__init__.py index e9e3575..22d5bd5 100644 --- a/devolo_plc_api/device_api/__init__.py +++ b/devolo_plc_api/device_api/__init__.py @@ -1,8 +1,7 @@ """The devolo device API.""" -import re - from .deviceapi import DeviceApi +from .diagnostics import CONFIGLAYER_FORMAT, DeviceDiagnostics, EthernetPortDiagnostics from .multiap_pb2 import WifiMultiApGetResponse from .support_pb2 import SupportInfoDump from .updatefirmware_pb2 import UpdateFirmwareCheck @@ -18,7 +17,6 @@ WifiRepeatedAPsGet, ) -CONFIGLAYER_FORMAT = re.compile(rb"(([A-Z][A-Z0-9._]+)=(.+?))(?=(([A-Z][A-Z0-9._]+)=(.+))|$)", re.DOTALL) SERVICE_TYPE = "_dvl-deviceapi._tcp.local." UPDATE_AVAILABLE = UpdateFirmwareCheck.UPDATE_AVAILABLE UPDATE_NOT_AVAILABLE = UpdateFirmwareCheck.UPDATE_NOT_AVAILABLE @@ -40,6 +38,8 @@ "WIFI_VAP_STATION", "ConnectedStationInfo", "DeviceApi", + "DeviceDiagnostics", + "EthernetPortDiagnostics", "NeighborAPInfo", "RepeatedAPInfo", "SupportInfoItem", diff --git a/devolo_plc_api/device_api/deviceapi.py b/devolo_plc_api/device_api/deviceapi.py index 930fcfe..4a10418 100644 --- a/devolo_plc_api/device_api/deviceapi.py +++ b/devolo_plc_api/device_api/deviceapi.py @@ -8,6 +8,7 @@ from devolo_plc_api.clients import Protobuf from devolo_plc_api.exceptions import FeatureNotSupported +from .diagnostics import DeviceDiagnostics, _parse_device_diagnostics from .factoryreset_pb2 import FactoryResetStart from .ledsettings_pb2 import LedSettingsGet, LedSettingsSet, LedSettingsSetResponse from .multiap_pb2 import WifiMultiApGetResponse @@ -197,12 +198,24 @@ async def async_get_support_info(self) -> SupportInfoDump: :return: The support info """ - self._logger.debug("Get uptime.") + self._logger.debug("Getting support info.") support_info = SupportInfoDumpResponse() response = await self._async_get("SupportInfoDump", timeout=LONG_RUNNING) support_info.ParseFromString(await response.aread()) return support_info.info + @_feature("support") + async def async_get_device_diagnostics(self) -> DeviceDiagnostics: + """ + Get typed diagnostic information from the device support dump. + + Missing or malformed values are returned as None for compatibility with different firmware versions. + + :return: Typed device diagnostic information + """ + self._logger.debug("Getting device diagnostics.") + return _parse_device_diagnostics(await self.async_get_support_info()) + @_feature("update") async def async_check_firmware_available(self) -> UpdateFirmwareCheck: """ diff --git a/devolo_plc_api/device_api/deviceapi.pyi b/devolo_plc_api/device_api/deviceapi.pyi index 3aeda26..7809116 100644 --- a/devolo_plc_api/device_api/deviceapi.pyi +++ b/devolo_plc_api/device_api/deviceapi.pyi @@ -2,6 +2,7 @@ @generated by stubgen. Do not edit manually! isort:skip_file """ +from .diagnostics import DeviceDiagnostics as DeviceDiagnostics from .multiap_pb2 import WifiMultiApGetResponse from .support_pb2 import SupportInfoDump from .updatefirmware_pb2 import UpdateFirmwareCheck @@ -25,6 +26,7 @@ class DeviceApi(Protobuf): async def async_restart(self) -> bool: ... async def async_uptime(self) -> int: ... async def async_get_support_info(self) -> SupportInfoDump: ... + async def async_get_device_diagnostics(self) -> DeviceDiagnostics: ... async def async_check_firmware_available(self) -> UpdateFirmwareCheck: ... async def async_start_firmware_update(self) -> bool: ... async def async_get_wifi_connected_station(self) -> list[WifiConnectedStationsGet.ConnectedStationInfo]: ... @@ -41,6 +43,7 @@ class DeviceApi(Protobuf): def restart(self) -> bool: ... def uptime(self) -> int: ... def get_support_info(self) -> SupportInfoDump: ... + def get_device_diagnostics(self) -> DeviceDiagnostics: ... def check_firmware_available(self) -> UpdateFirmwareCheck: ... def start_firmware_update(self) -> bool: ... def get_wifi_connected_station(self) -> list[WifiConnectedStationsGet.ConnectedStationInfo]: ... diff --git a/devolo_plc_api/device_api/diagnostics.py b/devolo_plc_api/device_api/diagnostics.py new file mode 100644 index 0000000..818dc9d --- /dev/null +++ b/devolo_plc_api/device_api/diagnostics.py @@ -0,0 +1,135 @@ +"""Typed diagnostic information from a device support dump.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from .support_pb2 import SupportInfoDump + +CONFIGLAYER_FORMAT = re.compile(rb"(([A-Z][A-Z0-9._]+)=(.+?))(?=(([A-Z][A-Z0-9._]+)=(.+))|$)", re.DOTALL) + +_ACTIVE_SLOTS = b"MSPS.L1.ACTIVE_SLOTS" +_ACTIVE_SLOTS_MAX = b"MSPS.L1.ACTIVE_SLOTS_MAX" +_ACTIVE_SLOTS_MIN = b"MSPS.L1.ACTIVE_SLOTS_MIN" +_CONNECTED_TO_GATEWAY = b"DEVOLO.ETHERNET.CONNECTED_TO_GATEWAY" +_CPU_USAGE = b"SYSTEM.STATS.CPU_USAGE" +_ETHERNET_FULL_DUPLEX = b"DEVOLO.ETHERNET.ETH_PORTS_FULLDUPLEX" +_ETHERNET_LINK = b"DEVOLO.ETHERNET.ETH_PORTS_LINK" +_ETHERNET_SPEED = b"DEVOLO.ETHERNET.ETH_PORTS_SPEED" +_FREE_MEMORY = b"SYSTEM.STATS.FREE_MEMORY" +_LOST_MAPS = b"MASTERSELECTION.DOMAIN.LOST_MAPS" +_TEMPERATURE = b"TEMPSENSORS.GENERAL.MEASURE" +_TOTAL_MEMORY = b"SYSTEM.STATS.TOTAL_MEMORY" + + +@dataclass(frozen=True) +class EthernetPortDiagnostics: + """Diagnostic information for one Ethernet port.""" + + port: int + link: bool | None + speed_mbps: int | None + full_duplex: bool | None + + +@dataclass(frozen=True) +class DeviceDiagnostics: + """Typed diagnostic information reported by a device.""" + + cpu_usage_percent: int | None = None + total_memory_kib: int | None = None + free_memory_kib: int | None = None + temperature_celsius: float | None = None + active_time_slots_percent: float | None = None + ethernet_ports: tuple[EthernetPortDiagnostics, ...] = () + connected_to_gateway: bool | None = None + lost_maps: int | None = None + + +def _get_int(values: dict[bytes, bytes], key: bytes) -> int | None: + try: + return int(values[key]) + except (KeyError, ValueError): + return None + + +def _get_bool(values: dict[bytes, bytes], key: bytes) -> bool | None: + value = values.get(key) + if value == b"YES": + return True + if value == b"NO": + return False + return None + + +def _get_list(values: dict[bytes, bytes], key: bytes) -> list[bytes]: + value = values.get(key) + return value.split(b",") if value else [] + + +def _get_list_int(values: list[bytes], index: int) -> int | None: + try: + return int(values[index]) + except (IndexError, ValueError): + return None + + +def _get_list_bool(values: list[bytes], index: int) -> bool | None: + try: + value = values[index] + except IndexError: + return None + if value == b"YES": + return True + if value == b"NO": + return False + return None + + +def _parse_configlayer(support_info: SupportInfoDump) -> dict[bytes, bytes]: + values: dict[bytes, bytes] = {} + for item in support_info.items: + if item.label != "configlayer": + continue + for match in CONFIGLAYER_FORMAT.finditer(item.content): + values[match.group(2)] = match.group(3).strip() + return values + + +def _parse_device_diagnostics(support_info: SupportInfoDump) -> DeviceDiagnostics: + values = _parse_configlayer(support_info) + + temperature = _get_int(values, _TEMPERATURE) + active_slots = _get_int(values, _ACTIVE_SLOTS) + active_slots_min = _get_int(values, _ACTIVE_SLOTS_MIN) + active_slots_max = _get_int(values, _ACTIVE_SLOTS_MAX) + active_time_slots_percent = None + if active_slots is not None and active_slots_max is not None and active_slots_max > 0: + active_time_slots_percent = max(active_slots, active_slots_min or 0) / active_slots_max * 100 + + links = _get_list(values, _ETHERNET_LINK) + speeds = _get_list(values, _ETHERNET_SPEED) + full_duplex = _get_list(values, _ETHERNET_FULL_DUPLEX) + ethernet_ports = tuple( + EthernetPortDiagnostics( + port=index + 1, + link=_get_list_bool(links, index), + speed_mbps=_get_list_int(speeds, index), + full_duplex=_get_list_bool(full_duplex, index), + ) + for index in range(max(len(links), len(speeds), len(full_duplex))) + ) + + return DeviceDiagnostics( + cpu_usage_percent=_get_int(values, _CPU_USAGE), + total_memory_kib=_get_int(values, _TOTAL_MEMORY), + free_memory_kib=_get_int(values, _FREE_MEMORY), + temperature_celsius=temperature / 100 if temperature is not None else None, + active_time_slots_percent=active_time_slots_percent, + ethernet_ports=ethernet_ports, + connected_to_gateway=_get_bool(values, _CONNECTED_TO_GATEWAY), + lost_maps=_get_int(values, _LOST_MAPS), + ) diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 80bf396..328533c 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -4,6 +4,12 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## Unreleased + +### Added + +- Get typed CPU, memory, temperature, powerline, and Ethernet diagnostic information from the device support dump + ## [v1.5.1] - 2025/04/14 ### Changed diff --git a/example_async.py b/example_async.py index f00aca5..edca375 100644 --- a/example_async.py +++ b/example_async.py @@ -40,6 +40,11 @@ async def run(): # Get support information from the device. print(await dpa.device.async_get_support_info()) + # Get an allowlisted, typed view of diagnostic information from the support dump. + diagnostics = await dpa.device.async_get_device_diagnostics() + print(diagnostics.temperature_celsius) + print(diagnostics.ethernet_ports) + # Check for new firmware versions firmware = await dpa.device.async_check_firmware_available() print(firmware.result) # devolo_plc_api.device_api.UPDATE_NOT_AVAILABLE diff --git a/example_sync.py b/example_sync.py index c552553..adf4698 100644 --- a/example_sync.py +++ b/example_sync.py @@ -38,6 +38,11 @@ def run(): # Get support information from the device. print(dpa.device.get_support_info()) + # Get an allowlisted, typed view of diagnostic information from the support dump. + diagnostics = dpa.device.get_device_diagnostics() + print(diagnostics.temperature_celsius) + print(diagnostics.ethernet_ports) + # Check for new firmware versions firmware = dpa.device.check_firmware_available() print(firmware.result) # devolo_plc_api.device_api.UPDATE_NOT_AVAILABLE diff --git a/tests/test_deviceapi.py b/tests/test_deviceapi.py index 082ebab..e71d008 100644 --- a/tests/test_deviceapi.py +++ b/tests/test_deviceapi.py @@ -10,6 +10,7 @@ import pytest from httpx import ConnectTimeout +from devolo_plc_api.device_api import DeviceDiagnostics, EthernetPortDiagnostics from devolo_plc_api.device_api.factoryreset_pb2 import FactoryResetStart from devolo_plc_api.device_api.ledsettings_pb2 import LedSettingsGet, LedSettingsSetResponse from devolo_plc_api.device_api.multiap_pb2 import WifiMultiApGetResponse @@ -214,6 +215,83 @@ def test_get_support_info(self, device_api: DeviceApi, httpx_mock: HTTPXMock, su httpx_mock.add_response(content=support_info.SerializeToString()) assert device_api.get_support_info() == support_info.info + @pytest.mark.asyncio + @pytest.mark.parametrize("feature", ["support"]) + async def test_async_get_device_diagnostics(self, device_api: DeviceApi, httpx_mock: HTTPXMock): + """Test getting typed device diagnostic information asynchronously.""" + configlayer = b"""SYSTEM.STATS.CPU_USAGE=29 +SYSTEM.STATS.TOTAL_MEMORY=8053 +SYSTEM.STATS.FREE_MEMORY=5036 +TEMPSENSORS.GENERAL.MEASURE=8156 +MSPS.L1.ACTIVE_SLOTS=8 +MSPS.L1.ACTIVE_SLOTS_MIN=2 +MSPS.L1.ACTIVE_SLOTS_MAX=16 +DEVOLO.ETHERNET.ETH_PORTS_LINK=NO,YES,NO +DEVOLO.ETHERNET.ETH_PORTS_SPEED=0,1000,0 +DEVOLO.ETHERNET.ETH_PORTS_FULLDUPLEX=NO,YES,NO +DEVOLO.ETHERNET.CONNECTED_TO_GATEWAY=YES +MASTERSELECTION.DOMAIN.LOST_MAPS=89""" + support_info = SupportInfoDumpResponse( + info=SupportInfoDump(items=[SupportInfoDump.SupportInfoItem(label="configlayer", content=configlayer)]) + ) + httpx_mock.add_response(content=support_info.SerializeToString()) + + assert await device_api.async_get_device_diagnostics() == DeviceDiagnostics( + cpu_usage_percent=29, + total_memory_kib=8053, + free_memory_kib=5036, + temperature_celsius=81.56, + active_time_slots_percent=50, + ethernet_ports=( + EthernetPortDiagnostics(port=1, link=False, speed_mbps=0, full_duplex=False), + EthernetPortDiagnostics(port=2, link=True, speed_mbps=1000, full_duplex=True), + EthernetPortDiagnostics(port=3, link=False, speed_mbps=0, full_duplex=False), + ), + connected_to_gateway=True, + lost_maps=89, + ) + + @pytest.mark.parametrize("feature", ["support"]) + def test_get_device_diagnostics_with_partial_malformed_data(self, device_api: DeviceApi, httpx_mock: HTTPXMock): + """Test missing and malformed diagnostic information synchronously.""" + configlayer = b"""SYSTEM.STATS.CPU_USAGE=invalid +SYSTEM.STATS.FREE_MEMORY=1024 +TEMPSENSORS.GENERAL.MEASURE=invalid +MSPS.L1.ACTIVE_SLOTS=8 +MSPS.L1.ACTIVE_SLOTS_MAX=0 +DEVOLO.ETHERNET.ETH_PORTS_LINK=YES,UNKNOWN +DEVOLO.ETHERNET.ETH_PORTS_SPEED=1000,invalid,100 +DEVOLO.ETHERNET.ETH_PORTS_FULLDUPLEX=YES +DEVOLO.ETHERNET.CONNECTED_TO_GATEWAY=NO""" + support_info = SupportInfoDumpResponse( + info=SupportInfoDump( + items=[ + SupportInfoDump.SupportInfoItem(label="other", content=b"SYSTEM.STATS.CPU_USAGE=99"), + SupportInfoDump.SupportInfoItem(label="configlayer", content=configlayer), + ] + ) + ) + httpx_mock.add_response(content=support_info.SerializeToString()) + + assert device_api.get_device_diagnostics() == DeviceDiagnostics( + free_memory_kib=1024, + ethernet_ports=( + EthernetPortDiagnostics(port=1, link=True, speed_mbps=1000, full_duplex=True), + EthernetPortDiagnostics(port=2, link=None, speed_mbps=None, full_duplex=None), + EthernetPortDiagnostics(port=3, link=None, speed_mbps=100, full_duplex=None), + ), + connected_to_gateway=False, + ) + + @pytest.mark.asyncio + @pytest.mark.parametrize("feature", ["support"]) + async def test_async_get_device_diagnostics_without_configlayer(self, device_api: DeviceApi, httpx_mock: HTTPXMock): + """Test getting empty diagnostics if the support dump has no configlayer item.""" + support_info = SupportInfoDumpResponse(info=SupportInfoDump(items=[])) + httpx_mock.add_response(content=support_info.SerializeToString()) + + assert await device_api.async_get_device_diagnostics() == DeviceDiagnostics() + @pytest.mark.asyncio @pytest.mark.parametrize("feature", ["update"]) async def test_async_check_firmware_available(