• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

2Fake / devolo_plc_api / 6080961227

05 Sep 2023 06:23AM UTC coverage: 97.468% (-1.1%) from 98.543%
6080961227

Pull #142

github-actions

Shutgun
Fix TCH in tests
Pull Request #142: Fix new ruff findings

14 of 14 new or added lines in 3 files covered. (100.0%)

539 of 553 relevant lines covered (97.47%)

0.97 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.79
/devolo_plc_api/device_api/deviceapi.py
1
"""Implementation of the devolo device API."""
2
from __future__ import annotations
1✔
3

4
import functools
1✔
5
from typing import TYPE_CHECKING, Callable, TypeVar
1✔
6

7
from devolo_plc_api.clients import Protobuf
1✔
8
from devolo_plc_api.exceptions import FeatureNotSupported
1✔
9

10
from .factoryreset_pb2 import FactoryResetStart
1✔
11
from .ledsettings_pb2 import LedSettingsGet, LedSettingsSet, LedSettingsSetResponse
1✔
12
from .multiap_pb2 import WifiMultiApGetResponse
1✔
13
from .restart_pb2 import RestartResponse, UptimeGetResponse
1✔
14
from .support_pb2 import SupportInfoDump, SupportInfoDumpResponse
1✔
15
from .updatefirmware_pb2 import UpdateFirmwareCheck, UpdateFirmwareStart
1✔
16
from .wifinetwork_pb2 import (
1✔
17
    WifiConnectedStationsGet,
18
    WifiGuestAccessGet,
19
    WifiGuestAccessSet,
20
    WifiGuestAccessSetResponse,
21
    WifiNeighborAPsGet,
22
    WifiRepeatedAPsGet,
23
    WifiRepeaterWpsClonePbcStart,
24
    WifiResult,
25
    WifiWpsPbcStart,
26
)
27

28
if TYPE_CHECKING:
1✔
29
    from httpx import AsyncClient
×
30
    from typing_extensions import Concatenate, ParamSpec
×
31

32
    from devolo_plc_api.zeroconf import ZeroconfServiceInfo
×
33

34
    _ReturnT = TypeVar("_ReturnT")
×
35
    _P = ParamSpec("_P")
×
36

37

38
def _feature(
1✔
39
    feature: str,
40
) -> Callable[[Callable[Concatenate[DeviceApi, _P], _ReturnT]], Callable[Concatenate[DeviceApi, _P], _ReturnT]]:
41
    """Filter unsupported features before querying the device."""
42

43
    def feature_decorator(method: Callable[Concatenate[DeviceApi, _P], _ReturnT]) -> Callable[..., _ReturnT]:
1✔
44
        @functools.wraps(method)
1✔
45
        def wrapper(deviceapi: DeviceApi, *args: _P.args, **kwargs: _P.kwargs) -> _ReturnT:
1✔
46
            if feature in deviceapi.features:
1✔
47
                return method(deviceapi, *args, **kwargs)
1✔
48
            raise FeatureNotSupported(method.__name__)
1✔
49

50
        return wrapper
1✔
51

52
    return feature_decorator
1✔
53

54

55
class DeviceApi(Protobuf):
1✔
56
    """
57
    Implementation of the devolo device API.
58

59
    :param ip: IP address of the device to communicate with
60
    :param session: HTTP client session
61
    :param info: Information collected from the mDNS query
62
    """
63

64
    def __init__(self, ip: str, session: AsyncClient, info: ZeroconfServiceInfo) -> None:
1✔
65
        """Initialize the device API."""
66
        super().__init__()
1✔
67

68
        self._ip = ip
1✔
69
        # HC gateway has no Path, it has a path.
70
        self._path = info.properties.get("Path") or info.properties.get("path")
1✔
71
        self._port = info.port
1✔
72
        self._session = session
1✔
73
        self._user = "devolo"
1✔
74
        self._version = info.properties["Version"]
1✔
75

76
        features: str = info.properties.get("Features", "")
1✔
77
        self.features = features.split(",") if features else ["reset", "update", "led", "intmtg"]
1✔
78
        self.password = ""
1✔
79

80
    @_feature("led")
1✔
81
    async def async_get_led_setting(self) -> bool:
1✔
82
        """
83
        Get LED setting asynchronously. This feature only works on devices, that announce the led feature.
84

85
        return: LED settings
86
        """
87
        self._logger.debug("Getting LED settings.")
1✔
88
        led_setting = LedSettingsGet()
1✔
89
        response = await self._async_get("LedSettingsGet")
1✔
90
        led_setting.ParseFromString(await response.aread())
1✔
91
        return led_setting.state == led_setting.LED_ON
1✔
92

93
    @_feature("led")
1✔
94
    async def async_set_led_setting(self, enable: bool) -> bool:
1✔
95
        """
96
        Set LED setting asynchronously. This feature only works on devices, that announce the led feature.
97

98
        :param enable: True to enable the LEDs, False to disable the LEDs
99
        :return: True, if LED state was successfully changed, otherwise False
100
        """
