Parse the entire NIB structure for both 3.0.1 and newer versions

Author: puddly

tests/test_nib.py

@@ -0,0 +1,112 @@
+import pytest
+
+import zigpy_znp.types as t
+
+from zigpy_znp.znp.nib import (
+    NIB,
+    OldNIB,
+    parse_nib,
+    NwkState8,
+    NwkState16,
+    PaddingByte,
+    Empty,
+)
+
+
+NEW_NIB = bytes.fromhex(
+    """
+    790502331433001e0000000105018f00070002051e000000190000000000000000000000958608000080
+    10020f0f0400010000000100000000a860ca53db3bc0a801000000000000000000000000000000000000
+    0000000000000000000000000000000000000f030001780a0100000020470000"""
+)
+
+OLD_NIB = bytes.fromhex(
+    """
+    5f0502101410001e0000000105018f070002051e0000140000000000000000000085d208000010020f0f
+    05000100000001000000008533ce1c004b12000100000000000000000000000000000000000000000000
+    00000000000000000000000000003c030001780a010000170000"""
+)
+
+
+def test_nwk_state():
+    assert NwkState8._member_type_ == t.uint8_t
+    assert NwkState16._member_type_ == t.uint16_t
+
+    # They should be otherwise identical
+    assert NwkState8._value2member_map_ == NwkState16._value2member_map_
+
+
+def test_padding_byte():
+    with pytest.raises(ValueError):
+        PaddingByte.deserialize(b"")
+
+    with pytest.raises(ValueError):
+        PaddingByte(b"ab")
+
+    with pytest.raises(ValueError):
+        PaddingByte(b"")
+
+    assert PaddingByte.deserialize(b"abc") == (PaddingByte(b"a"), b"bc")
+    assert PaddingByte.deserialize(b"a") == (PaddingByte(b"a"), b"")
+
+
+def test_empty():
+    with pytest.raises(ValueError):
+        Empty(b"a")
+
+    assert Empty.deserialize(b"abc") == (Empty(), b"abc")
+    assert Empty.deserialize(b"") == (Empty(), b"")
+
+
+def test_nib_classes():
+    old_nib_nwk_state = next(a for a in OldNIB.__attrs_attrs__ if a.type == NwkState8)
+
+    # The old NIB should be the new NIB, without padding and with a shorter struct
+    # integer type.
+    fixed_new_nib = [
+        a if a.type != NwkState16 else old_nib_nwk_state
+        for a in NIB.__attrs_attrs__
+        if a.type is not PaddingByte
+    ]
+
+    assert fixed_new_nib == list(OldNIB.__attrs_attrs__)
+
+
+def test_nib_detection():
+    assert isinstance(parse_nib(NEW_NIB), NIB)
+    assert isinstance(parse_nib(OLD_NIB), OldNIB)
+
+    with pytest.raises(ValueError):
+        parse_nib(NEW_NIB + b"\x00")
+
+    with pytest.raises(ValueError):
+        parse_nib(OLD_NIB + b"\x00")
+
+
+def test_nib_parsing():
+    new_nib, remaining = NIB.deserialize(NEW_NIB)
+    assert not remaining
+
+    old_nib, remaining = OldNIB.deserialize(OLD_NIB)
+    assert not remaining
+
+    assert new_nib.serialize() == NEW_NIB
+    assert old_nib.serialize() == OLD_NIB
+
+    # Superficially validate that the NIBs were parsed correctly
+    for nib in (new_nib, old_nib):
+        assert nib.nwkProtocolVersion == 2
+
+        # Make sure the channel list is valid
+        assert nib.channelList | t.Channels.ALL_CHANNELS == t.Channels.ALL_CHANNELS
+
+        # Make sure the logical channel is contained within the channel mask
+        assert (
+            t.Channels.from_channel_list([nib.nwkLogicalChannel]) | nib.channelList
+            == nib.channelList
+        )
+
+        assert nib.nwkKeyLoaded
+        assert nib.nwkIsConcentrator
+        assert nib.nwkManagerAddr == t.NWK(0x0000)
+        assert nib.nwkCoordAddress == t.NWK(0x0000)

tests/test_types_struct.py

