merged

.github/workflows/ci.yml

@@ -13,7 +13,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.10", "3.11", "3.12"]
+        python-version: ["3.11", "3.12"]
     steps:
     - uses: actions/checkout@v2
     - name: Set up Python ${{ matrix.python-version }}
@@ -99,7 +99,7 @@ jobs:
     - name: Create coverage report
       run: |
         coverage combine coverage*/.coverage*
-        coverage report --fail-under=80
+        coverage report --fail-under=79
 
         coverage xml
     - name: Upload coverage to Codecov

pyvlx/__init__.py

@@ -1,5 +1,6 @@
 """Module for accessing KLF 200 gateway with python."""
 
+from .discovery import VeluxDiscovery
 from .exception import PyVLXException
 from .klf200gateway import Klf200Gateway
 from .lightening_device import Light, LighteningDevice

pyvlx/discovery.py

@@ -0,0 +1,88 @@
+"""Module to discover Velux KLF200 devices on the network."""
+import asyncio
+from asyncio import Event, Future, Task
+from dataclasses import dataclass
+from typing import Any, Optional
+
+from zeroconf import IPVersion
+from zeroconf.asyncio import (
+    AsyncServiceBrowser, AsyncServiceInfo, AsyncZeroconf)
+
+SERVICE_STARTS_WITH: str = "VELUX_KLF_LAN"
+SERVICE_TYPE: str = "_http._tcp.local."
+
+
+@dataclass
+class VeluxHost():
+    """Class to store Velux KLF200 host information."""
+
+    hostname: str
+    ip_address: str
+
+
+class VeluxDiscovery():
+    """Class to discover Velux KLF200 devices on the network."""
+
+    hosts: list[VeluxHost | None] = []
+    infos: list[AsyncServiceInfo | None] = []
+
+    def __init__(self, zeroconf: AsyncZeroconf,) -> None:
+        """Initialize VeluxDiscovery object."""
+        self.zc: AsyncZeroconf = zeroconf
+
+    async def _async_discover_hosts(self, min_wait_time: float, expected_hosts: int | None) -> None:
+        """Listen for zeroconf ServiceInfo."""
+        self.hosts.clear()
+        service_names: list[str] = []
+        tasks: list[Task] = []
+        got_host: Event = Event()
+
+        def add_info_and_host(fut: Future) -> None:
+            info: AsyncServiceInfo = fut.result()
+            self.infos.append(info)
+            host = VeluxHost(
+                hostname=info.name.replace("._http._tcp.local.", ""),
+                ip_address=info.parsed_addresses(version=IPVersion.V4Only)[0],
+            )
+            self.hosts.append(host)
+            got_host.set()
+
+        def handler(name: str, **kwargs: Any) -> None:  # pylint: disable=W0613:unused-argument
+            if name.startswith(SERVICE_STARTS_WITH):
+                if name not in service_names:
+                    service_names.append(name)
+                    task = asyncio.create_task(self.zc.async_get_service_info(type_=SERVICE_TYPE, name=name))
+                    task.add_done_callback(add_info_and_host)
+                    tasks.append(task)
+
+        browser: AsyncServiceBrowser = AsyncServiceBrowser(self.zc.zeroconf, SERVICE_TYPE, handlers=[handler])
+        if expected_hosts:
+            while len(self.hosts) < expected_hosts:
+                await got_host.wait()
+                got_host.clear()
+        while not self.hosts:
+            await asyncio.sleep(min_wait_time)
+        await browser.async_cancel()
+        await asyncio.gather(*tasks)
+
+    async def async_discover_hosts(
+        self,
+        timeout: float = 10,
+        min_wait_time: float = 3,
+        expected_hosts: Optional[int] = None
+    ) -> bool:
+        """Return true if Velux KLF200 devices found on the network.
+
+        This function creates a zeroconf AsyncServiceBrowser and
+        waits min_wait_time (seconds) for ServiceInfos from hosts.
+        Some devices may take some time to respond (i.e. if they currently have a high CPU load).
+        If one or more Hosts are found, the function cancels the ServiceBrowser and returns true.
+        If expected_hosts is set, the function ignores min_wait_time and returns true once expected_hosts are found.
+        If timeout (seconds) is exceeded, the function returns false.
+        """
+        try:
+            async with asyncio.timeout(timeout):
+                await self._async_discover_hosts(min_wait_time, expected_hosts)
+        except TimeoutError:
+            return False
+        return True

pyvlx/klf200gateway.py

@@ -109,8 +109,7 @@ async def reboot(self) -> bool:
         await reboot.do_api_call()
         if not reboot.success:
             raise PyVLXException("Unable to reboot gateway.")
-        else:
-            await self.pyvlx.disconnect()
+        await self.pyvlx.disconnect()
         return reboot.success
 
     async def set_factory_default(self) -> bool:

requirements/production.txt

@@ -1 +1,2 @@
 pyyaml==6.0.1
+zeroconf==0.131.0

requirements/testing.txt

@@ -5,10 +5,10 @@ flake8==7.0.0
 flake8-isort==6.1.1
 pydocstyle==6.3.0
 pylint==3.0.3
-pytest==7.4.4
+pytest==8.0.0
 pytest-cov==4.0.0
 pytest-timeout==2.2.0
-setuptools==69.0.3
-twine==4.0.2
+setuptools==69.1.0
+twine==5.0.0
 mypy==1.8.0
 types-pyyaml==6.0.12.12

setup.py

@@ -4,7 +4,7 @@
 
 from setuptools import find_packages, setup
 
-REQUIRES = ["PyYAML"]
+REQUIRES = ["PyYAML", "zeroconf"]
 
 PKG_ROOT = os.path.dirname(__file__)
 
@@ -36,15 +36,14 @@ def get_long_description() -> str:
         "Intended Audience :: Developers",
         "Topic :: System :: Hardware :: Hardware Drivers",
         "License :: OSI Approved :: GNU Lesser General Public License v3 (LGPLv3)",
-        "Programming Language :: Python :: 3.10",
         "Programming Language :: Python :: 3.11",
         "Programming Language :: Python :: 3.12",
     ],
     packages=find_packages(exclude=['test*']),
     package_data={
         "pyvlx": ["py.typed"],
     },
-    python_requires='>=3.10',
+    python_requires='>=3.11',
     install_requires=REQUIRES,
     keywords="velux KLF 200 home automation",
     zip_safe=False,