Skip to content

Commit

Permalink
Add getting support information from the device (#103)
Browse files Browse the repository at this point in the history
* Add getting support information from the device

* Add tests
  • Loading branch information
Shutgun authored Jan 24, 2023
1 parent 2e4207f commit 59ba047
Show file tree
Hide file tree
Showing 11 changed files with 229 additions and 3 deletions.
7 changes: 7 additions & 0 deletions devolo_plc_api/device_api/__init__.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,8 @@
"""The devolo device API."""
import re

from .deviceapi import DeviceApi
from .support_pb2 import SupportInfoDump
from .updatefirmware_pb2 import UpdateFirmwareCheck
from .wifinetwork_pb2 import (
WIFI_BAND_2G,
Expand All @@ -13,20 +16,24 @@
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

RepeatedAPInfo = WifiRepeatedAPsGet.RepeatedAPInfo
ConnectedStationInfo = WifiConnectedStationsGet.ConnectedStationInfo
NeighborAPInfo = WifiNeighborAPsGet.NeighborAPInfo
SupportInfoItem = SupportInfoDump.SupportInfoItem

__all__ = [
"ConnectedStationInfo",
"DeviceApi",
"NeighborAPInfo",
"RepeatedAPInfo",
"SupportInfoItem",
"WifiGuestAccessGet",
"CONFIGLAYER_FORMAT",
"SERVICE_TYPE",
"UPDATE_AVAILABLE",
"UPDATE_NOT_AVAILABLE",
Expand Down
14 changes: 14 additions & 0 deletions devolo_plc_api/device_api/deviceapi.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
from .factoryreset_pb2 import FactoryResetStart
from .ledsettings_pb2 import LedSettingsGet, LedSettingsSet, LedSettingsSetResponse
from .restart_pb2 import RestartResponse, UptimeGetResponse
from .support_pb2 import SupportInfoDump, SupportInfoDumpResponse
from .updatefirmware_pb2 import UpdateFirmwareCheck, UpdateFirmwareStart
from .wifinetwork_pb2 import (
WifiConnectedStationsGet,
Expand Down Expand Up @@ -169,6 +170,19 @@ async def async_uptime(self) -> int:
uptime.ParseFromString(await response.aread())
return uptime.uptime

@_feature("support")
async def async_get_support_info(self) -> SupportInfoDump:
"""
Get support info from the device. This feature only works on devices, that announce the support feature.
:return: The support info
"""
self._logger.debug("Get uptime.")
support_info = SupportInfoDumpResponse()
response = await self._async_get("SupportInfoDump")
support_info.ParseFromString(await response.aread())
return support_info.info

@_feature("update")
async def async_check_firmware_available(self) -> UpdateFirmwareCheck:
"""
Expand Down
3 changes: 3 additions & 0 deletions devolo_plc_api/device_api/deviceapi.pyi
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ isort:skip_file
from __future__ import annotations
from ..clients import Protobuf
from ..zeroconf import ZeroconfServiceInfo as ZeroconfServiceInfo
from .support_pb2 import SupportInfoDump
from .updatefirmware_pb2 import UpdateFirmwareCheck
from .wifinetwork_pb2 import WifiConnectedStationsGet, WifiGuestAccessGet, WifiNeighborAPsGet, WifiRepeatedAPsGet
from httpx import AsyncClient as AsyncClient
Expand All @@ -20,6 +21,7 @@ class DeviceApi(Protobuf):
async def async_factory_reset(self) -> bool: ...
async def async_restart(self) -> bool: ...
async def async_uptime(self) -> int: ...
async def async_get_support_info(self) -> SupportInfoDump: ...
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]: ...
Expand All @@ -34,6 +36,7 @@ class DeviceApi(Protobuf):
def factory_reset(self) -> bool: ...
def restart(self) -> bool: ...
def uptime(self) -> int: ...
def get_support_info(self) -> SupportInfoDump: ...
def check_firmware_available(self) -> UpdateFirmwareCheck: ...
def start_firmware_update(self) -> bool: ...
def get_wifi_connected_station(self) -> list[WifiConnectedStationsGet.ConnectedStationInfo]: ...
Expand Down
59 changes: 59 additions & 0 deletions devolo_plc_api/device_api/support_pb2.py

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

108 changes: 108 additions & 0 deletions devolo_plc_api/device_api/support_pb2.pyi
Original file line number Diff line number Diff line change
@@ -0,0 +1,108 @@
"""
@generated by mypy-protobuf. Do not edit manually!
isort:skip_file
"""
import builtins
import collections.abc
import google.protobuf.descriptor
import google.protobuf.internal.containers
import google.protobuf.internal.enum_type_wrapper
import google.protobuf.message
import sys
import typing

if sys.version_info >= (3, 10):
import typing as typing_extensions
else:
import typing_extensions

DESCRIPTOR: google.protobuf.descriptor.FileDescriptor

class SupportInfoDump(google.protobuf.message.Message):
"""
The SupportInfoDump is a collection of SupportInfoItems.
Each SupportInfoItem is a pair of a unique label and any content.
The label describes the type of support info that is contained in
the content field. It can be used by the client to form a file name
for a file that the content is stored in.
Labels must be unique within the collection of SupportInfoItems.
The content can be any data that should be delivered to support for
the given label and is mostly opaque to the client.
"""

DESCRIPTOR: google.protobuf.descriptor.Descriptor

class SupportInfoItem(google.protobuf.message.Message):
DESCRIPTOR: google.protobuf.descriptor.Descriptor

LABEL_FIELD_NUMBER: builtins.int
CONTENT_FIELD_NUMBER: builtins.int
label: builtins.str
"""Label describing the content. Defined labels: configlayer, delosconfig, pib, syslog"""
content: builtins.bytes
"""Any type of support info content."""
def __init__(
self,
*,
label: builtins.str = ...,
content: builtins.bytes = ...,
) -> None: ...
def ClearField(self, field_name: typing_extensions.Literal["content", b"content", "label", b"label"]) -> None: ...

ITEMS_FIELD_NUMBER: builtins.int
@property
def items(self) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[global___SupportInfoDump.SupportInfoItem]:
"""A list of label-content pairs that make up multiple support info items for this device."""
def __init__(
self,
*,
items: collections.abc.Iterable[global___SupportInfoDump.SupportInfoItem] | None = ...,
) -> None: ...
def ClearField(self, field_name: typing_extensions.Literal["items", b"items"]) -> None: ...

global___SupportInfoDump = SupportInfoDump

class SupportInfoDumpResponse(google.protobuf.message.Message):
"""
Response to calling the "SupportInfoDump" endpoint to collect support info data from a device.
The response is made up of a result code, in case something went wrong, and a SupportInfoDump
which contains the collection of support info items that the client needs to store in a file
to be sent to devolo support.
"""

DESCRIPTOR: google.protobuf.descriptor.Descriptor

class _Result:
ValueType = typing.NewType("ValueType", builtins.int)
V: typing_extensions.TypeAlias = ValueType

class _ResultEnumTypeWrapper(google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[SupportInfoDumpResponse._Result.ValueType], builtins.type): # noqa: F821
DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor
SUCCESS: SupportInfoDumpResponse._Result.ValueType # 0
"""Successful operation, support items are returned."""
UNKNOWN_ERROR: SupportInfoDumpResponse._Result.ValueType # 255
"""Some error occured, returned info is unreliable."""

class Result(_Result, metaclass=_ResultEnumTypeWrapper): ...
SUCCESS: SupportInfoDumpResponse.Result.ValueType # 0
"""Successful operation, support items are returned."""
UNKNOWN_ERROR: SupportInfoDumpResponse.Result.ValueType # 255
"""Some error occured, returned info is unreliable."""

RESULT_FIELD_NUMBER: builtins.int
INFO_FIELD_NUMBER: builtins.int
result: global___SupportInfoDumpResponse.Result.ValueType
"""The return code of the opration, success or error."""
@property
def info(self) -> global___SupportInfoDump:
"""The collected support information."""
def __init__(
self,
*,
result: global___SupportInfoDumpResponse.Result.ValueType = ...,
info: global___SupportInfoDump | None = ...,
) -> None: ...
def HasField(self, field_name: typing_extensions.Literal["info", b"info"]) -> builtins.bool: ...
def ClearField(self, field_name: typing_extensions.Literal["info", b"info", "result", b"result"]) -> None: ...

global___SupportInfoDumpResponse = SupportInfoDumpResponse
2 changes: 2 additions & 0 deletions devolo_plc_api/plcnet_api/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,10 +10,12 @@
REMOTE = GetNetworkOverview.Device.REMOTE
SERVICE_TYPE = "_dvl-plcnetapi._tcp.local."

DataRate = GetNetworkOverview.DataRate
Device = GetNetworkOverview.Device
LogicalNetwork = GetNetworkOverview.LogicalNetwork

__all__ = [
"DataRate",
"Device",
"LogicalNetwork",
"PlcNetApi",
Expand Down
6 changes: 6 additions & 0 deletions docs/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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).

## [v1.1.0] - 2023/01/24

### Added

- Get support information from the device

## [v1.0.0] - 2023/01/05

### Changed
Expand Down
4 changes: 3 additions & 1 deletion example_async.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,9 @@ async def run():
# Get uptime of the device. This value can only be used as a strict monotonically increasing number and therefore has no unit.
print(await dpa.device.async_uptime())

# Get support information from the device.
print(await dpa.device.async_get_support_info())

# Check for new firmware versions
firmware = await dpa.device.async_check_firmware_available()
print(firmware.result) # devolo_plc_api.device_api.UPDATE_NOT_AVAILABLE
Expand Down Expand Up @@ -84,7 +87,6 @@ async def run():
print(network.devices[0].product_id) # "MT3056"
print(network.devices[0].friendly_version) # "7.12.5.124"
print(network.devices[0].full_version) # "magic-2-wifi-next 7.12.5.124_2022-08-29"
print(network.devices[0].full_version) # "magic-2-wifi-next 7.12.5.124_2022-08-29"
print(network.devices[0].user_device_name) # "Living Room"
print(network.devices[0].mac_address) # "AABBCCDDEEFF"
print(network.devices[0].topology) # devolo_plc_api.plcnet_api.LOCAL
Expand Down
4 changes: 3 additions & 1 deletion example_sync.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,9 @@ def run():
# Get uptime of the device. This value can only be used as a strict monotonically increasing number and therefore has no unit.
print(dpa.device.uptime())

# Get support information from the device.
print(dpa.device.get_support_info())

# Check for new firmware versions
firmware = dpa.device.check_firmware_available()
print(firmware.result) # devolo_plc_api.device_api.UPDATE_NOT_AVAILABLE
Expand Down Expand Up @@ -82,7 +85,6 @@ def run():
print(network.devices[0].product_id) # "MT3056"
print(network.devices[0].friendly_version) # "7.12.5.124"
print(network.devices[0].full_version) # "magic-2-wifi-next 7.12.5.124_2022-08-29"
print(network.devices[0].full_version) # "magic-2-wifi-next 7.12.5.124_2022-08-29"
print(network.devices[0].user_device_name) # "Living Room"
print(network.devices[0].mac_address) # "AABBCCDDEEFF"
print(network.devices[0].topology) # devolo_plc_api.plcnet_api.LOCAL
Expand Down
7 changes: 7 additions & 0 deletions tests/fixtures/device_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
DeviceApi,
NeighborAPInfo,
RepeatedAPInfo,
SupportInfoItem,
UpdateFirmwareCheck,
)

Expand Down Expand Up @@ -73,3 +74,9 @@ def repeated_ap() -> RepeatedAPInfo:
def runtime() -> int:
"""Generate mocked runtime of a device."""
return randbelow(65536)


@pytest.fixture(scope="session")
def support_item() -> SupportInfoItem:
"""Generate mocked support information."""
return SupportInfoItem(label="test", content=b"test")
18 changes: 17 additions & 1 deletion tests/test_deviceapi.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,11 @@
import pytest
from pytest_httpx import HTTPXMock

from devolo_plc_api.device_api import ConnectedStationInfo, DeviceApi, NeighborAPInfo, RepeatedAPInfo
from devolo_plc_api.device_api import ConnectedStationInfo, DeviceApi, NeighborAPInfo, RepeatedAPInfo, SupportInfoItem
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.restart_pb2 import RestartResponse, UptimeGetResponse
from devolo_plc_api.device_api.support_pb2 import SupportInfoDump, SupportInfoDumpResponse
from devolo_plc_api.device_api.updatefirmware_pb2 import UpdateFirmwareCheck, UpdateFirmwareStart
from devolo_plc_api.device_api.wifinetwork_pb2 import (
WifiConnectedStationsGet,
Expand Down Expand Up @@ -144,6 +145,21 @@ def test_uptime(self, device_api: DeviceApi, httpx_mock: HTTPXMock, runtime: int
httpx_mock.add_response(content=uptime.SerializeToString())
assert device_api.uptime() == runtime

@pytest.mark.asyncio
@pytest.mark.parametrize("feature", ["support"])
async def test_async_get_support_info(self, device_api: DeviceApi, httpx_mock: HTTPXMock, support_item: SupportInfoItem):
"""Test getting a device's support information asynchronously."""
support_info = SupportInfoDumpResponse(info=SupportInfoDump(items=[support_item]))
httpx_mock.add_response(content=support_info.SerializeToString())
assert await device_api.async_get_support_info() == support_info.info

@pytest.mark.parametrize("feature", ["support"])
def test_get_support_info(self, device_api: DeviceApi, httpx_mock: HTTPXMock, support_item: SupportInfoItem):
"""Test getting a device's support information synchronously."""
support_info = SupportInfoDumpResponse(info=SupportInfoDump(items=[support_item]))
httpx_mock.add_response(content=support_info.SerializeToString())
assert device_api.get_support_info() == support_info.info

@pytest.mark.asyncio
@pytest.mark.parametrize("feature", ["update"])
async def test_async_check_firmware_available(
Expand Down

0 comments on commit 59ba047

Please sign in to comment.