101
        self._logger.debug("Setting LED settings.")
1✔
102
        led_setting = LedSettingsSet()
1✔
103
        led_setting.state = led_setting.LED_ON if enable else led_setting.LED_OFF
1✔
104
        query = await self._async_post("LedSettingsSet", content=led_setting.SerializeToString())
1✔
105
        response = LedSettingsSetResponse()
1✔
106
        response.ParseFromString(await query.aread())
1✔
107
        return response.result == response.SUCCESS
1✔
108

109
    @_feature("multiap")
1✔
110
    async def async_get_wifi_multi_ap(self) -> WifiMultiApGetResponse:
1✔
111
        """
112
        Get MultiAP details asynchronously. This feature only works on devices, that announce the multiap feature.
113

114
        return: MultiAP details
115
        """
116
        self._logger.debug("Getting MultiAP details.")
1✔
117
        query = await self._async_get("WifiMultiApGet")
1✔
118
        response = WifiMultiApGetResponse()
1✔
119
        response.ParseFromString(await query.aread())
1✔
120
        return response
1✔
121

122
    @_feature("repeater0")
1✔
123
    async def async_get_wifi_repeated_access_points(self) -> list[WifiRepeatedAPsGet.RepeatedAPInfo]:
1✔
124
        """
125
        Get repeated wifi access point asynchronously. This feature only works on repeater devices, that announce the
126
        repeater0 feature.
127

128
        :return: Repeated access points in the neighborhood including connection rate data
129
        """
130
        self._logger.debug("Getting repeated access points.")
1✔
131
        repeated_aps = WifiRepeatedAPsGet()
1✔
132
        response = await self._async_get("WifiRepeatedAPsGet")
1✔
133
        repeated_aps.ParseFromString(await response.aread())
1✔
134
        return list(repeated_aps.repeated_aps)
1✔
135

136
    @_feature("repeater0")
1✔
137
    async def async_start_wps_clone(self) -> bool:
1✔
138
        """
139
        Start WPS clone mode. This feature only works on repeater devices, that announce the repeater0 feature.
140

141
        :return: True, if the wifi settings were successfully cloned, otherwise False
142
        """
143
        self._logger.debug("Starting WPS clone.")
1✔
144
        wps_clone = WifiRepeaterWpsClonePbcStart()
1✔
145
        response = await self._async_get("WifiRepeaterWpsClonePbcStart")
1✔
146
        wps_clone.ParseFromString(await response.aread())
1✔
147
        return wps_clone.result == WifiResult.WIFI_SUCCESS
1✔
148

149
    @_feature("reset")
1✔
150
    async def async_factory_reset(self) -> bool:
1✔
151
        """
152
        Factory-reset the device. This feature only works on devices, that announce the reset feature.
153

154
        :return: True if reset is started, otherwise False
155
        """
156
        self._logger.debug("Resetting the device.")
1✔
157
        reset = FactoryResetStart()
1✔
158
        response = await self._async_get("FactoryResetStart")
1✔
159
        reset.ParseFromString(await response.aread())
1✔
160
        return reset.result == reset.SUCCESS
1✔
161

162
    @_feature("restart")
1✔
163
    async def async_restart(self) -> bool:
1✔
164
        """
165
        Restart the device. This feature only works on devices, that announce the restart feature.
166

167
        :return: True if restart is started, otherwise False
168
        """
169
        self._logger.debug("Restarting the device.")
1✔
170
        query = await self._async_post("Restart", content=b"")
1✔
171
        response = RestartResponse()
1✔
172
        response.ParseFromString(await query.aread())
1✔
173
        return response.result == response.SUCCESS
1✔
174

175
    @_feature("restart")
1✔
176
    async def async_uptime(self) -> int:
1✔
177
        """
178
        Get the uptime of the device. This feature only works on devices, that announce the restart feature. It can only be
179
        used as a strict monotonically increasing number and therefore has no unit.
180

181
        :return: The uptime without unit
182
        """
183
        self._logger.debug("Get uptime.")
1✔
184
        uptime = UptimeGetResponse()
1✔
185
        response = await self._async_get("UptimeGet")
1✔
186
        uptime.ParseFromString(await response.aread())
1✔
187
        return uptime.uptime
1✔
188

189
    @_feature("support")
1✔
190
    async def async_get_support_info(self) -> SupportInfoDump:
1✔
191
        """
192
        Get support info from the device. This feature only works on devices, that announce the support feature.
193

194
        :return: The support info
195
        """
196
        self._logger.debug("Get uptime.")
1✔
197
        support_info = SupportInfoDumpResponse()
1✔
198
        response = await self._async_get("SupportInfoDump")
1✔
199
        support_info.ParseFromString(await response.aread())
1✔
200
        return support_info.info
1✔
201

202
    @_feature("update")
1✔
203
    async def async_check_firmware_available(self) -> UpdateFirmwareCheck:
