diff --git a/docs/changelog.md b/docs/changelog.md index df2d4ce8e..4d4ae868b 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -22,6 +22,15 @@ nav_order: 2 - `CEMIFlags.EXTENDED_FRAME_FORMAT` was removed; its value `0x0001` was reserved, not an "extended frame format" indicator - `0x0000` is used for standard frames as well as for long extended frames. `CEMIFlags.LTE_FRAME_FORMAT` and `CEMIFlags.EXTENDED_FRAME_FORMAT_MASK` were added instead. - Add explicit length checks to every remaining APCI `from_knx` (and the top-level `APCI.from_knx` dispatcher) as defense-in-depth on top of the broad `except (IndexError, struct.error, ValueError)` added in 3.17.0: each service now raises `ConversionError` with a specific "Invalid length for A_X in CEMI" message for a truncated, malformed or overlong frame instead of relying solely on the generic dispatcher-level catch. +### Breaking Changes + +- Remove the `nm_invididual_address_write` typo alias for `nm_individual_address_write` in `xknx.management.procedures` — the misspelled name is no longer exported. + +### New Features + +- Add `dm_restart_r_co(conn)` and `nm_individual_address_check_conn(conn)` to `xknx.management.procedures` — variants of `dm_restart`/`nm_individual_address_check` that operate on an already-open `P2PConnection` instead of opening and closing their own, for chaining several procedures over one connection. `dm_restart_r_co` is the actual KNX v02.01.02 - Management Procedures 03.05.02 - §3.7.3 procedure name; the `_conn` suffix on the others is an xknx-only naming convention, not a KNX spec name. All existing top-level procedures (`dm_restart`, `nm_individual_address_check`, `nm_individual_address_write`, `nm_individual_address_read`, `nm_individual_address_serial_number_read`/`_write`) keep their `(xknx, ...)` signature unchanged. +- Add `P2PConnection.send_data(payload, wait_for_ack=True)` for sending a telegram, optionally without waiting for an ACK (used internally by `dm_restart`/`dm_restart_r_co` and `nm_individual_address_write` instead of open-coding the same `TDataConnected` construction in multiple procedures). + # 3.17.0 APCIs and DPTs 2026-07-25 ### Deprecation notes diff --git a/test/management_tests/management_test.py b/test/management_tests/management_test.py index 9956d0846..e8fafe5bf 100644 --- a/test/management_tests/management_test.py +++ b/test/management_tests/management_test.py @@ -10,6 +10,7 @@ CommunicationError, ConfirmationError, ManagementConnectionError, + ManagementConnectionRefused, ManagementConnectionTimeout, ) from xknx.management.management import MANAGAMENT_ACK_TIMEOUT @@ -138,6 +139,28 @@ async def test_failed_connect_disconnect() -> None: await conn_1.disconnect() +async def test_send_on_disconnected_connection() -> None: + """Test send_data and request raise once the connection is closed.""" + xknx = XKNX() + xknx.cemi_handler = AsyncMock() + ia = IndividualAddress("4.0.1") + + conn = await xknx.management.connect(ia) + await conn.disconnect() + + with pytest.raises(ManagementConnectionRefused): + await conn.send_data(apci.Restart(), wait_for_ack=False) + + with pytest.raises(ManagementConnectionRefused): + await conn.send_data(apci.Restart()) + + with pytest.raises(ManagementConnectionRefused): + await conn.request( + payload=apci.DeviceDescriptorRead(descriptor=0), + expected=apci.DeviceDescriptorResponse, + ) + + async def test_reject_incoming_connection() -> None: """Test rejecting incoming transport connections.""" # Note: incoming L_DATA.ind indication connection requests are rejected diff --git a/test/management_tests/procedures/device/test_dm_restart_r_co.py b/test/management_tests/procedures/device/test_dm_restart_r_co.py index 2cb4e7ec8..ced9aeec4 100644 --- a/test/management_tests/procedures/device/test_dm_restart_r_co.py +++ b/test/management_tests/procedures/device/test_dm_restart_r_co.py @@ -1,14 +1,42 @@ -"""Tests for dm_restart — KNX 03.05.02 §3.7.3 DM_Restart_RCo.""" +"""Tests for dm_restart — KNX v02.01.02 - Management Procedures 03.05.02 - §3.7.3 DM_Restart_RCo.""" from unittest.mock import AsyncMock, call from xknx import XKNX -from xknx.management.procedures.device.dm_restart_r_co import dm_restart +from xknx.management.procedures.device.dm_restart_r_co import ( + dm_restart, + dm_restart_r_co, +) from xknx.telegram import IndividualAddress, Telegram, apci, tpci +async def test_dm_restart_r_co() -> None: + """Test dm_restart_r_co on an already-open connection.""" + xknx = XKNX() + xknx.cemi_handler = AsyncMock() + individual_address = IndividualAddress("4.0.10") + + connect = Telegram(destination_address=individual_address, tpci=tpci.TConnect()) + restart = Telegram( + destination_address=individual_address, + tpci=tpci.TDataConnected(0), + payload=apci.Restart(), + ) + disconnect = Telegram( + destination_address=individual_address, + tpci=tpci.TDisconnect(), + ) + async with xknx.management.connection(individual_address) as conn: + await dm_restart_r_co(conn) + assert xknx.cemi_handler.send_telegram.call_args_list == [ + call(connect), + call(restart), + call(disconnect), + ] + + async def test_dm_restart() -> None: - """Test dm_restart.""" + """Test dm_restart opens and closes its own connection.""" xknx = XKNX() xknx.cemi_handler = AsyncMock() individual_address = IndividualAddress("4.0.10") diff --git a/test/management_tests/procedures/network/test_nm_individual_address_check.py b/test/management_tests/procedures/network/test_nm_individual_address_check.py index 5463da5f3..ff3579807 100644 --- a/test/management_tests/procedures/network/test_nm_individual_address_check.py +++ b/test/management_tests/procedures/network/test_nm_individual_address_check.py @@ -1,11 +1,13 @@ -"""Tests for nm_individual_address_check — KNX 03.05.02 §2.19 NM_IndividualAddress_Check.""" +"""Tests for nm_individual_address_check — KNX v02.01.02 - Management Procedures 03.05.02 - §2.19 NM_IndividualAddress_Check.""" import asyncio from unittest.mock import AsyncMock, call from xknx import XKNX +from xknx.exceptions import ManagementConnectionRefused from xknx.management.procedures.network.nm_individual_address_check import ( nm_individual_address_check, + nm_individual_address_check_conn, ) from xknx.telegram import ( IndividualAddress, @@ -16,8 +18,84 @@ ) +async def test_nm_individual_address_check_conn_success() -> None: + """Test nm_individual_address_check_conn when device responds normally.""" + xknx = XKNX() + xknx.cemi_handler = AsyncMock() + individual_address = IndividualAddress("4.0.10") + + connect = Telegram(destination_address=individual_address, tpci=tpci.TConnect()) + device_desc_read = Telegram( + destination_address=individual_address, + tpci=tpci.TDataConnected(0), + payload=apci.DeviceDescriptorRead(descriptor=0), + ) + ack = Telegram( + source_address=individual_address, + destination_address=IndividualAddress(0), + direction=TelegramDirection.INCOMING, + tpci=tpci.TAck(0), + ) + device_desc_resp = Telegram( + source_address=individual_address, + destination_address=IndividualAddress(0), + direction=TelegramDirection.INCOMING, + tpci=tpci.TDataConnected(0), + payload=apci.DeviceDescriptorResponse(), + ) + async with xknx.management.connection(individual_address) as conn: + task = asyncio.create_task(nm_individual_address_check_conn(conn)) + await asyncio.sleep(0) + assert xknx.cemi_handler.send_telegram.call_args_list == [ + call(connect), + call(device_desc_read), + ] + xknx.management.process(ack) + xknx.management.process(device_desc_resp) + assert await task + + +async def test_nm_individual_address_check_conn_refused() -> None: + """Test nm_individual_address_check_conn when device refuses the connection.""" + xknx = XKNX() + xknx.cemi_handler = AsyncMock() + individual_address = IndividualAddress("4.0.10") + + connect = Telegram(destination_address=individual_address, tpci=tpci.TConnect()) + device_desc_read = Telegram( + destination_address=individual_address, + tpci=tpci.TDataConnected(0), + payload=apci.DeviceDescriptorRead(descriptor=0), + ) + ack = Telegram( + source_address=individual_address, + destination_address=IndividualAddress(0), + direction=TelegramDirection.INCOMING, + tpci=tpci.TAck(0), + ) + disconnect = Telegram( + source_address=individual_address, + destination_address=IndividualAddress(0), + direction=TelegramDirection.INCOMING, + tpci=tpci.TDisconnect(), + ) + try: + async with xknx.management.connection(individual_address) as conn: + task = asyncio.create_task(nm_individual_address_check_conn(conn)) + await asyncio.sleep(0) + assert xknx.cemi_handler.send_telegram.call_args_list == [ + call(connect), + call(device_desc_read), + ] + xknx.management.process(disconnect) + xknx.management.process(ack) + assert await task + except ManagementConnectionRefused: + pass + + async def test_nm_individual_address_check_success() -> None: - """Test nm_individual_address_check.""" + """Test nm_individual_address_check opens and closes its own connection.""" xknx = XKNX() xknx.cemi_handler = AsyncMock() individual_address = IndividualAddress("4.0.10") @@ -34,6 +112,11 @@ async def test_nm_individual_address_check_success() -> None: direction=TelegramDirection.INCOMING, tpci=tpci.TAck(0), ) + ack_out = Telegram( + source_address=IndividualAddress(0), + destination_address=individual_address, + tpci=tpci.TAck(0), + ) device_desc_resp = Telegram( source_address=individual_address, destination_address=IndividualAddress(0), @@ -41,20 +124,34 @@ async def test_nm_individual_address_check_success() -> None: tpci=tpci.TDataConnected(0), payload=apci.DeviceDescriptorResponse(), ) + disconnect = Telegram( + destination_address=individual_address, + tpci=tpci.TDisconnect(), + ) + task = asyncio.create_task(nm_individual_address_check(xknx, individual_address)) await asyncio.sleep(0) + xknx.management.process(ack) + xknx.management.process(device_desc_resp) + + assert await task assert xknx.cemi_handler.send_telegram.call_args_list == [ call(connect), call(device_desc_read), + call(ack_out), + call(disconnect), ] - # receive response - xknx.management.process(ack) - xknx.management.process(device_desc_resp) - assert await task -async def test_nm_individual_address_check_refused() -> None: - """Test nm_individual_address_check.""" +async def test_nm_individual_address_check_occupied_by_disconnect() -> None: + """ + Test nm_individual_address_check when the peer disconnects during the check. + + The device sends TDisconnect before answering, so nm_individual_address_check_conn + returns True internally; the connection context manager's own disconnect() then + raises ManagementConnectionRefused (peer already disconnected), which is swallowed + and the address is still reported as found/occupied. + """ xknx = XKNX() xknx.cemi_handler = AsyncMock() individual_address = IndividualAddress("4.0.10") @@ -65,24 +162,27 @@ async def test_nm_individual_address_check_refused() -> None: tpci=tpci.TDataConnected(0), payload=apci.DeviceDescriptorRead(descriptor=0), ) - ack = Telegram( + disconnect_from_device = Telegram( source_address=individual_address, destination_address=IndividualAddress(0), direction=TelegramDirection.INCOMING, - tpci=tpci.TAck(0), + tpci=tpci.TDisconnect(), ) - disconnect = Telegram( + ack_from_device = Telegram( source_address=individual_address, destination_address=IndividualAddress(0), direction=TelegramDirection.INCOMING, - tpci=tpci.TDisconnect(), + tpci=tpci.TAck(0), ) + task = asyncio.create_task(nm_individual_address_check(xknx, individual_address)) await asyncio.sleep(0) assert xknx.cemi_handler.send_telegram.call_args_list == [ call(connect), call(device_desc_read), ] - xknx.management.process(disconnect) - xknx.management.process(ack) + xknx.management.process(disconnect_from_device) + xknx.management.process(ack_from_device) + assert await task + assert len(xknx.cemi_handler.send_telegram.call_args_list) == 2 diff --git a/test/management_tests/procedures/network/test_nm_individual_address_write.py b/test/management_tests/procedures/network/test_nm_individual_address_write.py index 5e4a57962..7e24ab5b0 100644 --- a/test/management_tests/procedures/network/test_nm_individual_address_write.py +++ b/test/management_tests/procedures/network/test_nm_individual_address_write.py @@ -1,4 +1,4 @@ -"""Tests for nm_individual_address_write — KNX 03.05.02 §2.3 NM_IndividualAddress_Write.""" +"""Tests for nm_individual_address_write — KNX v02.01.02 - Management Procedures 03.05.02 - §2.3 NM_IndividualAddress_Write.""" import asyncio from unittest.mock import AsyncMock, call @@ -76,9 +76,7 @@ async def test_nm_individual_address_write(time_travel: EventLoopClockAdvancer) payload=apci.IndividualAddressWrite(address=individual_address_new), ) task = asyncio.create_task( - nm_individual_address_write( - xknx=xknx, individual_address=individual_address_new - ) + nm_individual_address_write(xknx, individual_address_new) ) # make sure first request (address check) times out @@ -138,9 +136,7 @@ async def test_nm_individual_address_write_two_devices_in_programming_mode( ) task = asyncio.create_task( - nm_individual_address_write( - xknx=xknx, individual_address=individual_address_new - ) + nm_individual_address_write(xknx, individual_address_new) ) # make sure first request (address check) times out @@ -189,9 +185,7 @@ async def test_nm_individual_address_write_no_device_programming_mode( ) task = asyncio.create_task( - nm_individual_address_write( - xknx=xknx, individual_address=individual_address_new - ) + nm_individual_address_write(xknx, individual_address_new) ) # make sure first request (address check) times out @@ -220,6 +214,73 @@ async def test_nm_individual_address_write_no_device_programming_mode( assert len(xknx.cemi_handler.send_telegram.call_args_list) == 5 +async def test_nm_individual_address_write_address_occupied_by_disconnect( + time_travel: EventLoopClockAdvancer, +) -> None: + """ + Test nm_individual_address_write when device refuses connection during address check. + + Device sends TDisconnect → nm_individual_address_check(xknx, address) already handles + this internally (its own connection context manager's finally raises + ManagementConnectionRefused from disconnect() after the peer already disconnected, and its + own except swallows it) → returns True. + """ + xknx = XKNX() + xknx.cemi_handler = AsyncMock() + individual_address = IndividualAddress("1.1.4") + + connect = Telegram(destination_address=individual_address, tpci=tpci.TConnect()) + device_desc_read = Telegram( + destination_address=individual_address, + tpci=tpci.TDataConnected(0), + payload=apci.DeviceDescriptorRead(descriptor=0), + ) + disconnect_from_device = Telegram( + source_address=individual_address, + destination_address=IndividualAddress(0), + direction=TelegramDirection.INCOMING, + tpci=tpci.TDisconnect(), + ) + ack_from_device = Telegram( + source_address=individual_address, + destination_address=IndividualAddress(0), + direction=TelegramDirection.INCOMING, + tpci=tpci.TAck(0), + ) + individual_address_read = Telegram( + GroupAddress("0/0/0"), payload=apci.IndividualAddressRead() + ) + + task = asyncio.create_task(nm_individual_address_write(xknx, individual_address)) + + # advance until address check is awaiting ACK + await time_travel(0) + assert xknx.cemi_handler.send_telegram.call_args_list == [ + call(connect), + call(device_desc_read), + ] + + # device sends disconnect (address occupied) then ACK for the DeviceDescriptorRead + xknx.management.process(disconnect_from_device) + xknx.management.process(ack_from_device) + + # nm_individual_address_check's own connection context manager finally raises + # ManagementConnectionRefused (peer already disconnected, no TDisconnect sent by us); + # its own except swallows it and returns True; IndividualAddressRead broadcast is sent + await time_travel(0) + assert xknx.cemi_handler.send_telegram.call_args_list[2:] == [ + call(individual_address_read), + ] + + # no device in programming mode → timeout → error + await time_travel(MANAGAMENT_CONNECTION_TIMEOUT) + with pytest.raises( + ManagementConnectionError, match="No device in programming mode" + ): + await task + assert len(xknx.cemi_handler.send_telegram.call_args_list) == 3 + + async def test_nm_individual_address_write_address_found( time_travel: EventLoopClockAdvancer, ) -> None: @@ -260,9 +321,7 @@ async def test_nm_individual_address_write_address_found( GroupAddress("0/0/0"), payload=apci.IndividualAddressRead() ) - task = asyncio.create_task( - nm_individual_address_write(xknx=xknx, individual_address=individual_address) - ) + task = asyncio.create_task(nm_individual_address_write(xknx, individual_address)) # first request (address check) succeeds await time_travel(0) @@ -323,9 +382,7 @@ async def test_nm_individual_address_write_programming_failed( payload=apci.IndividualAddressWrite(address=individual_address_new), ) task = asyncio.create_task( - nm_individual_address_write( - xknx=xknx, individual_address=individual_address_new - ) + nm_individual_address_write(xknx, individual_address_new) ) # make sure first request (address check) times out @@ -404,17 +461,20 @@ async def test_nm_individual_address_write_address_found_other_in_programming_mo payload=apci.IndividualAddressResponse(), ) - task = asyncio.create_task( - nm_individual_address_write(xknx=xknx, individual_address=individual_address) - ) + task = asyncio.create_task(nm_individual_address_write(xknx, individual_address)) - # make sure first request (address check) times out + # first request (address check) succeeds await time_travel(0) xknx.management.process(ack) xknx.management.process(device_desc_resp) - await time_travel(MANAGAMENT_CONNECTION_TIMEOUT) + await time_travel(0) + # a different device answers the programming-mode broadcast before its + # 3s window elapses, so dev_pgm_mode[0] != individual_address xknx.management.process(address_reply_message) + # fast-forward past the broadcast's 3s timeout via the mocked clock, + # instead of waiting on it for real + await time_travel(3) assert xknx.cemi_handler.send_telegram.call_args_list == [ call(connect), @@ -424,5 +484,8 @@ async def test_nm_individual_address_write_address_found_other_in_programming_mo call(individual_address_read), ] - with pytest.raises(ManagementConnectionError): + with pytest.raises( + ManagementConnectionError, + match=f"A device was found with {individual_address}, cannot continue with programming.", + ): await task diff --git a/xknx/management/management.py b/xknx/management/management.py index e2bcd7f1c..ed6806fa6 100644 --- a/xknx/management/management.py +++ b/xknx/management/management.py @@ -225,7 +225,13 @@ def _sequence_number_generator() -> Generator[int, None, None]: seq_num = seq_num + 1 & 0xF async def connect(self) -> None: - """Connect to the KNX device.""" + """ + Connect to the KNX device. + + Sends T_Connect-PDU (= A_Connect at wire level). KNX v02.01.02 - Application Layer + 03.03.07 - §3.5.1: "T_Connect.ind and T_Disconnect.ind are mapped transparently to + A_Connect.ind and A_Disconnect.ind service and passed to the user of Application Layer." + """ connect = Telegram( destination_address=self.address, source_address=self.xknx.current_address, @@ -244,7 +250,7 @@ async def connect(self) -> None: self._connected = True async def disconnect(self) -> None: - """Disconnect from the KNX device.""" + """Disconnect from the KNX device. Sends T_Disconnect-PDU (= A_Disconnect, see connect()).""" if not self._connected: self.disconnect_hook() # remove connection from management class raise ManagementConnectionRefused( @@ -304,17 +310,20 @@ def process(self, telegram: Telegram) -> None: self._response_waiter.set_result(telegram) self._expected_sequence_number = self._expected_sequence_number + 1 & 0xF - async def _send_data(self, payload: APCI) -> None: + async def send_data(self, payload: APCI, wait_for_ack: bool = True) -> None: """ Send a payload to the KNX device. - A response has to be processed by `_receive` before sending the next telegram. + If `wait_for_ack` is True (default), waits for the device to ACK the + telegram - resending once after a timeout - and validates the ACK's + sequence number. The response itself still has to be processed by + `_receive` before sending the next telegram. Set `wait_for_ack=False` + for commands the device doesn't ACK back. """ if not self._connected: raise ManagementConnectionRefused( "Management connection disconnected by the peer." ) - self._ack_waiter = asyncio.get_event_loop().create_future() seq_num = next(self.sequence_number) telegram = Telegram( destination_address=self.address, @@ -322,6 +331,11 @@ async def _send_data(self, payload: APCI) -> None: payload=payload, tpci=TDataConnected(sequence_number=seq_num), ) + if not wait_for_ack: + await self.xknx.cemi_handler.send_telegram(telegram) + return + + self._ack_waiter = asyncio.get_event_loop().create_future() try: await self.xknx.cemi_handler.send_telegram(telegram) async with asyncio_timeout(MANAGAMENT_ACK_TIMEOUT): @@ -392,7 +406,7 @@ async def request(self, payload: APCI, expected: type[APCI] | None) -> Telegram: if time_diff < wait_time: await asyncio.sleep(wait_time - time_diff) - await self._send_data(payload) + await self.send_data(payload) response = await self._receive(expected) self._last_response_time = time.time() return response diff --git a/xknx/management/procedures/__init__.py b/xknx/management/procedures/__init__.py index 978afc19e..3ecfea477 100644 --- a/xknx/management/procedures/__init__.py +++ b/xknx/management/procedures/__init__.py @@ -4,9 +4,12 @@ Subpackages are created when their first procedure lands. Naming mirrors the KNX spec prefix: - - ``network/`` for NM_* procedures (KNX 03.05.02 Network Management) - - ``device/`` for DM_* procedures (KNX 03.05.02 Device Management) - - ``ftp/`` for FTP_* procedures (KNX 03.05.02 §8 File Transfer) + - ``network/`` for NM_* procedures (KNX v02.01.02 - Management Procedures + 03.05.02 - Network Management) + - ``device/`` for DM_* procedures (KNX v02.01.02 - Management Procedures + 03.05.02 - Device Management) + - ``ftp/`` for FTP_* procedures (KNX v02.01.02 - Management Procedures + 03.05.02 - §8 File Transfer) Per-procedure files inside each subpackage host a single public ``async def`` function. This package re-exports every implemented procedure so callers can @@ -14,6 +17,17 @@ individual functions via ``from xknx.management.procedures import ``, or access them as attributes such as ``procedures.``. +Most procedures come in two forms: + + - ``(xknx: XKNX, ...)`` opens (and closes) whatever P2P + connections or broadcasts it needs on its own, via ``xknx.management``. + - ``_conn(conn: P2PConnection, ...)`` operates on an + already-open connection, for chaining several procedures over one + connection. This suffix is an xknx convention, not a KNX spec name. + ``dm_restart_r_co`` is the one exception — ``RCo`` is the actual KNX + v02.01.02 - Management Procedures 03.05.02 - §3.7.3 procedure name for + the connection-based variant of DM_Restart, not the xknx convention. + When adding a new procedure follow the workflow: 1. Create ``procedures//.py`` with the spec text embedded @@ -23,14 +37,12 @@ """ # ruff: noqa: F401 -from .device import dm_restart +from .device import dm_restart, dm_restart_r_co from .network import ( nm_individual_address_check, + nm_individual_address_check_conn, nm_individual_address_read, nm_individual_address_serial_number_read, nm_individual_address_serial_number_write, nm_individual_address_write, ) - -# Backwards-compatibility typo alias (the original module exposed both spellings). -nm_invididual_address_write = nm_individual_address_write diff --git a/xknx/management/procedures/device/__init__.py b/xknx/management/procedures/device/__init__.py index 4ee32f35c..493a9b5ce 100644 --- a/xknx/management/procedures/device/__init__.py +++ b/xknx/management/procedures/device/__init__.py @@ -1,4 +1,4 @@ -"""KNX 03.05.02 Device Management (DM_*) procedures.""" +"""KNX v02.01.02 - Management Procedures 03.05.02 - Device Management (DM_*) procedures.""" # ruff: noqa: F401 -from .dm_restart_r_co import dm_restart +from .dm_restart_r_co import dm_restart, dm_restart_r_co diff --git a/xknx/management/procedures/device/dm_restart_r_co.py b/xknx/management/procedures/device/dm_restart_r_co.py index 96737d34e..75bb03526 100644 --- a/xknx/management/procedures/device/dm_restart_r_co.py +++ b/xknx/management/procedures/device/dm_restart_r_co.py @@ -1,11 +1,12 @@ -"""DM_Restart_RCo — KNX 03.05.02 §3.7.3.""" +"""DM_Restart_RCo — KNX v02.01.02 - Management Procedures 03.05.02 - §3.7.3.""" from __future__ import annotations import logging from typing import TYPE_CHECKING -from xknx.telegram import Telegram, apci, tpci +from xknx.management.management import P2PConnection +from xknx.telegram import apci from xknx.telegram.address import IndividualAddress, IndividualAddressableType if TYPE_CHECKING: @@ -14,23 +15,24 @@ logger = logging.getLogger("xknx.management.procedures") +async def dm_restart_r_co(conn: P2PConnection) -> None: + """ + Restart the device on an already-open connection. + + :param conn: an established P2P connection to the device + """ + logger.debug("Requesting a Basic Restart of %s.", conn.address) + await conn.send_data(apci.Restart(), wait_for_ack=False) + + async def dm_restart(xknx: XKNX, individual_address: IndividualAddressableType) -> None: """ - Restart the device. + Restart a device, opening and closing a connection to it. - :param xknx: XKNX object - :param individual_address: address of device to reset + :param xknx: the XKNX object + :param individual_address: address of the device to restart """ async with xknx.management.connection( - address=IndividualAddress(individual_address) - ) as connection: - logger.debug("Requesting a Basic Restart of %s.", individual_address) - # A_Restart will not be ACKed by the device, so it is manually sent to avoid timeout and retry - seq_num = next(connection.sequence_number) - telegram = Telegram( - destination_address=connection.address, - source_address=xknx.current_address, - payload=apci.Restart(), - tpci=tpci.TDataConnected(sequence_number=seq_num), - ) - await xknx.cemi_handler.send_telegram(telegram) + IndividualAddress(individual_address) + ) as conn: + await dm_restart_r_co(conn) diff --git a/xknx/management/procedures/network/__init__.py b/xknx/management/procedures/network/__init__.py index db751ab7a..d9f19b5b0 100644 --- a/xknx/management/procedures/network/__init__.py +++ b/xknx/management/procedures/network/__init__.py @@ -1,7 +1,10 @@ -"""KNX 03.05.02 Network Management (NM_*) procedures.""" +"""KNX v02.01.02 - Management Procedures 03.05.02 - Network Management (NM_*) procedures.""" # ruff: noqa: F401 -from .nm_individual_address_check import nm_individual_address_check +from .nm_individual_address_check import ( + nm_individual_address_check, + nm_individual_address_check_conn, +) from .nm_individual_address_read import nm_individual_address_read from .nm_individual_address_serial_number_read import ( nm_individual_address_serial_number_read, diff --git a/xknx/management/procedures/network/nm_individual_address_check.py b/xknx/management/procedures/network/nm_individual_address_check.py index 6376371a5..23d4fc767 100644 --- a/xknx/management/procedures/network/nm_individual_address_check.py +++ b/xknx/management/procedures/network/nm_individual_address_check.py @@ -1,4 +1,4 @@ -"""NM_IndividualAddress_Check — KNX 03.05.02 §2.19.""" +"""NM_IndividualAddress_Check — KNX v02.01.02 - Management Procedures 03.05.02 - §2.19.""" from __future__ import annotations @@ -6,6 +6,7 @@ from typing import TYPE_CHECKING from xknx.exceptions import ManagementConnectionRefused, ManagementConnectionTimeout +from xknx.management.management import P2PConnection from xknx.telegram import apci from xknx.telegram.address import IndividualAddress, IndividualAddressableType @@ -15,35 +16,52 @@ logger = logging.getLogger("xknx.management.procedures") +async def nm_individual_address_check_conn(conn: P2PConnection) -> bool: + """ + Check if a device responds on an already-open connection. + + Returns True if the device answers, False on timeout. + ManagementConnectionRefused propagates to the caller. + + :param conn: an established P2P connection to the device + """ + try: + response = await conn.request( + payload=apci.DeviceDescriptorRead(descriptor=0), + expected=apci.DeviceDescriptorResponse, + ) + if isinstance(response.payload, apci.DeviceDescriptorResponse): + logger.debug("Device found at %s", conn.address) + return True + return False + except ManagementConnectionTimeout as ex: + logger.debug("No device answered to connection attempt. %s", ex) + return False + except ManagementConnectionRefused as ex: + # if Disconnect is received immediately, IA is occupied + logger.debug("Device does not support transport layer connections. %s", ex) + return True + + async def nm_individual_address_check( xknx: XKNX, individual_address: IndividualAddressableType ) -> bool: """ - Check if the individual address is occupied on the network. + Check if a device responds, opening and closing a connection to it. + + Returns True if the device answers or refuses the connection (per KNX + v02.01.02 - Management Procedures 03.05.02 - §2.3 step 1, an + A_Disconnect-PDU means the address is occupied), False on timeout. - :param xknx: XKNX object + :param xknx: the XKNX object :param individual_address: address to check """ + address_found = False try: async with xknx.management.connection( - address=IndividualAddress(individual_address) - ) as connection: - try: - response = await connection.request( - payload=apci.DeviceDescriptorRead(descriptor=0), - expected=apci.DeviceDescriptorResponse, - ) - - except ManagementConnectionTimeout as ex: - # if nothing is received (-> timeout) IA is free - logger.debug("No device answered to connection attempt. %s", ex) - return False - if isinstance(response.payload, apci.DeviceDescriptorResponse): - # if response is received IA is occupied - logger.debug("Device found at %s", individual_address) - return True - return False - except ManagementConnectionRefused as ex: - # if Disconnect is received immediately, IA is occupied - logger.debug("Device does not support transport layer connections. %s", ex) - return True + IndividualAddress(individual_address) + ) as conn: + address_found = await nm_individual_address_check_conn(conn) + except ManagementConnectionRefused: + address_found = True + return address_found diff --git a/xknx/management/procedures/network/nm_individual_address_write.py b/xknx/management/procedures/network/nm_individual_address_write.py index 50b4039ae..d12851edc 100644 --- a/xknx/management/procedures/network/nm_individual_address_write.py +++ b/xknx/management/procedures/network/nm_individual_address_write.py @@ -1,18 +1,20 @@ -"""NM_IndividualAddress_Write — KNX 03.05.02 §2.3.""" +"""NM_IndividualAddress_Write — KNX v02.01.02 - Management Procedures 03.05.02 - §2.3.""" from __future__ import annotations import logging from typing import TYPE_CHECKING -from xknx.exceptions import ManagementConnectionError, ManagementConnectionTimeout +from xknx.exceptions import ManagementConnectionError +from xknx.management.procedures.device.dm_restart_r_co import dm_restart_r_co from xknx.management.procedures.network.nm_individual_address_check import ( nm_individual_address_check, + nm_individual_address_check_conn, ) from xknx.management.procedures.network.nm_individual_address_read import ( nm_individual_address_read, ) -from xknx.telegram import Telegram, apci, tpci +from xknx.telegram import apci from xknx.telegram.address import IndividualAddress, IndividualAddressableType if TYPE_CHECKING: @@ -22,12 +24,13 @@ async def nm_individual_address_write( - xknx: XKNX, individual_address: IndividualAddressableType + xknx: XKNX, + individual_address: IndividualAddressableType, ) -> None: """ Write the individual address of a single device in programming mode. - :param xknx: XKNX object + :param xknx: the XKNX object :param individual_address: address to be written to KNX device """ logger.debug("Writing individual address %s to device.", individual_address) @@ -67,31 +70,13 @@ async def nm_individual_address_write( ) logger.debug("Wrote new address %s to device.", individual_address) - async with xknx.management.connection( - address=IndividualAddress(individual_address) - ) as connection: + async with xknx.management.connection(address=individual_address) as connection: logger.debug( "Checking if device exists at %s and restarting it.", individual_address ) - - try: - await connection.request( - payload=apci.DeviceDescriptorRead(descriptor=0), - expected=apci.DeviceDescriptorResponse, - ) - except ManagementConnectionTimeout as ex: - # if nothing is received (-> timeout) IA is free + if not await nm_individual_address_check_conn(connection): raise ManagementConnectionError( - f"No device answered to connection attempt after write address operation. {ex}" - ) from None - - logger.debug("Restating device, exiting programming mode.") - # A_Restart will not be ACKed by the device, so it is manually sent to avoid timeout and retry - seq_num = next(connection.sequence_number) - telegram = Telegram( - destination_address=connection.address, - source_address=xknx.current_address, - payload=apci.Restart(), - tpci=tpci.TDataConnected(sequence_number=seq_num), - ) - await xknx.cemi_handler.send_telegram(telegram) + "No device answered to connection attempt after write address operation." + ) + logger.debug("Restarting device, exiting programming mode.") + await dm_restart_r_co(connection)