Enhance Interface ABC with common connection management

xknx/core/connection_manager.py

@@ -88,7 +88,11 @@ def _connection_state_changed(
             return
 
         self._state = state
-        self.connection_type = connection_type
+        self.connection_type = (
+            XknxConnectionType.NOT_CONNECTED
+            if state == XknxConnectionState.DISCONNECTED
+            else connection_type
+        )
         if state == XknxConnectionState.CONNECTED:
             self.connected.set()
             self._reset_counters()

xknx/io/interface.py

@@ -10,20 +10,29 @@
 
 from abc import ABC, abstractmethod
 from collections.abc import Callable
+from typing import TYPE_CHECKING, ClassVar
 
 from xknx.cemi import CEMIFrame
+from xknx.core import XknxConnectionState, XknxConnectionType
+from xknx.telegram import IndividualAddress
 
 from .transport.ip_transport import KNXIPTransport
 
+if TYPE_CHECKING:
+    from xknx.xknx import XKNX
+
 CEMIBytesCallbackType = Callable[[bytes], None]
 
 
 class Interface(ABC):
     """Abstract base class for KNX/IP connections."""
 
-    __slots__ = ("transport",)
+    __slots__ = ("cemi_received_callback", "transport", "xknx")
 
+    connection_type: ClassVar[XknxConnectionType]
+    cemi_received_callback: CEMIBytesCallbackType
     transport: KNXIPTransport
+    xknx: XKNX
 
     @abstractmethod
     async def connect(self) -> None:
@@ -36,3 +45,13 @@ async def disconnect(self) -> None:
     @abstractmethod
     async def send_cemi(self, cemi: CEMIFrame) -> None:
         """Send CEMIFrame to KNX bus."""
+
+    def connection_state_changed(self, state: XknxConnectionState) -> None:
+        """Update connection state via connection manager."""
+        self.xknx.connection_manager.connection_state_changed(
+            state, self.connection_type
+        )
+
+    def _set_individual_address(self, address: IndividualAddress) -> None:
+        """Set current individual address."""
+        self.xknx.current_address = address

xknx/io/routing.py

@@ -11,7 +11,7 @@
 from contextlib import asynccontextmanager
 import logging
 import random
-from typing import TYPE_CHECKING, Final
+from typing import TYPE_CHECKING, ClassVar, Final
 
 from xknx.cemi import CEMIFrame, CEMIMessageCode
 from xknx.core import XknxConnectionState, XknxConnectionType
@@ -147,15 +147,13 @@ class Routing(Interface):
 
     __slots__ = (
         "_flow_control",
-        "cemi_received_callback",
         "individual_address",
         "local_ip",
         "multicast_group",
         "multicast_port",
-        "xknx",
     )
 
-    connection_type = XknxConnectionType.ROUTING
+    connection_type: ClassVar[XknxConnectionType] = XknxConnectionType.ROUTING
     transport: UDPTransport
 
     def __init__(
@@ -202,10 +200,8 @@ def _init_transport(self) -> None:
 
     async def connect(self) -> None:
         """Start routing."""
-        self.xknx.current_address = self.individual_address
-        self.xknx.connection_manager.connection_state_changed(
-            XknxConnectionState.CONNECTING, self.connection_type
-        )
+        self.current_address = self.individual_address
+        self.connection_state_changed(XknxConnectionState.CONNECTING)
         try:
             await self.transport.connect()
         except OSError as ex:
@@ -214,20 +210,16 @@ async def connect(self) -> None:
                 type(ex).__name__,
                 ex,
             )
-            self.xknx.connection_manager.connection_state_changed(
-                XknxConnectionState.DISCONNECTED
-            )
+            self.connection_state_changed(XknxConnectionState.DISCONNECTED)
             # close udp transport to prevent open file descriptors
             self.transport.stop()
             raise CommunicationError("Routing could not be started") from ex
-        self.xknx.connection_manager.connection_state_changed(
-            XknxConnectionState.CONNECTED, self.connection_type
-        )
+        self.connection_state_changed(XknxConnectionState.CONNECTED)
 
     async def disconnect(self) -> None:
         """Stop routing."""
         self.transport.stop()
-        self.xknx.connection_manager.connection_state_changed(
+        self.connection_state_changed(
             XknxConnectionState.DISCONNECTED
         )
         self._flow_control.cancel()
@@ -290,7 +282,7 @@ class SecureRouting(Routing):
         "latency_ms",
     )
 
-    connection_type = XknxConnectionType.ROUTING_SECURE
+    connection_type: ClassVar[XknxConnectionType] = XknxConnectionType.ROUTING_SECURE
     transport: SecureGroup
 
     def __init__(

xknx/io/tunnel.py

@@ -11,7 +11,7 @@
 from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 import logging
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, ClassVar
 
 from xknx.cemi import CEMIFrame
 from xknx.core import XknxConnectionState, XknxConnectionType
@@ -55,14 +55,12 @@ class _Tunnel(Interface):
         "_src_address",
         "auto_reconnect",
         "auto_reconnect_wait",
-        "cemi_received_callback",
         "communication_channel",
         "local_hpai",
         "sequence_number",
-        "xknx",
     )
 
-    connection_type: XknxConnectionType
+    connection_type: ClassVar[XknxConnectionType]
     transport: KNXIPTransport
 
     def __init__(
@@ -119,9 +117,7 @@ async def connect(self) -> None:
 
         Raise CommunicationError when not successful.
         """
-        self.xknx.connection_manager.connection_state_changed(
-            XknxConnectionState.CONNECTING, self.connection_type
-        )
+        self.connection_state_changed(XknxConnectionState.CONNECTING)
         try:
             await self.transport.connect()
             await self.setup_tunnel()
@@ -132,19 +128,15 @@ async def connect(self) -> None:
                 type(ex).__name__,
                 ex,
             )
-            self.xknx.connection_manager.connection_state_changed(
-                XknxConnectionState.DISCONNECTED
-            )
+            self.connection_state_changed(XknxConnectionState.DISCONNECTED)
             # close transport to prevent open file descriptors
             self.transport.stop()
             raise CommunicationError(
                 "Tunnel connection could not be established"
             ) from ex
 
         self._tunnel_established()
-        self.xknx.connection_manager.connection_state_changed(
-            XknxConnectionState.CONNECTED, self.connection_type
-        )
+        self.connection_state_changed(XknxConnectionState.CONNECTED)
 
     def _tunnel_established(self) -> None:
         """Set up interface when the tunnel is ready."""
@@ -209,7 +201,7 @@ async def _reconnect(self) -> None:
     def _prepare_disconnect(self) -> None:
         """Prepare for disconnect. Stop tunnel related tasks and set connection state."""
         self.stop_heartbeat()
-        self.xknx.connection_manager.connection_state_changed(
+        self.connection_state_changed(
             XknxConnectionState.DISCONNECTED
         )
 
@@ -251,7 +243,7 @@ async def _connect_request(self) -> bool:
             )
             # Use the individual address provided by the tunnelling server
             self._src_address = connect.crd.individual_address or IndividualAddress(0)
-            self.xknx.current_address = self._src_address
+            self.current_address = self._src_address
             logger.debug(
                 "Tunnel established. communication_channel=%s, address=%s",
                 connect.communication_channel,
@@ -456,7 +448,7 @@ class UDPTunnel(_Tunnel):
         "route_back",
     )
 
-    connection_type = XknxConnectionType.TUNNEL_UDP
+    connection_type: ClassVar[XknxConnectionType] = XknxConnectionType.TUNNEL_UDP
     transport: UDPTransport
 
     def __init__(
@@ -700,7 +692,7 @@ class TCPTunnel(_Tunnel):
 
     __slots__ = ("gateway_ip", "gateway_port")
 
-    connection_type = XknxConnectionType.TUNNEL_TCP
+    connection_type: ClassVar[XknxConnectionType] = XknxConnectionType.TUNNEL_TCP
     transport: TCPTransport
 
     def __init__(
@@ -746,7 +738,7 @@ class SecureTunnel(TCPTunnel):
 
     __slots__ = ("_device_authentication_password", "_user_id", "_user_password")
 
-    connection_type = XknxConnectionType.TUNNEL_SECURE
+    connection_type: ClassVar[XknxConnectionType] = XknxConnectionType.TUNNEL_SECURE
     transport: SecureSession
 
     def __init__(