1✔
204
        """
205
        Check asynchronously, if a firmware update is available for the device.
206

207
        :return: Result and new firmware version, if newer one is available
208
        """
209
        self._logger.debug("Checking for new firmware.")
1✔
210
        update_firmware_check = UpdateFirmwareCheck()
1✔
211
        response = await self._async_get("UpdateFirmwareCheck")
1✔
212
        update_firmware_check.ParseFromString(await response.aread())
1✔
213
        return update_firmware_check
1✔
214

215
    @_feature("update")
1✔
216
    async def async_start_firmware_update(self) -> bool:
1✔
217
        """
218
        Start firmware update asynchronously, if a firmware update is available for the device. Important: The response does
219
        not tell you anything about the success of the update itself.
220

221
        :return: True, if the firmware update was started, False if there is no update
222
        """
223
        self._logger.debug("Updating firmware.")
1✔
224
        update_firmware = UpdateFirmwareStart()
1✔
225
        query = await self._async_get("UpdateFirmwareStart")
1✔
226
        update_firmware.ParseFromString(await query.aread())
1✔
227
        return update_firmware.result == update_firmware.UPDATE_STARTED
1✔
228

229
    @_feature("wifi1")
1✔
230
    async def async_get_wifi_connected_station(self) -> list[WifiConnectedStationsGet.ConnectedStationInfo]:
1✔
231
        """
232
        Get wifi stations connected to the device asynchronously. This feature only works on devices, that announce the wifi1
233
        feature.
234

235
        :return: All connected wifi stations including connection rate data
236
        """
237
        self._logger.debug("Getting connected wifi stations.")
1✔
238
        wifi_connected = WifiConnectedStationsGet()
1✔
239
        response = await self._async_get("WifiConnectedStationsGet")
1✔
240
        wifi_connected.ParseFromString(await response.aread())
1✔
241
        return list(wifi_connected.connected_stations)
1✔
242

243
    @_feature("wifi1")
1✔
244
    async def async_get_wifi_guest_access(self) -> WifiGuestAccessGet:
1✔
245
        """
246
        Get details about wifi guest access asynchronously. This feature only works on devices, that announce the wifi1
247
        feature.
248

249
        :return: Details about the wifi guest access
250
        """
251
        self._logger.debug("Getting wifi guest access status.")
1✔
252
        wifi_guest = WifiGuestAccessGet()
1✔
253
        response = await self._async_get("WifiGuestAccessGet")
1✔
254
        wifi_guest.ParseFromString(await response.aread())
1✔
255
        return wifi_guest
1✔
256

257
    @_feature("wifi1")
1✔
258
    async def async_set_wifi_guest_access(self, enable: bool, duration: int = 0) -> bool:
1✔
259
        """
260
        Enable wifi guest access asynchronously. This feature only works on devices, that announce the wifi1 feature.
261

262
        :param enable: True to enable, False to disable wifi guest access
263
        :param duration: Duration in minutes to enable the guest wifi. 0 is infinite.
264
        :return: True, if the state of the wifi guest access was successfully changed, otherwise False
265
        """
266
        self._logger.debug("Setting wifi guest access status.")
1✔
267
        wifi_guest = WifiGuestAccessSet()
1✔
268
        wifi_guest.enable = enable
1✔
269
        wifi_guest.duration = duration
1✔
270
        query = await self._async_post("WifiGuestAccessSet", content=wifi_guest.SerializeToString())
1✔
271
        response = WifiGuestAccessSetResponse()
1✔
272
        response.ParseFromString(await query.aread())
1✔
273
        return response.result == WifiResult.WIFI_SUCCESS
1✔
274

275
    @_feature("wifi1")
1✔
276
    async def async_get_wifi_neighbor_access_points(self) -> list[WifiNeighborAPsGet.NeighborAPInfo]:
1✔
277
        """
278
        Get wifi access point in the neighborhood asynchronously. This feature only works on devices, that announce the wifi1
279
        feature.
280

281
        :return: Visible access points in the neighborhood including connection rate data
282
        """
283
        self._logger.debug("Getting neighbored access points.")
1✔
284
        wifi_neighbor_aps = WifiNeighborAPsGet()
1✔
285
        response = await self._async_get("WifiNeighborAPsGet", timeout=30.0)
1✔
286
        wifi_neighbor_aps.ParseFromString(await response.aread())
1✔
287
        return list(wifi_neighbor_aps.neighbor_aps)
1✔
288

289
    @_feature("wifi1")
1✔
290
    async def async_start_wps(self) -> bool:
1✔
291
        """
292
        Start WPS push button configuration.
293

294
        :return: True, if the WPS was successfully started, otherwise False
295
        """
296
        self._logger.debug("Starting WPS.")
1✔
297
        wps = WifiWpsPbcStart()
1✔
298
        response = await self._async_get("WifiWpsPbcStart")
1✔
299
        wps.ParseFromString(await response.aread())
1✔
300
        return wps.result == WifiResult.WIFI_SUCCESS
1✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc