From fd0278a744ad3f2af1af2828e250938f6bcd6115 Mon Sep 17 00:00:00 2001 From: farmio Date: Sun, 28 Mar 2021 21:24:59 +0200 Subject: [PATCH] move ConnectionConfig to new module xknx.io.connection --- xknx/io/__init__.py | 3 +- xknx/io/connection.py | 68 ++++++++++++++++++++++++++++++++++++++ xknx/io/knxip_interface.py | 65 +----------------------------------- 3 files changed, 71 insertions(+), 65 deletions(-) create mode 100644 xknx/io/connection.py diff --git a/xknx/io/__init__.py b/xknx/io/__init__.py index 4d757b6bd8..4785c26209 100644 --- a/xknx/io/__init__.py +++ b/xknx/io/__init__.py @@ -7,9 +7,10 @@ - Tunnel uses UDP packets and builds a static tunnel with KNX/IP device. """ # flake8: noqa +from .connection import ConnectionConfig, ConnectionType from .const import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT from .gateway_scanner import GatewayScanFilter, GatewayScanner -from .knxip_interface import ConnectionConfig, ConnectionType, KNXIPInterface +from .knxip_interface import KNXIPInterface 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 new file mode 100644 index 0000000000..304de03ad0 --- /dev/null +++ b/xknx/io/connection.py @@ -0,0 +1,68 @@ +"""Manages a connection to the KNX bus.""" +from enum import Enum +from typing import Optional + +from .const import DEFAULT_MCAST_PORT +from .gateway_scanner import GatewayScanFilter + + +class ConnectionType(Enum): + """Enum class for different types of KNX/IP Connections.""" + + AUTOMATIC = 0 + TUNNELING = 1 + ROUTING = 2 + + +class ConnectionConfig: + """ + Connection configuration. + + Handles: + * type of connection: + * AUTOMATIC for using GatewayScanner for searching and finding KNX/IP devices in the network. + * TUNNELING connect to a specific KNX/IP tunneling device. + * ROUTING use KNX/IP multicast routing. + * local_ip: Local ip of the interface though which KNXIPInterface should connect. + * gateway_ip: IP of KNX/IP tunneling device. + * gateway_port: Port of KNX/IP tunneling device. + * route_back: the KNXnet/IP Server shall use the IP address and the port number in the IP package + received as the target IP address or port number for the response to the KNXnet/IP Client. + * auto_reconnect: Auto reconnect to KNX/IP tunneling device if connection cannot be established. + * auto_reconnect_wait: Wait n seconds before trying to reconnect to KNX/IP tunneling device. + * scan_filter: For AUTOMATIC connection, limit scan with the given filter + """ + + # pylint: disable=too-few-public-methods,too-many-instance-attributes + + def __init__( + self, + connection_type: ConnectionType = ConnectionType.AUTOMATIC, + local_ip: Optional[str] = None, + local_port: int = 0, + gateway_ip: Optional[str] = None, + gateway_port: int = DEFAULT_MCAST_PORT, + route_back: bool = False, + auto_reconnect: bool = True, + auto_reconnect_wait: int = 3, + scan_filter: GatewayScanFilter = GatewayScanFilter(), + ): + """Initialize ConnectionConfig class.""" + # pylint: disable=too-many-arguments + self.connection_type = connection_type + self.local_ip = local_ip + self.local_port = local_port + self.gateway_ip = gateway_ip + self.gateway_port = gateway_port + 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 + + def __eq__(self, other: object) -> bool: + """Equality for ConnectionConfig class (used in unit tests).""" + return self.__dict__ == other.__dict__ diff --git a/xknx/io/knxip_interface.py b/xknx/io/knxip_interface.py index f52f7ea724..df1f2e6ecc 100644 --- a/xknx/io/knxip_interface.py +++ b/xknx/io/knxip_interface.py @@ -6,7 +6,6 @@ * provides callbacks after having received a telegram from the network. """ -from enum import Enum import ipaddress import logging from typing import TYPE_CHECKING, Optional, cast @@ -14,7 +13,7 @@ import netifaces from xknx.exceptions import CommunicationError, XKNXException -from .const import DEFAULT_MCAST_PORT +from .connection import ConnectionConfig, ConnectionType from .gateway_scanner import GatewayScanFilter, GatewayScanner from .routing import Routing from .tunnel import Tunnel @@ -28,68 +27,6 @@ logger = logging.getLogger("xknx.log") -class ConnectionType(Enum): - """Enum class for different types of KNX/IP Connections.""" - - AUTOMATIC = 0 - TUNNELING = 1 - ROUTING = 2 - - -class ConnectionConfig: - """ - Connection configuration. - - Handles: - * type of connection: - * AUTOMATIC for using GatewayScanner for searching and finding KNX/IP devices in the network. - * TUNNELING connect to a specific KNX/IP tunneling device. - * ROUTING use KNX/IP multicast routing. - * local_ip: Local ip of the interface though which KNXIPInterface should connect. - * gateway_ip: IP of KNX/IP tunneling device. - * gateway_port: Port of KNX/IP tunneling device. - * route_back: the KNXnet/IP Server shall use the IP address and the port number in the IP package - received as the target IP address or port number for the response to the KNXnet/IP Client. - * auto_reconnect: Auto reconnect to KNX/IP tunneling device if connection cannot be established. - * auto_reconnect_wait: Wait n seconds before trying to reconnect to KNX/IP tunneling device. - * scan_filter: For AUTOMATIC connection, limit scan with the given filter - """ - - # pylint: disable=too-few-public-methods,too-many-instance-attributes - - def __init__( - self, - connection_type: ConnectionType = ConnectionType.AUTOMATIC, - local_ip: Optional[str] = None, - local_port: int = 0, - gateway_ip: Optional[str] = None, - gateway_port: int = DEFAULT_MCAST_PORT, - route_back: bool = False, - auto_reconnect: bool = True, - auto_reconnect_wait: int = 3, - scan_filter: GatewayScanFilter = GatewayScanFilter(), - ): - """Initialize ConnectionConfig class.""" - # pylint: disable=too-many-arguments - self.connection_type = connection_type - self.local_ip = local_ip - self.local_port = local_port - self.gateway_ip = gateway_ip - self.gateway_port = gateway_port - 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 - - def __eq__(self, other: object) -> bool: - """Equality for ConnectionConfig class (used in unit tests).""" - return self.__dict__ == other.__dict__ - - class KNXIPInterface: """Class for managing KNX/IP Tunneling or Routing connections."""