Add basic zha_map support

tests/test_buffalo.py

@@ -1,7 +1,7 @@
 import zigpy_cc.types as t
-from zigpy.types import EUI64
+from zigpy.types import EUI64, NWK
 
-from zigpy_cc.buffalo import Buffalo
+from zigpy_cc.buffalo import Buffalo, BuffaloOptions
 
 ieeeAddr1 = {
  "string": EUI64.convert("ae:44:01:12:00:4b:12:00"),
@@ -36,3 +36,31 @@ def test_read_ieee2():
  data_in = Buffalo(ieeeAddr2["hex"])
  actual = data_in.read_parameter(t.ParameterType.IEEEADDR, {})
  assert ieeeAddr2["string"] == actual
+
+
+def test_list_nighbor_lqi():
+ value = [
+ {
+ "extPanId": EUI64.convert("d8:dd:dd:dd:d0:dd:ed:dd"),
+ "extAddr": EUI64.convert("00:15:8d:00:04:21:dc:b3"),
+ "nwkAddr": NWK(0xE961),
+ "deviceType": 1,
+ "rxOnWhenIdle": 2,
+ "relationship": 2,
+ "permitJoin": 2,
+ "depth": 255,
+ "lqi": 69,
+ }
+ ]
+ data_out = Buffalo(b"")
+ data_out.write_parameter(t.ParameterType.LIST_NEIGHBOR_LQI, value, {})
+ assert (
+ b"\xdd\xed\xdd\xd0\xdd\xdd\xdd\xd8\xb3\xdc!\x04\x00\x8d\x15\x00a\xe9)\x02\xffE"
+ == data_out.buffer
+ )
+
+ data_in = Buffalo(data_out.buffer)
+ options = BuffaloOptions()
+ options.length = len(value)
+ act = data_in.read_parameter(t.ParameterType.LIST_NEIGHBOR_LQI, options)
+ assert value == act

tests/test_types.py

@@ -512,3 +512,22 @@ def test_mgmt_lqi_rsp():
  "'deviceType': 2, 'rxOnWhenIdle': 0, 'relationship': 1, 'permitJoin': 2, 'depth': 1, 'lqi': 55}"
  "]}" == str(obj)
  )
+
+
+def test_mgmt_nwk_update_notify():
+ frame = uart.UnpiFrame(
+ 2,
+ 5,
+ 184,
+ b"\xe9\xc0\x00\x00\xf8\xff\x07\x14\x00\x0f\x00\x10\xd6\xac"
+ b"\xcc\xb0\xb5\xa2\xb2\xa5\xb3\xa5\xa2\xa9\xa1\xa1\xa5\xae",
+ 28,
+ 211,
+ )
+ obj = ZpiObject.from_unpi_frame(frame)
+ assert (
+ "AREQ ZDO mgmtNwkUpdateNotify tsn: None {'srcaddr': 0xc0e9, 'status': 0, 'scanchannels': 134215680, "
+ "'totaltransmissions': 20, 'transmissionfailures': 15, 'channelcount': 16, "
+ "'energyvalues': [214, 172, 204, 176, 181, 162, 178, 165, 179, 165, 162, 169, 161, 161, 165, 174]"
+ "}" == str(obj)
+ )

zigpy_cc/api.py

@@ -109,7 +109,12 @@ async def request_raw(self, obj: ZpiObject, expected_status=None):
  frame = obj.to_unpi_frame()
 
  if obj.type == CommandType.SREQ:
- timeout = 20000 if obj.command == 'bdbStartCommissioning' or obj.command == 'startupFromApp' else Timeouts.SREQ
+ timeout = (
+ 20000
+ if obj.command == "bdbStartCommissioning"
+ or obj.command == "startupFromApp"
+ else Timeouts.SREQ
+ )
  waiter = self.wait_for(
  CommandType.SRSP, obj.subsystem, obj.command, {}, timeout
  )
