"""
Module for serialization/deserialization and handling of KNX addresses.
The module can handle:
* individual addresses of devices.
* (logical) group addresses.
* xknx internal group addresses.
The module supports all different writings of group addresses:
* 3rn level: "1/2/3"
* 2nd level: "1/2"
* Free format: "123"
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from enum import Enum
from functools import total_ordering
from re import compile as re_compile
from typing import ClassVar, Union
from xknx.exceptions import CouldNotParseAddress
from xknx.typing import Self
GroupAddressableType = Union["GroupAddress", str, int]
IndividualAddressableType = Union["IndividualAddress", str, int]
InternalGroupAddressableType = Union["InternalGroupAddress", str]
DeviceAddressableType = GroupAddressableType | InternalGroupAddressableType
DeviceGroupAddress = Union["GroupAddress", "InternalGroupAddress"]
INVALID_PREFIX_MESSAGE = "Invalid prefix for internal group address"
def parse_device_group_address(
address: DeviceAddressableType,
) -> DeviceGroupAddress:
"""Parse an Addressable type to GroupAddress or InternalGroupAddress."""
try:
group_address = GroupAddress(address) # type: ignore[arg-type] # InternalGroupAddress will raise
except CouldNotParseAddress as ex:
if isinstance(address, str | InternalGroupAddress):
try:
return InternalGroupAddress(address)
except CouldNotParseAddress as internal_ex:
# prefer to raise original exception from GroupAddress
if internal_ex.message != INVALID_PREFIX_MESSAGE:
raise internal_ex
raise ex
if group_address.raw == 0:
raise CouldNotParseAddress(address, "Broadcast address invalid for devices")
return group_address
@total_ordering
class BaseAddress(ABC):
"""Base class for all knx address types."""
__slots__ = ("raw",)
raw: int
@abstractmethod
def __init__(
self, address: IndividualAddressableType | GroupAddressableType
) -> None:
"""Initialize Address instance. To be implemented in derived class."""
@classmethod
def from_knx(cls: type[Self], raw: bytes) -> Self:
"""Parse/deserialize from KNX/IP raw data."""
return cls(int.from_bytes(raw, "big"))
def to_knx(self) -> bytes:
"""
Serialize to KNX/IP raw data.
Returns a bytes object with length of 2 from the raw value.
"""
return int.to_bytes(self.raw, 2, "big")
def __eq__(self, other: object | None) -> bool:
"""
Implement the equal operator.
Returns `True` if we check against the same subclass and the
raw Value matches.
"""
return isinstance(other, self.__class__) and self.raw == other.raw
def __lt__(self, other: object) -> bool:
"""Implement less than operator for sorting addresses."""
if not isinstance(other, self.__class__):
raise TypeError(
f"'<' not supported between instances of '{self.__class__.__name__}' and '{other.__class__.__name__}'"
)
return self.raw < other.raw
def __hash__(self) -> int:
"""Hash Address so it can be used as dict key."""
return hash((self.__class__, self.raw))
class IndividualAddress(BaseAddress):
"""Class for handling KNX individual addresses."""
__slots__ = ()
MAX_AREA = 15
MAX_MAIN = 15
MAX_LINE = 255
ADDRESS_RE = re_compile(
r"^(?P\d{1,2})\.(?P\d{1,2})\.(?P\d{1,3})$"
)
def __init__(self, address: IndividualAddressableType) -> None:
"""Initialize IndividualAddress class."""
if isinstance(address, int):
self.raw = address
elif isinstance(address, IndividualAddress):
self.raw = address.raw
elif isinstance(address, str):
if address.isdigit():
self.raw = int(address)
else:
self.raw = self.__string_to_int(address)
else:
raise CouldNotParseAddress(address, message="Invalid type")
if not 0 <= self.raw <= 65535:
raise CouldNotParseAddress(
address, message="Address out of range (0..65535)"
)
def __string_to_int(self, address: str) -> int:
"""
Parse `address` as string to an integer and do some simple checks.
Returns the integer representation of `address` if all checks are valid:
* string matches against the regular expression
* area, main and line are inside its range
In any other case, we raise an `CouldNotParseAddress` exception.
"""
match = self.ADDRESS_RE.match(address)
if not match:
raise CouldNotParseAddress(address, message="Invalid format")
area = int(match.group("area"))
main = int(match.group("main"))
line = int(match.group("line"))
if area > self.MAX_AREA:
raise CouldNotParseAddress(
address, message=f"Area part out of range (0..{self.MAX_AREA})"
)
if main > self.MAX_MAIN:
raise CouldNotParseAddress(
address, message=f"Line part out of range (0..{self.MAX_MAIN})"
)
if line > self.MAX_LINE:
raise CouldNotParseAddress(
address, message=f"Device part out of range (0..{self.MAX_LINE})"
)
return (area << 12) + (main << 8) + line
@property
def area(self) -> int:
"""Return area part of individual address."""
return (self.raw >> 12) & self.MAX_AREA
@property
def main(self) -> int:
"""Return main part of individual address."""
return (self.raw >> 8) & self.MAX_MAIN
@property
def line(self) -> int:
"""Return line part of individual address."""
return self.raw & self.MAX_LINE
@property
def is_device(self) -> bool:
"""Return `True` if this address is a valid device address."""
return self.line != 0
@property
def is_line(self) -> bool:
"""Return `True` if this address is a valid line address."""
return not self.is_device
def __str__(self) -> str:
"""Return object as in KNX notation (e.g. '1.2.3')."""
return f"{self.area}.{self.main}.{self.line}"
def __repr__(self) -> str:
"""Return this object as parsable string."""
return f'IndividualAddress("{self}")'
class GroupAddressType(Enum):
"""
Possible types of `GroupAddress`.
KNX knows three types of group addresses:
* FREE, a integer or hex representation
* SHORT, a representation like '1/123', without middle groups
* LONG, a representation like '1/2/34', with middle groups
"""
FREE = 0
SHORT = 2
LONG = 3
class GroupAddress(BaseAddress):
"""Class for handling KNX group addresses."""
__slots__ = ()
# overridden by XKNX class on initialization to have consistent global string representation
address_format: ClassVar[GroupAddressType] = GroupAddressType.LONG
MAX_MAIN = 31
MAX_MIDDLE = 7
MAX_SUB_LONG = 255
MAX_SUB_SHORT = 2047
MAX_FREE = 65535
ADDRESS_RE = re_compile(
r"^(?P\d{1,2})(/(?P\d{1,2}))?/(?P\d{1,4})$"
)
def __init__(self, address: GroupAddressableType) -> None:
"""Initialize GroupAddress class."""
if isinstance(address, int):
self.raw = address
elif isinstance(address, GroupAddress):
self.raw = address.raw
elif isinstance(address, str):
if address.isdigit():
self.raw = int(address)
else:
self.raw = self.__string_to_int(address)
else:
raise CouldNotParseAddress(address, message="Invalid type")
if not 0 <= self.raw <= 65535:
raise CouldNotParseAddress(
address, message="Address out of range (0..65535)"
)
def __string_to_int(self, address: str) -> int:
"""
Parse `address` as string to an integer and do some simple checks.
Returns the integer representation of `address` if all checks are valid:
* string matches against the regular expression
* main, middle and sub are inside its range
In any other case, we raise an `CouldNotParseAddress` exception.
"""
match = self.ADDRESS_RE.match(address)
if not match:
raise CouldNotParseAddress(address, message="Invalid format")
main = int(match.group("main"))
middle = (
int(match.group("middle")) if match.group("middle") is not None else None
)
sub = int(match.group("sub"))
if main > self.MAX_MAIN:
raise CouldNotParseAddress(
address, message=f"Main group out of range (0..{self.MAX_MAIN})"
)
if middle is not None:
if middle > self.MAX_MIDDLE:
raise CouldNotParseAddress(
address, message=f"Middle group out of range (0..{self.MAX_MIDDLE})"
)
if sub > self.MAX_SUB_LONG:
raise CouldNotParseAddress(
address, message=f"Sub group out of range (0..{self.MAX_SUB_LONG})"
)
elif sub > self.MAX_SUB_SHORT:
raise CouldNotParseAddress(
address, message=f"Sub group out of range (0..{self.MAX_SUB_SHORT})"
)
return (
(main << 11) + (middle << 8) + sub
if middle is not None
else (main << 11) + sub
)
@property
def main(self) -> int | None:
"""
Return the main group part as an integer.
Works only if the group dont uses `GroupAddressType.FREE`, returns `None`
in any other case.
"""
return (
(self.raw >> 11) & self.MAX_MAIN
if self.address_format != GroupAddressType.FREE
else None
)
@property
def middle(self) -> int | None:
"""
Return the middle group part as an integer.
Works only if the group uses `GroupAddressType.LONG`, returns `None` in
any other case.
"""
return (
(self.raw >> 8) & self.MAX_MIDDLE
if self.address_format == GroupAddressType.LONG
else None
)
@property
def sub(self) -> int:
"""
Return the sub group part as an integer.
Works with any `GroupAddressType`, as we always have sub groups.
"""
if self.address_format == GroupAddressType.SHORT:
return self.raw & self.MAX_SUB_SHORT
if self.address_format == GroupAddressType.LONG:
return self.raw & self.MAX_SUB_LONG
return self.raw
def __str__(self) -> str:
"""
Return object as in KNX notation (e.g. '1/2/3').
Honors the used `GroupAddressType` of this group.
"""
if self.address_format == GroupAddressType.LONG:
return f"{self.main}/{self.middle}/{self.sub}"
if self.address_format == GroupAddressType.SHORT:
return f"{self.main}/{self.sub}"
return f"{self.sub}"
def __repr__(self) -> str:
"""Return object as parsable string."""
return f'GroupAddress("{self}")'
@total_ordering
class InternalGroupAddress:
"""Class for handling addresses used internally in xknx devices only."""
__slots__ = ("raw",)
def __init__(self, address: str | InternalGroupAddress) -> None:
"""Initialize InternalGroupAddress class."""
self.raw: str
if isinstance(address, InternalGroupAddress):
self.raw = address.raw
return
if not isinstance(address, str):
raise CouldNotParseAddress(address, message="Invalid type")
prefix_length = 1
if len(address) < 2 or address[0].lower() != "i":
raise CouldNotParseAddress(address, message=INVALID_PREFIX_MESSAGE)
if address[1] in "-_":
prefix_length = 2
_raw = address[prefix_length:].strip()
if not _raw:
raise CouldNotParseAddress(address, message="No chars after prefix")
self.raw = f"i-{_raw}"
def __str__(self) -> str:
"""Return object as readable string (e.g. 'i-123')."""
return self.raw
def __repr__(self) -> str:
"""Return object as parsable string."""
return f'InternalGroupAddress("{self.raw}")'
def __eq__(self, other: object | None) -> bool:
"""
Implement the equal operator.
Returns `True` if we check against the same subclass and the
raw Value matches.
"""
return isinstance(other, self.__class__) and self.raw == other.raw
def __lt__(self, other: object) -> bool:
"""Implement less than operator for sorting addresses."""
if not isinstance(other, self.__class__):
raise TypeError(
f"'<' not supported between instances of '{self.__class__.__name__}' and '{other.__class__.__name__}'"
)
return self.raw < other.raw
def __hash__(self) -> int:
"""Hash Address so it can be used as dict key."""
return hash((self.__class__, self.raw))