Repository: XKNX/xknx
Pull request #1395 (2024-01-26 11:16:12)

Serial number-based individual address read/write
## Description

Adds two new management procedures: `nm_individual_address_serial_number_read` and `nm_individual_address_serial_number_write`, which can be used to set and read the individual address based on the unique KNX serial number (if available).


Fixes # (issue)

## Type of change

- [ ] 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

- [x] The documentation has been adjusted accordingly
- [x] Tests have been added that prove the fix is effective or that the feature works
- [x] The changes are documented in the changelog (docs/changelog.md)

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/XKNX/xknx/pull/1395?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Comparison is base [(`2a9e95b`)](https://app.codecov.io/gh/XKNX/xknx/commit/2a9e95becbc3731a63546b4bdfcfe641a55c3d45?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) 96.60% compared to head [(`cde41bb`)](https://app.codecov.io/gh/XKNX/xknx/pull/1395?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) 96.61%.
> Report is 1 commits behind head on main.

<details><summary>Additional details and impacted files</summary>


[![Impacted file tree graph](https://app.codecov.io/gh/XKNX/xknx/pull/1395/graphs/tree.svg?width=650&height=150&src=pr&token=irWbIygS84&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)](https://app.codecov.io/gh/XKNX/xknx/pull/1395?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)

```diff
@@           Coverage Diff           @@
##             main    #1395   +/-   ##
=======================================
  Coverage   96.60%   96.61%           
=======================================
  Files         150      150           
  Lines        9754     9768   +14     
=======================================
+ Hits         9423     9437   +14     
  Misses        331      331           
```


| [Files](https://app.codecov.io/gh/XKNX/xknx/pull/1395?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) | Coverage Δ | |
|---|---|---|
| [xknx/management/procedures.py](https://app.codecov.io/gh/XKNX/xknx/pull/1395?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9tYW5hZ2VtZW50L3Byb2NlZHVyZXMucHk=) | `97.05% <100.00%> (+0.76%)` | :arrow_up: |

</details>
farmio: Have you tested this on real world devices? I am wondering if it can yield any race conditions / awkward responses when we send `A_IndividualAddressSerialNumber_Write-PDU` and `A_IndividualAddressSerialNumber_Read-PDU` without any wait time in between to let the devices process. The specs don't suggest to do so, I am just curious if this actually works the way it should 🙃.
opieters: > Have you tested this on real world devices? I am wondering if it can yield any race conditions / awkward responses when we send `A_IndividualAddressSerialNumber_Write-PDU` and `A_IndividualAddressSerialNumber_Read-PDU` without any wait time in between to let the devices process. The specs don't suggest to do so, I am just curious if this actually works the way it should 🙃.

So far, this was only tested on the KNX Sentido v4 from Basalte, which did not generate any weird error and/or response in the absence of a delay between writing and reading the individual address. Not sure about other devices though, but the spec does not indicate the need for a delay. Maybe it is assumed that the delay between frames should be sufficient for the device to process the write request before the read is transmitted?

Patch:
diff --git a/README.md b/README.md
index 040dfbdc..5faae532 100644
--- a/README.md
+++ b/README.md
@@ -24,6 +24,13 @@ Setting up your local environment:
 1. Install requirements: `pip install -r requirements/testing.txt`
 2. Install pre-commit hook: `pre-commit install`
 
+## Testing
+
+To run all tests, linters, formatters and type checker call `tox`
+
+Running only unit tests is possible with `pytest`
+Running specific unit tests can be invoked by: `pytest -vv test/management_tests/procedures_test.py -k test_nm_individual_address_serial_number_write_fail`
+
 ## Home-Assistant
 
 XKNX is the underlying library for the KNX integration in [Home Assistant](https://home-assistant.io/).
diff --git a/docs/changelog.md b/docs/changelog.md
index ad618290..f6b82de6 100644
--- a/docs/changelog.md
+++ b/docs/changelog.md
@@ -10,7 +10,7 @@ nav_order: 2
 
 ### Connection
 
-- Add new management routines for device management: `nm_invididual_address_write` and `nm_individual_address_read`.
+- Add new management routines for device management: `nm_invididual_address_write`,  `nm_individual_address_read`, `nm_individual_address_serial_number_read` and `nm_individual_address_serial_number_write`.
 - Add handling mechanism for broadcast telegrams in the management class.
 
 ### Secure
diff --git a/examples/example_write_serial_number.py b/examples/example_write_serial_number.py
new file mode 100644
index 00000000..06f1c9f5
--- /dev/null
+++ b/examples/example_write_serial_number.py
@@ -0,0 +1,33 @@
+"""Example for writing an individual address to a KNX device."""
+import asyncio
+import sys
+
+from xknx import XKNX
+from xknx.management import procedures
+from xknx.telegram import IndividualAddress
+
+
+async def main(argv: list[str]) -> int:
+    """Write the individual address to a device."""
+
+    if len(argv) != 3:
+        print(
+            f"{argv[0]}: required arguments: serial number formatted as f0:18:fa:23:d2:00 and individual address."
+        )
+        return 1
+
+    serial = bytes.fromhex(argv[1].replace(":", ""))
+    address = IndividualAddress(argv[2])
+
+    print(f"Setting address {address} to device with serial {serial}")
+
+    async with XKNX() as xknx:
+        await procedures.nm_individual_address_serial_number_write(
+            xknx, serial, address
+        )
+
+    return 0
+
+
+if __name__ == "__main__":
+    asyncio.run(main(sys.argv))
diff --git a/requirements/production.txt b/requirements/production.txt
index 1c882439..e6ef5ea0 100644
--- a/requirements/production.txt
+++ b/requirements/production.txt
@@ -1,3 +1,3 @@
 async-timeout==4.0.2;python_version<"3.11"
-cryptography==41.0.7
+cryptography==42.0.1
 ifaddr==0.2.0
diff --git a/xknx/management/procedures.py b/xknx/management/procedures.py
index fbff773f..fb225ded 100644
--- a/xknx/management/procedures.py
+++ b/xknx/management/procedures.py
@@ -184,3 +184,62 @@ async def nm_invididual_address_write(
             tpci=tpci.TDataConnected(sequence_number=seq_num),
         )
         await xknx.cemi_handler.send_telegram(telegram)
+
+
+async def nm_individual_address_serial_number_read(
+    xknx: XKNX, serial: bytes, timeout: float = 3
+) -> IndividualAddress | None:
+    """Read individual address from device with specified serial number."""
+
+    # initialize queue or event handler gathering broadcasts
+    async with xknx.management.broadcast() as bc_context:
+        broadcast_telegram = Telegram(
+            destination_address=GroupAddress("0/0/0"),
+            source_address=xknx.current_address,
+            payload=apci.IndividualAddressSerialRead(serial=serial),
+        )
+        await xknx.management.send_broadcast(broadcast_telegram)
+        async for result in bc_context.receive(timeout=timeout):
+            if (
+                isinstance(result.payload, apci.IndividualAddressSerialResponse)
+                and result.payload.serial == serial
+            ):
+                return result.source_address
+
+    return None
+
+
+async def nm_individual_address_serial_number_write(
+    xknx: XKNX, serial: bytes, individual_address: IndividualAddressableType
+) -> None:
+    """Write individual address to device with specified serial number."""
+    individual_address = IndividualAddress(individual_address)
+    await xknx.management.send_broadcast(
+        Telegram(
+            destination_address=GroupAddress("0/0/0"),
+            payload=apci.IndividualAddressSerialWrite(
+                address=individual_address, serial=serial
+            ),
+        )
+    )
+    logger.debug(
+        "Wrote new address %s to device with serial number %s.",
+        individual_address,
+        serial,
+    )
+
+    address = await nm_individual_address_serial_number_read(xknx=xknx, serial=serial)
+
+    if address is None:
+        raise ManagementConnectionError(f"No reply received from {serial!r}.")
+
+    if address != individual_address:
+        raise ManagementConnectionError(
+            f"Failed to write serial address {individual_address} to device with serial {serial!r}. Detected {address}"
+        )
+
+    logger.debug(
+        "New address %s validated on device with serial number %s.",
+        individual_address,
+        serial,
+    )


Test patch:
diff --git a/requirements/testing.txt b/requirements/testing.txt
index 464bed18..3274fdb2 100644
--- a/requirements/testing.txt
+++ b/requirements/testing.txt
@@ -2,7 +2,7 @@
 pre-commit==3.6.0
 flake8==7.0.0
 pylint==3.0.3
-pytest==7.4.4
+pytest==8.0.0
 pytest-asyncio==0.21.1
 pytest-cov==4.1.0
 pytest-icdiff==0.9
diff --git a/test/management_tests/procedures_test.py b/test/management_tests/procedures_test.py
index 43d62255..0434c055 100644
--- a/test/management_tests/procedures_test.py
+++ b/test/management_tests/procedures_test.py
@@ -594,3 +594,189 @@ async def test_nm_individual_address_write_address_found_other_in_programming_mo
 
     with pytest.raises(ManagementConnectionError):
         await task
+
+
+async def test_nm_individual_address_serial_number_read(time_travel):
+    """Test nm_individual_address_serial_number_read."""
+
+    xknx = XKNX()
+    xknx.cemi_handler = AsyncMock()
+    individual_address = IndividualAddress("1.1.5")
+    serial_number = b"aabbccddeeff"
+
+    task = asyncio.create_task(
+        procedures.nm_individual_address_serial_number_read(
+            xknx=xknx, serial=serial_number
+        )
+    )
+
+    read_address = Telegram(
+        destination_address=GroupAddress("0/0/0"),
+        payload=apci.IndividualAddressSerialRead(serial=serial_number),
+    )
+    address_reply = Telegram(
+        source_address=individual_address,
+        destination_address=GroupAddress("0/0/0"),
+        direction=TelegramDirection.INCOMING,
+        payload=apci.IndividualAddressSerialResponse(
+            address=individual_address, serial=serial_number
+        ),
+    )
+
+    await time_travel(0)
+    assert xknx.cemi_handler.send_telegram.call_args_list == [
+        call(read_address),
+    ]
+    xknx.management.process(address_reply)
+
+    assert await task == individual_address
+
+
+async def test_nm_individual_address_serial_number_read_fail(time_travel):
+    """Test nm_individual_address_serial_number_read."""
+
+    xknx = XKNX()
+    xknx.cemi_handler = AsyncMock()
+    serial_number = b"aabbccddeeff"
+
+    task = asyncio.create_task(
+        procedures.nm_individual_address_serial_number_read(
+            xknx=xknx, serial=serial_number
+        )
+    )
+
+    read_address = Telegram(
+        destination_address=GroupAddress("0/0/0"),
+        payload=apci.IndividualAddressSerialRead(serial=serial_number),
+    )
+
+    await time_travel(0)
+    assert xknx.cemi_handler.send_telegram.call_args_list == [
+        call(read_address),
+    ]
+    await time_travel(3)
+
+    assert await task is None
+
+
+async def test_nm_individual_address_serial_number_write(time_travel):
+    """Test nm_individual_address_serial_number_write."""
+
+    xknx = XKNX()
+    xknx.cemi_handler = AsyncMock()
+    serial_number = b"aabbccddeeff"
+    individual_address = IndividualAddress("1.1.5")
+
+    task = asyncio.create_task(
+        procedures.nm_individual_address_serial_number_write(
+            xknx=xknx, serial=serial_number, individual_address=individual_address
+        )
+    )
+
+    write_address = Telegram(
+        destination_address=GroupAddress("0/0/0"),
+        payload=apci.IndividualAddressSerialWrite(
+            serial=serial_number, address=individual_address
+        ),
+    )
+    read_address = Telegram(
+        destination_address=GroupAddress("0/0/0"),
+        payload=apci.IndividualAddressSerialRead(serial=serial_number),
+    )
+    address_reply = Telegram(
+        source_address=individual_address,
+        destination_address=GroupAddress("0/0/0"),
+        direction=TelegramDirection.INCOMING,
+        payload=apci.IndividualAddressSerialResponse(
+            address=individual_address, serial=serial_number
+        ),
+    )
+
+    await time_travel(0)
+    assert xknx.cemi_handler.send_telegram.call_args_list == [
+        call(write_address),
+        call(read_address),
+    ]
+    xknx.management.process(address_reply)
+    await task
+
+
+async def test_nm_individual_address_serial_number_write_fail_no_response(time_travel):
+    """Test nm_individual_address_serial_number_write."""
+
+    xknx = XKNX()
+    xknx.cemi_handler = AsyncMock()
+    serial_number = b"aabbccddeeff"
+    individual_address = IndividualAddress("1.1.5")
+
+    task = asyncio.create_task(
+        procedures.nm_individual_address_serial_number_write(
+            xknx=xknx, serial=serial_number, individual_address=individual_address
+        )
+    )
+
+    write_address = Telegram(
+        destination_address=GroupAddress("0/0/0"),
+        payload=apci.IndividualAddressSerialWrite(
+            serial=serial_number, address=individual_address
+        ),
+    )
+    read_address = Telegram(
+        destination_address=GroupAddress("0/0/0"),
+        payload=apci.IndividualAddressSerialRead(serial=serial_number),
+    )
+
+    await time_travel(0)
+    assert xknx.cemi_handler.send_telegram.call_args_list == [
+        call(write_address),
+        call(read_address),
+    ]
+    await time_travel(3)
+    with pytest.raises(ManagementConnectionError):
+        await task
+
+
+async def test_nm_individual_address_serial_number_write_fail_wrong_address(
+    time_travel,
+):
+    """Test nm_individual_address_serial_number_write."""
+
+    xknx = XKNX()
+    xknx.cemi_handler = AsyncMock()
+    serial_number = b"aabbccddeeff"
+    individual_address_tx = IndividualAddress("1.1.5")
+    individual_address_rx = IndividualAddress("1.1.6")
+
+    task = asyncio.create_task(
+        procedures.nm_individual_address_serial_number_write(
+            xknx=xknx, serial=serial_number, individual_address=individual_address_tx
+        )
+    )
+
+    write_address = Telegram(
+        destination_address=GroupAddress("0/0/0"),
+        payload=apci.IndividualAddressSerialWrite(
+            serial=serial_number, address=individual_address_tx
+        ),
+    )
+    read_address = Telegram(
+        destination_address=GroupAddress("0/0/0"),
+        payload=apci.IndividualAddressSerialRead(serial=serial_number),
+    )
+    address_reply = Telegram(
+        source_address=individual_address_rx,
+        destination_address=GroupAddress("0/0/0"),
+        direction=TelegramDirection.INCOMING,
+        payload=apci.IndividualAddressSerialResponse(
+            address=individual_address_rx, serial=serial_number
+        ),
+    )
+
+    await time_travel(0)
+    assert xknx.cemi_handler.send_telegram.call_args_list == [
+        call(write_address),
+        call(read_address),
+    ]
+    xknx.management.process(address_reply)
+    with pytest.raises(ManagementConnectionError):
+        await task