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
|
"""Implementation of KNX raw payload abstractions."""
from __future__ import annotations
from xknx.exceptions import ConversionError
class DPTBinary:
"""The DPTBinary is a base class for all datatypes encoded directly into the last 6 bit of the APCI/data octet."""
__slots__ = ("value",)
APCI_BITMASK = 0x3F # APCI uses first 2 bits
def __init__(self, value: int | tuple[int]) -> None:
"""Initialize DPTBinary class."""
if isinstance(value, tuple):
value = value[0]
if not isinstance(value, int):
raise TypeError()
if not 0 <= value <= DPTBinary.APCI_BITMASK:
raise ConversionError("Could not init DPTBinary", value=str(value))
self.value = value
def __eq__(self, other: object) -> bool:
"""Equal operator."""
return isinstance(other, DPTBinary) and self.value == other.value
def __repr__(self) -> str:
"""Return object representation."""
return f"DPTBinary({hex(self.value)})"
def __str__(self) -> str:
"""Return object as readable string."""
return f'<DPTBinary value="{self.value}" />'
class DPTArray:
"""The DPTArray is a base class for all datatypes appended to the KNX telegram."""
__slots__ = ("value",)
def __init__(self, value: int | bytes | tuple[int, ...] | list[int]) -> None:
"""Initialize DPTArray class."""
self.value: tuple[int, ...]
if isinstance(value, int):
self.value = (value,)
elif isinstance(value, list | bytes):
self.value = tuple(value)
elif isinstance(value, tuple):
self.value = value
else:
raise TypeError()
def __eq__(self, other: object) -> bool:
"""Equal operator."""
return isinstance(other, DPTArray) and self.value == other.value
def __repr__(self) -> str:
"""Return object representation."""
return f"DPTArray(({', '.join(hex(b) for b in self.value)}))"
def __str__(self) -> str:
"""Return object as readable string."""
return f'<DPTArray value="[{",".join(hex(b) for b in self.value)}]" />'
|
