EZSP v14

bellows/cli/backup.py

@@ -85,10 +85,10 @@ async def backup(ctx):
 async def _backup(ezsp):
     (status,) = await ezsp.networkInit()
     LOGGER.debug("Network init status: %s", status)
-    assert status == t.EmberStatus.SUCCESS
+    assert t.sl_Status.from_ember_status(status) == t.sl_Status.OK
 
     (status, node_type, network) = await ezsp.getNetworkParameters()
-    assert status == t.EmberStatus.SUCCESS
+    assert t.sl_Status.from_ember_status(status) == t.sl_Status.OK
     assert node_type == ezsp.types.EmberNodeType.COORDINATOR
     LOGGER.debug("Network params: %s", network)
 
@@ -112,7 +112,7 @@ async def _backup(ezsp):
         (ATTR_KEY_NWK, ezsp.types.EmberKeyType.CURRENT_NETWORK_KEY),
     ):
         (status, key) = await ezsp.getKey(key_type)
-        assert status == t.EmberStatus.SUCCESS
+        assert t.sl_Status.from_ember_status(status) == t.sl_Status.OK
         LOGGER.debug("%s key: %s", key_name, key)
         result[key_name] = key.as_dict()
         result[key_name][ATTR_KEY_PARTNER] = str(key.partnerEUI64)
@@ -248,7 +248,7 @@ async def _restore(
 
     (status,) = await ezsp.setInitialSecurityState(init_sec_state)
     LOGGER.debug("Set initial security state: %s", status)
-    assert status == t.EmberStatus.SUCCESS
+    assert t.sl_Status.from_ember_status(status) == t.sl_Status.OK
 
     if backup_data[ATTR_KEY_TABLE]:
         await _restore_keys(ezsp, backup_data[ATTR_KEY_TABLE])
@@ -259,15 +259,15 @@ async def _restore(
         t.uint32_t(network_key[ATTR_KEY_FRAME_COUNTER_OUT]).serialize(),
     )
     LOGGER.debug("Set network frame counter: %s", status)
-    assert status == t.EmberStatus.SUCCESS
+    assert t.sl_Status.from_ember_status(status) == t.sl_Status.OK
 
     tc_key = backup_data[ATTR_KEY_GLOBAL]
     (status,) = await ezsp.setValue(
         ezsp.types.EzspValueId.VALUE_APS_FRAME_COUNTER,
         t.uint32_t(tc_key[ATTR_KEY_FRAME_COUNTER_OUT]).serialize(),
     )
     LOGGER.debug("Set network frame counter: %s", status)
-    assert status == t.EmberStatus.SUCCESS
+    assert t.sl_Status.from_ember_status(status) == t.sl_Status.OK
 
     await _form_network(ezsp, backup_data)
     await asyncio.sleep(2)
@@ -279,7 +279,7 @@ async def _restore_keys(ezsp, key_table):
     (status,) = await ezsp.setConfigurationValue(
         ezsp.types.EzspConfigId.CONFIG_KEY_TABLE_SIZE, len(key_table)
     )
-    assert status == t.EmberStatus.SUCCESS
+    assert t.sl_Status.from_ember_status(status) == t.sl_Status.OK
 
     for key in key_table:
         is_link_key = key[ATTR_KEY_TYPE] in (
@@ -312,7 +312,7 @@ async def _form_network(ezsp, backup_data):
 
     (status,) = await ezsp.setValue(ezsp.types.EzspValueId.VALUE_STACK_TOKEN_WRITING, 1)
     LOGGER.debug("Set token writing: %s", status)
-    assert status == t.EmberStatus.SUCCESS
+    assert t.sl_Status.from_ember_status(status) == t.sl_Status.OK
 
 
 async def _update_nwk_id(ezsp, nwk_update_id):
@@ -338,7 +338,7 @@ async def _update_nwk_id(ezsp, nwk_update_id):
         0x01,
         payload,
     )
-    assert status == t.EmberStatus.SUCCESS
+    assert t.sl_Status.from_ember_status(status) == t.sl_Status.OK
     await asyncio.sleep(1)
 
 

bellows/ezsp/__init__.py

@@ -27,9 +27,9 @@
 import bellows.types as t
 import bellows.uart
 
-from . import v4, v5, v6, v7, v8, v9, v10, v11, v12, v13
+from . import v4, v5, v6, v7, v8, v9, v10, v11, v12, v13, v14
 
-EZSP_LATEST = v13.EZSPv13.VERSION
+EZSP_LATEST = v14.EZSPv14.VERSION
 LOGGER = logging.getLogger(__name__)
 MTOR_MIN_INTERVAL = 60
 MTOR_MAX_INTERVAL = 3600
@@ -56,6 +56,7 @@ class EZSP:
         v11.EZSPv11.VERSION: v11.EZSPv11,
         v12.EZSPv12.VERSION: v12.EZSPv12,
         v13.EZSPv13.VERSION: v13.EZSPv13,
+        v14.EZSPv14.VERSION: v14.EZSPv14,
     }
 
     def __init__(self, device_config: dict):
