Add typed device diagnostics

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",

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:
         """

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]: ...

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),
+    )

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

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

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

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(