From 7264e6a4b5649641b97140f439e74d5017712873 Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Fri, 10 Nov 2023 11:43:20 -0500 Subject: [PATCH] Rewrite command parsing/serialization (#231) * WIP * Replace implicit polling tasks with explicit polling loop * Get API unit tests passing * Fix remaining unit tests * Merge TX and RX schemas into a single object * Match command responses by both command ID and the seq * Increase test coverage * Parse firmware structure * Increase coverage * Restrict `cancelling` to 3.11+ * Add unit test for `utils` * Add unit tests for request retries * Unit test request locking --- .pre-commit-config.yaml | 2 +- tests/test_api.py | 1304 +++++++++++++++++----------- tests/test_application.py | 83 +- tests/test_network_state.py | 65 +- tests/test_send_receive.py | 36 +- tests/test_types.py | 145 +--- tests/test_utils.py | 24 + zigpy_deconz/api.py | 996 ++++++++++++--------- zigpy_deconz/types.py | 324 ++----- zigpy_deconz/uart.py | 2 +- zigpy_deconz/utils.py | 25 + zigpy_deconz/zigbee/application.py | 120 +-- 12 files changed, 1662 insertions(+), 1464 deletions(-) create mode 100644 tests/test_utils.py create mode 100644 zigpy_deconz/utils.py diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 3a2c439..d9ba037 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -19,7 +19,7 @@ repos: - --safe - repo: https://github.com/pycqa/flake8 - rev: 6.0.0 + rev: 6.1.0 hooks: - id: flake8 additional_dependencies: diff --git a/tests/test_api.py b/tests/test_api.py index c350ba6..93997bb 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1,467 +1,508 @@ """Test api module.""" import asyncio -import binascii -import enum +import collections +import inspect +import logging +import sys import pytest import zigpy.config +import zigpy.types as zigpy_t from zigpy_deconz import api as deconz_api, types as t, uart import zigpy_deconz.exception import zigpy_deconz.zigbee.application -from .async_mock import AsyncMock, MagicMock, patch, sentinel +from .async_mock import AsyncMock, MagicMock, call, patch DEVICE_CONFIG = {zigpy.config.CONF_DEVICE_PATH: "/dev/null"} @pytest.fixture -def uart_gw(): - gw = MagicMock(auto_spec=uart.Gateway(MagicMock())) - return gw +def gateway(): + return uart.Gateway(api=None) @pytest.fixture -def api(event_loop, uart_gw): - controller = MagicMock( - spec_set=zigpy_deconz.zigbee.application.ControllerApplication - ) - api = deconz_api.Deconz(controller, {zigpy.config.CONF_DEVICE_PATH: "/dev/null"}) - api._uart = uart_gw - return api - - -async def test_connect(): - controller = MagicMock( - spec_set=zigpy_deconz.zigbee.application.ControllerApplication - ) - api = deconz_api.Deconz(controller, {zigpy.config.CONF_DEVICE_PATH: "/dev/null"}) - - with patch.object(uart, "connect", new=AsyncMock()) as conn_mck: - await api.connect() - assert conn_mck.call_count == 1 - assert conn_mck.await_count == 1 - assert api._uart == conn_mck.return_value - - -def test_close(api): - uart = api._uart - api.close() - assert api._uart is None - assert uart.close.call_count == 1 - - -def test_commands(): - for cmd, cmd_opts in deconz_api.RX_COMMANDS.items(): - assert len(cmd_opts) == 2 - schema, solicited = cmd_opts - assert isinstance(cmd, int) is True - assert isinstance(schema, tuple) is True - assert isinstance(solicited, bool) +def api(gateway, mock_command_rsp): + async def mock_connect(config, api): + gateway._api = api + gateway.connection_made(MagicMock()) + return gateway + + with patch("zigpy_deconz.uart.connect", side_effect=mock_connect): + controller = MagicMock( + spec_set=zigpy_deconz.zigbee.application.ControllerApplication + ) + api = deconz_api.Deconz( + controller, {zigpy.config.CONF_DEVICE_PATH: "/dev/null"} + ) - for cmd, schema in deconz_api.TX_COMMANDS.items(): - assert isinstance(cmd, int) is True - assert isinstance(schema, tuple) is True + mock_command_rsp( + command_id=deconz_api.CommandId.device_state, + params={}, + rsp={ + "status": deconz_api.Status.SUCCESS, + "frame_length": t.uint16_t(8), + "device_state": deconz_api.DeviceState( + network_state=deconz_api.NetworkState2.CONNECTED, + device_state=( + deconz_api.DeviceStateFlags.APSDE_DATA_REQUEST_FREE_SLOTS_AVAILABLE + ), + ), + "reserved1": t.uint8_t(0), + "reserved2": t.uint8_t(0), + }, + ) + mock_command_rsp( + command_id=deconz_api.CommandId.read_parameter, + params={ + "parameter_id": deconz_api.NetworkParameter.protocol_version, + "parameter": t.Bytes(b""), + }, + rsp={ + "status": deconz_api.Status.SUCCESS, + "frame_length": t.uint16_t(10), + "payload_length": t.uint16_t(3), + "parameter_id": deconz_api.NetworkParameter.protocol_version, + "parameter": t.Bytes(t.uint16_t(270).serialize()), + }, + ) -async def test_command(api, monkeypatch): - def mock_api_frame(name, *args): - return sentinel.api_frame_data, api._seq + mock_command_rsp( + command_id=deconz_api.CommandId.version, + params={"reserved": t.uint8_t(0)}, + rsp={ + "status": deconz_api.Status.SUCCESS, + "frame_length": t.uint16_t(9), + "version": deconz_api.FirmwareVersion(645400320), + }, + ) - api._api_frame = MagicMock(side_effect=mock_api_frame) - api._uart.send = MagicMock() + yield api - async def mock_fut(): - return sentinel.cmd_result - monkeypatch.setattr(asyncio, "Future", mock_fut) +@pytest.fixture +async def mock_command_rsp(gateway): + def inner(command_id, params, rsp, *, replace=False): + if ( + getattr(getattr(gateway.send, "side_effect", None), "_handlers", None) + is None + ): - for cmd, cmd_opts in deconz_api.TX_COMMANDS.items(): - ret = await api._command(cmd, sentinel.cmd_data) - assert ret is sentinel.cmd_result - assert api._api_frame.call_count == 1 - assert api._api_frame.call_args[0][0] == cmd - assert api._api_frame.call_args[0][1] == sentinel.cmd_data - assert api._uart.send.call_count == 1 - assert api._uart.send.call_args[0][0] == sentinel.api_frame_data - api._api_frame.reset_mock() - api._uart.send.reset_mock() + def receiver(data): + command, _ = deconz_api.Command.deserialize(data) + tx_schema, _ = deconz_api.COMMAND_SCHEMAS[command.command_id] + schema = {} + for k, v in tx_schema.items(): + if v in (deconz_api.FRAME_LENGTH, deconz_api.PAYLOAD_LENGTH): + v = t.uint16_t + elif not inspect.isclass(v): + v = type(v) -async def test_command_queue(api, monkeypatch): - def mock_api_frame(name, *args): - return sentinel.api_frame_data, api._seq + schema[k] = v - api._api_frame = MagicMock(side_effect=mock_api_frame) - api._uart.send = MagicMock() + kwargs, rest = t.deserialize_dict(command.payload, schema) - monkeypatch.setattr(deconz_api, "COMMAND_TIMEOUT", 0.1) + for params, mock in receiver._handlers[command.command_id]: + if all(kwargs[k] == v for k, v in params.items()): + _, rx_schema = deconz_api.COMMAND_SCHEMAS[command.command_id] + ret = mock(**kwargs) - for cmd, cmd_opts in deconz_api.TX_COMMANDS.items(): - async with api._command_lock: - with pytest.raises(asyncio.TimeoutError): - await asyncio.wait_for(api._command(cmd, sentinel.cmd_data), 0.1) - assert api._api_frame.call_count == 0 - assert api._uart.send.call_count == 0 - api._api_frame.reset_mock() - api._uart.send.reset_mock() + asyncio.get_running_loop().call_soon( + gateway._api.data_received, + deconz_api.Command( + command_id=command.command_id, + seq=command.seq, + payload=t.serialize_dict(ret, rx_schema), + ).serialize(), + ) + receiver._handlers = collections.defaultdict(list) + gateway.send = MagicMock(side_effect=receiver) -async def test_command_timeout(api, monkeypatch): - def mock_api_frame(name, *args): - return sentinel.api_frame_data, api._seq + if replace: + gateway.send.side_effect._handlers[command_id].clear() - api._api_frame = MagicMock(side_effect=mock_api_frame) - api._uart.send = MagicMock() + mock = MagicMock(return_value=rsp) + gateway.send.side_effect._handlers[command_id].append((params, mock)) - monkeypatch.setattr(deconz_api, "COMMAND_TIMEOUT", 0.1) + return mock - for cmd, cmd_opts in deconz_api.TX_COMMANDS.items(): - with pytest.raises(asyncio.TimeoutError): - await api._command(cmd, sentinel.cmd_data) - assert api._api_frame.call_count == 1 - assert api._api_frame.call_args[0][0] == cmd - assert api._api_frame.call_args[0][1] == sentinel.cmd_data - assert api._uart.send.call_count == 1 - assert api._uart.send.call_args[0][0] == sentinel.api_frame_data - api._api_frame.reset_mock() - api._uart.send.reset_mock() + return inner -async def test_command_not_connected(api): - api._uart = None +def send_network_state( + api, + network_state: deconz_api.NetworkState2 = deconz_api.NetworkState2.CONNECTED, + device_state: deconz_api.DeviceStateFlags = ( + deconz_api.DeviceStateFlags.APSDE_DATA_CONFIRM + ), +): + _, rx_schema = deconz_api.COMMAND_SCHEMAS[deconz_api.CommandId.device_state_changed] - def mock_api_frame(name, *args): - return sentinel.api_frame_data, api._seq + data = deconz_api.Command( + command_id=deconz_api.CommandId.device_state_changed, + seq=api._seq, + payload=t.serialize_dict( + { + "status": deconz_api.Status.SUCCESS, + "frame_length": t.uint16_t(7), + "device_state": deconz_api.DeviceState( + network_state=network_state, + device_state=device_state, + ), + "reserved": t.uint8_t(0), + }, + rx_schema, + ), + ).serialize() - api._api_frame = MagicMock(side_effect=mock_api_frame) + asyncio.get_running_loop().call_later(0.01, api.data_received, data) - for cmd, cmd_opts in deconz_api.TX_COMMANDS.items(): - with pytest.raises(deconz_api.CommandError): - await api._command(cmd, sentinel.cmd_data) - assert api._api_frame.call_count == 0 - api._api_frame.reset_mock() +async def test_connect(api, mock_command_rsp): + await api.connect() -def _fake_args(arg_type): - if issubclass(arg_type, enum.Enum): - return list(arg_type)[0] # Pick the first enum value - elif issubclass(arg_type, t.DeconzAddressEndpoint): - addr = t.DeconzAddressEndpoint() - addr.address_mode = t.AddressMode.NWK - addr.address = t.uint8_t(0) - addr.endpoint = t.uint8_t(0) - return addr - elif issubclass(arg_type, t.EUI64): - return t.EUI64([0x01] * 8) - return arg_type() +async def test_close(api): + await api.connect() + uart = api._uart + uart.close = MagicMock(wraps=uart.close) -def test_api_frame(api): - for cmd, schema in deconz_api.TX_COMMANDS.items(): - if schema: - args = [_fake_args(a) for a in schema] - api._api_frame(cmd, *args) - else: - api._api_frame(cmd) + api.close() + assert api._uart is None + assert uart.close.call_count == 1 -def test_data_received(api, monkeypatch): - monkeypatch.setattr( - t, - "deserialize", - MagicMock(return_value=(sentinel.deserialize_data, b"")), - ) - my_handler = MagicMock() - - for cmd, cmd_opts in deconz_api.RX_COMMANDS.items(): - payload = b"\x01\x02\x03\x04" - data = cmd.serialize() + b"\x00\x00\x00\x00" + payload - setattr(api, f"_handle_{cmd.name}", my_handler) - api._awaiting[0] = MagicMock() - api.data_received(data) - assert t.deserialize.call_count == 1 - assert t.deserialize.call_args[0][0] == payload - assert my_handler.call_count == 1 - assert my_handler.call_args[0][0] == sentinel.deserialize_data - t.deserialize.reset_mock() - my_handler.reset_mock() - - -def test_data_received_unk_status(api, monkeypatch): - monkeypatch.setattr( - t, - "deserialize", - MagicMock(return_value=(sentinel.deserialize_data, b"")), - ) - my_handler = MagicMock() - - for cmd, cmd_opts in deconz_api.RX_COMMANDS.items(): - _, solicited = cmd_opts - payload = b"\x01\x02\x03\x04" - status = t.uint8_t(0xFE).serialize() - data = cmd.serialize() + b"\x00" + status + b"\x00\x00" + payload - setattr(api, f"_handle_{cmd.name}", my_handler) - api._awaiting[0] = MagicMock() - api.data_received(data) - if solicited: - assert my_handler.call_count == 0 - assert t.deserialize.call_count == 0 - else: - assert t.deserialize.call_count == 1 - assert my_handler.call_count == 1 - t.deserialize.reset_mock() - my_handler.reset_mock() - - -def test_data_received_unk_cmd(api, monkeypatch): - monkeypatch.setattr( - t, - "deserialize", - MagicMock(return_value=(sentinel.deserialize_data, b"")), +def test_commands(): + for cmd, (tx_schema, rx_schema) in deconz_api.COMMAND_SCHEMAS.items(): + assert isinstance(cmd, deconz_api.CommandId) + assert isinstance(tx_schema, dict) or tx_schema is None + assert isinstance(rx_schema, dict) + + +async def test_command(api): + await api.connect() + + addr = t.DeconzAddress() + addr.address_mode = t.AddressMode.NWK + addr.address = t.NWK(0x0000) + + params = { + "status": deconz_api.Status.SUCCESS, + "frame_length": t.uint16_t(61), + "payload_length": t.uint16_t(54), + "device_state": deconz_api.DeviceState( + network_state=deconz_api.NetworkState2.CONNECTED, + device_state=( + deconz_api.DeviceStateFlags.APSDE_DATA_REQUEST_FREE_SLOTS_AVAILABLE + ), + ), + "dst_addr": addr, + "dst_ep": t.uint8_t(0), + "src_addr": addr, + "src_ep": t.uint8_t(0), + "profile_id": t.uint16_t(0), + "cluster_id": t.uint16_t(32772), + "asdu": t.LongOctetString( + b"\x0f\x00\x00\x00\x1a\x01\x04\x01\x00\x04\x00\x05\x00\x00\x06" + b"\x00\n\x00\x19\x00\x01\x05\x04\x01\x00 \x00\x00\x05\x02\x05" + ), + "reserved1": t.uint8_t(0), + "reserved2": t.uint8_t(175), + "lqi": t.uint8_t(69), + "reserved3": t.uint8_t(189), + "reserved4": t.uint8_t(82), + "reserved5": t.uint8_t(0), + "reserved6": t.uint8_t(0), + "rssi": t.int8s(27), + } + + data = deconz_api.Command( + command_id=deconz_api.CommandId.aps_data_indication, + seq=api._seq, + payload=t.serialize_dict( + params, + deconz_api.COMMAND_SCHEMAS[deconz_api.CommandId.aps_data_indication][1], + ), + ).serialize() + + asyncio.get_running_loop().call_later(0.01, api.data_received, data) + + rsp = await api._command( + cmd=deconz_api.CommandId.aps_data_indication, + flags=t.DataIndicationFlags.Include_Both_NWK_And_IEEE, ) + assert rsp == params + + +async def test_command_lock(api, mock_command_rsp): + await api.connect() + + for i in range(4): + mock_command_rsp( + command_id=deconz_api.CommandId.version, + params={"reserved": t.uint8_t(i)}, + rsp={ + "status": deconz_api.Status.SUCCESS, + "frame_length": t.uint16_t(9), + "version": deconz_api.FirmwareVersion(i), + }, + replace=(i == 0), + ) - for cmd_id in range(253, 255): - payload = b"\x01\x02\x03\x04" - status = t.uint8_t(0x00).serialize() - data = cmd_id.to_bytes(1, "big") + b"\x00" + status + b"\x00\x00" + payload - api._awaiting[0] = (MagicMock(),) - api.data_received(data) - assert t.deserialize.call_count == 0 - t.deserialize.reset_mock() - - -def test_simplified_beacon(api): - api._handle_simplified_beacon((0x0007, 0x1234, 0x5678, 0x19, 0x00, 0x01)) + async with api._command_lock: + tasks = [ + asyncio.create_task( + api._command(cmd=deconz_api.CommandId.version, reserved=0) + ), + asyncio.create_task( + api._command(cmd=deconz_api.CommandId.version, reserved=1) + ), + asyncio.create_task( + api._command(cmd=deconz_api.CommandId.version, reserved=2) + ), + asyncio.create_task( + api._command(cmd=deconz_api.CommandId.version, reserved=3) + ), + ] + await asyncio.sleep(0.1) + assert not any(t.done() for t in tasks) -async def test_aps_data_confirm(api, monkeypatch): - monkeypatch.setattr(deconz_api, "COMMAND_TIMEOUT", 0.01) + responses = await asyncio.gather(*tasks) - success = True + for index, rsp in enumerate(responses): + assert rsp["version"] == index - async def mock_cmd(*args, **kwargs): - if not success: - raise asyncio.TimeoutError() - dst = t.DeconzAddressEndpoint() - dst.address_mode = t.AddressMode.NWK - dst.address = 0x26FF - dst.endpoint = 1 +async def test_command_timeout(api): + await api.connect() - rsp = [ - 12, - ( - deconz_api.DeviceState.APSDE_DATA_REQUEST_SLOTS_AVAILABLE - | deconz_api.DeviceState.APSDE_DATA_INDICATION - | deconz_api.DeviceState.APSDE_DATA_CONFIRM - | 2 - ), - 98, - dst, - 1, - deconz_api.TXStatus.SUCCESS, - 0, - 0, - 0, - 0, - ] - api._handle_aps_data_confirm(rsp) - return rsp - - api._command = mock_cmd - api._data_confirm = True - - res = await api._aps_data_confirm() - assert res is not None - assert api._data_confirm is False - - success = False - api._data_confirm = True - res = await api._aps_data_confirm() - assert res is None - assert api._data_confirm is False - - -async def test_aps_data_ind(api, monkeypatch): - monkeypatch.setattr(deconz_api, "COMMAND_TIMEOUT", 0.1) - - success = True - - def mock_cmd(*args, **kwargs): - res = asyncio.Future() - s = sentinel - if success: - res.set_result( - [ - s.len, - 0x22, - t.DeconzAddress(), - 1, - t.DeconzAddress(), - 1, - 0x0104, - 0x0000, - b"\x00\x01\x02", - ] + with patch.object(deconz_api, "COMMAND_TIMEOUT", 0.1): + with pytest.raises(asyncio.TimeoutError): + await api._command( + cmd=deconz_api.CommandId.change_network_state, + network_state=deconz_api.NetworkState.OFFLINE, ) - return asyncio.wait_for(res, timeout=deconz_api.COMMAND_TIMEOUT) - - api._command = mock_cmd - api._data_indication = True - - res = await api._aps_data_indication() - assert res is not None - assert api._data_indication is False - - success = False - api._data_indication = True - res = await api._aps_data_indication() - assert res is None - assert api._data_indication is False - - -async def test_aps_data_request(api): - params = [ - 0x00, # req id - t.DeconzAddressEndpoint.deserialize(b"\x02\xaa\x55\x01")[0], # dst + ep - 0x0104, # profile id - 0x0007, # cluster id - 0x01, # src ep - b"aps payload", - ] - mock_cmd = AsyncMock() - api._command = mock_cmd - - await api.aps_data_request(*params) - assert mock_cmd.call_count == 1 +async def test_command_not_connected(api): + api._uart = None -async def test_aps_data_request_timeout(api, monkeypatch): - params = [ - 0x00, # req id - t.DeconzAddressEndpoint.deserialize(b"\x02\xaa\x55\x01")[0], # dst + ep - 0x0104, # profile id - 0x0007, # cluster id - 0x01, # src ep - b"aps payload", - ] + with pytest.raises(deconz_api.CommandError): + await api._command(cmd=deconz_api.CommandId.version, reserved=0) + + +async def test_data_received(api, mock_command_rsp): + await api.connect() + + src_addr = t.DeconzAddress() + src_addr.address_mode = t.AddressMode.NWK + src_addr.address = t.NWK(0xE695) + + dst_addr = t.DeconzAddress() + dst_addr.address_mode = t.AddressMode.NWK + dst_addr.address = t.NWK(0x0000) + + mock_command_rsp( + command_id=deconz_api.CommandId.aps_data_indication, + params={}, + rsp={ + "status": deconz_api.Status.SUCCESS, + "frame_length": t.uint16_t(80), + "payload_length": t.uint16_t(73), + "device_state": deconz_api.DeviceState( + network_state=deconz_api.NetworkState2.CONNECTED, + device_state=( + deconz_api.DeviceStateFlags.APSDE_DATA_REQUEST_FREE_SLOTS_AVAILABLE + ), + ), + "dst_addr": dst_addr, + "dst_ep": t.uint8_t(1), + "src_addr": src_addr, + "src_ep": t.uint8_t(1), + "profile_id": t.uint16_t(260), + "cluster_id": t.uint16_t(0x0000), + "asdu": t.LongOctetString( + b"\x18\x1b\x01\x04\x00\x00B\x0eIKEA of Sweden" + b"\x05\x00\x00B\x17TRADFRI wireless dimmer" + ), + "reserved1": t.uint8_t(0), + "reserved2": t.uint8_t(175), + "lqi": t.uint8_t(255), + "reserved3": t.uint8_t(142), + "reserved4": t.uint8_t(98), + "reserved5": t.uint8_t(0), + "reserved6": t.uint8_t(0), + "rssi": t.int8s(-49), + }, + ) - monkeypatch.setattr(deconz_api, "COMMAND_TIMEOUT", 0.1) - mock_cmd = MagicMock( - return_value=asyncio.wait_for( - asyncio.Future(), timeout=deconz_api.COMMAND_TIMEOUT + # Unsolicited device_state_changed + api.data_received(bytes.fromhex("0e2f000700ae00")) + + await asyncio.sleep(0.1) + + api._app.packet_received.assert_called_once_with( + zigpy_t.ZigbeePacket( + src=zigpy_t.AddrModeAddress(addr_mode=zigpy_t.AddrMode.NWK, address=0xE695), + src_ep=1, + dst=zigpy_t.AddrModeAddress(addr_mode=zigpy_t.AddrMode.NWK, address=0x0000), + dst_ep=1, + source_route=None, + extended_timeout=False, + tsn=None, + profile_id=260, + cluster_id=0x0000, + data=zigpy_t.SerializableBytes( + b"\x18\x1b\x01\x04\x00\x00B\x0eIKEA of Sweden" + b"\x05\x00\x00B\x17TRADFRI wireless dimmer" + ), + tx_options=zigpy_t.TransmitOptions.NONE, + radius=0, + non_member_radius=0, + lqi=255, + rssi=-49, ) ) - api._command = mock_cmd - - with pytest.raises(asyncio.TimeoutError): - await api.aps_data_request(*params) - assert mock_cmd.call_count == 1 - - -async def test_aps_data_request_busy(api, monkeypatch): - params = [ - 0x00, # req id - t.DeconzAddressEndpoint.deserialize(b"\x02\xaa\x55\x01")[0], # dst + ep - 0x0104, # profile id - 0x0007, # cluster id - 0x01, # src ep - b"aps payload", - ] - - res = asyncio.Future() - exc = zigpy_deconz.exception.CommandError(deconz_api.Status.BUSY, "busy") - res.set_exception(exc) - mock_cmd = MagicMock(return_value=res) - - api._command = mock_cmd - monkeypatch.setattr(deconz_api, "COMMAND_TIMEOUT", 0.1) - sleep = AsyncMock() - monkeypatch.setattr(asyncio, "sleep", sleep) - - with pytest.raises(zigpy_deconz.exception.CommandError): - await api.aps_data_request(*params) - assert mock_cmd.call_count == 4 - - -def test_handle_read_parameter(api): - api._handle_read_parameter(sentinel.data) -async def test_read_parameter(api): - api._command = AsyncMock( - return_value=(sentinel.len, sentinel.param_id, b"\xaa\x55") +async def test_read_parameter(api, mock_command_rsp): + await api.connect() + + mock_command_rsp( + command_id=deconz_api.CommandId.read_parameter, + params={ + "parameter_id": deconz_api.NetworkParameter.nwk_update_id, + "parameter": t.Bytes(b""), + }, + rsp={ + "status": deconz_api.Status.SUCCESS, + "frame_length": t.uint16_t(9), + "payload_length": t.uint16_t(2), + "parameter_id": deconz_api.NetworkParameter.nwk_update_id, + "parameter": t.Bytes(b"\x00"), + }, ) - r = await api.read_parameter(deconz_api.NetworkParameter.nwk_panid) - assert api._command.call_count == 1 - assert r[0] == 0x55AA - - api._command.reset_mock() - r = await api.read_parameter(0x05) - assert api._command.call_count == 1 - assert r[0] == 0x55AA - - with pytest.raises(KeyError): - await api.read_parameter("unknown_param") - - unk_param = 0xFF - assert unk_param not in list(deconz_api.NetworkParameter) - with pytest.raises(KeyError): - await api.read_parameter(unk_param) + mock_command_rsp( + command_id=deconz_api.CommandId.read_parameter, + params={ + "parameter_id": deconz_api.NetworkParameter.network_key, + "parameter": t.Bytes(b"\x00"), + }, + rsp={ + "status": deconz_api.Status.SUCCESS, + "frame_length": t.uint16_t(25), + "payload_length": t.uint16_t(18), + "parameter_id": deconz_api.NetworkParameter.network_key, + "parameter": t.Bytes(b"\x00M\x07p\xb6\x0b|\x90\xad\\\x07\x8a8\xa9M\xf6["), + }, + ) + rsp = await api.read_parameter(deconz_api.NetworkParameter.nwk_update_id) + assert rsp == 0x00 -def test_handle_write_parameter(api): - param_id = 0x05 - api._handle_write_parameter([sentinel.len, param_id]) + rsp = await api.read_parameter(deconz_api.NetworkParameter.network_key, 0) + assert rsp == deconz_api.IndexedKey( + index=0, + key=deconz_api.KeyData.convert( + "4d:07:70:b6:0b:7c:90:ad:5c:07:8a:38:a9:4d:f6:5b" + ), + ) - unk_param = 0xFF - assert unk_param not in list(deconz_api.NetworkParameter) - api._handle_write_parameter([sentinel.len, unk_param]) +async def test_write_parameter(api, mock_command_rsp): + await api.connect() + + mock_command_rsp( + command_id=deconz_api.CommandId.write_parameter, + params={ + "parameter_id": deconz_api.NetworkParameter.watchdog_ttl, + "parameter": t.uint32_t(600).serialize(), + }, + rsp={ + "status": deconz_api.Status.SUCCESS, + "frame_length": t.uint16_t(8), + "payload_length": t.uint16_t(1), + "parameter_id": deconz_api.NetworkParameter.watchdog_ttl, + }, + ) -async def test_write_parameter(api): - api._command = AsyncMock() + await api.write_parameter(deconz_api.NetworkParameter.watchdog_ttl, 600) - await api.write_parameter(deconz_api.NetworkParameter.nwk_panid, 0x55AA) - assert api._command.call_count == 1 - api._command.reset_mock() - await api.write_parameter(0x05, 0x55AA) - assert api._command.call_count == 1 +async def test_write_parameter_failure(api, mock_command_rsp): + await api.connect() - with pytest.raises(KeyError): - await api.write_parameter("unknown_param", 0x55AA) + mock_command_rsp( + command_id=deconz_api.CommandId.write_parameter, + params={ + "parameter_id": deconz_api.NetworkParameter.watchdog_ttl, + "parameter": t.uint32_t(600).serialize(), + }, + rsp={ + "status": deconz_api.Status.INVALID_VALUE, + "frame_length": t.uint16_t(8), + "payload_length": t.uint16_t(1), + "parameter_id": deconz_api.NetworkParameter.watchdog_ttl, + }, + ) - unk_param = 0xFF - assert unk_param not in list(deconz_api.NetworkParameter) - with pytest.raises(KeyError): - await api.write_parameter(unk_param, 0x55AA) + with pytest.raises(deconz_api.CommandError): + await api.write_parameter(deconz_api.NetworkParameter.watchdog_ttl, 600) @pytest.mark.parametrize( - "protocol_ver, firmware_version, flags", + "protocol_ver, firmware_ver", [ - (0x010A, 0x123405DD, 0x01), - (0x010B, 0x123405DD, 0x04), - (0x010A, 0x123407DD, 0x01), - (0x010B, 0x123407DD, 0x01), + (0x010A, 0x123405DD), + (0x010B, 0x123405DD), + (0x010A, 0x123407DD), + (0x010B, 0x123407DD), ], ) -async def test_version(protocol_ver, firmware_version, flags, api): - api.read_parameter = AsyncMock(return_value=[protocol_ver]) - api._command = AsyncMock(return_value=[firmware_version]) - r = await api.version() - assert r == firmware_version - assert api._aps_data_ind_flags == flags +async def test_version(protocol_ver, firmware_ver, api, mock_command_rsp): + await api.connect() + + mock_command_rsp( + command_id=deconz_api.CommandId.read_parameter, + params={ + "parameter_id": deconz_api.NetworkParameter.protocol_version, + "parameter": t.Bytes(b""), + }, + rsp={ + "status": deconz_api.Status.SUCCESS, + "frame_length": t.uint16_t(10), + "payload_length": t.uint16_t(3), + "parameter_id": deconz_api.NetworkParameter.protocol_version, + "parameter": t.Bytes(t.uint16_t(protocol_ver).serialize()), + }, + replace=True, + ) + mock_command_rsp( + command_id=deconz_api.CommandId.version, + params={"reserved": t.uint8_t(0)}, + rsp={ + "status": deconz_api.Status.SUCCESS, + "frame_length": t.uint16_t(9), + "version": deconz_api.FirmwareVersion(firmware_ver), + }, + replace=True, + ) + + r = await api.version() + assert r == firmware_ver -def test_handle_version(api): - api._handle_version([sentinel.version]) + assert api.protocol_version == protocol_ver + assert api.firmware_version == firmware_ver @pytest.mark.parametrize( @@ -480,49 +521,6 @@ def test_device_state_network_state(data, network_state): assert state.serialize() == new_data -@patch.object(deconz_api.Deconz, "device_state", new_callable=AsyncMock) -@patch("zigpy_deconz.uart.connect", return_value=MagicMock(spec_set=uart.Gateway)) -async def test_probe_success(mock_connect, mock_device_state): - """Test device probing.""" - - res = await deconz_api.Deconz.probe(DEVICE_CONFIG) - assert res is True - assert mock_connect.call_count == 1 - assert mock_connect.await_count == 1 - assert mock_connect.call_args[0][0] is DEVICE_CONFIG - assert mock_device_state.call_count == 1 - assert mock_connect.return_value.close.call_count == 1 - - mock_connect.reset_mock() - mock_device_state.reset_mock() - mock_connect.reset_mock() - res = await deconz_api.Deconz.probe(DEVICE_CONFIG) - assert res is True - assert mock_connect.call_count == 1 - assert mock_connect.await_count == 1 - assert mock_connect.call_args[0][0] is DEVICE_CONFIG - assert mock_device_state.call_count == 1 - assert mock_connect.return_value.close.call_count == 1 - - -@patch.object(deconz_api.Deconz, "device_state", new_callable=AsyncMock) -@patch("zigpy_deconz.uart.connect", return_value=MagicMock(spec_set=uart.Gateway)) -@pytest.mark.parametrize("exception", (asyncio.TimeoutError,)) -async def test_probe_fail(mock_connect, mock_device_state, exception): - """Test device probing fails.""" - - mock_device_state.side_effect = exception - mock_device_state.reset_mock() - mock_connect.reset_mock() - res = await deconz_api.Deconz.probe(DEVICE_CONFIG) - assert res is False - assert mock_connect.call_count == 1 - assert mock_connect.await_count == 1 - assert mock_connect.call_args[0][0] is DEVICE_CONFIG - assert mock_device_state.call_count == 1 - assert mock_connect.return_value.close.call_count == 1 - - @pytest.mark.parametrize( "value, name", ( @@ -550,59 +548,176 @@ def test_tx_status(value, name): assert status.name == name -def test_handle_add_neighbour(api): - """Test handle_add_neighbour.""" - api._handle_add_neighbour((12, 1, 0x1234, sentinel.ieee, 0x80)) +@pytest.mark.parametrize("relays", (None, [], [0x1234, 0x5678])) +async def test_aps_data_request_relays(relays, api, mock_command_rsp): + await api.connect() + + mock_command_rsp( + command_id=deconz_api.CommandId.aps_data_request, + params={}, + rsp={ + "status": deconz_api.Status.SUCCESS, + "frame_length": t.uint16_t(9), + "payload_length": t.uint16_t(2), + "device_state": deconz_api.DeviceState( + network_state=deconz_api.NetworkState2.CONNECTED, + device_state=( + deconz_api.DeviceStateFlags.APSDE_DATA_REQUEST_FREE_SLOTS_AVAILABLE + ), + ), + "request_id": t.uint8_t(0x00), + }, + ) + await api.aps_data_request( + req_id=0x00, + dst_addr_ep=t.DeconzAddressEndpoint.deserialize(b"\x02\xaa\x55\x01")[0], + profile=0x0104, + cluster=0x0007, + src_ep=0x01, + aps_payload=b"aps payload", + relays=relays, + ) + + with pytest.raises(ValueError) as exc: + await api.aps_data_request( + req_id=0x00, + dst_addr_ep=t.DeconzAddressEndpoint.deserialize(b"\x02\xaa\x55\x01")[0], + profile=0x0104, + cluster=0x0007, + src_ep=None, # This is not possible + aps_payload=b"aps payload", + ) -@pytest.mark.parametrize("status", (0x00, 0x05)) -async def test_aps_data_req_deserialize_error(api, uart_gw, status, caplog): - """Test deserialization error.""" + assert "has non-trailing optional argument" in str(exc.value) - device_state = ( - deconz_api.DeviceState.APSDE_DATA_INDICATION - | deconz_api.DeviceState.APSDE_DATA_CONFIRM - | deconz_api.NetworkState.CONNECTED + +@patch( + "zigpy_deconz.api.REQUEST_RETRY_DELAYS", + [None if v is None else 0 for v in deconz_api.REQUEST_RETRY_DELAYS], +) +async def test_aps_data_request_retries_busy(api, mock_command_rsp): + await api.connect() + + mock_rsp = mock_command_rsp( + command_id=deconz_api.CommandId.aps_data_request, + params={}, + rsp={ + "status": deconz_api.Status.BUSY, + "frame_length": t.uint16_t(9), + "payload_length": t.uint16_t(2), + "device_state": deconz_api.DeviceState( + network_state=deconz_api.NetworkState2.CONNECTED, + device_state=( + deconz_api.DeviceStateFlags.APSDE_DATA_REQUEST_FREE_SLOTS_AVAILABLE + ), + ), + "request_id": t.uint8_t(0x00), + }, ) - api._handle_device_state_value(device_state) - await asyncio.sleep(0) - await asyncio.sleep(0) - await asyncio.sleep(0) - assert uart_gw.send.call_count == 1 - assert api._data_indication is True - - api.data_received( - uart_gw.send.call_args[0][0][0:2] - + bytes([status]) - + binascii.unhexlify("0800010022") + + with pytest.raises(deconz_api.CommandError): + await api.aps_data_request( + req_id=0x00, + dst_addr_ep=t.DeconzAddressEndpoint.deserialize(b"\x02\xaa\x55\x01")[0], + profile=0x0104, + cluster=0x0007, + src_ep=1, + aps_payload=b"aps payload", + ) + + assert len(mock_rsp.mock_calls) == 4 + + +async def test_aps_data_request_retries_failure(api, mock_command_rsp): + await api.connect() + + mock_rsp = mock_command_rsp( + command_id=deconz_api.CommandId.aps_data_request, + params={}, + rsp={ + "status": deconz_api.Status.FAILURE, + "frame_length": t.uint16_t(9), + "payload_length": t.uint16_t(2), + "device_state": deconz_api.DeviceState( + network_state=deconz_api.NetworkState2.CONNECTED, + device_state=( + deconz_api.DeviceStateFlags.APSDE_DATA_REQUEST_FREE_SLOTS_AVAILABLE + ), + ), + "request_id": t.uint8_t(0x00), + }, ) - await asyncio.sleep(0) - await asyncio.sleep(0) - await asyncio.sleep(0) - assert api._data_indication is False + with pytest.raises(deconz_api.CommandError): + await api.aps_data_request( + req_id=0x00, + dst_addr_ep=t.DeconzAddressEndpoint.deserialize(b"\x02\xaa\x55\x01")[0], + profile=0x0104, + cluster=0x0007, + src_ep=1, + aps_payload=b"aps payload", + ) -@pytest.mark.parametrize("relays", (None, [], [0x1234, 0x5678])) -async def test_aps_data_request_relays(relays, api): - mock_cmd = api._command = AsyncMock() + assert len(mock_rsp.mock_calls) == 1 - await api.aps_data_request( - 0x00, # req id - t.DeconzAddressEndpoint.deserialize(b"\x02\xaa\x55\x01")[0], # dst + ep - 0x0104, # profile id - 0x0007, # cluster id - 0x01, # src ep - b"aps payload", - relays=relays, + +async def test_aps_data_request_locking(caplog, api, mock_command_rsp): + await api.connect() + + # No free slots + send_network_state(api, device_state=deconz_api.DeviceStateFlags.APSDE_DATA_CONFIRM) + + await asyncio.sleep(0.1) + + mock_rsp = mock_command_rsp( + command_id=deconz_api.CommandId.aps_data_request, + params={}, + rsp={ + "status": deconz_api.Status.SUCCESS, + "frame_length": t.uint16_t(9), + "payload_length": t.uint16_t(2), + "device_state": deconz_api.DeviceState( + network_state=deconz_api.NetworkState2.CONNECTED, + device_state=( + deconz_api.DeviceStateFlags.APSDE_DATA_REQUEST_FREE_SLOTS_AVAILABLE + ), + ), + "request_id": t.uint8_t(0x00), + }, + ) + + with caplog.at_level(logging.DEBUG): + send = asyncio.create_task( + api.aps_data_request( + req_id=0x00, + dst_addr_ep=t.DeconzAddressEndpoint.deserialize(b"\x02\xaa\x55\x01")[0], + profile=0x0104, + cluster=0x0007, + src_ep=1, + aps_payload=b"aps payload", + ) + ) + + await asyncio.sleep(0.1) + + assert "Waiting for free slots to become available" in caplog.text + + assert len(mock_rsp.mock_calls) == 0 + + send_network_state( + api, + device_state=deconz_api.DeviceStateFlags.APSDE_DATA_REQUEST_FREE_SLOTS_AVAILABLE, ) - assert mock_cmd.call_count == 1 - if relays: - assert isinstance(mock_cmd.mock_calls[0][1][-1], t.NWKList) - assert mock_cmd.mock_calls[0][1][-1] == t.NWKList(relays) + await send + + assert len(mock_rsp.mock_calls) == 1 async def test_connection_lost(api): + await api.connect() + app = api._app = MagicMock() err = RuntimeError() @@ -611,51 +726,226 @@ async def test_connection_lost(api): app.connection_lost.assert_called_once_with(err) -async def test_aps_data_indication(api): - dst = t.DeconzAddress() - dst.address_mode = t.AddressMode.NWK - dst.address = 0x0000 - - src = t.DeconzAddress() - src.address_mode = t.AddressMode.NWK - src.address = 0xC643 - - data = b"\x18\x1f\x01\x04\x00\x00B\x12Third Reality, Inc\x05\x00\x00B\t3RSP019BZ" - - packet = [ - 63, - (deconz_api.DeviceState.APSDE_DATA_REQUEST_SLOTS_AVAILABLE | 2), - dst, - 1, - src, - 1, - 260, - 0, - data, - 0, - 175, - 255, - 186, - 25, - 78, - 3, - -47, +async def test_unknown_command(api, caplog): + await api.connect() + + assert 0xFF not in deconz_api.COMMAND_SCHEMAS + + with caplog.at_level(logging.WARNING): + api.data_received(b"\xFF\xAA\xBB") + + assert ( + "Unknown command received: Command(command_id=," + " seq=170, payload=b'\\xbb')" + ) in caplog.text + + +async def test_bad_command_parsing(api, caplog): + await api.connect() + + assert 0xFF not in deconz_api.COMMAND_SCHEMAS + + with caplog.at_level(logging.WARNING): + api.data_received( + bytes.fromhex( + "172c002f0028002e02000000020000000000" + "028011000300000010400f3511472b004000" + # "2b000000af45838600001b" # truncated + ) + ) + + assert ( + "Failed to parse command Command(command_id=" + "" + ) in caplog.text + + caplog.clear() + + with caplog.at_level(logging.DEBUG): + api.data_received(bytes.fromhex("0d03000d0000077826") + b"TEST") + + assert ( + "Unparsed data remains after frame" in caplog.text and "b'TEST'" in caplog.text + ) + + +async def test_bad_response_status(api, mock_command_rsp): + await api.connect() + + mock_command_rsp( + command_id=deconz_api.CommandId.write_parameter, + params={ + "parameter_id": deconz_api.NetworkParameter.nwk_update_id, + "parameter": t.uint8_t(123).serialize(), + }, + rsp={ + "status": deconz_api.Status.FAILURE, + "frame_length": t.uint16_t(8), + "payload_length": t.uint16_t(1), + "parameter_id": deconz_api.NetworkParameter.nwk_update_id, + }, + ) + + with pytest.raises(deconz_api.CommandError) as exc: + await api.write_parameter(deconz_api.NetworkParameter.nwk_update_id, 123) + + assert isinstance(exc.value, deconz_api.CommandError) + assert exc.value.status == deconz_api.Status.FAILURE + + +async def test_data_poller(api, mock_command_rsp): + await api.connect() + + dst_addr_ep = t.DeconzAddressEndpoint() + dst_addr_ep.address_mode = t.AddressMode.NWK + dst_addr_ep.address = t.NWK(0x0000) + dst_addr_ep.endpoint = t.uint8_t(0) + + src_addr = t.DeconzAddress() + src_addr.address_mode = t.AddressMode.NWK + src_addr.address = t.NWK(0xE695) + + dst_addr = t.DeconzAddress() + dst_addr.address_mode = t.AddressMode.NWK + dst_addr.address = t.NWK(0x0000) + + mock_command_rsp( + command_id=deconz_api.CommandId.aps_data_confirm, + params={}, + rsp={ + "status": deconz_api.Status.SUCCESS, + "frame_length": t.uint16_t(19), + "payload_length": t.uint16_t(12), + "device_state": deconz_api.DeviceState( + network_state=deconz_api.NetworkState2.CONNECTED, + device_state=( + # Include a data indication flag to trigger a poll + deconz_api.DeviceStateFlags.APSDE_DATA_REQUEST_FREE_SLOTS_AVAILABLE + | deconz_api.DeviceStateFlags.APSDE_DATA_INDICATION + ), + ), + "request_id": t.uint8_t(16), + "dst_addr": dst_addr_ep, + "src_ep": t.uint8_t(0), + "confirm_status": deconz_api.TXStatus.SUCCESS, + "reserved1": t.uint8_t(0), + "reserved2": t.uint8_t(0), + "reserved3": t.uint8_t(0), + "reserved4": t.uint8_t(0), + }, + ) + + mock_command_rsp( + command_id=deconz_api.CommandId.aps_data_indication, + params={}, + rsp={ + "status": deconz_api.Status.SUCCESS, + "frame_length": t.uint16_t(80), + "payload_length": t.uint16_t(73), + "device_state": deconz_api.DeviceState( + network_state=deconz_api.NetworkState2.CONNECTED, + device_state=( + deconz_api.DeviceStateFlags.APSDE_DATA_REQUEST_FREE_SLOTS_AVAILABLE + ), + ), + "dst_addr": dst_addr, + "dst_ep": t.uint8_t(1), + "src_addr": src_addr, + "src_ep": t.uint8_t(1), + "profile_id": t.uint16_t(260), + "cluster_id": t.uint16_t(0x0000), + "asdu": t.LongOctetString( + b"\x18\x1b\x01\x04\x00\x00B\x0eIKEA of Sweden" + b"\x05\x00\x00B\x17TRADFRI wireless dimmer" + ), + "reserved1": t.uint8_t(0), + "reserved2": t.uint8_t(175), + "lqi": t.uint8_t(255), + "reserved3": t.uint8_t(142), + "reserved4": t.uint8_t(98), + "reserved5": t.uint8_t(0), + "reserved6": t.uint8_t(0), + "rssi": t.int8s(-49), + }, + ) + + # Take us offline for a moment + send_network_state(api, network_state=deconz_api.NetworkState2.OFFLINE) + await asyncio.sleep(0.1) + + # Bring us back online with just a data confirmation to kick things off + send_network_state( + api, + network_state=deconz_api.NetworkState2.CONNECTED, + device_state=deconz_api.DeviceStateFlags.APSDE_DATA_CONFIRM, + ) + + await asyncio.sleep(0.1) + + # Both callbacks have been called + api._app.handle_tx_confirm.assert_called_once_with(16, deconz_api.TXStatus.SUCCESS) + assert len(api._app.packet_received.mock_calls) == 1 + + # The task is cancelled on close + task = api._data_poller_task + api.close() + assert api._data_poller_task is None + + if sys.version_info >= (3, 11): + assert task.cancelling() + + +async def test_get_device_state(api, mock_command_rsp): + await api.connect() + + device_state = deconz_api.DeviceState( + network_state=deconz_api.NetworkState2.CONNECTED, + device_state=( + deconz_api.DeviceStateFlags.APSDE_DATA_REQUEST_FREE_SLOTS_AVAILABLE + ), + ) + + mock_command_rsp( + command_id=deconz_api.CommandId.device_state, + params={}, + rsp={ + "status": deconz_api.Status.SUCCESS, + "frame_length": t.uint16_t(8), + "device_state": device_state, + "reserved1": t.uint8_t(0), + "reserved2": t.uint8_t(0), + }, + ) + + assert (await api.get_device_state()) == device_state + + +async def test_change_network_state(api, mock_command_rsp): + api._command = AsyncMock() + await api.change_network_state(new_state=deconz_api.NetworkState.OFFLINE) + + assert api._command.mock_calls == [ + call( + deconz_api.CommandId.change_network_state, + network_state=deconz_api.NetworkState.OFFLINE, + ) ] - api._handle_aps_data_indication(packet) - - api._app.handle_rx.assert_called_once_with( - src=src, - src_ep=1, - dst=dst, - dst_ep=1, - profile_id=260, - cluster_id=0x0000, - data=data, - lqi=255, - rssi=-47, + +async def test_add_neighbour(api, mock_command_rsp): + api._command = AsyncMock() + await api.add_neighbour( + nwk=0x1234, + ieee=t.EUI64.convert("aa:bb:cc:dd:11:22:33:44"), + mac_capability_flags=0x12, ) - # No error is thrown when the app is disconnected - api._app = None - api._handle_aps_data_indication(packet) + assert api._command.mock_calls == [ + call( + deconz_api.CommandId.add_neighbour, + unknown=0x01, + nwk=0x1234, + ieee=t.EUI64.convert("aa:bb:cc:dd:11:22:33:44"), + mac_capability_flags=0x12, + ) + ] diff --git a/tests/test_application.py b/tests/test_application.py index 8c64b70..45beee8 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -37,14 +37,14 @@ def device_path(): def api(): """Return API fixture.""" api = MagicMock(spec_set=zigpy_deconz.api.Deconz(None, None)) - api.device_state = AsyncMock( - return_value=(deconz_api.DeviceState(deconz_api.NetworkState.CONNECTED), 0, 0) + api.get_device_state = AsyncMock( + return_value=deconz_api.DeviceState(deconz_api.NetworkState.CONNECTED) ) api.write_parameter = AsyncMock() # So the protocol version is effectively infinite - api._proto_ver.__ge__.return_value = True - api._proto_ver.__lt__.return_value = False + api._protocol_version.__ge__.return_value = True + api._protocol_version.__lt__.return_value = False api.protocol_version.__ge__.return_value = True api.protocol_version.__lt__.return_value = False @@ -67,7 +67,7 @@ def app(device_path, api): device_state = MagicMock() device_state.network_state.__eq__.return_value = True - api.device_state = AsyncMock(return_value=(device_state, 0, 0)) + api.get_device_state = AsyncMock(return_value=device_state) p1 = patch.object(app, "_api", api) p2 = patch.object(app, "_delayed_neighbour_scan") @@ -127,11 +127,14 @@ async def test_start_network(app, proto_ver, target_state, returned_state): app.restore_neighbours = AsyncMock() app.add_endpoint = AsyncMock() - app._api.device_state = AsyncMock( - return_value=(deconz_api.DeviceState(returned_state), 0, 0) + app._api.get_device_state = AsyncMock( + return_value=deconz_api.DeviceState( + device_state=deconz_api.DeviceStateFlags.APSDE_DATA_REQUEST_FREE_SLOTS_AVAILABLE, + network_state=returned_state, + ) ) - app._api._proto_ver = proto_ver + app._api._protocol_version = proto_ver app._api.protocol_version = proto_ver if ( @@ -171,7 +174,6 @@ async def test_connect(app): def new_api(*args): api = MagicMock() api.connect = AsyncMock() - api.version = AsyncMock(return_value=sentinel.version) return api @@ -179,17 +181,13 @@ def new_api(*args): app._api = None await app.connect() assert app._api is not None - assert app._api.connect.await_count == 1 - assert app._api.version.await_count == 1 - assert app.version is sentinel.version async def test_connect_failure(app): with patch.object(application, "Deconz") as api_mock: api = api_mock.return_value = MagicMock() - api.connect = AsyncMock() - api.version = AsyncMock(side_effect=RuntimeError("Broken")) + api.connect = AsyncMock(side_effect=RuntimeError("Broken")) app._api = None @@ -198,7 +196,6 @@ async def test_connect_failure(app): assert app._api is None api.connect.assert_called_once() - api.version.assert_called_once() api.close.assert_called_once() @@ -237,7 +234,9 @@ async def test_deconz_dev_add_to_group(app, nwk, device_path): app._groups = MagicMock() app._groups.add_group.return_value = group - deconz = application.DeconzDevice(0, device_path, app, sentinel.ieee, nwk) + deconz = application.DeconzDevice( + deconz_api.FirmwareVersion(0), device_path, app, sentinel.ieee, nwk + ) deconz.endpoints = { 0: sentinel.zdo, 1: sentinel.ep1, @@ -255,7 +254,9 @@ async def test_deconz_dev_add_to_group(app, nwk, device_path): async def test_deconz_dev_remove_from_group(app, nwk, device_path): group = MagicMock() app.groups[sentinel.grp_id] = group - deconz = application.DeconzDevice(0, device_path, app, sentinel.ieee, nwk) + deconz = application.DeconzDevice( + deconz_api.FirmwareVersion(0), device_path, app, sentinel.ieee, nwk + ) deconz.endpoints = { 0: sentinel.zdo, 1: sentinel.ep1, @@ -267,7 +268,9 @@ async def test_deconz_dev_remove_from_group(app, nwk, device_path): def test_deconz_props(nwk, device_path): - deconz = application.DeconzDevice(0, device_path, app, sentinel.ieee, nwk) + deconz = application.DeconzDevice( + deconz_api.FirmwareVersion(0), device_path, app, sentinel.ieee, nwk + ) assert deconz.manufacturer is not None assert deconz.model is not None @@ -275,12 +278,12 @@ def test_deconz_props(nwk, device_path): @pytest.mark.parametrize( "name, firmware_version, device_path", [ - ("ConBee", 0x00000500, "/dev/ttyUSB0"), - ("ConBee II", 0x00000700, "/dev/ttyUSB0"), - ("RaspBee", 0x00000500, "/dev/ttyS0"), - ("RaspBee II", 0x00000700, "/dev/ttyS0"), - ("RaspBee", 0x00000500, "/dev/ttyAMA0"), - ("RaspBee II", 0x00000700, "/dev/ttyAMA0"), + ("ConBee", deconz_api.FirmwareVersion(0x00000500), "/dev/ttyUSB0"), + ("ConBee II", deconz_api.FirmwareVersion(0x00000700), "/dev/ttyUSB0"), + ("RaspBee", deconz_api.FirmwareVersion(0x00000500), "/dev/ttyS0"), + ("RaspBee II", deconz_api.FirmwareVersion(0x00000700), "/dev/ttyS0"), + ("RaspBee", deconz_api.FirmwareVersion(0x00000500), "/dev/ttyAMA0"), + ("RaspBee II", deconz_api.FirmwareVersion(0x00000700), "/dev/ttyAMA0"), ], ) def test_deconz_name(nwk, name, firmware_version, device_path): @@ -294,7 +297,9 @@ async def test_deconz_new(app, nwk, device_path, monkeypatch): mock_init = AsyncMock() monkeypatch.setattr(zigpy.device.Device, "_initialize", mock_init) - deconz = await application.DeconzDevice.new(app, sentinel.ieee, nwk, 0, device_path) + deconz = await application.DeconzDevice.new( + app, sentinel.ieee, nwk, deconz_api.FirmwareVersion(0), device_path + ) assert isinstance(deconz, application.DeconzDevice) assert mock_init.call_count == 1 mock_init.reset_mock() @@ -306,7 +311,9 @@ async def test_deconz_new(app, nwk, device_path, monkeypatch): 22: MagicMock(), } app.devices[sentinel.ieee] = mock_dev - deconz = await application.DeconzDevice.new(app, sentinel.ieee, nwk, 0, device_path) + deconz = await application.DeconzDevice.new( + app, sentinel.ieee, nwk, deconz_api.FirmwareVersion(0), device_path + ) assert isinstance(deconz, application.DeconzDevice) assert mock_init.call_count == 0 @@ -424,19 +431,19 @@ async def test_delayed_scan(): async def test_change_network_state(app, support_watchdog): app._reset_watchdog_task = MagicMock() - app._api.device_state = AsyncMock( + app._api.get_device_state = AsyncMock( side_effect=[ - (deconz_api.DeviceState(deconz_api.NetworkState.OFFLINE), 0, 0), - (deconz_api.DeviceState(deconz_api.NetworkState.JOINING), 0, 0), - (deconz_api.DeviceState(deconz_api.NetworkState.CONNECTED), 0, 0), + deconz_api.DeviceState(deconz_api.NetworkState.OFFLINE), + deconz_api.DeviceState(deconz_api.NetworkState.JOINING), + deconz_api.DeviceState(deconz_api.NetworkState.CONNECTED), ] ) if support_watchdog: - app._api._proto_ver = application.PROTO_VER_WATCHDOG + app._api._protocol_version = application.PROTO_VER_WATCHDOG app._api.protocol_version = application.PROTO_VER_WATCHDOG else: - app._api._proto_ver = application.PROTO_VER_WATCHDOG - 1 + app._api._protocol_version = application.PROTO_VER_WATCHDOG - 1 app._api.protocol_version = application.PROTO_VER_WATCHDOG - 1 old_watchdog_task = app._reset_watchdog_task @@ -486,14 +493,15 @@ async def read_param(param_id, index): deconz_api.Status.UNSUPPORTED, "Unsupported" ) else: - return index, slots[index] + return deconz_api.IndexedEndpoint(index=index, descriptor=slots[index]) app._api.read_parameter = AsyncMock(side_effect=read_param) app._api.write_parameter = AsyncMock() await app.add_endpoint(descriptor) app._api.write_parameter.assert_called_once_with( - deconz_api.NetworkParameter.configure_endpoint, target_slot, descriptor + deconz_api.NetworkParameter.configure_endpoint, + deconz_api.IndexedEndpoint(index=target_slot, descriptor=descriptor), ) @@ -526,7 +534,9 @@ async def read_param(param_id, index): deconz_api.Status.UNSUPPORTED, "Unsupported" ) - return index, ENDPOINT.replace(endpoint=1) + return deconz_api.IndexedEndpoint( + index=index, descriptor=ENDPOINT.replace(endpoint=1) + ) app._api.read_parameter = AsyncMock(side_effect=read_param) app._api.write_parameter = AsyncMock() @@ -537,7 +547,8 @@ async def read_param(param_id, index): # Writing another endpoint will cause a write await app.add_endpoint(ENDPOINT.replace(endpoint=2)) app._api.write_parameter.assert_called_once_with( - deconz_api.NetworkParameter.configure_endpoint, 1, ENDPOINT.replace(endpoint=2) + deconz_api.NetworkParameter.configure_endpoint, + deconz_api.IndexedEndpoint(index=1, descriptor=ENDPOINT.replace(endpoint=2)), ) diff --git a/tests/test_network_state.py b/tests/test_network_state.py index b469b8b..936818d 100644 --- a/tests/test_network_state.py +++ b/tests/test_network_state.py @@ -60,11 +60,7 @@ def network_info(node_info): nwk_addresses={}, stack_specific={}, source=f"zigpy-deconz@{importlib.metadata.version('zigpy-deconz')}", - metadata={ - "deconz": { - "version": 0, - } - }, + metadata={"deconz": {"version": "0x00000001"}}, ) @@ -128,6 +124,7 @@ async def write_parameter(param, *args): params[param.name] = args + app._change_network_state = AsyncMock() app._api.write_parameter = AsyncMock(side_effect=write_parameter) network_info = network_info.replace( @@ -171,9 +168,13 @@ async def write_parameter(param, *args): assert params["nwk_panid"] == (network_info.pan_id,) assert params["aps_extended_panid"] == (network_info.extended_pan_id,) assert params["nwk_update_id"] == (network_info.nwk_update_id,) - assert params["network_key"] == (0, network_info.network_key.key) + assert params["network_key"] == ( + zigpy_deconz.api.IndexedKey(index=0, key=network_info.network_key.key), + ) assert params["trust_center_address"] == (node_info.ieee,) - assert params["link_key"] == (node_info.ieee, network_info.tc_link_key.key) + assert params["link_key"] == ( + zigpy_deconz.api.LinkKey(ieee=node_info.ieee, key=network_info.tc_link_key.key), + ) if security_level == 0: assert params["security_mode"] == (zigpy_deconz.api.SecurityMode.NO_SECURITY,) @@ -188,20 +189,20 @@ async def write_parameter(param, *args): (None, {}, {}, {}), ( None, - {("aps_designed_coordinator",): [0x00]}, + {("aps_designed_coordinator",): 0x00}, {}, {"logical_type": zdo_t.LogicalType.Router}, ), ( None, { - ("aps_extended_panid",): [t.EUI64.convert("00:00:00:00:00:00:00:00")], - ("nwk_extended_panid",): [t.EUI64.convert("0D:49:91:99:AE:CD:3C:35")], + ("aps_extended_panid",): t.EUI64.convert("00:00:00:00:00:00:00:00"), + ("nwk_extended_panid",): t.EUI64.convert("0D:49:91:99:AE:CD:3C:35"), }, {}, {}, ), - (NetworkNotFormed, {("current_channel",): [0]}, {}, {}), + (NetworkNotFormed, {("current_channel",): 0}, {}, {}), ( None, { @@ -214,16 +215,16 @@ async def write_parameter(param, *args): ), ( None, - {("security_mode",): [zigpy_deconz.api.SecurityMode.NO_SECURITY]}, + {("security_mode",): zigpy_deconz.api.SecurityMode.NO_SECURITY}, {"security_level": 0}, {}, ), ( None, { - ("security_mode",): [ - zigpy_deconz.api.SecurityMode.PRECONFIGURED_NETWORK_KEY - ] + ( + "security_mode", + ): zigpy_deconz.api.SecurityMode.PRECONFIGURED_NETWORK_KEY }, {"security_level": 5}, {}, @@ -242,20 +243,24 @@ async def test_load_network_info( """Test that network info is correctly read.""" params = { - ("nwk_frame_counter",): [network_info.network_key.tx_counter], - ("aps_designed_coordinator",): [1], - ("nwk_address",): [node_info.nwk], - ("mac_address",): [node_info.ieee], - ("current_channel",): [network_info.channel], - ("channel_mask",): [t.Channels.from_channel_list([network_info.channel])], - ("use_predefined_nwk_panid",): [True], - ("nwk_panid",): [network_info.pan_id], - ("aps_extended_panid",): [network_info.extended_pan_id], - ("nwk_update_id",): [network_info.nwk_update_id], - ("network_key", 0): [0, network_info.network_key.key], - ("trust_center_address",): [node_info.ieee], - ("link_key", node_info.ieee): [node_info.ieee, network_info.tc_link_key.key], - ("security_mode",): [zigpy_deconz.api.SecurityMode.ONLY_TCLK], + ("nwk_frame_counter",): network_info.network_key.tx_counter, + ("aps_designed_coordinator",): 1, + ("nwk_address",): node_info.nwk, + ("mac_address",): node_info.ieee, + ("current_channel",): network_info.channel, + ("channel_mask",): t.Channels.from_channel_list([network_info.channel]), + ("use_predefined_nwk_panid",): True, + ("nwk_panid",): network_info.pan_id, + ("aps_extended_panid",): network_info.extended_pan_id, + ("nwk_update_id",): network_info.nwk_update_id, + ("network_key", 0): zigpy_deconz.api.IndexedKey( + index=0, key=network_info.network_key.key + ), + ("trust_center_address",): node_info.ieee, + ("link_key", node_info.ieee): zigpy_deconz.api.LinkKey( + ieee=node_info.ieee, key=network_info.tc_link_key.key + ), + ("security_mode",): zigpy_deconz.api.SecurityMode.ONLY_TCLK, } params.update(param_overrides) @@ -273,7 +278,7 @@ async def read_param(param, *args): return value - app._api.__getitem__ = app._api.read_parameter = AsyncMock(side_effect=read_param) + app._api.read_parameter = AsyncMock(side_effect=read_param) if error is not None: with pytest.raises(error): diff --git a/tests/test_send_receive.py b/tests/test_send_receive.py index 67d90c4..2f4db4a 100644 --- a/tests/test_send_receive.py +++ b/tests/test_send_receive.py @@ -2,7 +2,7 @@ import asyncio import contextlib -from unittest.mock import Mock, patch +from unittest.mock import patch import pytest import zigpy.exceptions @@ -204,37 +204,3 @@ async def test_send_packet_deliver_failure(app, tx_packet): # noqa: F811 await app.send_packet(tx_packet) assert "Failed to deliver" in str(e) - - -@pytest.fixture -def rx_packet(): - return zigpy_t.ZigbeePacket( - src=zigpy_t.AddrModeAddress(addr_mode=zigpy_t.AddrMode.NWK, address=0x1234), - src_ep=0x12, - dst=zigpy_t.AddrModeAddress(addr_mode=zigpy_t.AddrMode.NWK, address=0x0000), - dst_ep=0x34, - tsn=0x56, - profile_id=0x7890, - cluster_id=0xABCD, - data=zigpy_t.SerializableBytes(b"some data"), - tx_options=zigpy_t.TransmitOptions.NONE, - radius=0, - ) - - -async def test_receive_packet_nwk(app, rx_packet): # noqa: F811 - app.packet_received = Mock(spec_set=app.packet_received) - - app.handle_rx( - src=t.DeconzAddress.from_zigpy_type(rx_packet.src), - src_ep=rx_packet.src_ep, - dst=t.DeconzAddress.from_zigpy_type(rx_packet.dst), - dst_ep=rx_packet.dst_ep, - profile_id=rx_packet.profile_id, - cluster_id=rx_packet.cluster_id, - data=rx_packet.data.serialize(), - lqi=rx_packet.lqi, - rssi=rx_packet.rssi, - ) - - app.packet_received.assert_called_once_with(rx_packet.replace(tsn=None)) diff --git a/tests/test_types.py b/tests/test_types.py index 06f3ac9..17ad0e5 100644 --- a/tests/test_types.py +++ b/tests/test_types.py @@ -1,6 +1,5 @@ """Tests for zigpy_deconz.types module.""" -from unittest import mock import pytest import zigpy.types as zigpy_t @@ -126,25 +125,6 @@ def test_deconz_address_nwk_and_ieee(): assert addr.as_zigpy_type() == zigpy_addr -def test_pan_id(): - t.PanId() - - -def test_extended_pan_id(): - t.ExtendedPanId() - - -def test_key(): - data = b"\x31\x39\x63\x32\x30\x65\x61\x63\x36\x36\x32\x63\x61\x38\x30\x35" - extra = b"extra data" - - key, rest = t.Key.deserialize(data + extra) - assert rest == extra - assert key == [49, 57, 99, 50, 48, 101, 97, 99, 54, 54, 50, 99, 97, 56, 48, 53] - - assert key.serialize() == data - - def test_bytes(): data = b"abcde\x00\xff" @@ -155,117 +135,6 @@ def test_bytes(): assert r.serialize() == data -def test_lvbytes(): - data = b"abcde\x00\xff" - extra = b"\xffrest of the data\x00" - - r, rest = t.LVBytes.deserialize(len(data).to_bytes(2, "little") + data + extra) - assert rest == extra - assert r == data - - assert r.serialize() == len(data).to_bytes(2, "little") + data - - -def test_struct(): - class TestStruct(t.Struct): - _fields = [("a", t.uint8_t), ("b", t.uint8_t)] - - ts = TestStruct() - ts.a = t.uint8_t(0xAA) - ts.b = t.uint8_t(0xBB) - ts2 = TestStruct(ts) - assert ts2.a == ts.a - assert ts2.b == ts.b - assert ts == ts2 - assert ts != 123 - - r = repr(ts) - assert "TestStruct" in r - assert r.startswith("<") and r.endswith(">") - - s = ts2.serialize() - assert s == b"\xaa\xbb" - - extra = b"\x00extra data\xff" - d, rest = TestStruct.deserialize(s + extra) - assert rest == extra - assert d.a == ts.a - assert d.b == ts.b - - -def test_list(): - class TestList(t.List): - _itemtype = t.uint16_t - - r = TestList([1, 2, 3, 0x55AA]) - assert r.serialize() == b"\x01\x00\x02\x00\x03\x00\xaa\x55" - - -def test_list_deserialize(): - class TestList(t.List): - _itemtype = t.uint16_t - - data = b"\x34\x12\x55\xaa\x89\xab" - extra = b"\x00\xff" - - r, rest = TestList.deserialize(data + extra) - assert rest == b"" - assert r[0] == 0x1234 - assert r[1] == 0xAA55 - assert r[2] == 0xAB89 - assert r[3] == 0xFF00 - - -def test_fixed_list(): - class TestList(t.FixedList): - _length = 3 - _itemtype = t.uint16_t - - with pytest.raises(AssertionError): - r = TestList([1, 2, 3, 0x55AA]) - r.serialize() - - with pytest.raises(AssertionError): - r = TestList([1, 2]) - r.serialize() - - r = TestList([1, 2, 3]) - - assert r.serialize() == b"\x01\x00\x02\x00\x03\x00" - - -def test_fixed_list_deserialize(): - class TestList(t.FixedList): - _length = 3 - _itemtype = t.uint16_t - - data = b"\x34\x12\x55\xaa\x89\xab" - extra = b"\x00\xff" - - r, rest = TestList.deserialize(data + extra) - assert rest == extra - assert r[0] == 0x1234 - assert r[1] == 0xAA55 - assert r[2] == 0xAB89 - - -def test_eui64(): - r = t.EUI64([0x0F, 0x0E, 0x0D, 0x0C, 0x0B, 0x0A, 0x09, 0x08]) - ieee = "08:09:0a:0b:0c:0d:0e:0f" - assert repr(r) == ieee - i = {} - i[r] = mock.sentinel.data - - -def test_hexrepr(): - class TestHR(t.HexRepr, t.uint16_t): - pass - - i = TestHR(0xAA55) - assert repr(i) == "0xaa55" - assert str(i) == "0xaa55" - - def test_addr_ep_nwk(): data = b"\x02\xaa\x55\xcc" extra = b"\x00extra data\xff" @@ -312,7 +181,7 @@ def test_deconz_addr_ep(): a = t.DeconzAddressEndpoint() a.address_mode = 2 a.address = 0x55AA - with pytest.raises(AttributeError): + with pytest.raises(TypeError): a.serialize() a.endpoint = 0xCC assert a.serialize() == data @@ -327,7 +196,7 @@ def test_deconz_addr_ep(): a = t.DeconzAddressEndpoint() a.address_mode = 3 a.address = [0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37, 0x38] - with pytest.raises(AttributeError): + with pytest.raises(TypeError): a.serialize() a.endpoint = 0xCC assert a.serialize() == data @@ -347,3 +216,13 @@ def test_nwklist(): t.NWKList([0x1234, 0x5678]), b"abc", ) + + +def test_serialize_dict(): + assert ( + t.serialize_dict( + {"foo": 1, "bar": 2, "baz": None}, + {"foo": t.uint8_t, "bar": t.uint16_t, "baz": t.uint8_t}, + ) + == b"\x01\x02\x00" + ) diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 0000000..d3841d8 --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,24 @@ +"""Test utils module.""" + +import asyncio +import logging +from unittest.mock import AsyncMock + +from zigpy_deconz import utils + + +async def test_restart_forever(caplog): + mock = AsyncMock(side_effect=[None, RuntimeError(), RuntimeError(), None]) + func = utils.restart_forever( + mock, + restart_delay=0.1, + ) + + with caplog.at_level(logging.DEBUG): + task = asyncio.create_task(func()) + await asyncio.sleep(0.5) + task.cancel() + + assert caplog.text.count("failed, restarting...") >= 2 + assert caplog.text.count("RuntimeError") == 2 + assert len(mock.mock_calls) >= 4 diff --git a/zigpy_deconz/api.py b/zigpy_deconz/api.py index 1154bf4..a6fa81c 100644 --- a/zigpy_deconz/api.py +++ b/zigpy_deconz/api.py @@ -3,9 +3,7 @@ from __future__ import annotations import asyncio -import binascii -import enum -import functools +import itertools import logging import sys from typing import Any, Callable @@ -16,23 +14,33 @@ from asyncio import timeout as asyncio_timeout # pragma: no cover from zigpy.config import CONF_DEVICE_PATH -import zigpy.exceptions -from zigpy.types import APSStatus, Bool, Channels +from zigpy.types import ( + APSStatus, + Bool, + Channels, + KeyData, + SerializableBytes, + Struct, + ZigbeePacket, +) from zigpy.zdo.types import SimpleDescriptor from zigpy_deconz.exception import APIException, CommandError import zigpy_deconz.types as t import zigpy_deconz.uart +from zigpy_deconz.utils import restart_forever LOGGER = logging.getLogger(__name__) COMMAND_TIMEOUT = 1.8 PROBE_TIMEOUT = 2 -MIN_PROTO_VERSION = 0x010B REQUEST_RETRY_DELAYS = (0.5, 1.0, 1.5, None) +FRAME_LENGTH = object() +PAYLOAD_LENGTH = object() -class Status(t.uint8_t, enum.Enum): + +class Status(t.enum8): SUCCESS = 0 FAILURE = 1 BUSY = 2 @@ -43,36 +51,45 @@ class Status(t.uint8_t, enum.Enum): INVALID_VALUE = 7 -class DeviceState(enum.IntFlag): - APSDE_DATA_CONFIRM = 0x04 - APSDE_DATA_INDICATION = 0x08 - CONF_CHANGED = 0x10 - APSDE_DATA_REQUEST_SLOTS_AVAILABLE = 0x20 +class NetworkState2(t.enum2): + OFFLINE = 0 + JOINING = 1 + CONNECTED = 2 + LEAVING = 3 - @classmethod - def deserialize(cls, data) -> tuple[DeviceState, bytes]: - """Deserialize DevceState.""" - state, data = t.uint8_t.deserialize(data) - return cls(state), data - def serialize(self) -> bytes: - """Serialize data.""" - return t.uint8_t(self).serialize() +class DeviceStateFlags(t.bitmap6): + APSDE_DATA_CONFIRM = 0b00001 + APSDE_DATA_INDICATION = 0b000010 + CONF_CHANGED = 0b000100 + APSDE_DATA_REQUEST_FREE_SLOTS_AVAILABLE = 0b0001000 - @property - def network_state(self) -> NetworkState: - """Return network state.""" - return NetworkState(self & 0x03) + +class DeviceState(t.Struct): + network_state: NetworkState2 + device_state: DeviceStateFlags -class NetworkState(t.uint8_t, enum.Enum): +class FirmwarePlatform(t.enum8): + Conbee = 0x05 + Conbee_II = 0x07 + + +class FirmwareVersion(t.Struct, t.uint32_t): + reserved: t.uint8_t + platform: FirmwarePlatform + minor: t.uint8_t + major: t.uint8_t + + +class NetworkState(t.enum8): OFFLINE = 0 JOINING = 1 CONNECTED = 2 LEAVING = 3 -class SecurityMode(t.uint8_t, enum.Enum): +class SecurityMode(t.enum8): NO_SECURITY = 0x00 PRECONFIGURED_NETWORK_KEY = 0x01 NETWORK_KEY_FROM_TC = 0x02 @@ -84,7 +101,7 @@ class ZDPResponseHandling(t.bitmap16): NodeDescRsp = 0x0001 -class Command(t.uint8_t, enum.Enum): +class CommandId(t.enum8): aps_data_confirm = 0x04 device_state = 0x07 change_network_state = 0x08 @@ -97,10 +114,10 @@ class Command(t.uint8_t, enum.Enum): zigbee_green_power = 0x19 mac_poll = 0x1C add_neighbour = 0x1D - simplified_beacon = 0x1F + mac_beacon_indication = 0x1F -class TXStatus(t.uint8_t, enum.Enum): +class TXStatus(t.enum8): SUCCESS = 0x00 @classmethod @@ -112,89 +129,7 @@ def _missing_(cls, value): return status -TX_COMMANDS = { - Command.add_neighbour: (t.uint16_t, t.uint8_t, t.NWK, t.EUI64, t.uint8_t), - Command.aps_data_confirm: (t.uint16_t,), - Command.aps_data_indication: ( - t.uint16_t, - t.DataIndicationFlags, - ), - Command.aps_data_request: ( - t.uint16_t, - t.uint8_t, - t.DeconzSendDataFlags, - t.DeconzAddressEndpoint, - t.uint16_t, - t.uint16_t, - t.uint8_t, - t.LVBytes, - t.uint8_t, - t.uint8_t, - t.NWKList, # optional - ), - Command.change_network_state: (t.uint8_t,), - Command.device_state: (t.uint8_t, t.uint8_t, t.uint8_t), - Command.read_parameter: (t.uint16_t, t.uint8_t, t.Bytes), - Command.version: (t.uint32_t,), - Command.write_parameter: (t.uint16_t, t.uint8_t, t.Bytes), -} - -RX_COMMANDS = { - Command.add_neighbour: ((t.uint16_t, t.uint8_t, t.NWK, t.EUI64, t.uint8_t), True), - Command.aps_data_confirm: ( - ( - t.uint16_t, - DeviceState, - t.uint8_t, - t.DeconzAddressEndpoint, - t.uint8_t, - TXStatus, - t.uint8_t, - t.uint8_t, - t.uint8_t, - t.uint8_t, - ), - True, - ), - Command.aps_data_indication: ( - ( - t.uint16_t, - DeviceState, - t.DeconzAddress, - t.uint8_t, - t.DeconzAddress, - t.uint8_t, - t.uint16_t, - t.uint16_t, - t.LVBytes, - t.uint8_t, - t.uint8_t, - t.uint8_t, - t.uint8_t, - t.uint8_t, - t.uint8_t, - t.uint8_t, - t.int8s, - ), - True, - ), - Command.aps_data_request: ((t.uint16_t, DeviceState, t.uint8_t), True), - Command.change_network_state: ((t.uint8_t,), True), - Command.device_state: ((DeviceState, t.uint8_t, t.uint8_t), True), - Command.device_state_changed: ((DeviceState, t.uint8_t), False), - Command.mac_poll: ((t.uint16_t, t.DeconzAddress, t.uint8_t, t.int8s), False), - Command.read_parameter: ((t.uint16_t, t.uint8_t, t.Bytes), True), - Command.simplified_beacon: ( - (t.uint16_t, t.uint16_t, t.uint16_t, t.uint8_t, t.uint8_t, t.uint8_t), - False, - ), - Command.version: ((t.uint32_t,), True), - Command.write_parameter: ((t.uint16_t, t.uint8_t), True), - Command.zigbee_green_power: ((t.LVBytes,), False), -} - - -class NetworkParameter(t.uint8_t, enum.Enum): +class NetworkParameter(t.enum8): mac_address = 0x01 nwk_panid = 0x05 nwk_address = 0x07 @@ -217,27 +152,255 @@ class NetworkParameter(t.uint8_t, enum.Enum): app_zdp_response_handling = 0x28 -NETWORK_PARAMETER_SCHEMA = { - NetworkParameter.mac_address: (t.EUI64,), - NetworkParameter.nwk_panid: (t.PanId,), - NetworkParameter.nwk_address: (t.NWK,), - NetworkParameter.nwk_extended_panid: (t.ExtendedPanId,), - NetworkParameter.aps_designed_coordinator: (t.uint8_t,), - NetworkParameter.channel_mask: (Channels,), - NetworkParameter.aps_extended_panid: (t.ExtendedPanId,), - NetworkParameter.trust_center_address: (t.EUI64,), - NetworkParameter.security_mode: (t.uint8_t,), - NetworkParameter.use_predefined_nwk_panid: (Bool,), - NetworkParameter.network_key: (t.uint8_t, t.Key), - NetworkParameter.link_key: (t.EUI64, t.Key), - NetworkParameter.current_channel: (t.uint8_t,), - NetworkParameter.permit_join: (t.uint8_t,), - NetworkParameter.configure_endpoint: (t.uint8_t, SimpleDescriptor), - NetworkParameter.protocol_version: (t.uint16_t,), - NetworkParameter.nwk_update_id: (t.uint8_t,), - NetworkParameter.watchdog_ttl: (t.uint32_t,), - NetworkParameter.nwk_frame_counter: (t.uint32_t,), - NetworkParameter.app_zdp_response_handling: (ZDPResponseHandling,), +class IndexedKey(Struct): + index: t.uint8_t + key: KeyData + + +class LinkKey(Struct): + ieee: t.EUI64 + key: KeyData + + +class IndexedEndpoint(Struct): + index: t.uint8_t + descriptor: SimpleDescriptor + + +NETWORK_PARAMETER_TYPES = { + NetworkParameter.mac_address: (None, t.EUI64), + NetworkParameter.nwk_panid: (None, t.PanId), + NetworkParameter.nwk_address: (None, t.NWK), + NetworkParameter.nwk_extended_panid: (None, t.ExtendedPanId), + NetworkParameter.aps_designed_coordinator: (None, t.uint8_t), + NetworkParameter.channel_mask: (None, Channels), + NetworkParameter.aps_extended_panid: (None, t.ExtendedPanId), + NetworkParameter.trust_center_address: (None, t.EUI64), + NetworkParameter.security_mode: (None, t.uint8_t), + NetworkParameter.configure_endpoint: (t.uint8_t, IndexedEndpoint), + NetworkParameter.use_predefined_nwk_panid: (None, Bool), + NetworkParameter.network_key: (t.uint8_t, IndexedKey), + NetworkParameter.link_key: (t.EUI64, LinkKey), + NetworkParameter.current_channel: (None, t.uint8_t), + NetworkParameter.permit_join: (None, t.uint8_t), + NetworkParameter.protocol_version: (None, t.uint16_t), + NetworkParameter.nwk_update_id: (None, t.uint8_t), + NetworkParameter.watchdog_ttl: (None, t.uint32_t), + NetworkParameter.nwk_frame_counter: (None, t.uint32_t), + NetworkParameter.app_zdp_response_handling: (None, ZDPResponseHandling), +} + + +class Command(Struct): + command_id: CommandId + seq: t.uint8_t + payload: t.Bytes + + +COMMAND_SCHEMAS = { + CommandId.add_neighbour: ( + { + "status": Status.SUCCESS, + "frame_length": FRAME_LENGTH, + "payload_length": PAYLOAD_LENGTH, + "unknown": t.uint8_t, + "nwk": t.NWK, + "ieee": t.EUI64, + "mac_capability_flags": t.uint8_t, + }, + { + "status": Status, + "frame_length": t.uint16_t, + "payload_length": t.uint16_t, + "unknown": t.uint8_t, + "nwk": t.NWK, + "ieee": t.EUI64, + "mac_capability_flags": t.uint8_t, + }, + ), + CommandId.aps_data_confirm: ( + { + "status": Status.SUCCESS, + "frame_length": FRAME_LENGTH, + "payload_length": PAYLOAD_LENGTH, + }, + { + "status": Status, + "frame_length": t.uint16_t, + "payload_length": t.uint16_t, + "device_state": DeviceState, + "request_id": t.uint8_t, + "dst_addr": t.DeconzAddressEndpoint, + "src_ep": t.uint8_t, + "confirm_status": TXStatus, + "reserved1": t.uint8_t, + "reserved2": t.uint8_t, + "reserved3": t.uint8_t, + "reserved4": t.uint8_t, + }, + ), + CommandId.aps_data_indication: ( + { + "status": Status.SUCCESS, + "frame_length": FRAME_LENGTH, + "payload_length": PAYLOAD_LENGTH, + "flags": t.DataIndicationFlags, + }, + { + "status": Status, + "frame_length": t.uint16_t, + "payload_length": t.uint16_t, + "device_state": DeviceState, + "dst_addr": t.DeconzAddress, + "dst_ep": t.uint8_t, + "src_addr": t.DeconzAddress, + "src_ep": t.uint8_t, + "profile_id": t.uint16_t, + "cluster_id": t.uint16_t, + "asdu": t.LongOctetString, + "reserved1": t.uint8_t, + "reserved2": t.uint8_t, + "lqi": t.uint8_t, + "reserved3": t.uint8_t, + "reserved4": t.uint8_t, + "reserved5": t.uint8_t, + "reserved6": t.uint8_t, + "rssi": t.int8s, + }, + ), + CommandId.aps_data_request: ( + { + "status": Status.SUCCESS, + "frame_length": FRAME_LENGTH, + "payload_length": PAYLOAD_LENGTH, + "request_id": t.uint8_t, + "flags": t.DeconzSendDataFlags, + "dst": t.DeconzAddressEndpoint, + "profile_id": t.uint16_t, + "cluster_id": t.uint16_t, + "src_ep": t.uint8_t, + "asdu": t.LongOctetString, + "tx_options": t.DeconzTransmitOptions, + "radius": t.uint8_t, + "relays": t.NWKList, # optional + }, + { + "status": Status, + "frame_length": t.uint16_t, + "payload_length": t.uint16_t, + "device_state": DeviceState, + "request_id": t.uint8_t, + }, + ), + CommandId.change_network_state: ( + { + "status": Status.SUCCESS, + "frame_length": FRAME_LENGTH, + # "payload_length": PAYLOAD_LENGTH, + "network_state": NetworkState, + }, + { + "status": Status, + "frame_length": t.uint16_t, + # "payload_length": t.uint16_t, + "network_state": NetworkState, + }, + ), + CommandId.device_state: ( + { + "status": Status.SUCCESS, + "frame_length": FRAME_LENGTH, + # "payload_length": PAYLOAD_LENGTH, + "reserved1": t.uint8_t(0), + "reserved2": t.uint8_t(0), + "reserved3": t.uint8_t(0), + }, + { + "status": Status, + "frame_length": t.uint16_t, + # "payload_length": t.uint16_t, + "device_state": DeviceState, + "reserved1": t.uint8_t, + "reserved2": t.uint8_t, + }, + ), + CommandId.device_state_changed: ( + None, + { + "status": Status, + "frame_length": t.uint16_t, + # "payload_length": t.uint16_t, + "device_state": DeviceState, + "reserved": t.uint8_t, + }, + ), + CommandId.mac_poll: ( + None, + { + "status": Status, + "frame_length": t.uint16_t, + "payload_length": t.uint16_t, + "src_addr": t.DeconzAddress, + "lqi": t.uint8_t, + "rssi": t.int8s, + "life_time": t.uint32_t, # Optional + "device_timeout": t.uint32_t, # Optional + }, + ), + CommandId.read_parameter: ( + { + "status": Status.SUCCESS, + "frame_length": FRAME_LENGTH, + "payload_length": PAYLOAD_LENGTH, + "parameter_id": NetworkParameter, + "parameter": t.Bytes, + }, + { + "status": Status, + "frame_length": t.uint16_t, + "payload_length": t.uint16_t, + "parameter_id": NetworkParameter, + "parameter": t.Bytes, + }, + ), + CommandId.version: ( + { + "status": Status.SUCCESS, + "frame_length": FRAME_LENGTH, + # "payload_length": PAYLOAD_LENGTH, + "reserved": t.uint32_t(0), + }, + { + "status": Status, + "frame_length": t.uint16_t, + # "payload_length": t.uint16_t, + "version": FirmwareVersion, + }, + ), + CommandId.write_parameter: ( + { + "status": Status.SUCCESS, + "frame_length": FRAME_LENGTH, + "payload_length": PAYLOAD_LENGTH, + "parameter_id": NetworkParameter, + "parameter": t.Bytes, + }, + { + "status": Status, + "frame_length": t.uint16_t, + "payload_length": t.uint16_t, + "parameter_id": NetworkParameter, + }, + ), + CommandId.zigbee_green_power: ( + None, + { + "status": Status, + "frame_length": t.uint16_t, + "payload_length": t.uint16_t, + "reserved": t.LongOctetString, + }, + ), } @@ -247,22 +410,31 @@ class Deconz: def __init__(self, app: Callable, device_config: dict[str, Any]): """Init instance.""" self._app = app - self._aps_data_ind_flags: t.DataIndicationFlags = ( - t.DataIndicationFlags.Always_Use_NWK_Source_Addr - ) self._awaiting = {} self._command_lock = asyncio.Lock() self._config = device_config - self._data_indication: bool = False - self._data_confirm: bool = False - self._device_state = DeviceState(NetworkState.OFFLINE) + self._device_state = DeviceState( + network_state=NetworkState2.OFFLINE, + device_state=( + DeviceStateFlags.APSDE_DATA_CONFIRM + | DeviceStateFlags.APSDE_DATA_INDICATION + ), + ) + + self._free_slots_available_event = asyncio.Event() + self._free_slots_available_event.set() + + self._data_poller_event = asyncio.Event() + self._data_poller_event.set() + self._data_poller_task: asyncio.Task | None = None + self._seq = 1 - self._proto_ver: int | None = None - self._firmware_version: int | None = None + self._protocol_version = 0 + self._firmware_version = FirmwareVersion(0) self._uart: zigpy_deconz.uart.Gateway | None = None @property - def firmware_version(self) -> int | None: + def firmware_version(self) -> FirmwareVersion: """Return ConBee firmware version.""" return self._firmware_version @@ -272,14 +444,21 @@ def network_state(self) -> NetworkState: return self._device_state.network_state @property - def protocol_version(self) -> int | None: + def protocol_version(self) -> int: """Protocol Version.""" - return self._proto_ver + return self._protocol_version async def connect(self) -> None: assert self._uart is None self._uart = await zigpy_deconz.uart.connect(self._config, self) + await self.version() + + device_state_rsp = await self._command(CommandId.device_state) + self._device_state = device_state_rsp["device_state"] + + self._data_poller_task = asyncio.create_task(self._data_poller()) + def connection_lost(self, exc: Exception) -> None: """Lost serial connection.""" LOGGER.debug( @@ -294,20 +473,86 @@ def connection_lost(self, exc: Exception) -> None: def close(self): self._app = None + if self._data_poller_task is not None: + self._data_poller_task.cancel() + self._data_poller_task = None + if self._uart is not None: self._uart.close() self._uart = None - async def _command(self, cmd, *args): + async def _command(self, cmd, **kwargs): + payload = [] + tx_schema, _ = COMMAND_SCHEMAS[cmd] + trailing_optional = False + + for name, param_type in tx_schema.items(): + if isinstance(param_type, int): + if name not in kwargs: + # Default value + value = param_type.serialize() + else: + value = type(param_type)(kwargs[name]).serialize() + elif name in ("frame_length", "payload_length"): + value = param_type + elif kwargs.get(name) is None: + trailing_optional = True + value = None + elif not isinstance(kwargs[name], param_type): + value = param_type(kwargs[name]).serialize() + else: + value = kwargs[name].serialize() + + if value is None: + continue + + if trailing_optional: + raise ValueError( + f"Command {cmd} with kwargs {kwargs}" + f" has non-trailing optional argument" + ) + + payload.append(value) + + if PAYLOAD_LENGTH in payload: + payload = t.list_replace( + lst=payload, + old=PAYLOAD_LENGTH, + new=t.uint16_t( + sum(len(p) for p in payload[payload.index(PAYLOAD_LENGTH) + 1 :]) + ).serialize(), + ) + + if FRAME_LENGTH in payload: + payload = t.list_replace( + lst=payload, + old=FRAME_LENGTH, + new=t.uint16_t( + 2 + sum(len(p) if p is not FRAME_LENGTH else 2 for p in payload) + ).serialize(), + ) + + command = Command( + command_id=cmd, + seq=None, + payload=b"".join(payload), + ) + if self._uart is None: # connection was lost raise CommandError(Status.ERROR, "API is not running") + async with self._command_lock: - LOGGER.debug("Command %s %s", cmd, args) - data, seq = self._api_frame(cmd, *args) - self._uart.send(data) + seq = self._seq + + LOGGER.debug("Sending %s%s (seq=%s)", cmd, kwargs, seq) + self._uart.send(command.replace(seq=seq).serialize()) + + self._seq = (self._seq % 255) + 1 + fut = asyncio.Future() - self._awaiting[seq] = fut + self._awaiting[seq, cmd] = fut + try: async with asyncio_timeout(COMMAND_TIMEOUT): return await fut @@ -315,211 +560,200 @@ async def _command(self, cmd, *args): LOGGER.warning( "No response to '%s' command with seq id '0x%02x'", cmd, seq ) - self._awaiting.pop(seq, None) + self._awaiting.pop((seq, cmd), None) raise - def _api_frame(self, cmd, *args): - schema = TX_COMMANDS[cmd] - d = t.serialize(args, schema) - data = t.uint8_t(cmd).serialize() - self._seq = (self._seq % 255) + 1 - data += t.uint8_t(self._seq).serialize() - data += t.uint8_t(0).serialize() - data += t.uint16_t(len(d) + 5).serialize() - data += d - return data, self._seq - - def data_received(self, data): - try: - command = Command(data[0]) - schema, solicited = RX_COMMANDS[command] - except ValueError: - LOGGER.debug("Unknown command received: 0x%02x", data[0]) + def data_received(self, data: bytes) -> None: + command, _ = Command.deserialize(data) + + if command.command_id not in COMMAND_SCHEMAS: + LOGGER.warning("Unknown command received: %s", command) return - seq = data[1] - try: - status = Status(data[2]) - except ValueError: - status = data[2] - - fut = None - if solicited and seq in self._awaiting: - fut = self._awaiting.pop(seq) - if status != Status.SUCCESS: - try: - fut.set_exception( - CommandError(status, f"{command}, status: {status}") - ) - except asyncio.InvalidStateError: - LOGGER.warning( - "Duplicate or delayed response for 0x:%02x sequence", seq - ) - return + + _, rx_schema = COMMAND_SCHEMAS[command.command_id] + + fut = self._awaiting.pop((command.seq, command.command_id), None) try: - data, _ = t.deserialize(data[5:], schema) + params, rest = t.deserialize_dict(command.payload, rx_schema) except Exception: - LOGGER.warning("Failed to deserialize frame: %s", binascii.hexlify(data)) + LOGGER.warning("Failed to parse command %s", command, exc_info=True) + if fut is not None and not fut.done(): fut.set_exception( - APIException( - f"Failed to deserialize frame: {binascii.hexlify(data)}" - ) + APIException(f"Failed to deserialize command: {command}") ) + return + if rest: + LOGGER.debug("Unparsed data remains after frame: %s, %s", command, rest) + + assert params["frame_length"] == len(data) + + if "payload_length" in params: + running_length = itertools.accumulate( + len(v.serialize()) if v is not None else 0 for v in params.values() + ) + length_at_param = dict(zip(params.keys(), running_length)) + + assert ( + len(data) - length_at_param["payload_length"] - 2 + == params["payload_length"] + ) + + LOGGER.debug( + "Received command %s%s (seq %d)", command.command_id, params, command.seq + ) + status = params["status"] + + exc = None + + if status != Status.SUCCESS: + exc = CommandError(status, f"{command.command_id}, status: {status}") + if fut is not None: try: - fut.set_result(data) + if exc is None: + fut.set_result(params) + else: + fut.set_exception(exc) except asyncio.InvalidStateError: LOGGER.warning( - "Duplicate or delayed response for 0x:%02x sequence", seq + "Duplicate or delayed response for 0x:%02x sequence", + command.seq, ) - LOGGER.debug("Received command %s%r", command.name, data) - getattr(self, f"_handle_{command.name}")(data) + if exc is not None: + return - add_neighbour = functools.partialmethod(_command, Command.add_neighbour, 12) - device_state = functools.partialmethod(_command, Command.device_state, 0, 0, 0) - change_network_state = functools.partialmethod( - _command, Command.change_network_state - ) + if handler := getattr(self, f"_handle_{command.command_id.name}", None): + handler_params = { + k: v + for k, v in params.items() + if k not in ("frame_length", "payload_length") + } + + # Queue up the callback within the event loop + asyncio.get_running_loop().call_soon(lambda: handler(**handler_params)) + + @restart_forever + async def _data_poller(self): + while True: + await self._data_poller_event.wait() + self._data_poller_event.clear() + + if self._device_state.network_state == NetworkState2.OFFLINE: + continue + + # Poll data indication + if ( + DeviceStateFlags.APSDE_DATA_INDICATION + in self._device_state.device_state + ): + # Old Conbee I firmware has an addressing bug for incoming multicasts + if ( + self.protocol_version >= 0x010B + and self.firmware_version.platform == FirmwarePlatform.Conbee + ): + flags = t.DataIndicationFlags.Include_Both_NWK_And_IEEE + else: + flags = t.DataIndicationFlags.Always_Use_NWK_Source_Addr + + rsp = await self._command(CommandId.aps_data_indication, flags=flags) + self._handle_device_state_changed( + status=rsp["status"], device_state=rsp["device_state"] + ) - def _handle_device_state(self, data): - LOGGER.debug("Device state response: %s", data) - self._handle_device_state_value(data[0]) + self._app.packet_received( + ZigbeePacket( + src=rsp["src_addr"].as_zigpy_type(), + src_ep=rsp["src_ep"], + dst=rsp["dst_addr"].as_zigpy_type(), + dst_ep=rsp["dst_ep"], + tsn=None, + profile_id=rsp["profile_id"], + cluster_id=rsp["cluster_id"], + data=SerializableBytes(rsp["asdu"]), + lqi=rsp["lqi"], + rssi=rsp["rssi"], + ) + ) - def _handle_change_network_state(self, data): - LOGGER.debug("Change network state response: %s", NetworkState(data[0]).name) + # Poll data confirm + if DeviceStateFlags.APSDE_DATA_CONFIRM in self._device_state.device_state: + rsp = await self._command(CommandId.aps_data_confirm) - @classmethod - async def probe(cls, device_config: dict[str, Any]) -> bool: - """Probe port for the device presence.""" - api = cls(None, device_config) - try: - async with asyncio_timeout(PROBE_TIMEOUT): - await api._probe() - return True - except Exception as exc: + self._app.handle_tx_confirm(rsp["request_id"], rsp["confirm_status"]) + self._handle_device_state_changed( + status=rsp["status"], device_state=rsp["device_state"] + ) + + def _handle_device_state_changed( + self, + status: t.Status, + device_state: DeviceState, + reserved: t.uint8_t = 0, + ) -> None: + if device_state.network_state != self.network_state: LOGGER.debug( - "Unsuccessful radio probe of '%s' port", - device_config[CONF_DEVICE_PATH], - exc_info=exc, + "Network device_state transition: %s -> %s", + self.network_state.name, + device_state.network_state.name, ) - finally: - api.close() - - return False - async def _probe(self) -> None: - """Open port and try sending a command.""" - await self.connect() - await self.device_state() - self.close() - - async def read_parameter(self, id_, *args): - try: - if isinstance(id_, str): - param = NetworkParameter[id_] - else: - param = NetworkParameter(id_) - except (KeyError, ValueError): - raise KeyError(f"Unknown parameter id: {id_}") + if ( + DeviceStateFlags.APSDE_DATA_REQUEST_FREE_SLOTS_AVAILABLE + in device_state.device_state + ): + self._free_slots_available_event.set() + else: + self._free_slots_available_event.clear() - data = t.serialize(args, NETWORK_PARAMETER_SCHEMA[param]) - r = await self._command(Command.read_parameter, 1 + len(data), param, data) - data = t.deserialize(r[2], NETWORK_PARAMETER_SCHEMA[param])[0] - LOGGER.debug("Read parameter %s response: %s", param.name, data) - return data + self._device_state = device_state + self._data_poller_event.set() - def reconnect(self): - """Reconnect using saved parameters.""" - LOGGER.debug("Reconnecting '%s' serial port", self._config[CONF_DEVICE_PATH]) - return self.connect() + async def version(self): + self._protocol_version = await self.read_parameter( + NetworkParameter.protocol_version + ) - def _handle_read_parameter(self, data): - pass + version_rsp = await self._command(CommandId.version, reserved=0) + self._firmware_version = version_rsp["version"] - def write_parameter(self, id_, *args): - try: - if isinstance(id_, str): - param = NetworkParameter[id_] - else: - param = NetworkParameter(id_) - except (KeyError, ValueError): - raise KeyError(f"Unknown parameter id: {id_} write request") + return self.firmware_version - v = t.serialize(args, NETWORK_PARAMETER_SCHEMA[param]) - length = len(v) + 1 - return self._command(Command.write_parameter, length, param, v) + async def read_parameter( + self, parameter_id: NetworkParameter, parameter: Any = None + ) -> Any: + read_param_type, write_param_type = NETWORK_PARAMETER_TYPES[parameter_id] - def _handle_write_parameter(self, data): - try: - param = NetworkParameter(data[1]) - except ValueError: - LOGGER.error("Received unknown network param id '%s' response", data[1]) - return - LOGGER.debug("Write parameter %s: SUCCESS", param.name) + if parameter is None: + value = t.Bytes(b"") + else: + value = read_param_type(parameter).serialize() - async def version(self): - (self._proto_ver,) = await self.read_parameter( - NetworkParameter.protocol_version + rsp = await self._command( + CommandId.read_parameter, + parameter_id=parameter_id, + parameter=value, ) - (self._firmware_version,) = await self._command(Command.version, 0) - if ( - self.protocol_version >= MIN_PROTO_VERSION - and (self.firmware_version & 0x0000FF00) == 0x00000500 - ): - self._aps_data_ind_flags = t.DataIndicationFlags.Include_Both_NWK_And_IEEE - return self.firmware_version - def _handle_version(self, data): - LOGGER.debug("Version response: %x", data[0]) + assert rsp["parameter_id"] == parameter_id - def _handle_device_state_changed(self, data): - LOGGER.debug("Device state changed response: %s", data) - self._handle_device_state_value(data[0]) + result, _ = write_param_type.deserialize(rsp["parameter"]) + LOGGER.debug("Read parameter %s(%s)=%r", parameter_id.name, parameter, result) - async def _aps_data_indication(self): - try: - r = await self._command( - Command.aps_data_indication, 1, self._aps_data_ind_flags - ) - LOGGER.debug( - ( - "'aps_data_indication' response from %s, ep: %s, " - "profile: 0x%04x, cluster_id: 0x%04x, data: %s" - ), - r[4], - r[5], - r[6], - r[7], - binascii.hexlify(r[8]), - ) - return r - except (asyncio.TimeoutError, zigpy.exceptions.ZigbeeException): - pass - finally: - self._data_indication = False - - def _handle_aps_data_indication(self, data): - LOGGER.debug("APS data indication response: %s", data) - self._data_indication = False - self._handle_device_state_value(data[1]) - - if not self._app: - return + return result - self._app.handle_rx( - src=data[4], - src_ep=data[5], - dst=data[2], - dst_ep=data[3], - profile_id=data[6], - cluster_id=data[7], - data=data[8], - lqi=data[11], - rssi=data[16], + async def write_parameter( + self, parameter_id: NetworkParameter, parameter: Any + ) -> None: + read_param_type, write_param_type = NETWORK_PARAMETER_TYPES[parameter_id] + await self._command( + CommandId.write_parameter, + parameter_id=parameter_id, + parameter=write_param_type(parameter).serialize(), ) async def aps_data_request( @@ -534,37 +768,33 @@ async def aps_data_request( relays=None, tx_options=t.DeconzTransmitOptions.USE_NWK_KEY_SECURITY, radius=0, - ): - dst = dst_addr_ep.serialize() - length = len(dst) + len(aps_payload) + 11 - + ) -> None: flags = t.DeconzSendDataFlags.NONE - extras = [] # https://github.com/zigpy/zigpy-deconz/issues/180#issuecomment-1017932865 - if relays: + if relays is not None: # There is a max of 9 relays assert len(relays) <= 9 flags |= t.DeconzSendDataFlags.RELAYS - extras.append(t.NWKList(relays)) - - length += sum(len(e.serialize()) for e in extras) for delay in REQUEST_RETRY_DELAYS: + if not self._free_slots_available_event.is_set(): + LOGGER.debug("Waiting for free slots to become available") + await self._free_slots_available_event.wait() + try: - return await self._command( - Command.aps_data_request, - length, - req_id, - flags, - dst_addr_ep, - profile, - cluster, - src_ep, - aps_payload, - tx_options, - radius, - *extras, + rsp = await self._command( + CommandId.aps_data_request, + request_id=req_id, + flags=flags, + dst=dst_addr_ep, + profile_id=profile, + cluster_id=cluster, + src_ep=src_ep, + asdu=aps_payload, + tx_options=tx_options, + radius=radius, + relays=relays, ) except CommandError as ex: LOGGER.debug("'aps_data_request' failure: %s", ex) @@ -573,75 +803,27 @@ async def aps_data_request( LOGGER.debug("retrying 'aps_data_request' in %ss", delay) await asyncio.sleep(delay) + else: + self._handle_device_state_changed( + status=rsp["status"], device_state=rsp["device_state"] + ) + return - def _handle_aps_data_request(self, data): - LOGGER.debug("APS data request response: %s", data) - self._handle_device_state_value(data[1]) + async def get_device_state(self) -> DeviceState: + rsp = await self._command(CommandId.device_state) - async def _aps_data_confirm(self): - try: - r = await self._command(Command.aps_data_confirm, 0) - LOGGER.debug( - "Request id: 0x%02x 'aps_data_confirm' for %s, status: 0x%02x", - r[2], - r[3], - r[5], - ) - return r - except (asyncio.TimeoutError, zigpy.exceptions.ZigbeeException): - pass - finally: - self._data_confirm = False + return rsp["device_state"] - def _handle_add_neighbour(self, data) -> None: - """Handle add_neighbour response.""" - LOGGER.debug("add neighbour response: %s", data) + async def change_network_state(self, new_state: NetworkState) -> None: + await self._command(CommandId.change_network_state, network_state=new_state) - def _handle_aps_data_confirm(self, data): - LOGGER.debug( - "APS data confirm response for request with id %s: %02x", data[2], data[5] + async def add_neighbour( + self, nwk: t.NWK, ieee: t.EUI64, mac_capability_flags: t.uint8_t + ) -> None: + await self._command( + CommandId.add_neighbour, + unknown=0x01, + nwk=nwk, + ieee=ieee, + mac_capability_flags=mac_capability_flags, ) - self._data_confirm = False - self._handle_device_state_value(data[1]) - self._app.handle_tx_confirm(data[2], data[5]) - - def _handle_mac_poll(self, data): - pass - - def _handle_zigbee_green_power(self, data): - pass - - def _handle_simplified_beacon(self, data): - LOGGER.debug( - ( - "Received simplified beacon frame: source=0x%04x, " - "pan_id=0x%04x, channel=%s, flags=0x%02x, " - "update_id=0x%02x" - ), - data[1], - data[2], - data[3], - data[4], - data[5], - ) - - def _handle_device_state_value(self, state: DeviceState) -> None: - if state.network_state != self.network_state: - LOGGER.debug( - "Network state transition: %s -> %s", - self.network_state.name, - state.network_state.name, - ) - self._device_state = state - if DeviceState.APSDE_DATA_REQUEST_SLOTS_AVAILABLE not in state: - LOGGER.debug("Data request queue full.") - if DeviceState.APSDE_DATA_INDICATION in state and not self._data_indication: - self._data_indication = True - asyncio.create_task(self._aps_data_indication()) - if DeviceState.APSDE_DATA_CONFIRM in state and not self._data_confirm: - self._data_confirm = True - asyncio.create_task(self._aps_data_confirm()) - - def __getitem__(self, key): - """Access parameters via getitem.""" - return self.read_parameter(key) diff --git a/zigpy_deconz/types.py b/zigpy_deconz/types.py index da44e83..d08de8d 100644 --- a/zigpy_deconz/types.py +++ b/zigpy_deconz/types.py @@ -1,21 +1,64 @@ """Data types module.""" -import enum import zigpy.types as zigpy_t -from zigpy.types import bitmap8, bitmap16 # noqa: F401 - - -def deserialize(data, schema): - result = [] - for type_ in schema: - value, data = type_.deserialize(data) - result.append(value) +from zigpy.types import ( # noqa: F401 + EUI64, + NWK, + ExtendedPanId, + LongOctetString, + LVBytes, + LVList, + PanId, + Struct, + bitmap3, + bitmap5, + bitmap6, + bitmap8, + bitmap16, + enum2, + enum3, + enum8, + int8s, + uint8_t, + uint16_t, + uint32_t, + uint64_t, +) + + +def serialize_dict(data, schema): + chunks = [] + + for key in schema: + value = data[key] + if value is None: + break + + if not isinstance(value, schema[key]): + value = schema[key](value) + + chunks.append(value.serialize()) + + return b"".join(chunks) + + +def deserialize_dict(data, schema): + result = {} + for name, type_ in schema.items(): + try: + result[name], data = type_.deserialize(data) + except ValueError: + if data: + raise + + result[name] = None return result, data -def serialize(data, schema): - return b"".join(t(v).serialize() for t, v in zip(schema, data)) +def list_replace(lst: list, old: object, new: object) -> list: + """Replace all occurrences of `old` with `new` in `lst`.""" + return [new if x == old else x for x in lst] class Bytes(bytes): @@ -27,100 +70,7 @@ def deserialize(cls, data): return cls(data), b"" -class LVBytes(bytes): - def serialize(self): - return uint16_t(len(self)).serialize() + self - - @classmethod - def deserialize(cls, data, byteorder="little"): - length, data = uint16_t.deserialize(data) - return cls(data[:length]), data[length:] - - -class int_t(int): - _signed = True - _size = 0 - - def serialize(self, byteorder="little"): - return self.to_bytes(self._size, byteorder, signed=self._signed) - - @classmethod - def deserialize(cls, data, byteorder="little"): - # Work around https://bugs.python.org/issue23640 - r = cls(int.from_bytes(data[: cls._size], byteorder, signed=cls._signed)) - data = data[cls._size :] - return r, data - - -class int8s(int_t): - _size = 1 - - -class int16s(int_t): - _size = 2 - - -class int24s(int_t): - _size = 3 - - -class int32s(int_t): - _size = 4 - - -class int40s(int_t): - _size = 5 - - -class int48s(int_t): - _size = 6 - - -class int56s(int_t): - _size = 7 - - -class int64s(int_t): - _size = 8 - - -class uint_t(int_t): - _signed = False - - -class uint8_t(uint_t): - _size = 1 - - -class uint16_t(uint_t): - _size = 2 - - -class uint24_t(uint_t): - _size = 3 - - -class uint32_t(uint_t): - _size = 4 - - -class uint40_t(uint_t): - _size = 5 - - -class uint48_t(uint_t): - _size = 6 - - -class uint56_t(uint_t): - _size = 7 - - -class uint64_t(uint_t): - _size = 8 - - -class AddressMode(uint8_t, enum.Enum): +class AddressMode(enum8): # Address modes used in deconz protocol GROUP = 0x01 @@ -143,142 +93,9 @@ class DeconzTransmitOptions(bitmap8): ALLOW_FRAGMENTATION = 0x08 -class Struct: - _fields = [] - - def __init__(self, *args, **kwargs): - """Initialize instance.""" - - if len(args) == 1 and isinstance(args[0], self.__class__): - # copy constructor - for field in self._fields: - if hasattr(args[0], field[0]): - setattr(self, field[0], getattr(args[0], field[0])) - - def serialize(self): - r = b"" - for field in self._fields: - if hasattr(self, field[0]): - r += getattr(self, field[0]).serialize() - return r - - @classmethod - def deserialize(cls, data): - """Deserialize data.""" - r = cls() - for field_name, field_type in cls._fields: - v, data = field_type.deserialize(data) - setattr(r, field_name, v) - return r, data - - def __eq__(self, other): - """Check equality between structs.""" - if not isinstance(other, type(self)): - return NotImplemented - - return all(getattr(self, n) == getattr(other, n) for n, _ in self._fields) - - def __repr__(self): - """Instance representation.""" - r = f"<{self.__class__.__name__} " - r += " ".join([f"{f[0]}={getattr(self, f[0], None)}" for f in self._fields]) - r += ">" - return r - - -class List(list): - _length = None - _itemtype = None - - def serialize(self): - assert self._length is None or len(self) == self._length - return b"".join([self._itemtype(i).serialize() for i in self]) - - @classmethod - def deserialize(cls, data): - assert cls._itemtype is not None - r = cls() - while data: - item, data = cls._itemtype.deserialize(data) - r.append(item) - return r, data - - -class LVList(list): - _length_type = None - _itemtype = None - - def serialize(self): - return self._length_type(len(self)).serialize() + b"".join( - [self._itemtype(i).serialize() for i in self] - ) - - @classmethod - def deserialize(cls, data): - length, data = cls._length_type.deserialize(data) - r = cls() - for _ in range(length): - item, data = cls._itemtype.deserialize(data) - r.append(item) - return r, data - - -class FixedList(List): - _length = None - _itemtype = None - - @classmethod - def deserialize(cls, data): - assert cls._itemtype is not None - r = cls() - for i in range(cls._length): - item, data = cls._itemtype.deserialize(data) - r.append(item) - return r, data - - -class EUI64(FixedList): - _length = 8 - _itemtype = uint8_t - - def __repr__(self): - """Instance representation.""" - return ":".join("%02x" % i for i in self[::-1]) - - def __hash__(self): - """Hash magic method.""" - return hash(repr(self)) - - -class HexRepr: - def __repr__(self): - """Instance representation.""" - return ("0x{:0" + str(self._size * 2) + "x}").format(self) - - def __str__(self): - """Instance str method.""" - return ("0x{:0" + str(self._size * 2) + "x}").format(self) - - -class GroupId(HexRepr, uint16_t): - pass - - -class NWK(HexRepr, uint16_t): - pass - - -class PanId(HexRepr, uint16_t): - pass - - -class ExtendedPanId(EUI64): - pass - - class NWKList(LVList): _length_type = uint8_t - _itemtype = NWK + _item_type = NWK ZIGPY_ADDR_MODE_MAPPING = { @@ -292,7 +109,7 @@ class NWKList(LVList): ZIGPY_ADDR_TYPE_MAPPING = { zigpy_t.AddrMode.NWK: NWK, zigpy_t.AddrMode.IEEE: EUI64, - zigpy_t.AddrMode.Group: GroupId, + zigpy_t.AddrMode.Group: NWK, zigpy_t.AddrMode.Broadcast: NWK, } @@ -314,11 +131,9 @@ class NWKList(LVList): class DeconzAddress(Struct): - _fields = [ - # The address format (AddressMode) - ("address_mode", AddressMode), - ("address", EUI64), - ] + address_mode: AddressMode + address: EUI64 + ieee: EUI64 @classmethod def deserialize(cls, data): @@ -334,7 +149,7 @@ def deserialize(cls, data): return r, data def serialize(self): - r = super().serialize() + r = self.address_mode.serialize() + self.address.serialize() if self.address_mode == AddressMode.NWK_AND_IEEE: r += self.ieee.serialize() return r @@ -364,12 +179,10 @@ def from_zigpy_type(cls, addr): class DeconzAddressEndpoint(Struct): - _fields = [ - # The address format (AddressMode) - ("address_mode", AddressMode), - ("address", EUI64), - ("endpoint", uint8_t), - ] + address_mode: AddressMode + address: EUI64 + ieee: EUI64 + endpoint: uint8_t @classmethod def deserialize(cls, data): @@ -392,7 +205,7 @@ def serialize(self): if self.address_mode in (AddressMode.NWK, AddressMode.NWK_AND_IEEE): r += NWK(self.address).serialize() elif self.address_mode == AddressMode.GROUP: - r += GroupId(self.address).serialize() + r += NWK(self.address).serialize() if self.address_mode in (AddressMode.IEEE, AddressMode.NWK_AND_IEEE): r += EUI64(self.address).serialize() @@ -418,11 +231,6 @@ def from_zigpy_type(cls, addr, endpoint): return instance -class Key(FixedList): - _itemtype = uint8_t - _length = 16 - - class DataIndicationFlags(bitmap8): Always_Use_NWK_Source_Addr = 0b00000001 Last_Hop_In_Reserved_Bytes = 0b00000010 diff --git a/zigpy_deconz/uart.py b/zigpy_deconz/uart.py index 2bb25c9..b243de5 100644 --- a/zigpy_deconz/uart.py +++ b/zigpy_deconz/uart.py @@ -47,7 +47,7 @@ def close(self): def send(self, data): """Send data, taking care of escaping and framing.""" - LOGGER.debug("Send: 0x%s", binascii.hexlify(data).decode()) + LOGGER.debug("Send: %s", binascii.hexlify(data).decode()) checksum = bytes(self._checksum(data)) frame = self._escape(data + checksum) self._transport.write(self.END + frame + self.END) diff --git a/zigpy_deconz/utils.py b/zigpy_deconz/utils.py new file mode 100644 index 0000000..fa3ab2e --- /dev/null +++ b/zigpy_deconz/utils.py @@ -0,0 +1,25 @@ +"""deCONZ serial protocol API.""" + +from __future__ import annotations + +import asyncio +import functools +import logging + +LOGGER = logging.getLogger(__name__) + + +def restart_forever(func, *, restart_delay=1.0): + @functools.wraps(func) + async def replacement(*args, **kwargs): + while True: + try: + await func(*args, **kwargs) + except Exception: + LOGGER.debug( + "Endless task %s failed, restarting...", func, exc_info=True + ) + + await asyncio.sleep(restart_delay) + + return replacement diff --git a/zigpy_deconz/zigbee/application.py b/zigpy_deconz/zigbee/application.py index 077731d..ff77633 100644 --- a/zigpy_deconz/zigbee/application.py +++ b/zigpy_deconz/zigbee/application.py @@ -29,6 +29,11 @@ from zigpy_deconz import types as t from zigpy_deconz.api import ( Deconz, + FirmwarePlatform, + FirmwareVersion, + IndexedEndpoint, + IndexedKey, + LinkKey, NetworkParameter, NetworkState, SecurityMode, @@ -61,8 +66,8 @@ def __init__(self, config: dict[str, Any]): self._pending = zigpy.util.Requests() - self.version = 0 self._reset_watchdog_task = None + self._delayed_neighbor_scan_task = None self._reconnect_task = None self._written_endpoints = set() @@ -85,7 +90,6 @@ async def connect(self): try: await api.connect() - self.version = await api.version() except Exception: api.close() raise @@ -94,6 +98,10 @@ async def connect(self): self._written_endpoints.clear() def close(self): + if self._delayed_neighbor_scan_task is not None: + self._delayed_neighbor_scan_task.cancel() + self._delayed_neighbor_scan_task = None + if self._reset_watchdog_task is not None: self._reset_watchdog_task.cancel() self._reset_watchdog_task = None @@ -124,22 +132,26 @@ async def start_network(self): self, self.state.node_info.ieee, self.state.node_info.nwk, - self.version, + self._api.firmware_version, self._config[zigpy.config.CONF_DEVICE][zigpy.config.CONF_DEVICE_PATH], ) self.devices[self.state.node_info.ieee] = coordinator if self._api.protocol_version >= PROTO_VER_NEIGBOURS: await self.restore_neighbours() - asyncio.create_task(self._delayed_neighbour_scan()) + + self._delayed_neighbor_scan_task = asyncio.create_task( + self._delayed_neighbour_scan() + ) async def _change_network_state( self, target_state: NetworkState, *, timeout: int = 10 * CHANGE_NETWORK_WAIT ): async def change_loop(): while True: - (state, _, _) = await self._api.device_state() - if state.network_state == target_state: + device_state = await self._api.get_device_state() + + if NetworkState(device_state.network_state) == target_state: break await asyncio.sleep(CHANGE_NETWORK_WAIT) @@ -203,7 +215,7 @@ async def write_network_info(self, *, network_info, node_info): ) node_ieee = node_info.ieee else: - (ieee,) = await self._api[NetworkParameter.mac_address] + ieee = await self._api.read_parameter(NetworkParameter.mac_address) node_ieee = zigpy.types.EUI64(ieee) # There is no way to specify both a mask and the logical channel @@ -232,7 +244,8 @@ async def write_network_info(self, *, network_info, node_info): ) await self._api.write_parameter( - NetworkParameter.network_key, 0, network_info.network_key.key + NetworkParameter.network_key, + IndexedKey(index=0, key=network_info.network_key.key), ) if network_info.network_key.seq != 0: @@ -252,8 +265,10 @@ async def write_network_info(self, *, network_info, node_info): ) await self._api.write_parameter( NetworkParameter.link_key, - tc_link_key_partner_ieee, - network_info.tc_link_key.key, + LinkKey( + ieee=tc_link_key_partner_ieee, + key=network_info.tc_link_key.key, + ), ) if network_info.security_level == 0x00: @@ -279,64 +294,72 @@ async def load_network_info(self, *, load_devices=False): ) network_info.metadata = { "deconz": { - "version": self.version, + "version": f"{int(self._api.firmware_version):#010x}", } } - (ieee,) = await self._api[NetworkParameter.mac_address] + ieee = await self._api.read_parameter(NetworkParameter.mac_address) node_info.ieee = zigpy.types.EUI64(ieee) - (designed_coord,) = await self._api[NetworkParameter.aps_designed_coordinator] + designed_coord = await self._api.read_parameter( + NetworkParameter.aps_designed_coordinator + ) if designed_coord == 0x01: node_info.logical_type = zdo_t.LogicalType.Coordinator else: node_info.logical_type = zdo_t.LogicalType.Router - (node_info.nwk,) = await self._api[NetworkParameter.nwk_address] + node_info.nwk = await self._api.read_parameter(NetworkParameter.nwk_address) - (network_info.pan_id,) = await self._api[NetworkParameter.nwk_panid] - (network_info.extended_pan_id,) = await self._api[ + network_info.pan_id = await self._api.read_parameter(NetworkParameter.nwk_panid) + network_info.extended_pan_id = await self._api.read_parameter( NetworkParameter.aps_extended_panid - ] + ) if network_info.extended_pan_id == zigpy.types.EUI64.convert( "00:00:00:00:00:00:00:00" ): - (network_info.extended_pan_id,) = await self._api[ + network_info.extended_pan_id = await self._api.read_parameter( NetworkParameter.nwk_extended_panid - ] + ) - (network_info.channel,) = await self._api[NetworkParameter.current_channel] - (network_info.channel_mask,) = await self._api[NetworkParameter.channel_mask] - (network_info.nwk_update_id,) = await self._api[NetworkParameter.nwk_update_id] + network_info.channel = await self._api.read_parameter( + NetworkParameter.current_channel + ) + network_info.channel_mask = await self._api.read_parameter( + NetworkParameter.channel_mask + ) + network_info.nwk_update_id = await self._api.read_parameter( + NetworkParameter.nwk_update_id + ) if network_info.channel == 0: raise NetworkNotFormed("Network channel is zero") + indexed_key = await self._api.read_parameter(NetworkParameter.network_key, 0) + network_info.network_key = zigpy.state.Key() - ( - _, - network_info.network_key.key, - ) = await self._api.read_parameter(NetworkParameter.network_key, 0) + network_info.network_key.key = indexed_key.key try: - (network_info.network_key.tx_counter,) = await self._api[ + network_info.network_key.tx_counter = await self._api.read_parameter( NetworkParameter.nwk_frame_counter - ] + ) except zigpy_deconz.exception.CommandError as ex: assert ex.status == Status.UNSUPPORTED network_info.tc_link_key = zigpy.state.Key() - (network_info.tc_link_key.partner_ieee,) = await self._api[ + network_info.tc_link_key.partner_ieee = await self._api.read_parameter( NetworkParameter.trust_center_address - ] + ) - (_, network_info.tc_link_key.key) = await self._api.read_parameter( + link_key = await self._api.read_parameter( NetworkParameter.link_key, network_info.tc_link_key.partner_ieee, ) + network_info.tc_link_key.key = link_key.key - (security_mode,) = await self._api[NetworkParameter.security_mode] + security_mode = await self._api.read_parameter(NetworkParameter.security_mode) if security_mode == SecurityMode.NO_SECURITY: network_info.security_level = 0x00 @@ -380,14 +403,14 @@ async def add_endpoint(self, descriptor: zdo_t.SimpleDescriptor) -> None: # Read and count the current endpoints. Some firmwares have three, others four. for index in range(255 + 1): try: - _, current_descriptor = await self._api.read_parameter( + current_descriptor = await self._api.read_parameter( NetworkParameter.configure_endpoint, index ) except zigpy_deconz.exception.CommandError as ex: assert ex.status == Status.UNSUPPORTED break else: - endpoints[index] = current_descriptor + endpoints[index] = current_descriptor.descriptor LOGGER.debug("Got endpoint slots: %r", endpoints) @@ -419,7 +442,8 @@ async def add_endpoint(self, descriptor: zdo_t.SimpleDescriptor) -> None: LOGGER.debug("Writing %s to slot %r", descriptor, target_index) await self._api.write_parameter( - NetworkParameter.configure_endpoint, target_index, descriptor + NetworkParameter.configure_endpoint, + IndexedEndpoint(index=target_index, descriptor=descriptor), ) async def send_packet(self, packet): @@ -465,24 +489,6 @@ async def send_packet(self, packet): f"Failed to deliver packet: {status!r}", status ) - def handle_rx( - self, src, src_ep, dst, dst_ep, profile_id, cluster_id, data, lqi, rssi - ): - self.packet_received( - zigpy.types.ZigbeePacket( - src=src.as_zigpy_type(), - src_ep=src_ep, - dst=dst.as_zigpy_type(), - dst_ep=dst_ep, - tsn=None, - profile_id=profile_id, - cluster_id=cluster_id, - data=zigpy.types.SerializableBytes(data), - lqi=lqi, - rssi=rssi, - ) - ) - async def permit_ncp(self, time_s=60): assert 0 <= time_s <= 254 await self._api.write_parameter(NetworkParameter.permit_join, time_s) @@ -533,7 +539,9 @@ async def restore_neighbours(self) -> None: device.nwk, ) await self._api.add_neighbour( - 0x01, device.nwk, device.ieee, descr.mac_capability_flags + nwk=device.nwk, + ieee=device.ieee, + mac_capability_flags=descr.mac_capability_flags, ) async def _delayed_neighbour_scan(self) -> None: @@ -587,13 +595,13 @@ async def _reconnect_loop(self) -> None: class DeconzDevice(zigpy.device.Device): """Zigpy Device representing Coordinator.""" - def __init__(self, version: int, device_path: str, *args): + def __init__(self, version: FirmwareVersion, device_path: str, *args): """Initialize instance.""" super().__init__(*args) is_gpio_device = re.match(r"/dev/tty(S|AMA|ACM)\d+", device_path) self._model = "RaspBee" if is_gpio_device else "ConBee" - self._model += " II" if ((version & 0x0000FF00) == 0x00000700) else "" + self._model += " II" if version.platform == FirmwarePlatform.Conbee_II else "" async def add_to_group(self, grp_id: int, name: str = None) -> None: group = self.application.groups.add_group(grp_id, name)