@@ -68,7 +69,7 @@ def __init__(self, device_config: dict):
         self._send_sem = PriorityDynamicBoundedSemaphore(value=MAX_COMMAND_CONCURRENCY)
 
         self._stack_status_listeners: collections.defaultdict[
-            t.EmberStatus, list[asyncio.Future]
+            t.sl_Status, list[asyncio.Future]
         ] = collections.defaultdict(list)
 
         self.add_callback(self.stack_status_callback)
@@ -78,13 +79,13 @@ def stack_status_callback(self, frame_name: str, args: list[Any]) -> None:
         if frame_name != "stackStatusHandler":
             return
 
-        status = args[0]
+        status = t.sl_Status.from_ember_status(args[0])
 
         for listener in self._stack_status_listeners[status]:
             listener.set_result(status)
 
     @contextlib.contextmanager
-    def wait_for_stack_status(self, status: t.EmberStatus) -> Generator[asyncio.Future]:
+    def wait_for_stack_status(self, status: t.sl_Status) -> Generator[asyncio.Future]:
         """Waits for a `stackStatusHandler` to come in with the provided status."""
         listeners = self._stack_status_listeners[status]
 
@@ -228,10 +229,10 @@ def cb(frame_name, response):
         cbid = self.add_callback(cb)
         try:
             v = await self._command(name, *args)
-            if v[0] != t.EmberStatus.SUCCESS:
+            if t.sl_Status.from_ember_status(v[0]) != t.sl_Status.OK:
                 raise Exception(v)
             v = await fut
-            if v[spos] != t.EmberStatus.SUCCESS:
+            if t.sl_Status.from_ember_status(v[spos]) != t.sl_Status.OK:
                 raise Exception(v)
         finally:
             self.remove_callback(cbid)
@@ -267,9 +268,9 @@ async def leaveNetwork(self, timeout: float | int = NETWORK_OPS_TIMEOUT) -> None
         """Send leaveNetwork command and wait for stackStatusHandler frame."""
         stack_status = asyncio.Future()
 
-        with self.wait_for_stack_status(t.EmberStatus.NETWORK_DOWN) as stack_status:
+        with self.wait_for_stack_status(t.sl_Status.NETWORK_DOWN) as stack_status:
             (status,) = await self._command("leaveNetwork")
-            if status != t.EmberStatus.SUCCESS:
+            if status != t.sl_Status.OK:
                 raise EzspError(f"failed to leave network: {status.name}")
 
             async with asyncio_timeout(timeout):
@@ -302,10 +303,10 @@ def __getattr__(self, name: str) -> Callable:
         return functools.partial(self._command, name)
 
     async def formNetwork(self, parameters: t.EmberNetworkParameters) -> None:
-        with self.wait_for_stack_status(t.EmberStatus.NETWORK_UP) as stack_status:
+        with self.wait_for_stack_status(t.sl_Status.NETWORK_UP) as stack_status:
             v = await self._command("formNetwork", parameters)
 
-            if v[0] != self.types.EmberStatus.SUCCESS:
+            if t.sl_Status.from_ember_status(v[0]) != t.sl_Status.OK:
                 raise zigpy.exceptions.FormationFailure(f"Failure forming network: {v}")
 
             async with asyncio_timeout(NETWORK_OPS_TIMEOUT):
@@ -361,7 +362,7 @@ async def get_board_info(
         )
         version = None
 