@@ -188,7 +193,7 @@ def wait_for(
 
  def callback():
  if not waiter.future.done() or waiter.future.cancelled():
- LOGGER.warning('Waiter timeout: %s', waiter)
+ LOGGER.warning("Waiter timeout: %s", waiter)
  self._waiters.remove(waiter)
 
  asyncio.get_event_loop().call_later(timeout / 1000 + 0.1, callback)

zigpy_cc/buffalo.py

@@ -37,11 +37,31 @@ def write_parameter(self, type, value, options):
  elif type == ParameterType.LIST_UINT16:
  for v in value:
  self.write(v, 2)
+ elif type == ParameterType.LIST_NEIGHBOR_LQI:
+ for v in value:
+ self.write_neighbor_lqi(v)
  else:
- raise TODO("write %s, value: %s", ParameterType(type), value)
-
- def write(self, value, length=1):
- self.buffer += value.to_bytes(length, "little")
+ raise TODO(
+ "write %s, value: %s, options: %s", ParameterType(type), value, options
+ )
+
+ def write(self, value, length=1, signed=False):
+ self.buffer += value.to_bytes(length, "little", signed=signed)
+
+ def write_neighbor_lqi(self, value):
+ for i in value["extPanId"]:
+ self.write(i)
+ for i in value["extAddr"]:
+ self.write(i)
+ self.write(value["nwkAddr"], 2)
+ self.write(
+ value["deviceType"]
+ | (value["rxOnWhenIdle"] * 4)
+ | (value["relationship"] * 16)
+ )
+ self.write(value["permitJoin"])
+ self.write(value["depth"])
+ self.write(value["lqi"])
 
  def read_parameter(self, type, options):
  if type == ParameterType.UINT8:

zigpy_cc/definition.py

@@ -10,6 +10,7 @@
 parameterType = "parameterType"
 type = "type"
 
+
 Definition = {
  Subsystem.SYS: [
  {
@@ -1639,6 +1640,20 @@
  {name: "status", parameterType: ParameterType.UINT8},
  ],
  },
+ {
+ name: "mgmtNwkUpdateNotify",
+ ID: 184,
+ type: CommandType.AREQ,
+ request: [
+ {name: "srcaddr", parameterType: ParameterType.UINT16},
+ {name: "status", parameterType: ParameterType.UINT8},
+ {name: "scanchannels", parameterType: ParameterType.UINT32},
+ {name: "totaltransmissions", parameterType: ParameterType.UINT16},
+ {name: "transmissionfailures", parameterType: ParameterType.UINT16},
+ {name: "channelcount", parameterType: ParameterType.UINT8},
+ {name: "energyvalues", parameterType: ParameterType.LIST_UINT8},
+ ],
+ },
  {
  name: "mgmtPermitJoinRsp",
  ID: 182,

zigpy_cc/zigbee/application.py

@@ -26,6 +26,7 @@
  "ieeeAddrReq": (ZDOCmd.IEEE_addr_req, 0),
  "matchDescReq": (ZDOCmd.Match_Desc_req, 2),
  "endDeviceAnnceInd": (ZDOCmd.Device_annce, 2),
+ "mgmtLqiRsp": (ZDOCmd.Mgmt_Lqi_rsp, 2),
  "mgmtPermitJoinReq": (ZDOCmd.Mgmt_Permit_Joining_req, 3),
  "mgmtPermitJoinRsp": (ZDOCmd.Mgmt_Permit_Joining_rsp, 2),
  "nodeDescRsp": (ZDOCmd.Node_Desc_rsp, 2),
@@ -105,7 +106,7 @@ async def form_network(self, channel=15, pan_id=None, extended_pan_id=None):
  options = NetworkOptions()
  backupPath = ""
  status = await start_znp(
- self._api, self.version["product"], options, 0x0b84, backupPath
+ self._api, self.version["product"], options, 0x0B84, backupPath
  )
  LOGGER.debug("ZNP started, status: %s", status)
 
@@ -326,7 +327,10 @@ def handle_znp(self, obj: ZpiObject):
 
  else:
  LOGGER.warning(
- "Unhandled message: %s %s %s", t.CommandType(obj.type), t.Subsystem(obj.subsystem), obj.command
+ "Unhandled message: %s %s %s",
+ t.CommandType(obj.type),
+ t.Subsystem(obj.subsystem),
+ obj.command,
  )
  return
 

zigpy_cc/zigbee/start_znp.py

@@ -54,8 +54,9 @@ def __init__(self, **kwargs) -> None:
  appnuminclusters=1,
  appoutclusterlist=[Ota.cluster_id],
  ),
- # Insta/Jung/Gira: OTA fallback EP (since it's buggy in firmware 10023202 when it tries to find a matching EP for
- # OTA - it queries for ZLL profile, but then contacts with HA profile)
+ # Insta/Jung/Gira: OTA fallback EP (since it's buggy in firmware 10023202
+ # when it tries to find a matching EP for OTA - it queries for ZLL profile,
+ # but then contacts with HA profile)
  Endpoint(endpoint=47, appprofid=0x0104),
  Endpoint(endpoint=242, appprofid=0xA1E0),
 ]
@@ -113,9 +114,6 @@ async def needsToBeInitialised(znp: API, version, options):
  )
  except AssertionError as e:
  if version == ZnpVersion.zStack30x or version == ZnpVersion.zStack3x0:
- # Zigbee-herdsman =< 0.6.5 didn't set the panID and extendedPanID on zStack 3.
- # As we are now checking it, it would trigger a reinitialise which will cause users
- # to lose their network. Therefore we are ignoring this case.
  # When the panID has never been set, it will be [0xFF, 0xFF].
  result = await znp.request(
  Subsystem.SYS, "osalNvRead", Items.panID(options.panID)
@@ -242,15 +240,27 @@ async def addToGroup(znp: API, endpoint: int, group: int):
  await znp.request(
  Subsystem.ZDO,
  "extAddGroup",
- {"endpoint": endpoint, "groupid": group, "namelen": 0, "groupname": bytes([])},
+ {
+ "endpoint": endpoint,
+ "groupid": group,
+ "namelen": 0,
+ "groupname": bytes([]),
+ },
  )
 
 
-async def start_znp(znp: API, version, options: NetworkOptions, greenPowerGroup: int, backupPath=""):
+async def start_znp(
+ znp: API, version, options: NetworkOptions, greenPowerGroup: int, backupPath=""
+):
  result = "resumed"
 
  try:
- await validate_item(znp, Items.znpHasConfigured(version), "hasConfigured", expected_status=[0, 2])
+ await validate_item(
+ znp,
+ Items.znpHasConfigured(version),
+ "hasConfigured",
+ expected_status=[0, 2],
+ )
  hasConfigured = True
  except (AssertionError, CommandError):
  hasConfigured = False

zigpy_cc/zpi_object.py

@@ -108,7 +108,9 @@ def from_cluster(
  "data": data,
  }
  elif name == "mgmtPermitJoinReq":
- addrmode = 0x0F if nwk == BroadcastAddress.ALL_ROUTERS_AND_COORDINATOR else 0x02
+ addrmode = (
+ 0x0F if nwk == BroadcastAddress.ALL_ROUTERS_AND_COORDINATOR else 0x02
+ )
  payload = cls.read_parameters(
  bytes([addrmode]) + nwk.to_bytes(2, "little") + data[1:], parameters
  )