Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""
Module for Serialization and Deserialization of KNX Search Response.
Search Requests are used to search for KNX/IP devices within the network.
With a search response the receiving party acknowledges the valid processing of the request.
The search response contains all information of the found device (Name, serial number, supported features.).
It supports an array-style access to the DIBs (use classname as index).
"""
from .body import KNXIPBody
from .dib import DIB, DIBDeviceInformation
from .hpai import HPAI
from .knxip_enum import KNXIPServiceType
class SearchResponse(KNXIPBody):
"""Representation of a KNX Connect Request."""
# pylint: disable=too-many-instance-attributes
service_type = KNXIPServiceType.SEARCH_RESPONSE
def __init__(self, xknx):
"""Initialize SearchResponse object."""
super().__init__(xknx)
self.control_endpoint = HPAI()
self.dibs = []
def calculated_length(self):
"""Get length of KNX/IP body."""
return HPAI.LENGTH + \
sum([dib.calculated_length() for dib in self.dibs])
Documentation within:
Application Note 117/08 v02
KNX IP Communication Medium
File: AN117 v02.01 KNX IP Communication Medium DV.pdf
"""
from xknx.dpt import DPTArray, DPTBinary
from xknx.exceptions import ConversionError, CouldNotParseKNXIP
from xknx.telegram import GroupAddress, PhysicalAddress, Telegram, TelegramType
from .body import KNXIPBody
from .knxip_enum import APCICommand, CEMIFlags, CEMIMessageCode
class CEMIFrame(KNXIPBody):
"""Representation of a CEMI Frame."""
# pylint: disable=too-many-instance-attributes
def __init__(self, xknx):
"""Initialize CEMIFrame object."""
super().__init__(xknx)
self.code = CEMIMessageCode.L_DATA_IND
self.flags = 0
self.cmd = APCICommand.GROUP_READ
self.src_addr = GroupAddress(None)
self.dst_addr = GroupAddress(None)
self.mpdu_len = 0
self.payload = None
@property
"""
Module for Serialization and Deserialization of a KNX Disconnect Request information.
Connect requests are used to disconnect a tunnel from a KNX/IP device.
"""
from xknx.exceptions import CouldNotParseKNXIP
from .body import KNXIPBody
from .hpai import HPAI
from .knxip_enum import KNXIPServiceType
class DisconnectRequest(KNXIPBody):
"""Representation of a KNX Disconnect Request."""
# pylint: disable=too-many-instance-attributes
service_type = KNXIPServiceType.DISCONNECT_REQUEST
def __init__(self, xknx):
"""Initialize DisconnectRequest object."""
super().__init__(xknx)
self.communication_channel_id = 1
self.control_endpoint = HPAI()
def calculated_length(self):
"""Get length of KNX/IP body."""
return 2 + HPAI.LENGTH
"""
Module for Serialization and Deserialization of KNX Search Requests.
Search Requests are used to search for KNX/IP devices within the network.
"""
from .body import KNXIPBody
from .hpai import HPAI
from .knxip_enum import KNXIPServiceType
class SearchRequest(KNXIPBody):
"""Representation of a KNX Connect Request."""
# pylint: disable=too-many-instance-attributes
service_type = KNXIPServiceType.SEARCH_REQUEST
def __init__(self, xknx):
"""Initialize SearchRequest object."""
super().__init__(xknx)
self.discovery_endpoint = HPAI(ip_addr="224.0.23.12", port=3671)
def calculated_length(self):
"""Get length of KNX/IP body."""
return HPAI.LENGTH
def from_knx(self, raw):
"""
Module for Serialization and Deserialization of a KNX Connetionstate Request information.
Connectionstate requests are used to determine if a tunnel connection is still active and valid.
"""
from xknx.exceptions import CouldNotParseKNXIP
from .body import KNXIPBody
from .hpai import HPAI
from .knxip_enum import KNXIPServiceType
class ConnectionStateRequest(KNXIPBody):
"""Representation of a KNX Connection State Request."""
# pylint: disable=too-many-instance-attributes
service_type = KNXIPServiceType.CONNECTIONSTATE_REQUEST
def __init__(self, xknx):
"""Initialize ConnectionStateRequest object."""
super().__init__(xknx)
self.communication_channel_id = 1
self.control_endpoint = HPAI()
def calculated_length(self):
"""Get length of KNX/IP body."""
return 2 + HPAI.LENGTH
"""
Module for Serialization and Deserialization of a KNX Tunnel ACK information.
Connect requests are used to transmit a KNX telegram within an existing KNX tunnel connection.
With an Tunnel ACK the receiving party acknowledges the valid processing of the request.
"""
from xknx.exceptions import CouldNotParseKNXIP
from .body import KNXIPBody
from .error_code import ErrorCode
from .knxip_enum import KNXIPServiceType
class TunnellingAck(KNXIPBody):
"""Representation of a KNX Connect Request."""
# pylint: disable=too-many-instance-attributes
service_type = KNXIPServiceType.TUNNELLING_ACK
BODY_LENGTH = 4
def __init__(self, xknx):
"""Initialize TunnellingAck object."""
super().__init__(xknx)
self.communication_channel_id = 1
self.sequence_counter = 0
self.status_code = ErrorCode.E_NO_ERROR
def calculated_length(self):
"""
Module for Serialization and Deserialization of a KNX Connect Response information.
Connect requests are used to start a new tunnel connection on a KNX/IP device.
With an Connect Response the receiving party acknowledges the valid processing of the request.
"""
from xknx.exceptions import CouldNotParseKNXIP
from .body import KNXIPBody
from .error_code import ErrorCode
from .hpai import HPAI
from .knxip_enum import ConnectRequestType, KNXIPServiceType
class ConnectResponse(KNXIPBody):
"""Representation of a KNX Connect Response."""
# pylint: disable=too-many-instance-attributes
service_type = KNXIPServiceType.CONNECT_RESPONSE
CRD_LENGTH = 4
def __init__(self, xknx):
"""Initialize ConnectResponse class."""
super().__init__(xknx)
self.communication_channel = 0
self.status_code = ErrorCode.E_NO_ERROR
self.request_type = None
self.control_endpoint = HPAI()
"""
Module for Serialization and Deserialization of a KNX Connect Request information.
Connect requests are used to start a new tunnel connection on a KNX/IP device.
"""
from xknx.exceptions import CouldNotParseKNXIP
from .body import KNXIPBody
from .hpai import HPAI
from .knxip_enum import ConnectRequestType, KNXIPServiceType
class ConnectRequest(KNXIPBody):
"""Representation of a KNX Connect Request."""
# pylint: disable=too-many-instance-attributes
service_type = KNXIPServiceType.CONNECT_REQUEST
CRI_LENGTH = 4
def __init__(self, xknx):
"""Initialize ConnectRequest object."""
super().__init__(xknx)
self.request_type = None
self.control_endpoint = HPAI()
self.data_endpoint = HPAI()
# KNX layer, 0x02 = TUNNEL_LINKLAYER
self.flags = 0x02
"""
Module for Serialization and Deserialization of a KNX Tunnel Request information.
Connect requests are used to transmit a KNX telegram within an existing KNX tunnel connection.
"""
from xknx.exceptions import CouldNotParseKNXIP
from .body import KNXIPBody
from .cemi_frame import CEMIFrame, CEMIMessageCode
from .knxip_enum import KNXIPServiceType
class TunnellingRequest(KNXIPBody):
"""Representation of a KNX Connect Request."""
# pylint: disable=too-many-instance-attributes
service_type = KNXIPServiceType.TUNNELLING_REQUEST
HEADER_LENGTH = 4
def __init__(self, xknx):
"""Initialize TunnellingRequest object."""
super().__init__(xknx)
self.communication_channel_id = 1
self.sequence_counter = 0
self.cemi = CEMIFrame(xknx)
self.cemi.code = CEMIMessageCode.L_Data_REQ
"""
Module for Serialization and Deserialization of a KNX Connectionstate Response information.
Connectionstate requests are used to determine if a tunnel connection is still active and valid.
With a connectionstate response the receiving party acknowledges the valid processing of the request.
"""
from xknx.exceptions import CouldNotParseKNXIP
from .body import KNXIPBody
from .error_code import ErrorCode
from .knxip_enum import KNXIPServiceType
class ConnectionStateResponse(KNXIPBody):
"""Representation of a KNX Connection State Response."""
# pylint: disable=too-many-instance-attributes
service_type = KNXIPServiceType.CONNECTIONSTATE_RESPONSE
def __init__(self, xknx):
"""Initialize ConnectionStateResponse object."""
super().__init__(xknx)
self.communication_channel_id = 1
self.status_code = ErrorCode.E_NO_ERROR
def calculated_length(self):
"""Get length of KNX/IP body."""
return 2