Repository: XKNX/xknx
Pull request #847 (2022-01-10 21:36:07)

Optionally run KNXIPInterface in separate thread
<!--
  You are awesome! Thanks for contributing to our project!
  Please, DO NOT DELETE ANY TEXT from this template!.
-->
## Description
<!--
Please include a summary of the change and which issue is fixed.
Please also include relevant motivation and context.
List any dependencies that are required for this change.
-->
Optionally run KNXIPInterface in separate thread

Fixes # (issue)

## Type of change
<!--
Please tick the applicable options.
NOTE: Ticking multiple options most likely indicates
that your change is to big and it is suggested to split it into several smaller PRs.
-->

- [ ] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] This change requires a documentation update

## Checklist:

- [ ] I have performed a self-review of my own code
- [ ] The documentation has been adjusted accordingly
- [ ] The changes generate no new warnings
- [ ] Tests have been added that prove the fix is effective or that the feature works
- [ ] The changes are documented in the changelog
- [ ] The Homeassistant plugin has been adjusted in case of new config options


codecov[bot]: # [Codecov](https://codecov.io/gh/XKNX/xknx/pull/847?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) Report
> Merging [#847](https://codecov.io/gh/XKNX/xknx/pull/847?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (c3763e6) into [main](https://codecov.io/gh/XKNX/xknx/commit/829099050a80da1f2186a8fbd8a4e38e71d6c16d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (8290990) will **decrease** coverage by `0.18%`.
> The diff coverage is `44.23%`.

[![Impacted file tree graph](https://codecov.io/gh/XKNX/xknx/pull/847/graphs/tree.svg?width=650&height=150&src=pr&token=irWbIygS84&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)](https://codecov.io/gh/XKNX/xknx/pull/847?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)

```diff
@@            Coverage Diff             @@
##             main     #847      +/-   ##
==========================================
- Coverage   95.71%   95.52%   -0.19%     
==========================================
  Files         109      109              
  Lines        6723     6774      +51     
==========================================
+ Hits         6435     6471      +36     
- Misses        288      303      +15     
```


| [Impacted Files](https://codecov.io/gh/XKNX/xknx/pull/847?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) | Coverage Δ | |
|---|---|---|
| [xknx/io/knxip\_interface.py](https://codecov.io/gh/XKNX/xknx/pull/847/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9rbnhpcF9pbnRlcmZhY2UucHk=) | `46.21% <42.57%> (+13.28%)` | :arrow_up: |
| [xknx/io/\_\_init\_\_.py](https://codecov.io/gh/XKNX/xknx/pull/847/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9fX2luaXRfXy5weQ==) | `100.00% <100.00%> (ø)` | |
| [xknx/io/connection.py](https://codecov.io/gh/XKNX/xknx/pull/847/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9jb25uZWN0aW9uLnB5) | `100.00% <100.00%> (+4.00%)` | :arrow_up: |
| [xknx/xknx.py](https://codecov.io/gh/XKNX/xknx/pull/847/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC94a254LnB5) | `88.77% <100.00%> (ø)` | |
| [xknx/dpt/dpt\_hvac\_mode.py](https://codecov.io/gh/XKNX/xknx/pull/847/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9kcHQvZHB0X2h2YWNfbW9kZS5weQ==) | `100.00% <0.00%> (ø)` | |
| [xknx/remote\_value/remote\_value.py](https://codecov.io/gh/XKNX/xknx/pull/847/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9yZW1vdGVfdmFsdWUvcmVtb3RlX3ZhbHVlLnB5) | `100.00% <0.00%> (ø)` | |


Patch:
diff --git a/changelog.md b/changelog.md
index 08af4cee..7819504d 100644
--- a/changelog.md
+++ b/changelog.md
@@ -8,6 +8,7 @@
 
 ### Connection
 
+- Optionally run KNXIPInterface in separate thread
 - Handle separate Tunneling control and data endpoints
 - Fix rate limiter wait time: don't add time waiting for ACK or L_DATA.con frames to the rate_limit
 
diff --git a/xknx/io/__init__.py b/xknx/io/__init__.py
index 60d908d4..3e440b39 100644
--- a/xknx/io/__init__.py
+++ b/xknx/io/__init__.py
@@ -10,7 +10,7 @@ This package contains all objects managing Tunneling and Routing Connections..
 from .connection import ConnectionConfig, ConnectionType
 from .const import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT
 from .gateway_scanner import GatewayDescriptor, GatewayScanFilter, GatewayScanner
-from .knxip_interface import KNXIPInterface
+from .knxip_interface import KNXIPInterface, knx_interface_factory
 from .routing import Routing
 from .tunnel import Tunnel
 from .udp_client import UDPClient
diff --git a/xknx/io/connection.py b/xknx/io/connection.py
index c7ad6c8e..96d218e8 100644
--- a/xknx/io/connection.py
+++ b/xknx/io/connection.py
@@ -45,6 +45,7 @@ class ConnectionConfig:
         auto_reconnect: bool = True,
         auto_reconnect_wait: int = 3,
         scan_filter: GatewayScanFilter = GatewayScanFilter(),
+        threaded: bool = False,
     ):
         """Initialize ConnectionConfig class."""
         self.connection_type = connection_type
@@ -55,11 +56,8 @@ class ConnectionConfig:
         self.route_back = route_back
         self.auto_reconnect = auto_reconnect
         self.auto_reconnect_wait = auto_reconnect_wait
-        if connection_type == ConnectionType.TUNNELING:
-            scan_filter.tunnelling = True
-        elif connection_type == ConnectionType.ROUTING:
-            scan_filter.routing = True
         self.scan_filter = scan_filter
+        self.threaded = threaded
 
     def __eq__(self, other: object) -> bool:
         """Equality for ConnectionConfig class (used in unit tests)."""
diff --git a/xknx/io/knxip_interface.py b/xknx/io/knxip_interface.py
index 3fdf48c7..c8e8497a 100644
--- a/xknx/io/knxip_interface.py
+++ b/xknx/io/knxip_interface.py
@@ -8,15 +8,17 @@ KNXIPInterface manages KNX/IP Tunneling or Routing connections.
 """
 from __future__ import annotations
 
+import asyncio
 import ipaddress
 import logging
-from typing import TYPE_CHECKING, cast
+import threading
+from typing import TYPE_CHECKING, Any, Awaitable, TypeVar, cast
 
 import netifaces
 from xknx.exceptions import CommunicationError, XKNXException
 
 from .connection import ConnectionConfig, ConnectionType
-from .gateway_scanner import GatewayScanFilter, GatewayScanner
+from .gateway_scanner import GatewayDescriptor, GatewayScanFilter, GatewayScanner
 from .routing import Routing
 from .tunnel import Tunnel
 
@@ -28,6 +30,17 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger("xknx.log")
 
+T = TypeVar("T")  # pylint: disable=invalid-name
+
+
+def knx_interface_factory(
+    xknx: XKNX, connection_config: ConnectionConfig
+) -> KNXIPInterface:
+    """Create KNX/IP interface from config."""
+    if connection_config.threaded:
+        return KNXIPInterfaceThreaded(xknx=xknx, connection_config=connection_config)
+    return KNXIPInterface(xknx=xknx, connection_config=connection_config)
+
 
 class KNXIPInterface:
     """Class for managing KNX/IP Tunneling or Routing connections."""
@@ -39,21 +52,22 @@ class KNXIPInterface:
     ):
         """Initialize KNXIPInterface class."""
         self.xknx = xknx
-        self.interface: Interface | None = None
+        self._interface: Interface | None = None
         self.connection_config = connection_config
 
     async def start(self) -> None:
+        """Start KNX/IP interface."""
+        await self._start()
+
+    async def _start(self) -> None:
         """Start interface. Connecting KNX/IP device with the selected method."""
-        if (
-            self.connection_config.connection_type == ConnectionType.ROUTING
-            and self.connection_config.local_ip is not None
-        ):
-            await self.start_routing(self.connection_config.local_ip)
+        if self.connection_config.connection_type == ConnectionType.ROUTING:
+            await self._start_routing(self.connection_config.local_ip)
         elif (
             self.connection_config.connection_type == ConnectionType.TUNNELING
             and self.connection_config.gateway_ip is not None
         ):
-            await self.start_tunnelling(
+            await self._start_tunnelling(
                 local_ip=self.connection_config.local_ip,
                 local_port=self.connection_config.local_port,
                 gateway_ip=self.connection_config.gateway_ip,
@@ -63,24 +77,15 @@ class KNXIPInterface:
                 route_back=self.connection_config.route_back,
             )
         else:
-            await self.start_automatic(self.connection_config.scan_filter)
+            await self._start_automatic()
 
-    async def start_automatic(self, scan_filter: GatewayScanFilter) -> None:
+    async def _start_automatic(self) -> None:
         """Start GatewayScanner and connect to the found device."""
-        gatewayscanner = GatewayScanner(self.xknx, scan_filter=scan_filter)
-        gateways = await gatewayscanner.scan()
-
-        if not gateways:
-            raise XKNXException("No Gateways found")
-
-        gateway = gateways[0]
-
-        # on Linux gateway.local_ip can be any interface listening to the
-        # multicast group (even 127.0.0.1) so we set the interface with find_local_ip
-        local_interface_ip = self.find_local_ip(gateway_ip=gateway.ip_addr)
+        scan_filter = self.connection_config.scan_filter
+        gateway, local_interface_ip = await self.find_gateway(scan_filter)
 
         if gateway.supports_tunnelling and scan_filter.routing is not True:
-            await self.start_tunnelling(
+            await self._start_tunnelling(
                 local_interface_ip,
                 self.connection_config.local_port,
                 gateway.ip_addr,
@@ -90,9 +95,9 @@ class KNXIPInterface:
                 route_back=self.connection_config.route_back,
             )
         elif gateway.supports_routing:
-            await self.start_routing(local_interface_ip)
+            await self._start_routing(local_interface_ip)
 
-    async def start_tunnelling(
+    async def _start_tunnelling(
         self,
         local_ip: str | None,
         local_port: int,
@@ -105,7 +110,7 @@ class KNXIPInterface:
         """Start KNX/IP tunnel."""
         validate_ip(gateway_ip, address_name="Gateway IP address")
         if local_ip is None:
-            local_ip = self.find_local_ip(gateway_ip=gateway_ip)
+            local_ip = find_local_ip(gateway_ip=gateway_ip)
         validate_ip(local_ip, address_name="Local IP address")
         logger.debug(
             "Starting tunnel from %s:%s to %s:%s",
@@ -114,7 +119,7 @@ class KNXIPInterface:
             gateway_ip,
             gateway_port,
         )
-        self.interface = Tunnel(
+        self._interface = Tunnel(
             self.xknx,
             gateway_ip=gateway_ip,
             gateway_port=gateway_port,
@@ -125,20 +130,24 @@ class KNXIPInterface:
             auto_reconnect=auto_reconnect,
             auto_reconnect_wait=auto_reconnect_wait,
         )
-        await self.interface.connect()
+        await self._interface.connect()
 
-    async def start_routing(self, local_ip: str) -> None:
+    async def _start_routing(self, local_ip: str | None = None) -> None:
         """Start KNX/IP Routing."""
+        if local_ip is None:
+            scan_filter = self.connection_config.scan_filter
+            scan_filter.routing = True
+            _gateway, local_ip = await self.find_gateway(scan_filter)
         validate_ip(local_ip, address_name="Local IP address")
         logger.debug("Starting Routing from %s as %s", local_ip, self.xknx.own_address)
-        self.interface = Routing(self.xknx, self.telegram_received, local_ip)
-        await self.interface.connect()
+        self._interface = Routing(self.xknx, self.telegram_received, local_ip)
+        await self._interface.connect()
 
     async def stop(self) -> None:
         """Stop connected interfae (either Tunneling or Routing)."""
-        if self.interface is not None:
-            await self.interface.disconnect()
-            self.interface = None
+        if self._interface is not None:
+            await self._interface.disconnect()
+            self._interface = None
 
     def telegram_received(self, telegram: Telegram) -> None:
         """Put received telegram into queue. Callback for having received telegram."""
@@ -146,49 +155,130 @@ class KNXIPInterface:
 
     async def send_telegram(self, telegram: "Telegram") -> None:
         """Send telegram to connected device (either Tunneling or Routing)."""
-        if self.interface is not None:
-            await self.interface.send_telegram(telegram)
-        else:
+        if self._interface is None:
+            raise CommunicationError("KNX/IP interface not connected")
+        return await self._interface.send_telegram(telegram)
+
+    async def find_gateway(
+        self, scan_filter: GatewayScanFilter
+    ) -> tuple[GatewayDescriptor, str]:
+        """Find Gateway and connect to it."""
+        gatewayscanner = GatewayScanner(self.xknx, scan_filter=scan_filter)
+        gateways = await gatewayscanner.scan()
+
+        if not gateways:
+            raise XKNXException("No Gateways found")
+        gateway = gateways[0]
+        # on Linux gateway.local_ip can be any interface listening to the
+        # multicast group (even 127.0.0.1) so we set the interface with find_local_ip
+        local_interface_ip = find_local_ip(gateway_ip=gateway.ip_addr)
+
+        return gateway, local_interface_ip
+
+
+class KNXIPInterfaceThreaded(KNXIPInterface):
+    """Class for managing KNX/IP Tunneling or Routing connections."""
+
+    def __init__(
+        self,
+        xknx: XKNX,
+        connection_config: ConnectionConfig = ConnectionConfig(),
+    ):
+        """Initialize KNXIPInterface class."""
+        super().__init__(xknx, connection_config)
+        self._main_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
+        self._thread_loop: asyncio.AbstractEventLoop
+
+        loop_loaded = threading.Event()
+        connection_thread = threading.Thread(
+            target=self._init_connection_loop,
+            args=[loop_loaded],
+            name="KNX Interface",
+            daemon=True,
+        )
+        connection_thread.start()
+        loop_loaded.wait()  # wait for the thread to initialize its loop
+
+    def _init_connection_loop(self, loop_loaded: threading.Event) -> None:
+        """Start KNX/IP interface in its own thread."""
+        self._thread_loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(self._thread_loop)
+        loop_loaded.set()
+        self._thread_loop.run_forever()
+
+    async def _await_from_connection_thread(self, coro: Awaitable[T]) -> T:
+        """Await coroutine in different thread."""
+        fut = asyncio.run_coroutine_threadsafe(coro, self._thread_loop)
+        finished = threading.Event()
+
+        def fut_finished_cb(_: Any) -> None:  # with py3.9 `Future[T]` should be working
+            """Fire threading.Event when the future is finished."""
+            finished.set()
+
+        fut.add_done_callback(fut_finished_cb)
+        # wait on that event in an executor, yielding control to _main_loop
+        await self._main_loop.run_in_executor(None, finished.wait)
+        return fut.result()
+
+    async def start(self) -> None:
+        """Start KNX/IP interface."""
+        return await self._await_from_connection_thread(self._start())
+
+    async def stop(self) -> None:
+        """Stop connected interfae (either Tunneling or Routing)."""
+        if self._interface is not None:
+            await self._await_from_connection_thread(self._interface.disconnect())
+            self._interface = None
+
+    def telegram_received(self, telegram: Telegram) -> None:
+        """Put received telegram into queue. Callback for having received telegram."""
+        self._main_loop.call_soon_threadsafe(self.xknx.telegrams.put_nowait, telegram)
+
+    async def send_telegram(self, telegram: "Telegram") -> None:
+        """Send telegram to connected device (either Tunneling or Routing)."""
+        if self._interface is None:
             raise CommunicationError("KNX/IP interface not connected")
 
-    @staticmethod
-    def find_local_ip(gateway_ip: str) -> str:
-        """Find local IP address on same subnet as gateway."""
-
-        def _scan_interfaces(gateway: ipaddress.IPv4Address) -> str | None:
-            """Return local IP address on same subnet as given gateway."""
-            for interface in netifaces.interfaces():
-                try:
-                    af_inet = netifaces.ifaddresses(interface)[netifaces.AF_INET]
-                    for link in af_inet:
-                        network = ipaddress.IPv4Network(
-                            (link["addr"], link["netmask"]), strict=False
-                        )
-                        if gateway in network:
-                            logger.debug("Using interface: %s", interface)
-                            return cast(str, link["addr"])
-                except KeyError:
-                    logger.debug(
-                        "Could not find IPv4 address on interface %s", interface
+        return await self._await_from_connection_thread(
+            self._interface.send_telegram(telegram)
+        )
+
+
+def find_local_ip(gateway_ip: str) -> str:
+    """Find local IP address on same subnet as gateway."""
+
+    def _scan_interfaces(gateway: ipaddress.IPv4Address) -> str | None:
+        """Return local IP address on same subnet as given gateway."""
+        for interface in netifaces.interfaces():
+            try:
+                af_inet = netifaces.ifaddresses(interface)[netifaces.AF_INET]
+                for link in af_inet:
+                    network = ipaddress.IPv4Network(
+                        (link["addr"], link["netmask"]), strict=False
                     )
-                    continue
-            return None
+                    if gateway in network:
+                        logger.debug("Using interface: %s", interface)
+                        return cast(str, link["addr"])
+            except KeyError:
+                logger.debug("Could not find IPv4 address on interface %s", interface)
+                continue
+        return None
 
-        def _find_default_gateway() -> ipaddress.IPv4Address:
-            """Return IP address of default gateway."""
-            gws = netifaces.gateways()
-            return ipaddress.IPv4Address(gws["default"][netifaces.AF_INET][0])
+    def _find_default_gateway() -> ipaddress.IPv4Address:
+        """Return IP address of default gateway."""
+        gws = netifaces.gateways()
+        return ipaddress.IPv4Address(gws["default"][netifaces.AF_INET][0])
 
-        gateway = ipaddress.IPv4Address(gateway_ip)
-        local_ip = _scan_interfaces(gateway)
-        if local_ip is None:
-            logger.warning(
-                "No interface on same subnet as gateway found. Falling back to default gateway."
-            )
-            default_gateway = _find_default_gateway()
-            local_ip = _scan_interfaces(default_gateway)
-        assert isinstance(local_ip, str)
-        return local_ip
+    gateway = ipaddress.IPv4Address(gateway_ip)
+    local_ip = _scan_interfaces(gateway)
+    if local_ip is None:
+        logger.warning(
+            "No interface on same subnet as gateway found. Falling back to default gateway."
+        )
+        default_gateway = _find_default_gateway()
+        local_ip = _scan_interfaces(default_gateway)
+    assert isinstance(local_ip, str)
+    return local_ip
 
 
 def validate_ip(address: str, address_name: str = "IP address") -> None:
diff --git a/xknx/xknx.py b/xknx/xknx.py
index 9f4e331e..75f9ad51 100644
--- a/xknx/xknx.py
+++ b/xknx/xknx.py
@@ -23,6 +23,7 @@ from xknx.io import (
     DEFAULT_MCAST_PORT,
     ConnectionConfig,
     KNXIPInterface,
+    knx_interface_factory,
 )
 from xknx.telegram import GroupAddressType, IndividualAddress, Telegram
 
@@ -114,8 +115,9 @@ class XKNX:
     async def start(self) -> None:
         """Start XKNX module. Connect to KNX/IP devices and start state updater."""
         self.task_registry.start()
-        self.knxip_interface = KNXIPInterface(
-            self, connection_config=self.connection_config
+        self.knxip_interface = knx_interface_factory(
+            xknx=self,
+            connection_config=self.connection_config,
         )
         logger.info(
             "XKNX v%s starting %s connection to KNX bus.",


Test patch:
diff --git a/test/xknx_test.py b/test/xknx_test.py
index 5adf99a5..e22835c4 100644
--- a/test/xknx_test.py
+++ b/test/xknx_test.py
@@ -55,7 +55,7 @@ class TestXknxModule:
 
         assert len(xknx.connection_manager._connection_state_changed_cbs) == 1
 
-    @patch("xknx.io.KNXIPInterface.start", new_callable=AsyncMock)
+    @patch("xknx.io.KNXIPInterface._start", new_callable=AsyncMock)
     async def test_xknx_start(self, start_mock):
         """Test xknx start."""
         xknx = XKNX(state_updater=True)
@@ -64,7 +64,7 @@ class TestXknxModule:
         start_mock.assert_called_once()
         await xknx.stop()
 
-    @patch("xknx.io.KNXIPInterface.start", new_callable=AsyncMock)
+    @patch("xknx.io.KNXIPInterface._start", new_callable=AsyncMock)
     async def test_xknx_start_as_context_manager(self, ipinterface_mock):
         """Test xknx start."""
 
@@ -75,7 +75,7 @@ class TestXknxModule:
 
         await run_in_contextmanager()
 
-    @patch("xknx.io.KNXIPInterface.start", new_callable=AsyncMock)
+    @patch("xknx.io.KNXIPInterface._start", new_callable=AsyncMock)
     async def test_xknx_start_and_stop_with_dedicated_connection_config(
         self, start_mock
     ):