1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
|
"""
Module for Serialization and Deserialization of a KNX Tunnelling Feature service.
The Tunnelling Client shall use the Tunnelling Feature services TUNNELLING_FEATURE_GET and
TUNNELLING_FEATURE_SET to read and respectively write features related to the Tunnelling
interface and the Tunnelling host device.
The Tunnelling Server shall use the service TUNNELLING_FEATURE_INFO to spontaneously report
Tunnelling Clients about relevant changes in the state of itself or the Tunnelling connection.
"""
from __future__ import annotations
import struct
from xknx.exceptions import CouldNotParseKNXIP
from xknx.management.application_layer_enum import ReturnCode
from .body import KNXIPBody, KNXIPBodyResponse
from .error_code import ErrorCode
from .knxip_enum import KNXIPServiceType, TunnellingFeatureType
class _TunnellingFeature:
"""Base representation of a KNX Tunnelling Feature interface request."""
HEADER_LENGTH = 4
FEATURE_ID_LENGTH = 2
def __init__(
self,
communication_channel_id: int = 1,
sequence_counter: int = 0,
status_code: ErrorCode = ErrorCode.E_NO_ERROR,
feature_type: TunnellingFeatureType = TunnellingFeatureType.SUPPORTED_EMI_TYPE,
data: bytes = b"",
) -> None:
"""Initialize _TunnellingFeature object."""
self.communication_channel_id = communication_channel_id
self.sequence_counter = sequence_counter
self.status_code = status_code
self.feature_type = feature_type
self.data = data
def _has_data(self) -> bool:
return True
def calculated_length(self) -> int:
"""Get length of KNX/IP body."""
data_size = len(self.data) + (len(self.data) % 2) if self._has_data() else 0
return (
_TunnellingFeature.HEADER_LENGTH
+ _TunnellingFeature.FEATURE_ID_LENGTH
+ data_size
)
def from_knx(self, raw: bytes) -> int:
"""Parse/deserialize from KNX/IP raw data."""
if raw[0] != _TunnellingFeature.HEADER_LENGTH: # structure_length field
raise CouldNotParseKNXIP("TunnellingFeature header has invalid length")
self.communication_channel_id = raw[1]
self.sequence_counter = raw[2]
self.status_code = ErrorCode(raw[3])
self.feature_type = TunnellingFeatureType(raw[4])
self.data = raw[6:]
if self._has_data() and len(self.data) == 0:
raise CouldNotParseKNXIP("TunnellingFeature missing data")
if not self._has_data() and len(self.data) > 0:
raise CouldNotParseKNXIP("TunnellingFeature unexpected data")
return len(raw)
def to_knx(self) -> bytes:
"""Serialize to KNX/IP raw data."""
data_size = len(self.data) + (len(self.data) % 2) if self._has_data() else 0
return struct.pack(
# Append a NUL byte if data size is uneven
f"!BBBBBx{data_size}s",
_TunnellingFeature.HEADER_LENGTH,
self.communication_channel_id,
self.sequence_counter,
self.status_code.value,
self.feature_type.value,
self.data,
)
def __repr__(self) -> str:
"""Return object as readable string."""
_data = f'data="{self.data.hex()}" ' if self._has_data() else ""
return (
f"<{self.__class__.__name__} "
f'communication_channel_id="{self.communication_channel_id}" '
f'sequence_counter="{self.sequence_counter}" '
f'feature_type="{self.feature_type}" '
f"{_data}/>"
)
class TunnellingFeatureGet(_TunnellingFeature, KNXIPBody):
"""Representation of a KNX Tunnelling Feature Get request."""
SERVICE_TYPE = KNXIPServiceType.TUNNELLING_FEATURE_GET
def __init__(
self,
communication_channel_id: int = 1,
sequence_counter: int = 0,
feature_type: TunnellingFeatureType = TunnellingFeatureType.SUPPORTED_EMI_TYPE,
) -> None:
"""Initialize TunnellingFeatureGet object."""
super().__init__(
communication_channel_id=communication_channel_id,
sequence_counter=sequence_counter,
feature_type=feature_type,
)
def _has_data(self) -> bool:
return False
class TunnellingFeatureSet(_TunnellingFeature, KNXIPBody):
"""Representation of a KNX Tunnelling Feature Set request."""
SERVICE_TYPE = KNXIPServiceType.TUNNELLING_FEATURE_SET
def __init__(
self,
communication_channel_id: int = 1,
sequence_counter: int = 0,
feature_type: TunnellingFeatureType = TunnellingFeatureType.SUPPORTED_EMI_TYPE,
data: bytes = b"",
) -> None:
"""Initialize TunnellingFeatureSet object."""
super().__init__(
communication_channel_id=communication_channel_id,
sequence_counter=sequence_counter,
feature_type=feature_type,
data=data,
)
class TunnellingFeatureInfo(_TunnellingFeature, KNXIPBody):
"""Representation of a KNX Tunnelling Feature Info indication."""
SERVICE_TYPE = KNXIPServiceType.TUNNELLING_FEATURE_INFO
def __init__(
self,
communication_channel_id: int = 1,
sequence_counter: int = 0,
feature_type: TunnellingFeatureType = TunnellingFeatureType.SUPPORTED_EMI_TYPE,
data: bytes = b"",
) -> None:
"""Initialize TunnellingFeatureInfo object."""
super().__init__(
communication_channel_id=communication_channel_id,
sequence_counter=sequence_counter,
feature_type=feature_type,
data=data,
)
class TunnellingFeatureResponse(_TunnellingFeature, KNXIPBodyResponse):
"""Representation of a KNX Tunnelling Feature response."""
SERVICE_TYPE = KNXIPServiceType.TUNNELLING_FEATURE_RESPONSE
def __init__(
self,
communication_channel_id: int = 1,
sequence_counter: int = 0,
feature_type: TunnellingFeatureType = TunnellingFeatureType.SUPPORTED_EMI_TYPE,
return_code: ReturnCode = ReturnCode.E_SUCCESS,
data: bytes = b"",
) -> None:
"""Initialize TunnellingFeatureSet object."""
super().__init__(
communication_channel_id=communication_channel_id,
sequence_counter=sequence_counter,
feature_type=feature_type,
data=data,
)
self.return_code = return_code
def from_knx(self, raw: bytes) -> int:
"""Parse/deserialize from KNX/IP raw data."""
if raw[0] != _TunnellingFeature.HEADER_LENGTH: # structure_length field
raise CouldNotParseKNXIP("TunnellingFeature header has invalid length")
self.communication_channel_id = raw[1]
self.sequence_counter = raw[2]
self.status_code = ErrorCode(raw[3])
self.feature_type = TunnellingFeatureType(raw[4])
self.return_code = ReturnCode(raw[5])
self.data = raw[6:]
if self.return_code is ReturnCode.E_SUCCESS and len(self.data) == 0:
# Data may be omitted by some servers when an error occurred
raise CouldNotParseKNXIP("TunnellingFeature missing data.")
return len(raw)
def to_knx(self) -> bytes:
"""Serialize to KNX/IP raw data."""
data_size = len(self.data) + (len(self.data) % 2) if self._has_data() else 0
return struct.pack(
# Append a NUL byte if data size is uneven
f"!BBBBBB{data_size}s",
_TunnellingFeature.HEADER_LENGTH,
self.communication_channel_id,
self.sequence_counter,
self.status_code.value,
self.feature_type.value,
self.return_code.value,
self.data,
)
def __repr__(self) -> str:
"""Return object as readable string."""
return (
"<TunnellingFeatureResponse "
f'communication_channel_id="{self.communication_channel_id}" '
f'sequence_counter="{self.sequence_counter}" '
f'status_code="{self.status_code}" '
f'feature_type="{self.feature_type}" '
f'return_code="{self.return_code}" '
f'data="{self.data.hex()}" />'
)
|
