Refactor reading/writing network parameters. (#79)

* Network parameteres specific types.
* List types class.
Refactor EUI64 to use new List classes.

* Refactor read_parameter()/write_parameter().
Some parameters (like Network Key) require an extra argument eg index.
Modify readin/writing parameters schema.

* Update tests.
* Update dependencies.

Author: Adminiuga

setup.py

@@ -15,7 +15,7 @@
     packages=find_packages(exclude=['*.tests']),
     install_requires=[
         'pyserial-asyncio',
-        'zigpy-homeassistant>=0.9.0',
+        'zigpy-homeassistant>=0.10.0',
     ],
     tests_require=[
         'pytest',

tests/test_api.py

@@ -305,12 +305,12 @@ async def test_read_parameter(api):
 
     r = await api.read_parameter(deconz_api.NetworkParameter.nwk_panid)
     assert api._command.call_count == 1
-    assert r == 0x55aa
+    assert r[0] == 0x55aa
 
     api._command.reset_mock()
     r = await api.read_parameter(0x05)
     assert api._command.call_count == 1
-    assert r == 0x55aa
+    assert r[0] == 0x55aa
 
     with pytest.raises(KeyError):
         await api.read_parameter('unknown_param')
@@ -365,7 +365,7 @@ async def test_write_parameter(api):
 async def test_version(protocol_ver, firmware_version, flags, api):
     api.read_parameter = mock.MagicMock()
     api.read_parameter.side_effect = asyncio.coroutine(
-        mock.MagicMock(return_value=protocol_ver))
+        mock.MagicMock(return_value=[protocol_ver]))
     api._command = mock.MagicMock()
     api._command.side_effect = asyncio.coroutine(
         mock.MagicMock(return_value=[firmware_version]))

tests/test_application.py

@@ -195,7 +195,7 @@ async def test_startup(protocol_ver, watchdog_cc, app, monkeypatch, version=0):
 
     async def _version():
         app._api._proto_ver = protocol_ver
-        return version
+        return [version]
 
     app._reset_watchdog = mock.MagicMock(
         side_effect=asyncio.coroutine(mock.MagicMock()))
@@ -204,7 +204,8 @@ async def _version():
     app._api._command = mock.MagicMock(
         side_effect=asyncio.coroutine(mock.MagicMock()))
     app._api.read_parameter = mock.MagicMock(
-        side_effect=asyncio.coroutine(mock.MagicMock()))
+        side_effect=asyncio.coroutine(
+            mock.MagicMock(return_value=[[0]])))
     app._api.version = mock.MagicMock(
         side_effect=_version)
     app._api.write_parameter = mock.MagicMock(

tests/test_types.py

@@ -35,7 +35,14 @@ def test_deconz_address_ieee():
     assert rest == extra
     assert addr.address_mode == t.ADDRESS_MODE.IEEE
     assert addr.address_mode == 3
-    assert addr.address == [0xbe, 0xef, 0xee, 0xdd, 0xcc, 0xbb, 0xaa, 0x55]
+    assert addr.address[0] == 0x55
+    assert addr.address[1] == 0xaa
+    assert addr.address[2] == 0xbb
+    assert addr.address[3] == 0xcc
+    assert addr.address[4] == 0xdd
+    assert addr.address[5] == 0xee
+    assert addr.address[6] == 0xef
+    assert addr.address[7] == 0xbe
 
     assert addr.serialize() == data
 
@@ -48,7 +55,33 @@ def test_deconz_address_nwk_and_ieee():
     assert rest == extra
     assert addr.address_mode == t.ADDRESS_MODE.NWK_AND_IEEE
     assert addr.address_mode == 4
-    assert addr.ieee == [0xbe, 0xef, 0xee, 0xdd, 0xcc, 0xbb, 0x99, 0x88]
+    assert addr.ieee[0] == 0x88
+    assert addr.ieee[1] == 0x99
+    assert addr.ieee[2] == 0xbb
+    assert addr.ieee[3] == 0xcc
+    assert addr.ieee[4] == 0xdd
+    assert addr.ieee[5] == 0xee
+    assert addr.ieee[6] == 0xef
+    assert addr.ieee[7] == 0xbe
     assert addr.address == 0xaa55
 
     assert addr.serialize() == data
+
+
+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

zigpy_deconz/api.py

@@ -38,7 +38,7 @@ class Command(t.uint8_t, enum.Enum):
     ),
     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),
+    Command.read_parameter: (t.uint16_t, t.uint8_t, t.Bytes),
     Command.version: (),
     Command.write_parameter: (t.uint16_t, t.uint8_t, t.Bytes),
 }
