Merge pull request #10 from 2Fake/development

Development

.github/workflows/pythonpublish.yml

@@ -0,0 +1,26 @@
+name: Upload Python Package
+
+on:
+  release:
+    types: [created]
+
+jobs:
+  deploy:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v2
+    - name: Set up Python
+      uses: actions/setup-python@v1
+      with:
+        python-version: "3.7"
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install setuptools wheel twine
+    - name: Build and publish
+      env:
+        TWINE_USERNAME: ${{ secrets.PYPI_USERNAME }}
+        TWINE_PASSWORD: ${{ secrets.PYPI_PASSWORD }}
+      run: |
+        python setup.py sdist bdist_wheel
+        twine upload dist/*

README.md

@@ -1,6 +1,7 @@
 # devolo PLC API
 
 [![GitHub Workflow Status](https://img.shields.io/github/workflow/status/2Fake/devolo_plc_api/Python%20package)](https://github.com/2Fake/devolo_plc_api/actions?query=workflow%3A%22Python+package%22)
+[![PyPI - Downloads](https://img.shields.io/pypi/dd/devolo-plc-api)](https://pypi.org/project/devolo-plc-api/)
 [![Code Climate maintainability](https://img.shields.io/codeclimate/maintainability/2Fake/devolo_plc_api)](https://codeclimate.com/github/2Fake/devolo_plc_api)
 [![Coverage Status](https://coveralls.io/repos/github/2Fake/devolo_plc_api/badge.svg?branch=development)](https://coveralls.io/github/2Fake/devolo_plc_api?branch=development)
 
@@ -59,4 +60,23 @@ python setup.py test
 
 ## Usage
 
-All features we currently support are shown in our [example.py](https://github.com/2Fake/devolo_plc_api/blob/master/example.py)
+All features we currently support are shown in our examples. If you want to use the package asynchronously, please check [example_async.py](https://github.com/2Fake/devolo_plc_api/blob/master/example_async.py). If you want to use it synchronously, please check [example_sync.py](https://github.com/2Fake/devolo_plc_api/blob/master/example_sync.py).
+
+## Supported device
+
+The following devolo devices were queried with at least one call to verify functionality:
+
+* Magic 2 WiFi next
+* Magic 2 WiFi 2-1
+* Magic 2 LAN triple
+* Magic 2 DinRail
+* Magic 2 LAN 1-1
+* Magic 1 WiFi mini
+* Magic 1 WiFi 2-1
+* Magic 1 LAN 1-1
+* dLAN 1200+ WiFi ac
+* dLAN 550+ Wifi
+* dLAN 550 WiFi
+* dLAN 500 WiFi
+
+However, other devices might work, some might have a limited functionality. Also firmware version will matter. If you discover something weird, [we want to know](https://github.com/2Fake/devolo_plc_api/issues).

devolo_plc_api/__init__.py

@@ -1 +1 @@
-__version__ = "0.1.0"
+__version__ = "0.2.0"

devolo_plc_api/device.py

@@ -83,6 +83,7 @@ async def _get_device_info(self):
         self.product = self._info[service_type].get("Product", "")
 
         self.device = DeviceApi(ip=self.ip,
+                                port=self._info[service_type]['Port'],
                                 session=self._session,
                                 path=self._info[service_type]['Path'],
                                 version=self._info[service_type]['Version'],
@@ -101,6 +102,7 @@ async def _get_plcnet_info(self):
         self.technology = self._info[service_type].get("PlcTechnology", "")
 
         self.plcnet = PlcNetApi(ip=self.ip,
+                                port=self._info[service_type]['Port'],
                                 session=self._session,
                                 path=self._info[service_type]['Path'],
                                 version=self._info[service_type]['Version'],
@@ -121,6 +123,8 @@ def _state_change(self, zeroconf: Zeroconf, service_type: str, name: str, state_
                 self.ip in [socket.inet_ntoa(address) for address in service_info.addresses]:
             self._logger.debug(f"Adding service info of {service_type}")
 
+            self._info[service_type]['Port'] = service_info.port
+
             # The answer is a byte string, that concatenates key-value pairs with their length as two byte hex value.
             total_length = len(service_info.text)
             offset = 0

devolo_plc_api/device_api/deviceapi.py

@@ -7,6 +7,7 @@
 from ..exceptions.feature import FeatureNotSupported
 from . import devolo_idl_proto_deviceapi_wifinetwork_pb2
 from . import devolo_idl_proto_deviceapi_ledsettings_pb2
+from . import devolo_idl_proto_deviceapi_updatefirmware_pb2
 
 
 class DeviceApi(Protobuf):
@@ -20,13 +21,13 @@ class DeviceApi(Protobuf):
     :param features: Feature, the device has
     """
 
-    def __init__(self, ip: str, session: Client, path: str, version: str, features: str, password: str):
+    def __init__(self, ip: str, port: int, session: Client, path: str, version: str, features: str, password: str):
         self._ip = ip
-        self._port = 14791
+        self._port = port
         self._session = session
         self._path = path
         self._version = version
-        self._features = features.split(",") if features else []
+        self._features = features.split(",") if features else ['reset', 'update', 'led', 'intmtg']
         self._user = "devolo"
         self._password = password
         self._logger = logging.getLogger(self.__class__.__name__)
@@ -45,24 +46,37 @@ def wrapper(self, *args, **kwargs):
 
 
     @_feature("led")
-    async def async_get_led_setting(self):
-        """ Get LED setting asynchronously. """
+    async def async_get_led_setting(self) -> dict:
+        """
+        Get LED setting asynchronously. This feature only works on devices, that announce the led feature.
+
+        return: LED settings
+        """
         led_setting = devolo_idl_proto_deviceapi_ledsettings_pb2.LedSettingsGet()
         response = await self._async_get("LedSettingsGet")
         led_setting.FromString(await response.aread())
         return self._message_to_dict(led_setting)
 
     @_feature("led")
     def get_led_setting(self):
-        """ Get LED setting synchronously. """
+        """
+        Get LED setting synchronously. This feature only works on devices, that announce the led feature.
+
+        return: LED settings
+        """
         led_setting = devolo_idl_proto_deviceapi_ledsettings_pb2.LedSettingsGet()
         response = self._get("LedSettingsGet")
         led_setting.FromString(response.read())
         return self._message_to_dict(led_setting)
 
     @_feature("led")
     async def async_set_led_setting(self, enable: bool) -> bool:
-        """ Set LED setting asynchronously. """
+        """
+        Set LED setting asynchronously. This feature only works on devices, that announce the led feature.
+
+        :param enable: True to enable the LEDs, False to disable the LEDs
+        :return: True, if LED state was successfully changed, otherwise False
+        """
         led_setting = devolo_idl_proto_deviceapi_ledsettings_pb2.LedSettingsSet()
         led_setting.state = int(not enable)
         query = await self._async_post("LedSettingsSet", data=led_setting.SerializeToString())
@@ -72,33 +86,105 @@ async def async_set_led_setting(self, enable: bool) -> bool:
 
     @_feature("led")
     def set_led_setting(self, enable: bool) -> bool:
-        """ Set LED setting synchronously. """
+        """
+        Set LED setting synchronously. This feature only works on devices, that announce the led feature.
+
+        :param enable: True to enable the LEDs, False to disable the LEDs
+        :return: True, if LED state was successfully changed, otherwise False
+        """
         led_setting = devolo_idl_proto_deviceapi_ledsettings_pb2.LedSettingsSet()
         led_setting.state = int(not enable)
         query = self._post("LedSettingsSet", data=led_setting.SerializeToString())
         response = devolo_idl_proto_deviceapi_ledsettings_pb2.LedSettingsSetResponse()
         response.FromString(query.read())
         return bool(not response.result)
 
+
+    @_feature("update")
+    async def async_check_firmware_available(self) -> dict:
+        """
+        Check asynchronously, if a firmware update is available for the device.
+
+        :return: Result and new firmware version, if newer one is available
+        """
+        update_firmware_check = devolo_idl_proto_deviceapi_updatefirmware_pb2.UpdateFirmwareCheck()
+        response = await self._async_get("UpdateFirmwareCheck")
+        update_firmware_check.ParseFromString(await response.aread())
+        return self._message_to_dict(update_firmware_check)
+
+    @_feature("update")
+    def check_firmware_available(self) -> dict:
+        """
+        Check synchronously, if a firmware update is available for the device.
+
+        :return: Result and new firmware version, if newer one is available
+        """
+        update_firmware_check = devolo_idl_proto_deviceapi_updatefirmware_pb2.UpdateFirmwareCheck()
+        response = self._get("UpdateFirmwareCheck")
+        update_firmware_check.ParseFromString(response.read())
+        return self._message_to_dict(update_firmware_check)
+
+    @_feature("update")
+    async def async_start_firmware_update(self) -> bool:
+        """
+        Start firmware update asynchronously, if a firmware update is available for the device. Important: The response does
+        not tell you anything about the success of the update itself.
+
+        :return: True, if the firmware update was started, False if there is no update
+        """
+        update_firmware = devolo_idl_proto_deviceapi_updatefirmware_pb2.UpdateFirmwareStart()
+        response = await self._async_get("UpdateFirmwareStart")
+        update_firmware.FromString(await response.aread())
+        return bool(not update_firmware.result)
+
+    @_feature("update")
+    def start_firmware_update(self) -> bool:
+        """
+        Start firmware update synchronously, if a firmware update is available for the device. Important: The response does
+        not tell you anything about the success of the update itself.
+
+        :return: True, if the firmware update was started, False if there is no update
+        """
+        update_firmware = devolo_idl_proto_deviceapi_updatefirmware_pb2.UpdateFirmwareStart()
+        response = self._get("UpdateFirmwareStart")
+        update_firmware.FromString(response.read())
+        return bool(not update_firmware.result)
+
+
     @_feature("wifi1")
     async def async_get_wifi_connected_station(self) -> dict:
-        """ Get wifi stations connected to the device asynchronously. """
+        """
+        Get wifi stations connected to the device asynchronously. This feature only works on devices, that announce the wifi1
+        feature.
+
+        :return: All connected wifi stations including connection rate data
+        """
         wifi_connected_proto = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiConnectedStationsGet()
         response = await self._async_get("WifiConnectedStationsGet")
         wifi_connected_proto.ParseFromString(await response.aread())
         return self._message_to_dict(wifi_connected_proto)
 
     @_feature("wifi1")
     def get_wifi_connected_station(self) -> dict:
-        """ Get wifi stations connected to the device synchronously. """
+        """
+        Get wifi stations connected to the device synchronously. This feature only works on devices, that announce the wifi1
+        feature.
+
+        :return: All connected wifi stations including connection rate data
+        """
         wifi_connected_proto = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiConnectedStationsGet()
         response = self._get("WifiConnectedStationsGet")
         wifi_connected_proto.ParseFromString(response.read())
         return self._message_to_dict(wifi_connected_proto)
 
     @_feature("wifi1")
     async def async_get_wifi_guest_access(self) -> dict:
-        """ Get details about wifi guest access asynchronously. """
+        """
+        Get details about wifi guest access asynchronously. This feature only works on devices, that announce the wifi1
+        feature.
+
+        :return: Details about the wifi guest access
+        """
         self._logger.debug("Getting wifi guest access")
         wifi_guest_proto = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiGuestAccessGet()
         response = await self._async_get("WifiGuestAccessGet")
@@ -107,7 +193,12 @@ async def async_get_wifi_guest_access(self) -> dict:
 
     @_feature("wifi1")
     def get_wifi_guest_access(self) -> dict:
-        """ Get details about wifi guest access synchronously. """
+        """
+        Get details about wifi guest access synchronously. This feature only works on devices, that announce the wifi1
+        feature.
+
+        :return: Details about the wifi guest access
+        """
         self._logger.debug("Getting wifi guest access")
         wifi_guest_proto = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiGuestAccessGet()
         response = self._get("WifiGuestAccessGet")
@@ -116,7 +207,12 @@ def get_wifi_guest_access(self) -> dict:
 
     @_feature("wifi1")
     async def async_set_wifi_guest_access(self, enable: bool) -> bool:
-        """ Enable wifi guest access asynchronously. """
+        """
+        Enable wifi guest access asynchronously. This feature only works on devices, that announce the wifi1 feature.
+
+        :param enable: True to enable, False to disable wifi guest access
+        :return: True, if the state of the wifi guest access was successfully changed, otherwise False
+        """
         wifi_guest_proto = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiGuestAccessSet()
         wifi_guest_proto.enable = enable
         query = await self._async_post("WifiGuestAccessSet", data=wifi_guest_proto.SerializeToString())
@@ -126,7 +222,12 @@ async def async_set_wifi_guest_access(self, enable: bool) -> bool:
 
     @_feature("wifi1")
     def set_wifi_guest_access(self, enable: bool) -> bool:
-        """ Enable wifi guest access synchronously. """
+        """
+        Enable wifi guest access synchronously. This feature only works on devices, that announce the wifi1 feature.
+
+        :param enable: True to enable, False to disable wifi guest access
+        :return: True, if the state of the wifi guest access was successfully changed, otherwise False
+        """
         wifi_guest_proto = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiGuestAccessSet()
         wifi_guest_proto.enable = enable
         query = self._post("WifiGuestAccessSet", data=wifi_guest_proto.SerializeToString())
@@ -136,30 +237,76 @@ def set_wifi_guest_access(self, enable: bool) -> bool:
 
     @_feature("wifi1")
     async def async_get_wifi_neighbor_access_points(self) -> dict:
-        """ Get wifi access point in the neighborhood asynchronously. """
+        """
+        Get wifi access point in the neighborhood asynchronously. This feature only works on devices, that announce the wifi1
+        feature.
+
+        :return: Visible access points in the neighborhood including connection rate data
+        """
         wifi_neighbor_aps = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiNeighborAPsGet()
         response = await self._async_get("WifiNeighborAPsGet", timeout=15.0)
         wifi_neighbor_aps.ParseFromString(await response.aread())
         return self._message_to_dict(wifi_neighbor_aps)
 
     @_feature("wifi1")
     def get_wifi_neighbor_access_points(self) -> dict:
-        """ Get wifi access point in the neighborhood synchronously. """
+        """
+        Get wifi access point in the neighborhood synchronously. This feature only works on devices, that announce the wifi1
+        feature.
+
+        :return: Visible access points in the neighborhood including connection rate data
+        """
         wifi_neighbor_aps = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiNeighborAPsGet()
         response = self._get("WifiNeighborAPsGet", timeout=15.0)
         wifi_neighbor_aps.ParseFromString(response.read())
         return self._message_to_dict(wifi_neighbor_aps)
 
     @_feature("wifi1")
     async def async_get_wifi_repeated_access_points(self):
+        """
+        Get repeated wifi access point asynchronously. This feature only works on repeater devices, that announce the wifi1
+        feature.
+
+        :return: Repeated access points in the neighborhood including connection rate data
+        """
         wifi_connected_proto = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiRepeatedAPsGet()
         response = await self._async_get("WifiRepeatedAPsGet")
         wifi_connected_proto.ParseFromString(await response.aread())
         return self._message_to_dict(wifi_connected_proto)
 
     @_feature("wifi1")
     def get_wifi_repeated_access_points(self):
+        """
+        Get repeated wifi access point synchronously. This feature only works on repeater devices, that announce the wifi1
+        feature.
+
+        :return: Repeated access points in the neighborhood including connection rate data
+        """
         wifi_connected_proto = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiRepeatedAPsGet()
         response = self._get("WifiRepeatedAPsGet")
         wifi_connected_proto.ParseFromString(response.read())
         return self._message_to_dict(wifi_connected_proto)
+
+    @_feature("wifi1")
+    async def async_start_wps(self):
+        """
+        Start WPS push button configuration.
+
+        :return: True, if the WPS was successfully started, otherwise False
+        """
+        wps_proto = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiWpsPbcStart()
+        response = await self._async_get("WifiWpsPbcStart")
+        wps_proto.FromString(await response.aread())
+        return bool(not wps_proto.result)
+
+    @_feature("wifi1")
+    def start_wps(self):
+        """
+        Start WPS push button configuration.
+
+        :return: True, if the WPS was successfully started, otherwise False
+        """
+        wps_proto = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiWpsPbcStart()
+        response = self._get("WifiWpsPbcStart")
+        wps_proto.FromString(response.read())
+        return bool(not wps_proto.result)