Repository: XKNX/xknx
Pull request #1362 (2023-11-27 17:39:31)

Add MemoryExtended services
<!--
  You are awesome! Thanks for contributing to our project!
  Please, DO NOT DELETE ANY TEXT from this template!.
-->
## Description
<!--
Please include a summary of the change and which issue is fixed.
Please also include relevant motivation and context.
List any dependencies that are required for this change.
-->
add MemoryExtended services : Read, ReadResponse, Write, WriteResponse

Fixes # (issue)

## Type of change
<!--
Please tick the applicable options.
NOTE: Ticking multiple options most likely indicates
that your change is to big and it is suggested to split it into several smaller PRs.
-->

- [ ] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] Code quality improvements to existing code or addition of tests

## Checklist

- [ ] The documentation has been adjusted accordingly
- [x] Tests have been added that prove the fix is effective or that the feature works
- [ ] The changes are documented in the changelog (docs/changelog.md)

RealByron: I had some problem with mypy, so I need some help
RealByron: ![image](https://github.com/XKNX/xknx/assets/1749192/59eddbd2-6cad-47f8-9bdf-81d3d441f8d8)
I don't know how to do that properly. I will dig on knx handbook, I probably miss something

Patch:
diff --git a/examples/example_fan_percent_mode.py b/examples/example_fan_percent_mode.py
new file mode 100644
index 00000000..6eb2f3b4
--- /dev/null
+++ b/examples/example_fan_percent_mode.py
@@ -0,0 +1,35 @@
+"""Example for Fan device."""
+import asyncio
+
+from xknx import XKNX
+from xknx.devices import Fan
+
+
+async def main():
+    """Connect to KNX/IP bus, control a fan, and turn it off afterwards."""
+    xknx = XKNX()
+    await xknx.start()
+
+    fan = Fan(
+        xknx,
+        name="TestFan",
+        group_address_switch="1/0/12",
+        group_address_speed="1/0/14",
+        max_step=3,
+    )
+
+    # Turn on the fan
+    await fan.turn_on()
+
+    # Set fan speed to different levels
+    for speed in [0, 33, 66, 100]:
+        await fan.set_speed(speed)
+        await asyncio.sleep(1)
+
+    # Turn off the fan
+    await fan.turn_off()
+
+    await xknx.stop()
+
+
+asyncio.run(main())
diff --git a/examples/example_fan_step_mode.py b/examples/example_fan_step_mode.py
new file mode 100644
index 00000000..7f779f1c
--- /dev/null
+++ b/examples/example_fan_step_mode.py
@@ -0,0 +1,35 @@
+"""Example for Fan device."""
+import asyncio
+
+from xknx import XKNX
+from xknx.devices import Fan
+
+
+async def main():
+    """Connect to KNX/IP bus, control a fan, and turn it off afterwards."""
+    xknx = XKNX()
+    await xknx.start()
+
+    fan = Fan(
+        xknx,
+        name="TestFan",
+        group_address_switch="1/0/12",
+        group_address_speed="1/0/14",
+        max_step=3,
+    )
+
+    # Turn on the fan
+    await fan.turn_on()
+
+    # Set fan speed in steps
+    for step in range(1, fan.max_step + 1):
+        await fan.set_speed(step)
+        await asyncio.sleep(1)
+
+    # Turn off the fan
+    await fan.turn_off()
+
+    await xknx.stop()
+
+
+asyncio.run(main())
diff --git a/xknx/telegram/apci.py b/xknx/telegram/apci.py
index 80e183d1..d5da00ee 100644
--- a/xknx/telegram/apci.py
+++ b/xknx/telegram/apci.py
@@ -50,6 +50,11 @@ class APCIService(Enum):
     ADC_READ = 0x0180
     ADC_RESPONSE = 0x1C0
 
+    MEMORY_EXTENDED_WRITE = 0x1FB
+    MEMORY_EXTENDED_WRITE_RESPONSE = 0x1FC
+    MEMORY_EXTENDED_READ = 0x1FD
+    MEMORY_EXTENDED_READ_RESPONSE = 0x1FE
+
     MEMORY_READ = 0x0200
     MEMORY_RESPONSE = 0x0240
     MEMORY_WRITE = 0x0280
@@ -158,6 +163,14 @@ class APCI(ABC):
         if service == APCIService.ADC_READ.value:
             return ADCRead.from_knx(raw)
         if service == APCIService.ADC_RESPONSE.value:
+            if apci == APCIService.MEMORY_EXTENDED_WRITE.value:
+                return MemoryExtendedWrite.from_knx(raw)
+            if apci == APCIService.MEMORY_EXTENDED_WRITE_RESPONSE.value:
+                return MemoryExtendedWriteResponse.from_knx(raw)
+            if apci == APCIService.MEMORY_EXTENDED_READ.value:
+                return MemoryExtendedRead.from_knx(raw)
+            if apci == APCIService.MEMORY_EXTENDED_READ_RESPONSE.value:
+                return MemoryExtendedReadResponse.from_knx(raw)
             return ADCResponse.from_knx(raw)
         if service == APCIService.MEMORY_READ.value:
             return MemoryRead.from_knx(raw)
@@ -492,6 +505,215 @@ class ADCResponse(APCI):
         return f'<ADCResponse channel="{self.channel}" count="{self.count}" value="{self.value}" />'
 
 
+class MemoryExtendedWrite(APCI):
+    """
+    MemoryExtendedWrite service.
+
+    Payload indicates address (16 MiB), count (1-255 bytes) and data.
+    """
+
+    CODE = APCIService.MEMORY_EXTENDED_WRITE
+
+    def __init__(self, address: int, data: bytes, count: int | None = None) -> None:
+        """Initialize a new instance of MemoryExtendedWrite."""
+        if count is None:
+            count = len(data)
+        self.address = address
+        self.count = count
+        self.data = data
+
+    def calculated_length(self) -> int:
+        """Get length of APCI payload."""
+        return 5 + len(self.data)
+
+    @classmethod
+    def from_knx(cls, raw: bytes) -> MemoryExtendedWrite:
+        """Parse/deserialize from KNX/IP raw data."""
+        count = raw[2]
+        address = int.from_bytes(raw[3:6], "big")
+        data = raw[6:]
+
+        return cls(
+            count=count,
+            address=address,
+            data=data,
+        )
+
+    def to_knx(self) -> bytearray:
+        """Serialize to KNX/IP raw data."""
+        if not 0 <= self.address <= 0xFFFFFF:
+            raise ConversionError("Address out of range.")
+        if not 0 <= self.count <= 250:
+            raise ConversionError("Count out of range.")
+
+        size = len(self.data)
+        payload = struct.pack(f"!BI{size}s", self.count, self.address, self.data)
+        # suppress first byte of address
+        payload = payload[:1] + payload[2:]
+
+        return encode_cmd_and_payload(self.CODE, 0, appended_payload=payload)
+
+    def __str__(self) -> str:
+        """Return object as readable string."""
+        return f'<MemoryExtendedWrite address="{hex(self.address)}" count="{self.count}" data="{self.data.hex()}" />'
+
+
+class MemoryExtendedWriteResponse(APCI):
+    """
+    MemoryExtendedWriteResponse service.
+
+    Payload indicates return code, address (16 MiB) and confirmation data.
+    """
+
+    CODE = APCIService.MEMORY_EXTENDED_WRITE_RESPONSE
+
+    def __init__(
+        self, return_code: int, address: int, confirmation_data: bytes = b""
+    ) -> None:
+        """Initialize a new instance of MemoryExtendedWriteResponse."""
+        self.return_code = return_code
+        self.address = address
+        self.confirmation_data = confirmation_data
+
+    def calculated_length(self) -> int:
+        """Get length of APCI payload."""
+        return 5 + len(self.confirmation_data)
+
+    @classmethod
+    def from_knx(cls, raw: bytes) -> MemoryExtendedWriteResponse:
+        """Parse/deserialize from KNX/IP raw data."""
+        return_code = raw[2]
+        address = int.from_bytes(raw[3:6], "big")
+        confirmation_data = raw[6:]
+
+        return cls(
+            return_code=return_code,
+            address=address,
+            confirmation_data=confirmation_data,
+        )
+
+    def to_knx(self) -> bytearray:
+        """Serialize to KNX/IP raw data."""
+        if not 0 <= self.address <= 0xFFFFFF:
+            raise ConversionError("Address out of range.")
+        if not 0 <= self.return_code <= 255:
+            raise ConversionError("Return code out of range.")
+
+        size = len(self.confirmation_data)
+        payload = struct.pack(
+            f"!BI{size}s", self.return_code, self.address, self.confirmation_data
+        )
+        # suppress first byte of address
+        payload = payload[:1] + payload[2:]
+
+        return encode_cmd_and_payload(self.CODE, 0, appended_payload=payload)
+
+    def __str__(self) -> str:
+        """Return object as readable string."""
+        return f'<MemoryExtendedWriteResponse return_code="{self.return_code}" address="{hex(self.address)}" confirmation_data="{self.confirmation_data.hex()}" />'
+
+
+class MemoryExtendedRead(APCI):
+    """
+    MemoryExtendedRead service.
+
+    Payload indicates count and address (16 MiB).
+    """
+
+    CODE = APCIService.MEMORY_EXTENDED_READ
+
+    def __init__(self, count: int, address: int) -> None:
+        """Initialize a new instance of MemoryExtendedRead."""
+        self.count = count
+        self.address = address
+
+    def calculated_length(self) -> int:
+        """Get length of APCI payload."""
+        return 5
+
+    @classmethod
+    def from_knx(cls, raw: bytes) -> MemoryExtendedRead:
+        """Parse/deserialize from KNX/IP raw data."""
+        count = raw[2]
+        address = int.from_bytes(raw[3:6], "big")
+
+        return cls(
+            count=count,
+            address=address,
+        )
+
+    def to_knx(self) -> bytearray:
+        """Serialize to KNX/IP raw data."""
+        if not 0 <= self.address <= 0xFFFFFF:
+            raise ConversionError("Address out of range.")
+        if not 0 <= self.count <= 250:
+            raise ConversionError("Count out of range.")
+
+        payload = struct.pack("!BI", self.count, self.address)
+        # suppress first byte of address
+        payload = payload[:1] + payload[2:]
+
+        return encode_cmd_and_payload(self.CODE, 0, appended_payload=payload)
+
+    def __str__(self) -> str:
+        """Return object as readable string."""
+        return (
+            f'<MemoryExtendedRead count="{self.count}" address="{hex(self.address)}" />'
+        )
+
+
+class MemoryExtendedReadResponse(APCI):
+    """
+    MemoryExtendedReadResponse service.
+
+    Payload indicates return code, address (16 MiB) and data.
+    """
+
+    CODE = APCIService.MEMORY_EXTENDED_READ_RESPONSE
+
+    def __init__(self, return_code: int, address: int, data: bytes = b"") -> None:
+        """Initialize a new instance of MemoryExtendedReadResponse."""
+        self.return_code = return_code
+        self.address = address
+        self.data = data
+
+    def calculated_length(self) -> int:
+        """Get length of APCI payload."""
+        return 5 + len(self.data)
+
+    @classmethod
+    def from_knx(cls, raw: bytes) -> MemoryExtendedReadResponse:
+        """Parse/deserialize from KNX/IP raw data."""
+        return_code = raw[2]
+        address = int.from_bytes(raw[3:6], "big")
+        data = raw[6:]
+
+        return cls(
+            return_code=return_code,
+            address=address,
+            data=data,
+        )
+
+    def to_knx(self) -> bytearray:
+        """Serialize to KNX/IP raw data."""
+        if not 0 <= self.address <= 0xFFFFFF:
+            raise ConversionError("Address out of range.")
+
+        if not 0 <= self.return_code <= 255:
+            raise ConversionError("Return code out of range.")
+
+        size = len(self.data)
+        payload = struct.pack(f"!BI{size}s", self.return_code, self.address, self.data)
+        # suppress first byte of address
+        payload = payload[:1] + payload[2:]
+
+        return encode_cmd_and_payload(self.CODE, 0, appended_payload=payload)
+
+    def __str__(self) -> str:
+        """Return object as readable string."""
+        return f'<MemoryExtendedReadResponse return_code="{self.return_code}" address="{hex(self.address)}" data="{self.data.hex()}" />'
+
+
 class MemoryRead(APCI):
     """
     MemoryRead service.


Test patch:
diff --git a/test/telegram_tests/apci_test.py b/test/telegram_tests/apci_test.py
index 8ddc3cdb..f8d85772 100644
--- a/test/telegram_tests/apci_test.py
+++ b/test/telegram_tests/apci_test.py
@@ -24,6 +24,10 @@ from xknx.telegram.apci import (
     IndividualAddressSerialResponse,
     IndividualAddressSerialWrite,
     IndividualAddressWrite,
+    MemoryExtendedRead,
+    MemoryExtendedReadResponse,
+    MemoryExtendedWrite,
+    MemoryExtendedWriteResponse,
     MemoryRead,
     MemoryResponse,
     MemoryWrite,
@@ -298,6 +302,241 @@ class TestADCResponse:
         assert str(payload) == '<ADCResponse channel="1" count="3" value="456" />'
 
 
+class TestMemoryExtendedWrite:
+    """Test class for MemoryExtendedWrite objects."""
+
+    def test_calculated_length(self):
+        """Test the test_calculated_length method."""
+        payload = MemoryExtendedWrite(
+            address=0x123456, count=3, data=bytes([0xAA, 0xBB, 0xCC])
+        )
+        assert payload.calculated_length() == 8
+
+    def test_from_knx(self):
+        """Test the from_knx method."""
+        payload = APCI.from_knx(
+            bytes([0x01, 0xFB, 0x03, 0x12, 0x34, 0x56, 0xAA, 0xBB, 0xCC])
+        )
+
+        assert payload == MemoryExtendedWrite(
+            address=0x123456, count=3, data=bytes([0xAA, 0xBB, 0xCC])
+        )
+
+    def test_to_knx(self):
+        """Test the to_knx method."""
+        payload = MemoryExtendedWrite(
+            address=0x123456, count=3, data=bytes([0xAA, 0xBB, 0xCC])
+        )
+
+        assert payload.to_knx() == bytes(
+            [0x01, 0xFB, 0x03, 0x12, 0x34, 0x56, 0xAA, 0xBB, 0xCC]
+        )
+
+    def test_to_knx_conversion_error(self):
+        """Test the to_knx method for conversion errors."""
+        payload = MemoryExtendedWrite(
+            address=0xAABBCCDD, count=3, data=bytes([0xAA, 0xBB, 0xCC])
+        )
+
+        with pytest.raises(ConversionError, match=r".*Address.*"):
+            payload.to_knx()
+
+        payload = MemoryExtendedWrite(
+            address=0x123456, count=256, data=bytes([0xAA, 0xBB, 0xCC])
+        )
+
+        with pytest.raises(ConversionError, match=r".*Count.*"):
+            payload.to_knx()
+
+    def test_str(self):
+        """Test the __str__ method."""
+        payload = MemoryExtendedWrite(
+            address=0x123456, count=3, data=bytes([0xAA, 0xBB, 0xCC])
+        )
+
+        assert (
+            str(payload)
+            == '<MemoryExtendedWrite address="0x123456" count="3" data="aabbcc" />'
+        )
+
+
+class TestMemoryExtendedWriteResponse:
+    """Test class for MemoryExtendedWrite objects."""
+
+    def test_calculated_length(self):
+        """Test the test_calculated_length method."""
+        payload = MemoryExtendedWriteResponse(return_code=0, address=0x123456)
+        assert payload.calculated_length() == 5
+
+    def test_calculated_lengt_with_confirmation_data(self):
+        """Test the test_calculated_length method."""
+        payload = MemoryExtendedWriteResponse(
+            return_code=0, address=0x123456, confirmation_data=bytes([0xAA, 0xBB])
+        )
+        assert payload.calculated_length() == 7
+
+    def test_from_knx(self):
+        """Test the from_knx method."""
+        payload = APCI.from_knx(bytes([0x01, 0xFC, 0x00, 0x12, 0x34, 0x56]))
+
+        assert payload == MemoryExtendedWriteResponse(
+            return_code=0, address=0x123456, confirmation_data=b""
+        )
+
+    def test_from_knx_with_confirmation_data(self):
+        """Test the from_knx method."""
+        payload = APCI.from_knx(bytes([0x01, 0xFC, 0x01, 0x12, 0x34, 0x56, 0xAA, 0xBB]))
+
+        assert payload == MemoryExtendedWriteResponse(
+            return_code=1, address=0x123456, confirmation_data=bytes([0xAA, 0xBB])
+        )
+
+    def test_to_knx(self):
+        """Test the to_knx method."""
+        payload = MemoryExtendedWriteResponse(return_code=0, address=0x123456)
+
+        assert payload.to_knx() == bytes([0x01, 0xFC, 0x00, 0x12, 0x34, 0x56])
+
+    def test_to_knx_with_confirmation_data(self):
+        """Test the to_knx method."""
+        payload = MemoryExtendedWriteResponse(
+            return_code=1, address=0x123456, confirmation_data=bytes([0xAA, 0xBB])
+        )
+
+        assert payload.to_knx() == bytes(
+            [0x01, 0xFC, 0x01, 0x12, 0x34, 0x56, 0xAA, 0xBB]
+        )
+
+    def test_to_knx_conversion_error(self):
+        """Test the to_knx method for conversion errors."""
+        payload = MemoryExtendedWriteResponse(return_code=0, address=0xAABBCCDD)
+
+        with pytest.raises(ConversionError, match=r".*Address.*"):
+            payload.to_knx()
+
+        payload = MemoryExtendedWriteResponse(return_code=0x100, address=0x123456)
+
+        with pytest.raises(ConversionError, match=r".*Return code.*"):
+            payload.to_knx()
+
+    def test_str(self):
+        """Test the __str__ method."""
+        payload = MemoryExtendedWriteResponse(return_code=0, address=0x123456)
+
+        assert (
+            str(payload)
+            == '<MemoryExtendedWriteResponse return_code="0" address="0x123456" confirmation_data="" />'
+        )
+
+    def test_str_with_confirmation_data(self):
+        """Test the __str__ method."""
+        payload = MemoryExtendedWriteResponse(
+            return_code=1, address=0x123456, confirmation_data=bytes([0xAA, 0xBB])
+        )
+
+        assert (
+            str(payload)
+            == '<MemoryExtendedWriteResponse return_code="1" address="0x123456" confirmation_data="aabb" />'
+        )
+
+
+class TestMemoryExtendedRead:
+    """Test class for MemoryExtendedRead objects."""
+
+    def test_calculated_length(self):
+        """Test the test_calculated_length method."""
+        payload = MemoryExtendedRead(address=0x123456, count=3)
+        assert payload.calculated_length() == 5
+
+    def test_from_knx(self):
+        """Test the from_knx method."""
+        payload = APCI.from_knx(bytes([0x01, 0xFD, 0x03, 0x12, 0x34, 0x56]))
+
+        assert payload == MemoryExtendedRead(address=0x123456, count=3)
+
+    def test_to_knx(self):
+        """Test the to_knx method."""
+        payload = MemoryExtendedRead(address=0x123456, count=3)
+
+        assert payload.to_knx() == bytes([0x01, 0xFD, 0x03, 0x12, 0x34, 0x56])
+
+    def test_to_knx_conversion_error(self):
+        """Test the to_knx method for conversion errors."""
+        payload = MemoryExtendedRead(address=0xAABBCCDD, count=3)
+
+        with pytest.raises(ConversionError, match=r".*Address.*"):
+            payload.to_knx()
+
+        payload = MemoryExtendedRead(address=0x123456, count=256)
+
+        with pytest.raises(ConversionError, match=r".*Count.*"):
+            payload.to_knx()
+
+    def test_str(self):
+        """Test the __str__ method."""
+        payload = MemoryExtendedRead(address=0x123456, count=3)
+
+        assert str(payload) == '<MemoryExtendedRead count="3" address="0x123456" />'
+
+
+class TestMemoryExtendedReadResponse:
+    """Test class for MemoryExtendedReadResponse objects."""
+
+    def test_calculated_length(self):
+        """Test the test_calculated_length method."""
+        payload = MemoryExtendedReadResponse(
+            return_code=0, address=0x123456, data=bytes([0xAA, 0xBB, 0xCC])
+        )
+        assert payload.calculated_length() == 8
+
+    def test_from_knx(self):
+        """Test the from_knx method."""
+        payload = APCI.from_knx(
+            bytes([0x01, 0xFE, 0x00, 0x12, 0x34, 0x56, 0xAA, 0xBB, 0xCC])
+        )
+
+        assert payload == MemoryExtendedReadResponse(
+            return_code=0, address=0x123456, data=bytes([0xAA, 0xBB, 0xCC])
+        )
+
+    def test_to_knx(self):
+        """Test the to_knx method."""
+        payload = MemoryExtendedReadResponse(
+            return_code=0, address=0x123456, data=bytes([0xAA, 0xBB, 0xCC])
+        )
+
+        assert payload.to_knx() == bytes(
+            [0x01, 0xFE, 0x00, 0x12, 0x34, 0x56, 0xAA, 0xBB, 0xCC]
+        )
+
+    def test_to_knx_conversion_error(self):
+        """Test the to_knx method for conversion errors."""
+        payload = MemoryExtendedReadResponse(
+            return_code=0, address=0xAABBCCDD, data=bytes([0xAA, 0xBB, 0xCC])
+        )
+
+        with pytest.raises(ConversionError, match=r".*Address.*"):
+            payload.to_knx()
+
+        payload = MemoryExtendedReadResponse(
+            return_code=0x100, address=0x123456, data=bytes([0xAA, 0xBB, 0xCC])
+        )
+
+        with pytest.raises(ConversionError, match=r".*Return code.*"):
+            payload.to_knx()
+
+    def test_str(self):
+        """Test the __str__ method."""
+        payload = MemoryExtendedReadResponse(
+            return_code=0, address=0x123456, data=bytes([0xAA, 0xBB, 0xCC])
+        )
+
+        assert (
+            str(payload)
+            == '<MemoryExtendedReadResponse return_code="0" address="0x123456" data="aabbcc" />'
+        )
+
+
 class TestMemoryRead:
     """Test class for MemoryRead objects."""