From 7833647083f27f55b7ad345f4aaa7dffaa369abc Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Sun, 14 Apr 2024 17:18:56 -0400 Subject: [PATCH] Fix Python 3.12.3 regression with enums (#616) * Migrate bellows to directly use zigpy types * Convert `data` to bytes when dumping packets * Fix LVBytes32 prefix type * Fix failing EZSP protocol tests * Fix broken data types * Add undocumented `radius` field * Revert "Add undocumented `radius` field" This reverts commit 95e83bb62b027ff44ba816e9bbe196682c382b5f. * Remove unnecessary "bogus" value from unit tests --- bellows/cli/dump.py | 2 +- bellows/cli/network.py | 2 +- bellows/cli/util.py | 2 +- bellows/ezsp/v10/commands.py | 22 +- bellows/ezsp/v10/types/named.py | 4 +- bellows/ezsp/v10/types/struct.py | 6 +- bellows/ezsp/v13/commands.py | 1 - bellows/ezsp/v4/commands.py | 22 +- bellows/ezsp/v4/types/struct.py | 10 +- bellows/ezsp/v5/commands.py | 22 +- bellows/ezsp/v5/types/named.py | 4 +- bellows/ezsp/v5/types/struct.py | 10 +- bellows/ezsp/v6/commands.py | 22 +- bellows/ezsp/v6/types/named.py | 4 +- bellows/ezsp/v6/types/struct.py | 6 +- bellows/ezsp/v7/commands.py | 22 +- bellows/ezsp/v7/types/named.py | 4 +- bellows/ezsp/v7/types/struct.py | 6 +- bellows/ezsp/v8/commands.py | 22 +- bellows/ezsp/v8/types/named.py | 4 +- bellows/ezsp/v8/types/struct.py | 6 +- bellows/ezsp/v9/commands.py | 22 +- bellows/ezsp/v9/types/named.py | 4 +- bellows/ezsp/v9/types/struct.py | 6 +- bellows/types/basic.py | 266 +++--------------------- bellows/types/named.py | 34 +-- bellows/types/struct.py | 6 +- bellows/zigbee/application.py | 4 +- tests/test_application_network_state.py | 2 - tests/test_ezsp.py | 2 +- tests/test_ezsp_protocol.py | 4 +- tests/test_types.py | 204 ------------------ 32 files changed, 170 insertions(+), 587 deletions(-) diff --git a/bellows/cli/dump.py b/bellows/cli/dump.py index 65c6d67f..e0d4acbb 100644 --- a/bellows/cli/dump.py +++ b/bellows/cli/dump.py @@ -89,7 +89,7 @@ def cb(frame_name, response): hdr = pure_pcapy.Pkthdr(ts_sec, ts_usec, len(data), len(data)) try: - pcap.dump(hdr, data) + pcap.dump(hdr, bytes(data)) except BrokenPipeError: done_event.set() diff --git a/bellows/cli/network.py b/bellows/cli/network.py index f44762a4..51266c74 100644 --- a/bellows/cli/network.py +++ b/bellows/cli/network.py @@ -72,7 +72,7 @@ def cb(fut, frame_name, response): if isinstance(extended_pan_id, str): extended_pan_id = util.parse_epan(extended_pan_id) if extended_pan_id is None: - extended_pan_id = t.fixed_list(8, t.uint8_t)([t.uint8_t(0)] * 8) + extended_pan_id = t.FixedList[t.uint8_t, 8]([t.uint8_t(0)] * 8) v = await util.network_init(s) diff --git a/bellows/cli/util.py b/bellows/cli/util.py index bcb9e0b7..4f2e97cf 100644 --- a/bellows/cli/util.py +++ b/bellows/cli/util.py @@ -147,7 +147,7 @@ async def network_init(s): def parse_epan(epan): """Parse a user specified extended PAN ID""" epan_list = [t.uint8_t(x, 16) for x in epan.split(":")] - return t.fixed_list(8, t.uint8_t)(epan_list) + return t.FixedList[t.uint8_t, 8](epan_list) async def basic_tc_permits(s): diff --git a/bellows/ezsp/v10/commands.py b/bellows/ezsp/v10/commands.py index 0fd70340..b70c5292 100644 --- a/bellows/ezsp/v10/commands.py +++ b/bellows/ezsp/v10/commands.py @@ -14,8 +14,8 @@ t.uint8_t, t.uint8_t, t.uint8_t, - t.List(t.uint16_t), - t.List(t.uint16_t), + t.List[t.uint16_t], + t.List[t.uint16_t], ), (t.EzspStatus,), ), @@ -35,8 +35,8 @@ "invalidCommand": (0x0058, (), (t.EzspStatus,)), "callback": (0x0006, (), ()), "noCallbacks": (0x0007, (), ()), - "setToken": (0x0009, (t.uint8_t, t.fixed_list(8, t.uint8_t)), (t.EmberStatus,)), - "getToken": (0x000A, (t.uint8_t,), (t.EmberStatus, t.fixed_list(8, t.uint8_t))), + "setToken": (0x0009, (t.uint8_t, t.FixedList[t.uint8_t, 8]), (t.EmberStatus,)), + "getToken": (0x000A, (t.uint8_t,), (t.EmberStatus, t.FixedList[t.uint8_t, 8])), "getMfgToken": (0x000B, (t.EzspMfgTokenId,), (t.LVBytes,)), "setMfgToken": (0x000C, (t.EzspMfgTokenId, t.LVBytes), (t.EmberStatus,)), "stackTokenChangedHandler": (0x000D, (), (t.uint16_t,)), @@ -52,12 +52,12 @@ "readAndClearCounters": ( 0x0065, (), - (t.fixed_list(len(t.EmberCounterType), t.uint16_t),), + (t.FixedList[t.uint16_t, len(t.EmberCounterType)],), ), "readCounters": ( 0x00F1, (), - (t.fixed_list(len(t.EmberCounterType), t.uint16_t),), + (t.FixedList[t.uint16_t, len(t.EmberCounterType)],), ), "counterRolloverHandler": (0x00F2, (), (t.EmberCounterType,)), "delayTest": (0x009D, (t.uint16_t,), ()), @@ -181,7 +181,7 @@ "getCurrentDutyCycle": ( 0x004C, (t.uint8_t,), - (t.EmberStatus, t.fixed_list(134, t.uint8_t)), + (t.EmberStatus, t.FixedList[t.uint8_t, 134]), ), "dutyCycleHandler": ( 0x004D, @@ -306,7 +306,7 @@ "incomingRouteRecordHandler": ( 0x0059, (), - (t.EmberNodeId, t.EUI64, t.uint8_t, t.int8s, t.LVList(t.EmberNodeId)), + (t.EmberNodeId, t.EUI64, t.uint8_t, t.int8s, t.LVList[t.EmberNodeId]), ), "incomingNetworkStatusHandler": ( 0x00C4, @@ -315,7 +315,7 @@ ), "setSourceRoute": ( 0x00AE, - (t.EmberNodeId, t.LVList(t.EmberNodeId)), + (t.EmberNodeId, t.LVList[t.EmberNodeId]), (t.EmberStatus,), ), "setSourceRouteDiscoveryMode": (0x005A, (t.uint8_t,), (t.uint32_t,)), @@ -550,8 +550,8 @@ "bootloadTransmitCompleteHandler": (0x0093, (), (t.EmberStatus, t.LVBytes)), "aesEncrypt": ( 0x0094, - (t.fixed_list(16, t.uint8_t), t.fixed_list(16, t.uint8_t)), - (t.fixed_list(16, t.uint8_t),), + (t.FixedList[t.uint8_t, 16], t.FixedList[t.uint8_t, 16]), + (t.FixedList[t.uint8_t, 16],), ), # 14. ZLL Frames "zllNetworkOps": ( diff --git a/bellows/ezsp/v10/types/named.py b/bellows/ezsp/v10/types/named.py index 72e735de..e0ffe429 100644 --- a/bellows/ezsp/v10/types/named.py +++ b/bellows/ezsp/v10/types/named.py @@ -702,7 +702,7 @@ class SecureEzspSecurityLevel(basic.uint8_t): """Security level of the Secure EZSP Protocol.""" -class SecureEzspRandomNumber(basic.fixed_list(16, basic.uint8_t)): +class SecureEzspRandomNumber(basic.FixedList[basic.uint8_t, 16]): """Randomly generated 64-bit number. Both NCP and Host contribute this number to create the Session ID, @@ -710,7 +710,7 @@ class SecureEzspRandomNumber(basic.fixed_list(16, basic.uint8_t)): """ -class SecureEzspSessionId(basic.fixed_list(8, basic.uint8_t)): +class SecureEzspSessionId(basic.FixedList[basic.uint8_t, 8]): """Generated 64-bit Session ID, using random numbers from Host and NCP. It is generated at each reboot (during negotiation phase). Having both sides diff --git a/bellows/ezsp/v10/types/struct.py b/bellows/ezsp/v10/types/struct.py index 65329a12..cfdd9204 100644 --- a/bellows/ezsp/v10/types/struct.py +++ b/bellows/ezsp/v10/types/struct.py @@ -140,7 +140,7 @@ class EmberGpProxyTableEntry(EzspStruct): # The security frame counter of the GPD. gpdSecurityFrameCounter: named.EmberGpSecurityFrameCounter # The list of sinks (hardcoded to 2 which is the spec minimum). - sinkList: basic.fixed_list(2, EmberGpSinkListEntry) + sinkList: basic.FixedList[EmberGpSinkListEntry, 2] # The groupcast radius. groupcastRadius: basic.uint8_t # The search counter @@ -160,7 +160,7 @@ class EmberGpSinkTableEntry(EzspStruct): # The device id for the GPD. deviceId: basic.uint8_t # The list of sinks (hardcoded to 2 which is the spec minimum). - sinkList: basic.fixed_list(2, EmberGpSinkListEntry) + sinkList: basic.FixedList[EmberGpSinkListEntry, 2] # The assigned alias for the GPD. assignedAlias: named.EmberNodeId # The groupcast radius. @@ -185,7 +185,7 @@ class EmberDutyCycleLimits(EzspStruct): # The vendor identifier field shall contain the vendor identifier of the node. vendorId: basic.uint16_t # The vendor string field shall contain the vendor string of the node. - vendorString: basic.fixed_list(7, basic.uint8_t) + vendorString: basic.FixedList[basic.uint8_t, 7] class EmberPerDeviceDutyCycle(EzspStruct): diff --git a/bellows/ezsp/v13/commands.py b/bellows/ezsp/v13/commands.py index 8e9d8559..47168b14 100644 --- a/bellows/ezsp/v13/commands.py +++ b/bellows/ezsp/v13/commands.py @@ -9,7 +9,6 @@ tuple( { "status": t.sl_Status, - "bogus": t.uint16_t, "network_key_info": t.sl_zb_sec_man_network_key_info_t, }.values() ), diff --git a/bellows/ezsp/v4/commands.py b/bellows/ezsp/v4/commands.py index c5086f56..eb93e9b7 100644 --- a/bellows/ezsp/v4/commands.py +++ b/bellows/ezsp/v4/commands.py @@ -14,8 +14,8 @@ t.uint8_t, t.uint8_t, t.uint8_t, - t.List(t.uint16_t), - t.List(t.uint16_t), + t.List[t.uint16_t], + t.List[t.uint16_t], ), (t.EzspStatus,), ), @@ -47,8 +47,8 @@ "invalidCommand": (0x58, (), (t.EzspStatus,)), "callback": (0x06, (), ()), "noCallbacks": (0x07, (), ()), - "setToken": (0x09, (t.uint8_t, t.fixed_list(8, t.uint8_t)), (t.EmberStatus,)), - "getToken": (0x0A, (t.uint8_t,), (t.EmberStatus, t.fixed_list(8, t.uint8_t))), + "setToken": (0x09, (t.uint8_t, t.FixedList[t.uint8_t, 8]), (t.EmberStatus,)), + "getToken": (0x0A, (t.uint8_t,), (t.EmberStatus, t.FixedList[t.uint8_t, 8])), "getMfgToken": (0x0B, (t.EzspMfgTokenId,), (t.LVBytes,)), "setMfgToken": (0x0C, (t.EzspMfgTokenId, t.LVBytes), (t.EmberStatus,)), "stackTokenChangedHandler": (0x0D, (), (t.uint16_t,)), @@ -64,12 +64,12 @@ "readAndClearCounters": ( 0x65, (), - (t.fixed_list(len(t.EmberCounterType), t.uint16_t),), + (t.FixedList[t.uint16_t, len(t.EmberCounterType)],), ), "readCounters": ( 0xF1, (), - (t.fixed_list(len(t.EmberCounterType), t.uint16_t),), + (t.FixedList[t.uint16_t, len(t.EmberCounterType)],), ), "counterRolloverHandler": (0xF2, (), (t.EmberCounterType,)), "delayTest": (0x9D, (t.uint16_t,), ()), @@ -220,11 +220,11 @@ "incomingRouteRecordHandler": ( 0x59, (), - (t.EmberNodeId, t.EUI64, t.uint8_t, t.int8s, t.LVList(t.EmberNodeId)), + (t.EmberNodeId, t.EUI64, t.uint8_t, t.int8s, t.LVList[t.EmberNodeId]), ), "setSourceRoute": ( 0x5A, - (t.EmberNodeId, t.LVList(t.EmberNodeId)), + (t.EmberNodeId, t.LVList[t.EmberNodeId]), (t.EmberStatus,), ), "incomingManyToOneRouteRequestHandler": ( @@ -412,8 +412,8 @@ "bootloadTransmitCompleteHandler": (0x93, (), (t.EmberStatus, t.LVBytes)), "aesEncrypt": ( 0x94, - (t.fixed_list(16, t.uint8_t), t.fixed_list(16, t.uint8_t)), - (t.fixed_list(16, t.uint8_t),), + (t.FixedList[t.uint8_t, 16], t.FixedList[t.uint8_t, 16]), + (t.FixedList[t.uint8_t, 16],), ), "overrideCurrentChannel": (0x95, (t.uint8_t,), (t.EmberStatus,)), # 14. ZLL Frames @@ -501,7 +501,7 @@ t.uint8_t, t.uint16_t, t.uint8_t, - t.LVList(t.uint8_t), + t.LVList[t.uint8_t], ), (t.EmberStatus,), ), diff --git a/bellows/ezsp/v4/types/struct.py b/bellows/ezsp/v4/types/struct.py index e9b04ccf..3a343e80 100644 --- a/bellows/ezsp/v4/types/struct.py +++ b/bellows/ezsp/v4/types/struct.py @@ -52,7 +52,7 @@ class EmberRf4ceVendorInfo(EzspStruct): # the node. vendorId: basic.uint16_t # The vendor string field shall contain the vendor string of the node. - vendorString: basic.fixed_list(7, basic.uint8_t) + vendorString: basic.FixedList[basic.uint8_t, 7] class EmberRf4ceApplicationInfo(EzspStruct): @@ -62,13 +62,13 @@ class EmberRf4ceApplicationInfo(EzspStruct): capabilities: named.EmberRf4ceApplicationCapabilities # The user string field shall contain the user specified identification # string. - userString: basic.fixed_list(15, basic.uint8_t) + userString: basic.FixedList[basic.uint8_t, 15] # The device type list field shall contain the list of device types # supported by the node. - deviceTypeList: basic.fixed_list(3, basic.uint8_t) + deviceTypeList: basic.FixedList[basic.uint8_t, 3] # The profile ID list field shall contain the list of profile # identifiers disclosed as supported by the node. - profileIdList: basic.fixed_list(7, basic.uint8_t) + profileIdList: basic.FixedList[basic.uint8_t, 7] class EmberRf4cePairingTableEntry(EzspStruct): @@ -88,7 +88,7 @@ class EmberRf4cePairingTableEntry(EzspStruct): # The vendor ID of the destination device. destVendorId: basic.uint16_t # The list of profiles supported by the destination device. - destProfileIdList: basic.fixed_list(7, basic.uint8_t) + destProfileIdList: basic.FixedList[basic.uint8_t, 7] # The length of the list of supported profiles. destProfileIdListLength: basic.uint8_t # Info byte. diff --git a/bellows/ezsp/v5/commands.py b/bellows/ezsp/v5/commands.py index 211cf5ac..87f3190d 100644 --- a/bellows/ezsp/v5/commands.py +++ b/bellows/ezsp/v5/commands.py @@ -14,8 +14,8 @@ t.uint8_t, t.uint8_t, t.uint8_t, - t.List(t.uint16_t), - t.List(t.uint16_t), + t.List[t.uint16_t], + t.List[t.uint16_t], ), (t.EzspStatus,), ), @@ -48,8 +48,8 @@ "invalidCommand": (0x58, (), (t.EzspStatus,)), "callback": (0x06, (), ()), "noCallbacks": (0x07, (), ()), - "setToken": (0x09, (t.uint8_t, t.fixed_list(8, t.uint8_t)), (t.EmberStatus,)), - "getToken": (0x0A, (t.uint8_t,), (t.EmberStatus, t.fixed_list(8, t.uint8_t))), + "setToken": (0x09, (t.uint8_t, t.FixedList[t.uint8_t, 8]), (t.EmberStatus,)), + "getToken": (0x0A, (t.uint8_t,), (t.EmberStatus, t.FixedList[t.uint8_t, 8])), "getMfgToken": (0x0B, (t.EzspMfgTokenId,), (t.LVBytes,)), "setMfgToken": (0x0C, (t.EzspMfgTokenId, t.LVBytes), (t.EmberStatus,)), "stackTokenChangedHandler": (0x0D, (), (t.uint16_t,)), @@ -65,12 +65,12 @@ "readAndClearCounters": ( 0x65, (), - (t.fixed_list(len(t.EmberCounterType), t.uint16_t),), + (t.FixedList[t.uint16_t, len(t.EmberCounterType)],), ), "readCounters": ( 0xF1, (), - (t.fixed_list(len(t.EmberCounterType), t.uint16_t),), + (t.FixedList[t.uint16_t, len(t.EmberCounterType)],), ), "counterRolloverHandler": (0xF2, (), (t.EmberCounterType,)), "delayTest": (0x9D, (t.uint16_t,), ()), @@ -221,12 +221,12 @@ "incomingRouteRecordHandler": ( 0x59, (), - (t.EmberNodeId, t.EUI64, t.uint8_t, t.int8s, t.LVList(t.EmberNodeId)), + (t.EmberNodeId, t.EUI64, t.uint8_t, t.int8s, t.LVList[t.EmberNodeId]), ), "changeSourceRouteHandler": (0xC4, (), (t.EmberNodeId, t.EmberNodeId, t.Bool)), "setSourceRoute": ( 0x5A, - (t.EmberNodeId, t.LVList(t.EmberNodeId)), + (t.EmberNodeId, t.LVList[t.EmberNodeId]), (t.EmberStatus,), ), "incomingManyToOneRouteRequestHandler": ( @@ -419,8 +419,8 @@ "bootloadTransmitCompleteHandler": (0x93, (), (t.EmberStatus, t.LVBytes)), "aesEncrypt": ( 0x94, - (t.fixed_list(16, t.uint8_t), t.fixed_list(16, t.uint8_t)), - (t.fixed_list(16, t.uint8_t),), + (t.FixedList[t.uint8_t, 16], t.FixedList[t.uint8_t, 16]), + (t.FixedList[t.uint8_t, 16],), ), "overrideCurrentChannel": (0x95, (t.uint8_t,), (t.EmberStatus,)), # 14. ZLL Frames @@ -508,7 +508,7 @@ t.uint8_t, t.uint16_t, t.uint8_t, - t.LVList(t.uint8_t), + t.LVList[t.uint8_t], ), (t.EmberStatus,), ), diff --git a/bellows/ezsp/v5/types/named.py b/bellows/ezsp/v5/types/named.py index aa63a9a1..0e22e958 100644 --- a/bellows/ezsp/v5/types/named.py +++ b/bellows/ezsp/v5/types/named.py @@ -624,7 +624,7 @@ class SecureEzspSecurityLevel(basic.uint8_t): """Security level of the Secure EZSP Protocol.""" -class SecureEzspRandomNumber(basic.fixed_list(16, basic.uint8_t)): +class SecureEzspRandomNumber(basic.FixedList[basic.uint8_t, 16]): """Randomly generated 64-bit number. Both NCP and Host contribute this number to create the Session ID, @@ -632,7 +632,7 @@ class SecureEzspRandomNumber(basic.fixed_list(16, basic.uint8_t)): """ -class SecureEzspSessionId(basic.fixed_list(8, basic.uint8_t)): +class SecureEzspSessionId(basic.FixedList[basic.uint8_t, 8]): """Generated 64-bit Session ID, using random numbers from Host and NCP. It is generated at each reboot (during negotiation phase). Having both sides diff --git a/bellows/ezsp/v5/types/struct.py b/bellows/ezsp/v5/types/struct.py index 8e2147e2..e9094faa 100644 --- a/bellows/ezsp/v5/types/struct.py +++ b/bellows/ezsp/v5/types/struct.py @@ -52,7 +52,7 @@ class EmberRf4ceVendorInfo(EzspStruct): # the node. vendorId: basic.uint16_t # The vendor string field shall contain the vendor string of the node. - vendorString: basic.fixed_list(7, basic.uint8_t) + vendorString: basic.FixedList[basic.uint8_t, 7] class EmberRf4ceApplicationInfo(EzspStruct): @@ -62,13 +62,13 @@ class EmberRf4ceApplicationInfo(EzspStruct): capabilities: named.EmberRf4ceApplicationCapabilities # The user string field shall contain the user specified identification # string. - userString: basic.fixed_list(15, basic.uint8_t) + userString: basic.FixedList[basic.uint8_t, 15] # The device type list field shall contain the list of device types # supported by the node. - deviceTypeList: basic.fixed_list(3, basic.uint8_t) + deviceTypeList: basic.FixedList[basic.uint8_t, 3] # The profile ID list field shall contain the list of profile # identifiers disclosed as supported by the node. - profileIdList: basic.fixed_list(7, basic.uint8_t) + profileIdList: basic.FixedList[basic.uint8_t, 7] class EmberRf4cePairingTableEntry(EzspStruct): @@ -88,7 +88,7 @@ class EmberRf4cePairingTableEntry(EzspStruct): # The vendor ID of the destination device. destVendorId: basic.uint16_t # The list of profiles supported by the destination device. - destProfileIdList: basic.fixed_list(7, basic.uint8_t) + destProfileIdList: basic.FixedList[basic.uint8_t, 7] # The length of the list of supported profiles. destProfileIdListLength: basic.uint8_t # Info byte. diff --git a/bellows/ezsp/v6/commands.py b/bellows/ezsp/v6/commands.py index 72c15135..39ab7254 100644 --- a/bellows/ezsp/v6/commands.py +++ b/bellows/ezsp/v6/commands.py @@ -14,8 +14,8 @@ t.uint8_t, t.uint8_t, t.uint8_t, - t.List(t.uint16_t), - t.List(t.uint16_t), + t.List[t.uint16_t], + t.List[t.uint16_t], ), (t.EzspStatus,), ), @@ -47,8 +47,8 @@ "invalidCommand": (0x58, (), (t.EzspStatus,)), "callback": (0x06, (), ()), "noCallbacks": (0x07, (), ()), - "setToken": (0x09, (t.uint8_t, t.fixed_list(8, t.uint8_t)), (t.EmberStatus,)), - "getToken": (0x0A, (t.uint8_t,), (t.EmberStatus, t.fixed_list(8, t.uint8_t))), + "setToken": (0x09, (t.uint8_t, t.FixedList[t.uint8_t, 8]), (t.EmberStatus,)), + "getToken": (0x0A, (t.uint8_t,), (t.EmberStatus, t.FixedList[t.uint8_t, 8])), "getMfgToken": (0x0B, (t.EzspMfgTokenId,), (t.LVBytes,)), "setMfgToken": (0x0C, (t.EzspMfgTokenId, t.LVBytes), (t.EmberStatus,)), "stackTokenChangedHandler": (0x0D, (), (t.uint16_t,)), @@ -64,12 +64,12 @@ "readAndClearCounters": ( 0x65, (), - (t.fixed_list(len(t.EmberCounterType), t.uint16_t),), + (t.FixedList[t.uint16_t, len(t.EmberCounterType)],), ), "readCounters": ( 0xF1, (), - (t.fixed_list(len(t.EmberCounterType), t.uint16_t),), + (t.FixedList[t.uint16_t, len(t.EmberCounterType)],), ), "counterRolloverHandler": (0xF2, (), (t.EmberCounterType,)), "delayTest": (0x9D, (t.uint16_t,), ()), @@ -175,7 +175,7 @@ "getCurrentDutyCycle": ( 0x4C, (t.uint8_t,), - (t.EmberStatus, t.fixed_list(134, t.uint8_t)), + (t.EmberStatus, t.FixedList[t.uint8_t, 134]), ), "dutyCycleHandler": ( 0x4D, @@ -286,12 +286,12 @@ "incomingRouteRecordHandler": ( 0x59, (), - (t.EmberNodeId, t.EUI64, t.uint8_t, t.int8s, t.LVList(t.EmberNodeId)), + (t.EmberNodeId, t.EUI64, t.uint8_t, t.int8s, t.LVList[t.EmberNodeId]), ), "changeSourceRouteHandler": (0xC4, (), (t.EmberNodeId, t.EmberNodeId, t.Bool)), "setSourceRoute": ( 0x5A, - (t.EmberNodeId, t.LVList(t.EmberNodeId)), + (t.EmberNodeId, t.LVList[t.EmberNodeId]), (t.EmberStatus,), ), "incomingManyToOneRouteRequestHandler": ( @@ -486,8 +486,8 @@ "bootloadTransmitCompleteHandler": (0x93, (), (t.EmberStatus, t.LVBytes)), "aesEncrypt": ( 0x94, - (t.fixed_list(16, t.uint8_t), t.fixed_list(16, t.uint8_t)), - (t.fixed_list(16, t.uint8_t),), + (t.FixedList[t.uint8_t, 16], t.FixedList[t.uint8_t, 16]), + (t.FixedList[t.uint8_t, 16],), ), "overrideCurrentChannel": (0x95, (t.uint8_t,), (t.EmberStatus,)), # 14. ZLL Frames diff --git a/bellows/ezsp/v6/types/named.py b/bellows/ezsp/v6/types/named.py index ac7223c1..8ee8faf4 100644 --- a/bellows/ezsp/v6/types/named.py +++ b/bellows/ezsp/v6/types/named.py @@ -614,7 +614,7 @@ class SecureEzspSecurityLevel(basic.uint8_t): """Security level of the Secure EZSP Protocol.""" -class SecureEzspRandomNumber(basic.fixed_list(16, basic.uint8_t)): +class SecureEzspRandomNumber(basic.FixedList[basic.uint8_t, 16]): """Randomly generated 64-bit number. Both NCP and Host contribute this number to create the Session ID, @@ -622,7 +622,7 @@ class SecureEzspRandomNumber(basic.fixed_list(16, basic.uint8_t)): """ -class SecureEzspSessionId(basic.fixed_list(8, basic.uint8_t)): +class SecureEzspSessionId(basic.FixedList[basic.uint8_t, 8]): """Generated 64-bit Session ID, using random numbers from Host and NCP. It is generated at each reboot (during negotiation phase). Having both sides diff --git a/bellows/ezsp/v6/types/struct.py b/bellows/ezsp/v6/types/struct.py index dcc2bf30..ce9bf78f 100644 --- a/bellows/ezsp/v6/types/struct.py +++ b/bellows/ezsp/v6/types/struct.py @@ -77,7 +77,7 @@ class EmberGpProxyTableEntry(EzspStruct): # The key to use for GPD. gpdKey: named.KeyData # The list of sinks (hardcoded to 2 which is the spec minimum). - sinkList: basic.fixed_list(2, EmberGpSinkListEntry) + sinkList: basic.FixedList[EmberGpSinkListEntry, 2] # The groupcast radius. groupcastRadius: basic.uint8_t # The search counter @@ -97,7 +97,7 @@ class EmberGpSinkTableEntry(EzspStruct): # The device id for the GPD. deviceId: basic.uint8_t # The list of sinks (hardcoded to 2 which is the spec minimum). - sinkList: basic.fixed_list(2, EmberGpSinkListEntry) + sinkList: basic.FixedList[EmberGpSinkListEntry, 2] # The assigned alias for the GPD. assignedAlias: named.EmberNodeId # The groupcast radius. @@ -122,7 +122,7 @@ class EmberDutyCycleLimits(EzspStruct): # The vendor identifier field shall contain the vendor identifier of the node. vendorId: basic.uint16_t # The vendor string field shall contain the vendor string of the node. - vendorString: basic.fixed_list(7, basic.uint8_t) + vendorString: basic.FixedList[basic.uint8_t, 7] class EmberPerDeviceDutyCycle(EzspStruct): diff --git a/bellows/ezsp/v7/commands.py b/bellows/ezsp/v7/commands.py index 98eaf15b..0fced8f7 100644 --- a/bellows/ezsp/v7/commands.py +++ b/bellows/ezsp/v7/commands.py @@ -14,8 +14,8 @@ t.uint8_t, t.uint8_t, t.uint8_t, - t.List(t.uint16_t), - t.List(t.uint16_t), + t.List[t.uint16_t], + t.List[t.uint16_t], ), (t.EzspStatus,), ), @@ -48,8 +48,8 @@ "invalidCommand": (0x58, (), (t.EzspStatus,)), "callback": (0x06, (), ()), "noCallbacks": (0x07, (), ()), - "setToken": (0x09, (t.uint8_t, t.fixed_list(8, t.uint8_t)), (t.EmberStatus,)), - "getToken": (0x0A, (t.uint8_t,), (t.EmberStatus, t.fixed_list(8, t.uint8_t))), + "setToken": (0x09, (t.uint8_t, t.FixedList[t.uint8_t, 8]), (t.EmberStatus,)), + "getToken": (0x0A, (t.uint8_t,), (t.EmberStatus, t.FixedList[t.uint8_t, 8])), "getMfgToken": (0x0B, (t.EzspMfgTokenId,), (t.LVBytes,)), "setMfgToken": (0x0C, (t.EzspMfgTokenId, t.LVBytes), (t.EmberStatus,)), "stackTokenChangedHandler": (0x0D, (), (t.uint16_t,)), @@ -65,12 +65,12 @@ "readAndClearCounters": ( 0x65, (), - (t.fixed_list(len(t.EmberCounterType), t.uint16_t),), + (t.FixedList[t.uint16_t, len(t.EmberCounterType)],), ), "readCounters": ( 0xF1, (), - (t.fixed_list(len(t.EmberCounterType), t.uint16_t),), + (t.FixedList[t.uint16_t, len(t.EmberCounterType)],), ), "counterRolloverHandler": (0xF2, (), (t.EmberCounterType,)), "delayTest": (0x9D, (t.uint16_t,), ()), @@ -182,7 +182,7 @@ "getCurrentDutyCycle": ( 0x4C, (t.uint8_t,), - (t.EmberStatus, t.fixed_list(134, t.uint8_t)), + (t.EmberStatus, t.FixedList[t.uint8_t, 134]), ), "dutyCycleHandler": ( 0x4D, @@ -297,12 +297,12 @@ "incomingRouteRecordHandler": ( 0x59, (), - (t.EmberNodeId, t.EUI64, t.uint8_t, t.int8s, t.LVList(t.EmberNodeId)), + (t.EmberNodeId, t.EUI64, t.uint8_t, t.int8s, t.LVList[t.EmberNodeId]), ), "changeSourceRouteHandler": (0xC4, (), (t.EmberNodeId, t.EmberNodeId, t.Bool)), "setSourceRoute": ( 0x5A, - (t.EmberNodeId, t.LVList(t.EmberNodeId)), + (t.EmberNodeId, t.LVList[t.EmberNodeId]), (t.EmberStatus,), ), "incomingManyToOneRouteRequestHandler": ( @@ -519,8 +519,8 @@ "bootloadTransmitCompleteHandler": (0x93, (), (t.EmberStatus, t.LVBytes)), "aesEncrypt": ( 0x94, - (t.fixed_list(16, t.uint8_t), t.fixed_list(16, t.uint8_t)), - (t.fixed_list(16, t.uint8_t),), + (t.FixedList[t.uint8_t, 16], t.FixedList[t.uint8_t, 16]), + (t.FixedList[t.uint8_t, 16],), ), "overrideCurrentChannel": (0x95, (t.uint8_t,), (t.EmberStatus,)), # 14. ZLL Frames diff --git a/bellows/ezsp/v7/types/named.py b/bellows/ezsp/v7/types/named.py index 0885316f..5f24bd6f 100644 --- a/bellows/ezsp/v7/types/named.py +++ b/bellows/ezsp/v7/types/named.py @@ -688,7 +688,7 @@ class SecureEzspSecurityLevel(basic.uint8_t): """Security level of the Secure EZSP Protocol.""" -class SecureEzspRandomNumber(basic.fixed_list(16, basic.uint8_t)): +class SecureEzspRandomNumber(basic.FixedList[basic.uint8_t, 16]): """Randomly generated 64-bit number. Both NCP and Host contribute this number to create the Session ID, @@ -696,7 +696,7 @@ class SecureEzspRandomNumber(basic.fixed_list(16, basic.uint8_t)): """ -class SecureEzspSessionId(basic.fixed_list(8, basic.uint8_t)): +class SecureEzspSessionId(basic.FixedList[basic.uint8_t, 8]): """Generated 64-bit Session ID, using random numbers from Host and NCP. It is generated at each reboot (during negotiation phase). Having both sides diff --git a/bellows/ezsp/v7/types/struct.py b/bellows/ezsp/v7/types/struct.py index a370dae3..177625e2 100644 --- a/bellows/ezsp/v7/types/struct.py +++ b/bellows/ezsp/v7/types/struct.py @@ -131,7 +131,7 @@ class EmberGpProxyTableEntry(EzspStruct): # The key to use for GPD. gpdKey: named.KeyData # The list of sinks (hardcoded to 2 which is the spec minimum). - sinkList: basic.fixed_list(2, EmberGpSinkListEntry) + sinkList: basic.FixedList[EmberGpSinkListEntry, 2] # The groupcast radius. groupcastRadius: basic.uint8_t # The search counter @@ -151,7 +151,7 @@ class EmberGpSinkTableEntry(EzspStruct): # The device id for the GPD. deviceId: basic.uint8_t # The list of sinks (hardcoded to 2 which is the spec minimum). - sinkList: basic.fixed_list(2, EmberGpSinkListEntry) + sinkList: basic.FixedList[EmberGpSinkListEntry, 2] # The assigned alias for the GPD. assignedAlias: named.EmberNodeId # The groupcast radius. @@ -176,7 +176,7 @@ class EmberDutyCycleLimits(EzspStruct): # The vendor identifier field shall contain the vendor identifier of the node. vendorId: basic.uint16_t # The vendor string field shall contain the vendor string of the node. - vendorString: basic.fixed_list(7, basic.uint8_t) + vendorString: basic.FixedList[basic.uint8_t, 7] class EmberPerDeviceDutyCycle(EzspStruct): diff --git a/bellows/ezsp/v8/commands.py b/bellows/ezsp/v8/commands.py index 35a4940b..a1c5a11c 100644 --- a/bellows/ezsp/v8/commands.py +++ b/bellows/ezsp/v8/commands.py @@ -14,8 +14,8 @@ t.uint8_t, t.uint8_t, t.uint8_t, - t.List(t.uint16_t), - t.List(t.uint16_t), + t.List[t.uint16_t], + t.List[t.uint16_t], ), (t.EzspStatus,), ), @@ -45,8 +45,8 @@ "invalidCommand": (0x0058, (), (t.EzspStatus,)), "callback": (0x0006, (), ()), "noCallbacks": (0x0007, (), ()), - "setToken": (0x0009, (t.uint8_t, t.fixed_list(8, t.uint8_t)), (t.EmberStatus,)), - "getToken": (0x000A, (t.uint8_t,), (t.EmberStatus, t.fixed_list(8, t.uint8_t))), + "setToken": (0x0009, (t.uint8_t, t.FixedList[t.uint8_t, 8]), (t.EmberStatus,)), + "getToken": (0x000A, (t.uint8_t,), (t.EmberStatus, t.FixedList[t.uint8_t, 8])), "getMfgToken": (0x000B, (t.EzspMfgTokenId,), (t.LVBytes,)), "setMfgToken": (0x000C, (t.EzspMfgTokenId, t.LVBytes), (t.EmberStatus,)), "stackTokenChangedHandler": (0x000D, (), (t.uint16_t,)), @@ -62,12 +62,12 @@ "readAndClearCounters": ( 0x0065, (), - (t.fixed_list(len(t.EmberCounterType), t.uint16_t),), + (t.FixedList[t.uint16_t, len(t.EmberCounterType)],), ), "readCounters": ( 0x00F1, (), - (t.fixed_list(len(t.EmberCounterType), t.uint16_t),), + (t.FixedList[t.uint16_t, len(t.EmberCounterType)],), ), "counterRolloverHandler": (0x00F2, (), (t.EmberCounterType,)), "delayTest": (0x009D, (t.uint16_t,), ()), @@ -184,7 +184,7 @@ "getCurrentDutyCycle": ( 0x004C, (t.uint8_t,), - (t.EmberStatus, t.fixed_list(134, t.uint8_t)), + (t.EmberStatus, t.FixedList[t.uint8_t, 134]), ), "dutyCycleHandler": ( 0x004D, @@ -309,12 +309,12 @@ "incomingRouteRecordHandler": ( 0x0059, (), - (t.EmberNodeId, t.EUI64, t.uint8_t, t.int8s, t.LVList(t.EmberNodeId)), + (t.EmberNodeId, t.EUI64, t.uint8_t, t.int8s, t.LVList[t.EmberNodeId]), ), "changeSourceRouteHandler": (0x00C4, (), (t.EmberNodeId, t.EmberNodeId, t.Bool)), "setSourceRoute": ( 0x00AE, - (t.EmberNodeId, t.LVList(t.EmberNodeId)), + (t.EmberNodeId, t.LVList[t.EmberNodeId]), (t.EmberStatus,), ), "setSourceRouteDiscoveryMode": (0x005A, (t.uint8_t,), (t.uint32_t,)), @@ -549,8 +549,8 @@ "bootloadTransmitCompleteHandler": (0x0093, (), (t.EmberStatus, t.LVBytes)), "aesEncrypt": ( 0x0094, - (t.fixed_list(16, t.uint8_t), t.fixed_list(16, t.uint8_t)), - (t.fixed_list(16, t.uint8_t),), + (t.FixedList[t.uint8_t, 16], t.FixedList[t.uint8_t, 16]), + (t.FixedList[t.uint8_t, 16],), ), "overrideCurrentChannel": (0x0095, (t.uint8_t,), (t.EmberStatus,)), # 14. ZLL Frames diff --git a/bellows/ezsp/v8/types/named.py b/bellows/ezsp/v8/types/named.py index b8a4e61c..d44d06d4 100644 --- a/bellows/ezsp/v8/types/named.py +++ b/bellows/ezsp/v8/types/named.py @@ -693,7 +693,7 @@ class SecureEzspSecurityLevel(basic.uint8_t): """Security level of the Secure EZSP Protocol.""" -class SecureEzspRandomNumber(basic.fixed_list(16, basic.uint8_t)): +class SecureEzspRandomNumber(basic.FixedList[basic.uint8_t, 16]): """Randomly generated 64-bit number. Both NCP and Host contribute this number to create the Session ID, @@ -701,7 +701,7 @@ class SecureEzspRandomNumber(basic.fixed_list(16, basic.uint8_t)): """ -class SecureEzspSessionId(basic.fixed_list(8, basic.uint8_t)): +class SecureEzspSessionId(basic.FixedList[basic.uint8_t, 8]): """Generated 64-bit Session ID, using random numbers from Host and NCP. It is generated at each reboot (during negotiation phase). Having both sides diff --git a/bellows/ezsp/v8/types/struct.py b/bellows/ezsp/v8/types/struct.py index f10daa0e..89de3ebf 100644 --- a/bellows/ezsp/v8/types/struct.py +++ b/bellows/ezsp/v8/types/struct.py @@ -130,7 +130,7 @@ class EmberGpProxyTableEntry(EzspStruct): # The key to use for GPD. gpdKey: named.KeyData # The list of sinks (hardcoded to 2 which is the spec minimum). - sinkList: basic.fixed_list(2, EmberGpSinkListEntry) + sinkList: basic.FixedList[EmberGpSinkListEntry, 2] # The groupcast radius. groupcastRadius: basic.uint8_t # The search counter @@ -150,7 +150,7 @@ class EmberGpSinkTableEntry(EzspStruct): # The device id for the GPD. deviceId: basic.uint8_t # The list of sinks (hardcoded to 2 which is the spec minimum). - sinkList: basic.fixed_list(2, EmberGpSinkListEntry) + sinkList: basic.FixedList[EmberGpSinkListEntry, 2] # The assigned alias for the GPD. assignedAlias: named.EmberNodeId # The groupcast radius. @@ -175,7 +175,7 @@ class EmberDutyCycleLimits(EzspStruct): # The vendor identifier field shall contain the vendor identifier of the node. vendorId: basic.uint16_t # The vendor string field shall contain the vendor string of the node. - vendorString: basic.fixed_list(7, basic.uint8_t) + vendorString: basic.FixedList[basic.uint8_t, 7] class EmberPerDeviceDutyCycle(EzspStruct): diff --git a/bellows/ezsp/v9/commands.py b/bellows/ezsp/v9/commands.py index ada069f2..ee92562f 100644 --- a/bellows/ezsp/v9/commands.py +++ b/bellows/ezsp/v9/commands.py @@ -14,8 +14,8 @@ t.uint8_t, t.uint8_t, t.uint8_t, - t.List(t.uint16_t), - t.List(t.uint16_t), + t.List[t.uint16_t], + t.List[t.uint16_t], ), (t.EzspStatus,), ), @@ -35,8 +35,8 @@ "invalidCommand": (0x0058, (), (t.EzspStatus,)), "callback": (0x0006, (), ()), "noCallbacks": (0x0007, (), ()), - "setToken": (0x0009, (t.uint8_t, t.fixed_list(8, t.uint8_t)), (t.EmberStatus,)), - "getToken": (0x000A, (t.uint8_t,), (t.EmberStatus, t.fixed_list(8, t.uint8_t))), + "setToken": (0x0009, (t.uint8_t, t.FixedList[t.uint8_t, 8]), (t.EmberStatus,)), + "getToken": (0x000A, (t.uint8_t,), (t.EmberStatus, t.FixedList[t.uint8_t, 8])), "getMfgToken": (0x000B, (t.EzspMfgTokenId,), (t.LVBytes,)), "setMfgToken": (0x000C, (t.EzspMfgTokenId, t.LVBytes), (t.EmberStatus,)), "stackTokenChangedHandler": (0x000D, (), (t.uint16_t,)), @@ -52,12 +52,12 @@ "readAndClearCounters": ( 0x0065, (), - (t.fixed_list(len(t.EmberCounterType), t.uint16_t),), + (t.FixedList[t.uint16_t, len(t.EmberCounterType)],), ), "readCounters": ( 0x00F1, (), - (t.fixed_list(len(t.EmberCounterType), t.uint16_t),), + (t.FixedList[t.uint16_t, len(t.EmberCounterType)],), ), "counterRolloverHandler": (0x00F2, (), (t.EmberCounterType,)), "delayTest": (0x009D, (t.uint16_t,), ()), @@ -181,7 +181,7 @@ "getCurrentDutyCycle": ( 0x004C, (t.uint8_t,), - (t.EmberStatus, t.fixed_list(134, t.uint8_t)), + (t.EmberStatus, t.FixedList[t.uint8_t, 134]), ), "dutyCycleHandler": ( 0x004D, @@ -306,12 +306,12 @@ "incomingRouteRecordHandler": ( 0x0059, (), - (t.EmberNodeId, t.EUI64, t.uint8_t, t.int8s, t.LVList(t.EmberNodeId)), + (t.EmberNodeId, t.EUI64, t.uint8_t, t.int8s, t.LVList[t.EmberNodeId]), ), "changeSourceRouteHandler": (0x00C4, (), (t.EmberNodeId, t.EmberNodeId, t.Bool)), "setSourceRoute": ( 0x00AE, - (t.EmberNodeId, t.LVList(t.EmberNodeId)), + (t.EmberNodeId, t.LVList[t.EmberNodeId]), (t.EmberStatus,), ), "setSourceRouteDiscoveryMode": (0x005A, (t.uint8_t,), (t.uint32_t,)), @@ -546,8 +546,8 @@ "bootloadTransmitCompleteHandler": (0x0093, (), (t.EmberStatus, t.LVBytes)), "aesEncrypt": ( 0x0094, - (t.fixed_list(16, t.uint8_t), t.fixed_list(16, t.uint8_t)), - (t.fixed_list(16, t.uint8_t),), + (t.FixedList[t.uint8_t, 16], t.FixedList[t.uint8_t, 16]), + (t.FixedList[t.uint8_t, 16],), ), # 14. ZLL Frames "zllNetworkOps": ( diff --git a/bellows/ezsp/v9/types/named.py b/bellows/ezsp/v9/types/named.py index b3a6e43f..cc437b2c 100644 --- a/bellows/ezsp/v9/types/named.py +++ b/bellows/ezsp/v9/types/named.py @@ -701,7 +701,7 @@ class SecureEzspSecurityLevel(basic.uint8_t): """Security level of the Secure EZSP Protocol.""" -class SecureEzspRandomNumber(basic.fixed_list(16, basic.uint8_t)): +class SecureEzspRandomNumber(basic.FixedList[basic.uint8_t, 16]): """Randomly generated 64-bit number. Both NCP and Host contribute this number to create the Session ID, @@ -709,7 +709,7 @@ class SecureEzspRandomNumber(basic.fixed_list(16, basic.uint8_t)): """ -class SecureEzspSessionId(basic.fixed_list(8, basic.uint8_t)): +class SecureEzspSessionId(basic.FixedList[basic.uint8_t, 8]): """Generated 64-bit Session ID, using random numbers from Host and NCP. It is generated at each reboot (during negotiation phase). Having both sides diff --git a/bellows/ezsp/v9/types/struct.py b/bellows/ezsp/v9/types/struct.py index b575ff9f..92ddc5a9 100644 --- a/bellows/ezsp/v9/types/struct.py +++ b/bellows/ezsp/v9/types/struct.py @@ -140,7 +140,7 @@ class EmberGpProxyTableEntry(EzspStruct): # The security frame counter of the GPD. gpdSecurityFrameCounter: named.EmberGpSecurityFrameCounter # The list of sinks (hardcoded to 2 which is the spec minimum). - sinkList: basic.fixed_list(2, EmberGpSinkListEntry) + sinkList: basic.FixedList[EmberGpSinkListEntry, 2] # The groupcast radius. groupcastRadius: basic.uint8_t # The search counter @@ -160,7 +160,7 @@ class EmberGpSinkTableEntry(EzspStruct): # The device id for the GPD. deviceId: basic.uint8_t # The list of sinks (hardcoded to 2 which is the spec minimum). - sinkList: basic.fixed_list(2, EmberGpSinkListEntry) + sinkList: basic.FixedList[EmberGpSinkListEntry, 2] # The assigned alias for the GPD. assignedAlias: named.EmberNodeId # The groupcast radius. @@ -185,7 +185,7 @@ class EmberDutyCycleLimits(EzspStruct): # The vendor identifier field shall contain the vendor identifier of the node. vendorId: basic.uint16_t # The vendor string field shall contain the vendor string of the node. - vendorString: basic.fixed_list(7, basic.uint8_t) + vendorString: basic.FixedList[basic.uint8_t, 7] class EmberPerDeviceDutyCycle(EzspStruct): diff --git a/bellows/types/basic.py b/bellows/types/basic.py index eddfe3d1..4075a5d9 100644 --- a/bellows/types/basic.py +++ b/bellows/types/basic.py @@ -1,243 +1,33 @@ from __future__ import annotations -import enum -from typing import Callable, TypeVar - -CALLABLE_T = TypeVar("CALLABLE_T", bound=Callable) # pylint: disable=invalid-name - - -class int_t(int): # noqa: N801 - _signed = True - - def serialize(self): - return self.to_bytes(self._size, "little", signed=self._signed) - - @classmethod - def deserialize(cls, data): - # Work around https://bugs.python.org/issue23640 - r = cls(int.from_bytes(data[: cls._size], "little", signed=cls._signed)) - data = data[cls._size :] - return r, data - - -class int8s(int_t): # noqa: N801 - _size = 1 - - -class int16s(int_t): # noqa: N801 - _size = 2 - - -class int24s(int_t): # noqa: N801 - _size = 3 - - -class int32s(int_t): # noqa: N801 - _size = 4 - - -class int40s(int_t): # noqa: N801 - _size = 5 - - -class int48s(int_t): # noqa: N801 - _size = 6 - - -class int56s(int_t): # noqa: N801 - _size = 7 - - -class int64s(int_t): # noqa: N801 - _size = 8 - - -class uint_t(int_t): # noqa: N801 - _signed = False - - -class uint8_t(uint_t): # noqa: N801 - _size = 1 - - -class uint16_t(uint_t): # noqa: N801 - _size = 2 - - -class uint24_t(uint_t): # noqa: N801 - _size = 3 - - -class uint32_t(uint_t): # noqa: N801 - _size = 4 - - -class uint40_t(uint_t): # noqa: N801 - _size = 5 - - -class uint48_t(uint_t): # noqa: N801 - _size = 6 - - -class uint56_t(uint_t): # noqa: N801 - _size = 7 - - -class uint64_t(uint_t): # noqa: N801 - _size = 8 - - -class LVBytes(bytes): - _length_type = uint8_t - - def serialize(self) -> LVBytes: - return self._length_type(len(self)).serialize() + self - - @classmethod - def deserialize(cls, data: bytes) -> tuple[LVBytes, bytes]: - length, data = cls._length_type.deserialize(data) - - if len(data) < length: - raise ValueError(f"Data is too short: expected {length}, got {len(data)}") - - return cls(data[:length]), data[length:] +from zigpy.types import ( # noqa: F401 + FixedList, + List, + LVBytes, + LVList, + bitmap8, + bitmap16, + enum8, + enum16, + enum32, + int8s, + int16s, + int24s, + int32s, + int40s, + int48s, + int56s, + int64s, + uint8_t, + uint16_t, + uint24_t, + uint32_t, + uint40_t, + uint48_t, + uint56_t, + uint64_t, +) class LVBytes32(LVBytes): - _length_type = uint32_t - - -class _List(list): - _length = 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): - r = cls() - while data: - item, data = r._itemtype.deserialize(data) - r.append(item) - return r, data - - -class _LVList(_List): - def serialize(self): - head = len(self).to_bytes(1, "little") - data = super().serialize() - return head + data - - @classmethod - def deserialize(cls, data): - r = cls() - length, data = data[0], data[1:] - for _i in range(length): - item, data = r._itemtype.deserialize(data) - r.append(item) - return r, data - - -def List(itemtype): # noqa: N802 - class List(_List): - _itemtype = itemtype - - return List - - -def LVList(itemtype): # noqa: N802 - class LVList(_LVList): - _itemtype = itemtype - - return LVList - - -class _FixedList(_List): - @classmethod - def deserialize(cls, data): - r = cls() - for _i in range(r._length): - item, data = r._itemtype.deserialize(data) - r.append(item) - return r, data - - -def fixed_list(length, itemtype): - class FixedList(_FixedList): - _length = length - _itemtype = itemtype - - return FixedList - - -class HexRepr: - _hex_len = 2 - - def __repr__(self): - return ("0x{:0" + str(self._hex_len) + "x}").format(self) - - def __str__(self): - return ("0x{:0" + str(self._hex_len) + "x}").format(self) - - -def bitmap_factory(int_type: CALLABLE_T = uint8_t) -> CALLABLE_T: - """Bitmap factory.""" - - class _NewBitmap(enum.IntFlag): - def serialize(self): - """Serialize enum.""" - return int_type(self.value).serialize() - - @classmethod - def deserialize(cls, data: bytes) -> (bytes, bytes): - """Deserialize data.""" - val, data = int_type.deserialize(data) - return cls(val), data - - return _NewBitmap - - -class bitmap8(bitmap_factory(uint8_t)): # noqa: N801 - """8 bit bitmap class.""" - - -class bitmap16(bitmap_factory(uint16_t)): # noqa: N801 - """16 bit bitmap class.""" - - -class _IntEnumMeta(enum.EnumMeta): - def __call__(cls, value, names=None, *args, **kwargs): # noqa: N805 - if isinstance(value, str) and value.startswith("0x"): - value = int(value, base=16) - else: - value = int(value) - return super().__call__(value, names, *args, **kwargs) - - -def enum_factory(int_type: CALLABLE_T, undefined: str = "undefined") -> CALLABLE_T: - """Enum factory.""" - - class _NewEnum(int_type, enum.Enum, metaclass=_IntEnumMeta): - @classmethod - def _missing_(cls, value): - new = int_type.__new__(cls, value) - name = f"{undefined}_0x{{:0{int_type._size * 2}x}}" # pylint: disable=protected-access - new._name_ = name.format(value) - new._value_ = value - return new - - return _NewEnum - - -class enum8(enum_factory(uint8_t)): # noqa: N801 - pass - - -class enum16(enum_factory(uint16_t)): # noqa: N801 - pass - - -class enum32(enum_factory(uint16_t)): # noqa: N801 - pass + _prefix_length = 4 diff --git a/bellows/types/named.py b/bellows/types/named.py index ae8522a5..15ff4055 100644 --- a/bellows/types/named.py +++ b/bellows/types/named.py @@ -25,19 +25,19 @@ class ExtendedPanId(ztypes.ExtendedPanId): """Extended PAN ID.""" -class EmberNodeId(basic.HexRepr, basic.uint16_t): +class EmberNodeId(basic.uint16_t, repr="hex"): # 16-bit ZigBee network address. - _hex_len = 4 + pass -class EmberPanId(basic.HexRepr, basic.uint16_t): +class EmberPanId(basic.uint16_t, repr="hex"): # 802.15.4 PAN ID. - _hex_len = 4 + pass -class EmberMulticastId(basic.HexRepr, basic.uint16_t): +class EmberMulticastId(basic.uint16_t, repr="hex"): # 16-bit ZigBee multicast group identifier. - _hex_len = 4 + pass class EmberLibraryId(basic.uint8_t): @@ -1243,43 +1243,43 @@ class EzspSourceRouteOverheadInformation(basic.enum8): KeyData = ztypes.KeyData -class EmberCertificateData(basic.fixed_list(48, basic.uint8_t)): +class EmberCertificateData(basic.FixedList[basic.uint8_t, 48]): """The implicit certificate used in CBKE.""" -class EmberPublicKeyData(basic.fixed_list(22, basic.uint8_t)): +class EmberPublicKeyData(basic.FixedList[basic.uint8_t, 22]): """The public key data used in CBKE.""" -class EmberPrivateKeyData(basic.fixed_list(21, basic.uint8_t)): +class EmberPrivateKeyData(basic.FixedList[basic.uint8_t, 21]): """The private key data used in CBKE.""" -class EmberSmacData(basic.fixed_list(16, basic.uint8_t)): +class EmberSmacData(basic.FixedList[basic.uint8_t, 16]): """The Shared Message Authentication Code data used in CBKE.""" -class EmberSignatureData(basic.fixed_list(42, basic.uint8_t)): +class EmberSignatureData(basic.FixedList[basic.uint8_t, 42]): """An ECDSA signature.""" -class EmberCertificate283k1Data(basic.fixed_list(74, basic.uint8_t)): +class EmberCertificate283k1Data(basic.FixedList[basic.uint8_t, 74]): """The implicit certificate used in CBKE.""" -class EmberPublicKey283k1Data(basic.fixed_list(37, basic.uint8_t)): +class EmberPublicKey283k1Data(basic.FixedList[basic.uint8_t, 37]): """The 283k1 public key data used in CBKE.""" -class EmberPrivateKey283k1Data(basic.fixed_list(36, basic.uint8_t)): +class EmberPrivateKey283k1Data(basic.FixedList[basic.uint8_t, 36]): """The 283k1 private key data used in CBKE.""" -class EmberSignature283k1Data(basic.fixed_list(72, basic.uint8_t)): +class EmberSignature283k1Data(basic.FixedList[basic.uint8_t, 72]): """An 283k1 ECDSA signature data.""" -class EmberMessageDigest(basic.fixed_list(16, basic.uint8_t)): +class EmberMessageDigest(basic.FixedList[basic.uint8_t, 16]): """The calculated digest of a message""" @@ -1870,7 +1870,7 @@ class EmberDistinguishedNodeId(basic.enum16): TABLE_ENTRY_UNUSED = 0xFFFF -class EmberStackError(basic.enum16): +class EmberStackError(basic.enum8): """Stack error codes.""" ROUTE_ERROR_NO_ROUTE_AVAILABLE = 0x00 diff --git a/bellows/types/struct.py b/bellows/types/struct.py index 83d91e03..e2559933 100644 --- a/bellows/types/struct.py +++ b/bellows/types/struct.py @@ -1,4 +1,4 @@ -from zigpy.types import Struct as EzspStruct +from zigpy.types import Struct as EzspStruct, StructField from . import basic, named @@ -97,7 +97,7 @@ class EmberMulticastTableEntry(EzspStruct): # ZDO is not a member of any multicast groups.) endpoint: basic.uint8_t # The network index of the network the entry is related to. - networkIndex: basic.uint8_t + networkIndex: basic.uint8_t = StructField(optional=True) class EmberTransientKeyData(EzspStruct): @@ -116,7 +116,7 @@ class EmberTransientKeyData(EzspStruct): class EmberAesMmoHashContext(EzspStruct): # The hash context for an ongoing hash operation. # The result of ongoing the hash operation. - result: basic.fixed_list(16, basic.uint8_t) + result: basic.FixedList[basic.uint8_t, 16] # The total length of the data that has been hashed so far. length: basic.uint32_t diff --git a/bellows/zigbee/application.py b/bellows/zigbee/application.py index 7832f841..5b757bf0 100644 --- a/bellows/zigbee/application.py +++ b/bellows/zigbee/application.py @@ -294,7 +294,7 @@ async def load_network_info(self, *, load_devices=False) -> None: ) assert status == t.EmberStatus.SUCCESS - (status, _, network_key_info) = await ezsp.getNetworkKeyInfo() + (status, network_key_info) = await ezsp.getNetworkKeyInfo() assert status == t.EmberStatus.SUCCESS if not network_key_info.network_key_set: @@ -1042,7 +1042,7 @@ def handle_route_record( ieee: t.EUI64, lqi: t.uint8_t, rssi: t.int8s, - relays: t.LVList(t.EmberNodeId), + relays: t.LVList[t.EmberNodeId], ) -> None: LOGGER.debug( "Processing route record request: %s", (nwk, ieee, lqi, rssi, relays) diff --git a/tests/test_application_network_state.py b/tests/test_application_network_state.py index 1c33deea..0b020740 100644 --- a/tests/test_application_network_state.py +++ b/tests/test_application_network_state.py @@ -187,7 +187,6 @@ def export_key(security_context): ezsp.getNetworkKeyInfo = AsyncMock( return_value=[ ezsp.types.sl_Status.SL_STATUS_OK, - 0x1234, ezsp.types.sl_zb_sec_man_network_key_info_t( network_key_set=True, alternate_network_key_set=False, @@ -221,7 +220,6 @@ async def test_load_network_info_no_key_set(app, network_info, node_info): app._ezsp.getNetworkKeyInfo = AsyncMock( return_value=[ app._ezsp.types.sl_Status.SL_STATUS_OK, - 0x1234, app._ezsp.types.sl_zb_sec_man_network_key_info_t( network_key_set=False, # Not set alternate_network_key_set=False, diff --git a/tests/test_ezsp.py b/tests/test_ezsp.py index 5c2933eb..6121d84f 100644 --- a/tests/test_ezsp.py +++ b/tests/test_ezsp.py @@ -183,7 +183,7 @@ async def test_form_network_fail_stack_status(ezsp_f): def test_receive_new(ezsp_f): callback = MagicMock() ezsp_f.add_callback(callback) - ezsp_f.frame_received(b"\x00\xff\x00\x04\x05\x06") + ezsp_f.frame_received(b"\x00\xff\x00\x04\x05\x06\x00") assert callback.call_count == 1 diff --git a/tests/test_ezsp_protocol.py b/tests/test_ezsp_protocol.py index f9dd25d2..6b727754 100644 --- a/tests/test_ezsp_protocol.py +++ b/tests/test_ezsp_protocol.py @@ -29,7 +29,7 @@ async def test_command(prot_hndl): def test_receive_reply(prot_hndl): callback_mock = MagicMock(spec_set=asyncio.Future) prot_hndl._awaiting[0] = (0, prot_hndl.COMMANDS["version"][2], callback_mock) - prot_hndl(b"\x00\xff\x00\x04\x05\x06") + prot_hndl(b"\x00\xff\x00\x04\x05\x06\x00") assert 0 not in prot_hndl._awaiting assert callback_mock.set_exception.call_count == 0 @@ -42,7 +42,7 @@ def test_receive_reply_after_timeout(prot_hndl): callback_mock = MagicMock(spec_set=asyncio.Future) callback_mock.set_result.side_effect = asyncio.InvalidStateError() prot_hndl._awaiting[0] = (0, prot_hndl.COMMANDS["version"][2], callback_mock) - prot_hndl(b"\x00\xff\x00\x04\x05\x06") + prot_hndl(b"\x00\xff\x00\x04\x05\x06\x00") assert 0 not in prot_hndl._awaiting assert callback_mock.set_exception.call_count == 0 diff --git a/tests/test_types.py b/tests/test_types.py index 98356c6e..a724d96a 100644 --- a/tests/test_types.py +++ b/tests/test_types.py @@ -4,29 +4,6 @@ import bellows.types as t -def test_basic(): - assert t.uint8_t.deserialize(b"\x08") == (8, b"") - - -def test_extra_data(): - assert t.uint8_t.deserialize(b"\x08extra") == (8, b"extra") - - -def test_multibyte(): - assert t.uint16_t.deserialize(b"\x08\x01") == (0x0108, b"") - - -def test_lvbytes(): - d, r = t.LVBytes.deserialize(b"\x0412345") - assert r == b"5" - assert d == b"1234" - - assert t.LVBytes.serialize(d) == b"\x041234" - - with pytest.raises(ValueError): - t.LVBytes.deserialize(b"\x04123") - - def test_lvbytes32(): d, r = t.LVBytes32.deserialize(b"\x04\x00\x00\x0012345") assert r == b"5" @@ -35,187 +12,6 @@ def test_lvbytes32(): assert t.LVBytes32.serialize(d) == b"\x04\x00\x00\x001234" -def test_lvlist(): - d, r = t.LVList(t.uint8_t).deserialize(b"\x0412345") - assert r == b"5" - assert d == list(map(ord, "1234")) - assert t.LVList(t.uint8_t).serialize(d) == b"\x041234" - - -def test_list(): - expected = list(map(ord, "\x0123")) - assert t.List(t.uint8_t).deserialize(b"\x0123") == (expected, b"") - - -def test_struct(): - class TestStruct(t.EzspStruct): - a: t.uint8_t - b: t.uint8_t - - @property - def test(self): - return None - - 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 - - r = repr(ts) - assert "TestStruct" in r - assert r.startswith("TestStruct") and r.endswith(")") - - s = ts2.serialize() - assert s == b"\xaa\xbb" - - -def test_str(): - assert str(t.EzspStatus.deserialize(b"\0")[0]) == "EzspStatus.SUCCESS" - - -def test_ember_eui64(): - serialized = b"\x00\x01\x02\x03\x04\x05\x06\x07" - eui64, data = t.EUI64.deserialize(serialized) - assert data == b"" - assert eui64.serialize() == serialized - - -def test_hex_repr(): - class NwkAsHex(t.HexRepr, t.uint16_t): - _hex_len = 4 - - nwk = NwkAsHex(0x1234) - assert str(nwk) == "0x1234" - assert repr(nwk) == "0x1234" - - -def test_bitmap(): - """Test bitmaps.""" - - class TestBitmap(t.bitmap16): - CH_1 = 0x0010 - CH_2 = 0x0020 - CH_3 = 0x0040 - CH_4 = 0x0080 - ALL = 0x00F0 - - extra = b"extra data\xaa\55" - data = b"\xf0\x00" - r, rest = TestBitmap.deserialize(data + extra) - assert rest == extra - assert r is TestBitmap.ALL - assert r.name == "ALL" - assert r.value == 0x00F0 - assert r.serialize() == data - - data = b"\x60\x00" - r, rest = TestBitmap.deserialize(data + extra) - assert rest == extra - assert TestBitmap.CH_1 not in r - assert TestBitmap.CH_2 in r - assert TestBitmap.CH_3 in r - assert TestBitmap.CH_4 not in r - assert TestBitmap.ALL not in r - assert r.value == 0x0060 - assert r.serialize() == data - - -def test_bitmap_undef(): - """Test bitmaps with some undefined flags.""" - - class TestBitmap(t.bitmap16): - CH_1 = 0x0010 - CH_2 = 0x0020 - CH_3 = 0x0040 - CH_4 = 0x0080 - ALL = 0x00F0 - - extra = b"extra data\xaa\55" - data = b"\x60\x0f" - r, rest = TestBitmap.deserialize(data + extra) - assert rest == extra - assert TestBitmap.CH_1 not in r - assert TestBitmap.CH_2 in r - assert TestBitmap.CH_3 in r - assert TestBitmap.CH_4 not in r - assert TestBitmap.ALL not in r - assert r.value == 0x0F60 - assert r.serialize() == data - - -def test_enum_undef(): - class TestEnum(t.enum8): - ALL = 0xAA - - data = b"\x55" - extra = b"extra" - - r, rest = TestEnum.deserialize(data + extra) - assert rest == extra - assert r == 0x55 - assert r.value == 0x55 - assert r.name == "undefined_0x55" - assert r.serialize() == data - assert isinstance(r, TestEnum) - - r = TestEnum("85") - assert r == 0x55 - assert r.value == 0x55 - assert r.name == "undefined_0x55" - assert r.serialize() == data - assert isinstance(r, TestEnum) - - r = TestEnum("0x55") - assert r == 0x55 - assert r.value == 0x55 - assert r.name == "undefined_0x55" - assert r.serialize() == data - assert isinstance(r, TestEnum) - - -def test_enum(): - class TestEnum(t.enum8): - ALL = 0x55 - ERR = 1 - - data = b"\x55" - extra = b"extra" - - r, rest = TestEnum.deserialize(data + extra) - assert rest == extra - assert r == 0x55 - assert r.value == 0x55 - assert r.name == "ALL" - assert isinstance(r, TestEnum) - assert TestEnum.ALL + TestEnum.ERR == 0x56 - - r = TestEnum("85") - assert r == 0x55 - assert r.value == 0x55 - assert r.name == "ALL" - assert isinstance(r, TestEnum) - assert TestEnum.ALL + TestEnum.ERR == 0x56 - - r = TestEnum("0x55") - assert r == 0x55 - assert r.value == 0x55 - assert r.name == "ALL" - assert isinstance(r, TestEnum) - assert TestEnum.ALL + TestEnum.ERR == 0x56 - - -def test_fixed_list(): - list_type = t.fixed_list(2, t.uint8_t) - data = b"\x01\xFE" - extra = b"extarlist \xaa\55" - - res, rest = list_type.deserialize(data + extra) - assert rest == extra - assert res == [1, 254] - - @pytest.mark.parametrize( "node_type, logical_type", (