-        if status == t.EmberStatus.SUCCESS:
+        if t.sl_Status.from_ember_status(status) == t.sl_Status.OK:
             build, ver_info_bytes = t.uint16_t.deserialize(ver_info_bytes)
             major, ver_info_bytes = t.uint8_t.deserialize(ver_info_bytes)
             minor, ver_info_bytes = t.uint8_t.deserialize(ver_info_bytes)
@@ -388,7 +389,7 @@ async def _get_nv3_restored_eui64_key(self) -> t.NV3KeyId | None:
                 # is not implemented in the firmware
                 return None
 
-            if rsp.status == t.EmberStatus.SUCCESS:
+            if t.sl_Status.from_ember_status(rsp.status) == t.sl_Status.OK:
                 nv3_restored_eui64, _ = t.EUI64.deserialize(rsp.value)
                 LOGGER.debug("NV3 restored EUI64: %s=%s", key, nv3_restored_eui64)
 
@@ -434,7 +435,7 @@ async def reset_custom_eui64(self) -> None:
             0,
             t.LVBytes32(t.EUI64.convert("FF:FF:FF:FF:FF:FF:FF:FF").serialize()),
         )
-        assert status == t.EmberStatus.SUCCESS
+        assert t.sl_Status.from_ember_status(status) == t.sl_Status.OK
 
     async def write_custom_eui64(
         self, ieee: t.EUI64, *, burn_into_userdata: bool = False
@@ -460,12 +461,12 @@ async def write_custom_eui64(
                 0,
                 t.LVBytes32(ieee.serialize()),
             )
-            assert status == t.EmberStatus.SUCCESS
+            assert t.sl_Status.from_ember_status(status) == t.sl_Status.OK
         elif mfg_custom_eui64 is None and burn_into_userdata:
             (status,) = await self.setMfgToken(
                 t.EzspMfgTokenId.MFG_CUSTOM_EUI_64, ieee.serialize()
             )
-            assert status == t.EmberStatus.SUCCESS
+            assert t.sl_Status.from_ember_status(status) == t.sl_Status.OK
         elif mfg_custom_eui64 is None and not burn_into_userdata:
             raise EzspError(
                 f"Firmware does not support NV3 tokens. Custom IEEE {ieee} will not be"
@@ -507,7 +508,7 @@ async def set_source_routing(self) -> None:
             0,
         )
         LOGGER.debug("Set concentrator type: %s", res)
-        if res[0] != self.types.EmberStatus.SUCCESS:
+        if t.sl_Status.from_ember_status(res[0]) != t.sl_Status.OK:
             LOGGER.warning("Couldn't set concentrator type %s: %s", True, res)
 
         if self._ezsp_version >= 8:
@@ -579,7 +580,7 @@ async def write_config(self, config: dict) -> None:
             # XXX: A read failure does not mean the value is not writeable!
             status, current_value = await self.getValue(cfg.value_id)
 
-            if status == self.types.EmberStatus.SUCCESS:
+            if t.sl_Status.from_ember_status(status) == t.sl_Status.OK:
                 current_value, _ = type(cfg.value).deserialize(current_value)
             else:
                 current_value = None
@@ -593,7 +594,7 @@ async def write_config(self, config: dict) -> None:
 
             (status,) = await self.setValue(cfg.value_id, cfg.value.serialize())
 
-            if status != self.types.EmberStatus.SUCCESS:
+            if t.sl_Status.from_ember_status(status) != t.sl_Status.OK:
                 LOGGER.debug(
                     "Could not set value %s = %s: %s",
                     cfg.value_id.name,
@@ -608,7 +609,7 @@ async def write_config(self, config: dict) -> None:
 
             # Only grow some config entries, all others should be set
             if (
-                status == self.types.EmberStatus.SUCCESS
+                t.sl_Status.from_ember_status(status) == t.sl_Status.OK
                 and cfg.minimum
                 and current_value >= cfg.value
             ):
@@ -628,7 +629,7 @@ async def write_config(self, config: dict) -> None:
             )
 
             (status,) = await self.setConfigurationValue(cfg.config_id, cfg.value)
-            if status != self.types.EmberStatus.SUCCESS:
+            if t.sl_Status.from_ember_status(status) != t.sl_Status.OK:
                 LOGGER.debug(
                     "Could not set config %s = %s: %s",
                     cfg.config_id,

bellows/ezsp/config.py

@@ -131,4 +131,5 @@ class ValueConfig:
     11: DEFAULT_CONFIG_NEW,
     12: DEFAULT_CONFIG_NEW,
     13: DEFAULT_CONFIG_NEW,
+    14: DEFAULT_CONFIG_NEW,
 }

bellows/ezsp/protocol.py

@@ -6,7 +6,9 @@
 import functools
 import logging
 import sys
-from typing import TYPE_CHECKING, Any, Callable
+from typing import TYPE_CHECKING, Any, AsyncGenerator, Callable, Iterable
+
+import zigpy.state
 
 if sys.version_info[:2] < (3, 11):
     from async_timeout import timeout as asyncio_timeout  # pragma: no cover
@@ -56,14 +58,6 @@ def _ezsp_frame_rx(self, data: bytes) -> tuple[int, int, bytes]:
     def _ezsp_frame_tx(self, name: str) -> bytes:
         """Serialize the named frame."""
 
-    async def pre_permit(self, time_s: int) -> None:
-        """Schedule task before allowing new joins."""
-
-    async def add_transient_link_key(
-        self, ieee: t.EUI64, key: t.KeyData
-    ) -> t.EmberStatus:
-        """Add a transient link key."""
-
     async def command(self, name, *args) -> Any:
         """Serialize command and send it."""
         LOGGER.debug("Sending command  %s: %s", name, args)
@@ -85,7 +79,9 @@ async def update_policies(self, policy_config: dict) -> None:
 
         for policy, value in policies.items():
             (status,) = await self.setPolicy(self.types.EzspPolicyId[policy], value)
-            assert status == self.types.EmberStatus.SUCCESS  # TODO: Better check
+            assert (
+                t.sl_Status.from_ember_status(status) == t.sl_Status.OK
+            )  # TODO: Better check
 
     def __call__(self, data: bytes) -> None:
         """Handler for received data frame."""
@@ -148,3 +144,90 @@ def __getattr__(self, name: str) -> Callable:
             raise AttributeError(f"{name} not found in COMMANDS")
 
         return functools.partial(self.command, name)
+
+    async def pre_permit(self, time_s: int) -> None:
+        """Schedule task before allowing new joins."""
+
+    async def add_transient_link_key(
+        self, ieee: t.EUI64, key: t.KeyData
+    ) -> t.sl_Status:
+        """Add a transient link key."""
+
+    @abc.abstractmethod
+    async def read_child_data(
+        self,
+    ) -> AsyncGenerator[tuple[t.NWK, t.EUI64, t.EmberNodeType], None]:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    async def read_link_keys(self) -> AsyncGenerator[zigpy.state.Key, None]:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    async def read_address_table(self) -> AsyncGenerator[tuple[t.NWK, t.EUI64], None]:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    async def get_network_key(self) -> zigpy.state.Key:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    async def get_tc_link_key(self) -> zigpy.state.Key:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    async def write_nwk_frame_counter(self, frame_counter: t.uint32_t) -> None:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    async def write_aps_frame_counter(self, frame_counter: t.uint32_t) -> None:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    async def write_link_keys(self, keys: Iterable[zigpy.state.Key]) -> None:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    async def write_child_data(self, children: dict[t.EUI64, t.NWK]) -> None:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    async def initialize_network(self) -> t.sl_Status:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    async def factory_reset(self) -> None:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    async def send_unicast(
+        self,
+        nwk: t.NWK,
+        aps_frame: t.EmberApsFrame,
+        message_tag: t.uint8_t,
+        data: bytes,
+    ) -> tuple[t.sl_Status, t.uint8_t]:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    async def send_multicast(
+        self,
+        aps_frame: t.EmberApsFrame,
+        radius: t.uint8_t,
+        non_member_radius: t.uint8_t,
+        message_tag: t.uint8_t,
+        data: bytes,
+    ) -> tuple[t.sl_Status, t.uint8_t]:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    async def send_broadcast(
+        self,
+        address: t.BroadcastAddress,
+        aps_frame: t.EmberApsFrame,
+        radius: t.uint8_t,
+        message_tag: t.uint8_t,
+        aps_sequence: t.uint8_t,
+        data: bytes,
+    ) -> tuple[t.sl_Status, t.uint8_t]:
+        raise NotImplementedError