@@ -88,21 +88,21 @@ class NetworkParameter(t.uint8_t, enum.Enum):
 
 
 NETWORK_PARAMETER_SCHEMA = {
-    NetworkParameter.mac_address: t.EUI64,
-    NetworkParameter.nwk_panid: t.uint16_t,
-    NetworkParameter.nwk_address: t.uint16_t,
-    NetworkParameter.nwk_extended_panid: t.uint64_t,
-    NetworkParameter.aps_designed_coordinator: t.uint8_t,
-    NetworkParameter.channel_mask: t.uint32_t,
-    NetworkParameter.aps_extended_panid: t.uint64_t,
-    NetworkParameter.trust_center_address: t.uint64_t,
-    NetworkParameter.security_mode: t.uint8_t,
-    NetworkParameter.network_key: t.uint8_t,
-    NetworkParameter.current_channel: t.uint8_t,
-    NetworkParameter.permit_join: t.uint8_t,
-    NetworkParameter.protocol_version: t.uint16_t,
-    NetworkParameter.nwk_update_id: t.uint8_t,
-    NetworkParameter.watchdog_ttl: t.uint32_t,
+    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: (t.uint32_t, ),
+    NetworkParameter.aps_extended_panid: (t.ExtendedPanId, ),
+    NetworkParameter.trust_center_address: (t.EUI64, ),
+    NetworkParameter.security_mode: (t.uint8_t, ),
+    NetworkParameter.network_key: (t.uint8_t, t.Key, ),
+    NetworkParameter.current_channel: (t.uint8_t, ),
+    NetworkParameter.permit_join: (t.uint8_t, ),
+    NetworkParameter.protocol_version: (t.uint16_t, ),
+    NetworkParameter.nwk_update_id: (t.uint8_t, ),
+    NetworkParameter.watchdog_ttl: (t.uint32_t, ),
 }
 
 
@@ -231,7 +231,7 @@ def change_network_state(self, state):
     def _handle_change_network_state(self, data):
         LOGGER.debug("Change network state response: %s", NetworkState(data[0]).name)
 
-    async def read_parameter(self, id_):
+    async def read_parameter(self, id_, *args):
         try:
             if isinstance(id_, str):
                 param = NetworkParameter[id_]
@@ -240,15 +240,16 @@ async def read_parameter(self, id_):
         except (KeyError, ValueError):
             raise KeyError("Unknown parameter id: %s" % (id_, ))
 
-        r = await self._command(Command.read_parameter, 1, param)
-        data = NETWORK_PARAMETER_SCHEMA[param].deserialize(r[2])[0]
+        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
 
     def _handle_read_parameter(self, data):
         pass
 
-    def write_parameter(self, id_, value):
+    def write_parameter(self, id_, *args):
         try:
             if isinstance(id_, str):
                 param = NetworkParameter[id_]
@@ -257,7 +258,7 @@ def write_parameter(self, id_, value):
         except (KeyError, ValueError):
             raise KeyError("Unknown parameter id: %s write request" % (id_,))
 
-        v = NETWORK_PARAMETER_SCHEMA[param](value).serialize()
+        v = t.serialize(args, NETWORK_PARAMETER_SCHEMA[param])
         length = len(v) + 1
         return self._command(Command.write_parameter, length, param, v)
 
@@ -271,7 +272,7 @@ def _handle_write_parameter(self, data):
         LOGGER.debug("Write parameter %s: SUCCESS", param.name)
 
     async def version(self):
-        self._proto_ver = await self[NetworkParameter.protocol_version]
+        self._proto_ver, = await self[NetworkParameter.protocol_version]
         version = await self._command(Command.version)
         if self.protocol_version >= MIN_PROTO_VERSION and \
                 (version[0] & 0x0000FF00) == 0x00000500:

zigpy_deconz/types.py

@@ -159,38 +159,67 @@ def __repr__(self):
         return r
 
 
-class EUI64(list):
+class List(list):
+    _length = None
+    _itemtype = None
+
     def serialize(self):
-        assert len(self) == 8
-        return b''.join([i.serialize() for i in self[::-1]])
+        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 FixedList(List):
+    _length = None
+    _itemtype = None
 
     @classmethod
     def deserialize(cls, data):
-        r = []
-        for i in range(8):
-            item, data = uint8_t.deserialize(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 cls(r[::-1]), data
+        return r, data
+
+
+class EUI64(FixedList):
+    _length = 8
+    _itemtype = uint8_t
 
     def __repr__(self):
-        return ':'.join('%02x' % i for i in self)
+        return ':'.join('%02x' % i for i in self[::-1])
 
     def __hash__(self):
         return hash(repr(self))
 
 
 class HexRepr:
-    _hex_len = 2
-
     def __repr__(self):
-        return ('0x{:0' + str(self._hex_len) + 'x}').format(self)
+        return ('0x{:0' + str(self._size * 2) + 'x}').format(self)
 
     def __str__(self):
-        return ('0x{:0' + str(self._hex_len) + 'x}').format(self)
+        return ('0x{:0' + str(self._size * 2) + 'x}').format(self)
 
 
 class NWK(HexRepr, uint16_t):
-    _hex_len = 4
+    pass
+
+
+class PanId(HexRepr, uint16_t):
+    pass
+
+
+class ExtendedPanId(EUI64):
+    pass
 
 
 class DeconzAddress(Struct):
@@ -245,3 +274,8 @@ def deserialize(cls, data):
             e, data = uint8_t.deserialize(data)
         setattr(r, cls._fields[2][0], e)
         return r, data
+
+
+class Key(FixedList):
+    _itemtype = uint8_t
+    _length = 16

zigpy_deconz/zigbee/application.py

@@ -44,7 +44,7 @@ async def startup(self, auto_form=False):
         """Perform a complete application startup"""
         self.version = await self._api.version()
         await self._api.device_state()
-        ieee = await self._api[NetworkParameter.mac_address]
+        ieee, = await self._api[NetworkParameter.mac_address]
         self._ieee = zigpy.types.EUI64(ieee)
         await self._api[NetworkParameter.nwk_panid]
         await self._api[NetworkParameter.nwk_address]