Implement new radio API

setup.py

@@ -3,7 +3,6 @@
 from setuptools import setup, find_packages
 
 import zigpy_cli
-import zigpy_cli.common
 
 setup(
     name="zigpy-cli",
@@ -18,14 +17,16 @@
     entry_points={"console_scripts": ["zigpy=zigpy_cli.__main__:cli"]},
     packages=find_packages(exclude=["tests", "tests.*"]),
     install_requires=[
-        "zigpy",
         "click",
         "coloredlogs",
         "scapy",
+        "zigpy>=0.47.1",
+        "bellows>=0.31.0",
+        "zigpy-deconz>=0.18.0",
+        "zigpy-znp>=0.8.0",
     ],
     extras_require={
         # [all] pulls in all radio libraries
-        "all": zigpy_cli.common.RADIO_TO_PYPI.values(),
         "testing": [
             "pytest>=5.4.5",
             "pytest-asyncio>=0.12.0",

zigpy_cli/cli.py

@@ -7,7 +7,7 @@
 import click
 import coloredlogs
 
-from zigpy_cli.common import LOG_LEVELS
+from zigpy_cli.const import LOG_LEVELS
 
 LOGGER = logging.getLogger(__name__)
 

zigpy_cli/common.py

@@ -1,77 +1,5 @@
-import logging
-
 import click
 
-TRACE = logging.DEBUG - 5
-logging.addLevelName(TRACE, "TRACE")
-
-
-LOG_LEVELS = [logging.WARNING, logging.INFO, logging.DEBUG, TRACE]
-
-
-RADIO_TO_PACKAGE = {
-    "ezsp": "bellows",
-    "deconz": "zigpy_deconz",
-    "xbee": "zigpy_xbee",
-    "zigate": "zigpy_zigate",
-    "znp": "zigpy_znp",
-}
-
-
-RADIO_LOGGING_CONFIGS = {
-    "ezsp": [
-        {
-            "bellows.zigbee.application": logging.INFO,
-            "bellows.ezsp": logging.INFO,
-        },
-        {
-            "bellows.zigbee.application": logging.DEBUG,
-            "bellows.ezsp": logging.DEBUG,
-        },
-    ],
-    "deconz": [
-        {
-            "zigpy_deconz.zigbee.application": logging.INFO,
-            "zigpy_deconz.api": logging.INFO,
-        },
-        {
-            "zigpy_deconz.zigbee.application": logging.DEBUG,
-            "zigpy_deconz.api": logging.DEBUG,
-        },
-    ],
-    "xbee": [
-        {
-            "zigpy_xbee.zigbee.application": logging.INFO,
-            "zigpy_xbee.api": logging.INFO,
-        },
-        {
-            "zigpy_xbee.zigbee.application": logging.DEBUG,
-            "zigpy_xbee.api": logging.DEBUG,
-        },
-    ],
-    "zigate": [
-        {
-            "zigpy_zigate": logging.INFO,
-        },
-        {
-            "zigpy_zigate": logging.DEBUG,
-        },
-    ],
-    "znp": [
-        {
-            "zigpy_znp": logging.INFO,
-        },
-        {
-            "zigpy_znp": logging.DEBUG,
-        },
-        {
-            "zigpy_znp": TRACE,
-        },
-    ],
-}
-
-RADIO_TO_PYPI = {name: mod.replace("_", "-") for name, mod in RADIO_TO_PACKAGE.items()}
-
 
 class HexOrDecIntParamType(click.ParamType):
     name = "integer"

zigpy_cli/const.py

@@ -0,0 +1,71 @@
+import logging
+
+TRACE = logging.DEBUG - 5
+logging.addLevelName(TRACE, "TRACE")
+
+
+LOG_LEVELS = [logging.WARNING, logging.INFO, logging.DEBUG, TRACE]
+
+
+RADIO_TO_PACKAGE = {
+    "ezsp": "bellows",
+    "deconz": "zigpy_deconz",
+    "xbee": "zigpy_xbee",
+    "zigate": "zigpy_zigate",
+    "znp": "zigpy_znp",
+}
+
+
+RADIO_LOGGING_CONFIGS = {
+    "ezsp": [
+        {
+            "bellows.zigbee.application": logging.INFO,
+            "bellows.ezsp": logging.INFO,
+        },
+        {
+            "bellows.zigbee.application": logging.DEBUG,
+            "bellows.ezsp": logging.DEBUG,
+        },
+    ],
+    "deconz": [
+        {
+            "zigpy_deconz.zigbee.application": logging.INFO,
+            "zigpy_deconz.api": logging.INFO,
+        },
+        {
+            "zigpy_deconz.zigbee.application": logging.DEBUG,
+            "zigpy_deconz.api": logging.DEBUG,
+        },
+    ],
+    "xbee": [
+        {
+            "zigpy_xbee.zigbee.application": logging.INFO,
+            "zigpy_xbee.api": logging.INFO,
+        },
+        {
+            "zigpy_xbee.zigbee.application": logging.DEBUG,
+            "zigpy_xbee.api": logging.DEBUG,
+        },
+    ],
+    "zigate": [
+        {
+            "zigpy_zigate": logging.INFO,
+        },
+        {
+            "zigpy_zigate": logging.DEBUG,
+        },
+    ],
+    "znp": [
+        {
+            "zigpy_znp": logging.INFO,
+        },
+        {
+            "zigpy_znp": logging.DEBUG,
+        },
+        {
+            "zigpy_znp": TRACE,
+        },
+    ],
+}
+
+RADIO_TO_PYPI = {name: mod.replace("_", "-") for name, mod in RADIO_TO_PACKAGE.items()}

zigpy_cli/radio.py

@@ -1,23 +1,19 @@
 from __future__ import annotations
 
+import json
 import logging
 import importlib
 import collections
+import importlib.util
 
 import click
 import zigpy.state
 import zigpy.types
-import zigpy.config as conf
 import zigpy.zdo.types
 
 from zigpy_cli.cli import cli, click_coroutine
-from zigpy_cli.utils import format_bytes
-from zigpy_cli.common import (
-    RADIO_TO_PYPI,
-    HEX_OR_DEC_INT,
-    RADIO_TO_PACKAGE,
-    RADIO_LOGGING_CONFIGS,
-)
+from zigpy_cli.const import RADIO_TO_PYPI, RADIO_TO_PACKAGE, RADIO_LOGGING_CONFIGS
+from zigpy_cli.common import HEX_OR_DEC_INT
 
 LOGGER = logging.getLogger(__name__)
 
@@ -26,8 +22,9 @@
 @click.pass_context
 @click.argument("radio", type=click.Choice(list(RADIO_TO_PACKAGE.keys())))
 @click.argument("port", type=str)
+@click.option("--baudrate", type=int, default=None)
 @click_coroutine
-async def radio(ctx, radio, port):
+async def radio(ctx, radio, port, baudrate=None):
     # Setup logging for the radio
     verbose = ctx.parent.params["verbose"]
     logging_configs = RADIO_LOGGING_CONFIGS[radio]
@@ -36,26 +33,25 @@ async def radio(ctx, radio, port):
     for logger, level in logging_config.items():
         logging.getLogger(logger).setLevel(level)
 
-    # Import the radio library
     module = RADIO_TO_PACKAGE[radio] + ".zigbee.application"
 
-    try:
-        radio_module = importlib.import_module(module)
-    except ImportError:
+    # Catching just `ImportError` masks dependency errors and is annoying
+    if importlib.util.find_spec(module) is None:
         raise click.ClickException(
             f"Radio module for {radio!r} is not installed."
             f" Install it with `pip install {RADIO_TO_PYPI[radio]}`."
         )
 
+    # Import the radio library
+    radio_module = importlib.import_module(module)
+
     # Start the radio
     app_cls = radio_module.ControllerApplication
-    config = app_cls.SCHEMA(
-        {
-            conf.CONF_DEVICE: {
-                conf.CONF_DEVICE_PATH: port,
-            },
-        }
-    )
+    config = app_cls.SCHEMA({"device": {"path": port}})
+
+    if baudrate is not None:
+        config["device"]["baudrate"] = baudrate
+
     app = app_cls(config)
 
     ctx.obj = app
@@ -66,36 +62,59 @@ async def radio(ctx, radio, port):
 @click_coroutine
 async def radio_cleanup(app):
     try:
-        await app.pre_shutdown()
+        await app.shutdown()
     except RuntimeError:
         LOGGER.warning("Caught an exception when shutting down app", exc_info=True)
 
 
-def dump_app_info(app):
-    if app.pan_id is not None:
-        print(f"PAN ID:                0x{app.pan_id:04X}")
+@radio.command()
+@click.pass_obj
+@click_coroutine
+async def info(app):
+    await app.connect()
+    await app.load_network_info(load_devices=False)
+
+    print(f"PAN ID:                0x{app.state.network_info.pan_id:04X}")
+    print(f"Extended PAN ID:       {app.state.network_info.extended_pan_id}")
+    print(f"Channel:               {app.state.network_info.channel}")
+    print(f"Channel mask:          {list(app.state.network_info.channel_mask)}")
+    print(f"NWK update ID:         {app.state.network_info.nwk_update_id}")
+    print(f"Device IEEE:           {app.state.node_info.ieee}")
+    print(f"Device NWK:            0x{app.state.node_info.nwk:04X}")
+    print(f"Network key:           {app.state.network_info.network_key.key}")
+    print(f"Network key sequence:  {app.state.network_info.network_key.seq}")
+    print(f"Network key counter:   {app.state.network_info.network_key.tx_counter}")
 
-    print(f"Extended PAN ID:       {app.extended_pan_id}")
-    print(f"Channel:               {app.channel}")
 
-    if app.channels is not None:
-        print(f"Channel mask:          {list(app.channels)}")
+@radio.command()
+@click.argument("output", type=click.File("w"))
+@click.pass_obj
+@click_coroutine
+async def backup(app, output):
+    await app.connect()
+    await app.load_network_info(load_devices=True)
 
-    print(f"NWK update ID:         {app.nwk_update_id}")
-    print(f"Device IEEE:           {app.ieee}")
-    print(f"Device NWK:            0x{app.nwk:04X}")
+    obj = zigpy.state.network_state_to_json(
+        network_info=app.state.network_info,
+        node_info=app.state.node_info,
+    )
 
-    if getattr(app, "network_key", None) is not None:
-        print(f"Network key:           {format_bytes(app.network_key)}")
-        print(f"Network key sequence:  {app.network_key_seq}")
+    output.write(json.dumps(obj, indent=4))
 
 
 @radio.command()
+@click.argument("input", type=click.File("r"))
+@click.option("-c", "--frame-counter-increment", type=int, default=5000)
 @click.pass_obj
 @click_coroutine
-async def info(app):
-    await app.startup(auto_form=False)
-    dump_app_info(app)
+async def restore(app, frame_counter_increment, input):
+    obj = json.load(input)
+
+    network_info, node_info = zigpy.state.json_to_network_state(obj)
+    network_info.network_key.tx_counter += frame_counter_increment
+
+    await app.connect()
+    await app.write_network_info(network_info=network_info, node_info=node_info)
 
 
 @radio.command()
@@ -104,7 +123,6 @@ async def info(app):
 async def form(app):
     await app.startup(auto_form=True)
     await app.form_network()
-    dump_app_info(app)
 
 
 @radio.command()
@@ -149,14 +167,23 @@ async def energy_scan(app, nwk):
         print(" + TX on 26 in North America may be with lower power due to regulations")
         print(" + Zigbee channels 15, 20, 25 fall between WiFi channels 1, 6, 11")
         print(" + Some Zigbee devices only join networks on channels 15, 20, and 25")
+        print(" + Current channel is enclosed in [square brackets]")
         print("------------------------------------------------")
 
         for channel, energies in channel_energies.items():
             count = sum(energies)
             asterisk = "*" if channel == 26 else " "
 
+            if channel == app.state.network_info.channel:
+                bracket_open = "["
+                bracket_close = "]"
+            else:
+                bracket_open = " "
+                bracket_close = " "
+
             print(
-                f" - {channel:>02}{asterisk}  {count / total:>7.2%}  "
+                f" - {bracket_open}{channel:>02}{asterisk}{bracket_close}"
+                + f"   {count / total:>7.2%}  "
                 + "#" * int(100 * count / total)
             )