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
|
"""
Module for serialization and deserialization of TPCI payloads.
TPCI stands for Transport Layer Protocol Control Information.
"""
from __future__ import annotations
from abc import ABC
from typing import ClassVar
from xknx.exceptions import ConversionError
CONTROL_BIT_MASK = 0x80
NUMBERED_BIT_MASK = 0x40
class TPCI(ABC):
"""
Base class for TCPI services.
This base class is only the interface for the derived classes.
"""
__slots__ = ()
control: ClassVar[bool]
numbered: ClassVar[bool]
sequence_number: int = 0
control_flags: ClassVar[int | None] = None
def to_knx(self) -> int:
"""Serialize to KNX/IP raw data."""
return (
self.control << 7
| self.numbered << 6
| (self.sequence_number & 0xF) << 2
| (self.control_flags or 0)
)
def __eq__(self, other: object) -> bool:
"""Equal operator."""
return (
isinstance(other, self.__class__)
and self.sequence_number == other.sequence_number
)
def __repr__(self) -> str:
"""Return object as readable string."""
_sequence_number = (
f"sequence_number={self.sequence_number}" if self.numbered else ""
)
return f"{self.__class__.__name__}({_sequence_number})"
@staticmethod
def resolve(raw_tpci: int, dst_is_group_address: bool, dst_is_zero: bool) -> TPCI:
"""
Return TPCI instance from TPCI command.
See KNX Specifications 03_03_04 Transport Layer v01.02.02 AS ยง2 TPDU
"""
control = raw_tpci & CONTROL_BIT_MASK
numbered = raw_tpci & NUMBERED_BIT_MASK
sequence_number = (raw_tpci >> 2) & 0xF
if dst_is_group_address:
if control or numbered:
raise ConversionError("Invalid TPCI flags in group addressed frame.")
if not sequence_number:
if dst_is_zero:
return TDataBroadcast()
return TDataGroup()
if sequence_number == 1:
# TDataTagGroup uses sequence number field as flag
return TDataTagGroup()
if not numbered and sequence_number:
raise ConversionError("Sequence number not allowed for unnumbered TPCI")
if not control:
# data - last 2 bits are part of APCI
if numbered:
return TDataConnected(sequence_number=sequence_number)
return TDataIndividual()
# unnumbered control
control_flags = raw_tpci & 0b11
if not numbered:
if control_flags == 0:
return TConnect()
if control_flags == 1:
return TDisconnect()
# numbered control
if control_flags == 0b10:
return TAck(sequence_number=sequence_number)
if control_flags == 0b11:
return TNak(sequence_number=sequence_number)
raise ConversionError(f"Unknown TPCI {raw_tpci:#10b}.")
class TDataGroup(TPCI):
"""T_Data_Group class."""
__slots__ = ()
control = False
numbered = False
def to_knx(self) -> int:
"""Serialize to KNX/IP raw data."""
return 0
class TDataBroadcast(TPCI):
"""T_Data_Broadcast class."""
__slots__ = ()
control = False
numbered = False
def to_knx(self) -> int:
"""Serialize to KNX/IP raw data."""
return 0
class TDataTagGroup(TPCI):
"""T_Data_Tag_Group class."""
__slots__ = ()
control = False
numbered = False
sequence_number = 0b0001
class TDataIndividual(TPCI):
"""T_Data_Individual class."""
__slots__ = ()
control = False
numbered = False
def to_knx(self) -> int:
"""Serialize to KNX/IP raw data."""
return 0
class TDataConnected(TPCI):
"""T_Data_Connected class."""
__slots__ = ("sequence_number",)
control = False
numbered = True
def __init__(self, sequence_number: int) -> None:
"""Initialize TDataConnected."""
self.sequence_number = sequence_number
class TConnect(TPCI):
"""T_Connect class."""
__slots__ = ()
control = True
numbered = False
control_flags = 0b00
class TDisconnect(TPCI):
"""T_Disconnect class."""
__slots__ = ()
control = True
numbered = False
control_flags = 0b01
class TAck(TPCI):
"""T_Ack class."""
__slots__ = ("sequence_number",)
control = True
numbered = True
control_flags = 0b10
def __init__(self, sequence_number: int) -> None:
"""Initialize TAck."""
self.sequence_number = sequence_number
class TNak(TPCI):
"""T_Nak class."""
__slots__ = ("sequence_number",)
control = True
numbered = True
control_flags = 0b11
def __init__(self, sequence_number: int) -> None:
"""Initialize TNak."""
self.sequence_number = sequence_number
|
