Skip to content

Commit

Permalink
move ConnectionConfig to new module xknx.io.connection
Browse files Browse the repository at this point in the history
  • Loading branch information
farmio committed Mar 28, 2021
1 parent c445259 commit fd0278a
Show file tree
Hide file tree
Showing 3 changed files with 71 additions and 65 deletions.
3 changes: 2 additions & 1 deletion xknx/io/__init__.py
Expand Up @@ -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
Expand Down
68 changes: 68 additions & 0 deletions 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__
65 changes: 1 addition & 64 deletions xknx/io/knxip_interface.py
Expand Up @@ -6,15 +6,14 @@
* 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

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
Expand All @@ -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."""

Expand Down

0 comments on commit fd0278a

Please sign in to comment.