@@ -84,10 +84,6 @@ class Nested(t.Struct):
     assert nstd.cmd.id_0 == 2
     assert nstd.cmd.id_1 == 3
 
-    test_2 = Nested((4, 5))
-    assert test_2.cmd.id_0 == 4
-    assert test_2.cmd.id_1 == 5
-
 
 def test_struct_converter_2():
     @attr.s

zigpy_znp/types/struct.py

@@ -17,10 +17,14 @@ def deserialize(cls, data: bytes):
 
         return cls(*args), data
 
+    @classmethod
+    def fields(cls):
+        return attr.fields(cls)
+
     @classmethod
     def schema(cls):
         """Return schema of the class."""
-        return (a.type for a in attr.fields(cls))
+        return (a.type for a in cls.fields())
 
     def serialize(self):
         """Serialize Structure."""
@@ -29,7 +33,7 @@ def serialize(self):
 
     def as_tuple(self):
         """Return tuple of fields values."""
-        names = (f.name for f in attr.fields(self.__class__))
+        names = (f.name for f in self.fields())
         return (getattr(self, n) for n in names)
 
     @staticmethod
@@ -39,9 +43,7 @@ def converter(_type) -> typing.Callable:
         def converter(input):
             if isinstance(input, _type):
                 return input
-            try:
-                return _type(*input)
-            except TypeError:
-                return _type(input)
+
+            return _type(input)
 
         return converter

zigpy_znp/zigbee/application.py

@@ -21,9 +21,11 @@
 import zigpy_znp.config as conf
 import zigpy_znp.types as t
 import zigpy_znp.commands as c
+
 from zigpy_znp.exceptions import InvalidCommandResponse
 
 from zigpy_znp.api import ZNP
+from zigpy_znp.znp.nib import parse_nib
 from zigpy_znp.types.nvids import NwkNvIds
 
 
@@ -420,22 +422,20 @@ async def startup(self, auto_form=False):
             endpoint=100, profile_id=zigpy.profiles.zll.PROFILE_ID, device_id=0x0005
         )
 
-        # Structure is in `zstack/stack/nwk/nwk.h`.
-        #
-        # This is hacky but since there are exactly two distinct firmwares, it doesn't
-        # make sense to try and write some fancys struct padding guesser.
-        nib = await self._znp.nvram_read(NwkNvIds.NIB)
-
-        if len(nib) == 116:
-            # Structs aligned to 16 bits: CC26x2, CC13X2
-            self._channel = nib[24]
-            self._channels, _ = t.Channels.deserialize(nib[40:44])
-        elif len(nib) == 110:
-            # No struct alignment: CC2531
-            self._channel = nib[22]
-            self._channels, _ = t.Channels.deserialize(nib[36:40])
-        else:
-            LOGGER.warning("Could not extract channel information from NIB: %r", nib)
+        nib = parse_nib(await self._znp.nvram_read(NwkNvIds.NIB))
+        LOGGER.debug("Parsed NIB: %s", nib)
+
+        self._channel = nib.nwkLogicalChannel
+        self._channels = nib.channelList
+        self._pan_id = nib.nwkPanId
+        self._ext_pan_id = nib.extendedPANID
+        self._nwk = nib.nwkDevAddress
+
+        LOGGER.info(
+            "Using channel mask %s, currently on channel %d",
+            self.channels,
+            self.channel,
+        )
 
     async def update_network(
         self,
@@ -481,17 +481,11 @@ async def update_network(
             self._channels = channels
 
         if channel is not None:
-            # XXX: We modify the logical channel value directly in the NIB
-            nib = bytearray(await self._znp.nvram_read(NwkNvIds.NIB))
-
-            if len(nib) == 116:
-                nib[24] = channel
-            elif len(nib) == 110:
-                nib[22] = channel
-            else:
-                raise RuntimeError(f"Cannot set channel in unknown NIB struct: {nib!r}")
-
-            await self._znp.nvram_write(NwkNvIds.NIB, nib)
+            # We modify the logical channel value directly in the NIB.
+            # Does this actually work?
+            nib = parse_nib(await self._znp.nvram_read(NwkNvIds.NIB))
+            nib.nwkLogicalChannel = channel
+            await self._znp.nvram_write(NwkNvIds.NIB, nib.serialize())
 
             self._channel = channel