Replace port_path with znp in tools that can run independently

Author: puddly

zigpy_znp/tools/flash_read.py

@@ -13,16 +13,7 @@
 LOGGER = logging.getLogger(__name__)
 
 
-async def read_firmware(radio_path: str) -> bytearray:
-    znp = ZNP(
-        CONFIG_SCHEMA(
-            {"znp_config": {"skip_bootloader": False}, "device": {"path": radio_path}}
-        )
-    )
-
-    # The bootloader handshake must be the very first command
-    await znp.connect(test_port=False)
-
+async def read_firmware(znp: ZNP) -> bytearray:
     try:
         async with async_timeout.timeout(5):
             handshake_rsp = await znp.request_callback_rsp(
@@ -60,8 +51,6 @@ async def read_firmware(radio_path: str) -> bytearray:
 
         data.extend(read_rsp.Data)
 
-    znp.close()
-
     return data
 
 
@@ -76,7 +65,19 @@ async def main(argv):
     )
 
     args = parser.parse_args(argv)
-    data = await read_firmware(args.serial)
+
+    znp = ZNP(
+        CONFIG_SCHEMA(
+            {"znp_config": {"skip_bootloader": False}, "device": {"path": args.serial}}
+        )
+    )
+
+    # The bootloader handshake must be the very first command
+    await znp.connect(test_port=False)
+
+    data = await read_firmware(znp)
+    znp.close()
+
     args.output.write(data)
 
     LOGGER.info("Unplug your adapter to leave bootloader mode!")

zigpy_znp/tools/flash_write.py

@@ -55,7 +55,7 @@ def get_firmware_crcs(firmware: bytes) -> tuple[int, int]:
     return real_crc, compute_crc16(firmware_without_crcs)
 
 
-async def write_firmware(firmware: bytes, radio_path: str, reset_nvram: bool):
+async def write_firmware(znp: ZNP, firmware: bytes, reset_nvram: bool):
     if len(firmware) != c.ubl.IMAGE_SIZE:
         raise ValueError(
             f"Firmware is the wrong size."
@@ -70,15 +70,6 @@ async def write_firmware(firmware: bytes, radio_path: str, reset_nvram: bool):
             f" Expected 0x{expected_crc:04X}, got 0x{computed_crc:04X}"
         )
 
-    znp = ZNP(
-        CONFIG_SCHEMA(
-            {"znp_config": {"skip_bootloader": False}, "device": {"path": radio_path}}
-        )
-    )
-
-    # The bootloader handshake must be the very first command
-    await znp.connect(test_port=False)
-
     try:
         async with async_timeout.timeout(5):
             handshake_rsp = await znp.request_callback_rsp(
@@ -146,8 +137,6 @@ async def write_firmware(firmware: bytes, radio_path: str, reset_nvram: bool):
     else:
         LOGGER.info("Unplug your adapter to leave bootloader mode!")
 
-    znp.close()
-
 
 async def main(argv):
     parser = setup_parser("Write firmware to a radio")
@@ -168,7 +157,18 @@ async def main(argv):
 
     args = parser.parse_args(argv)
 
-    await write_firmware(args.input.read(), args.serial, args.reset)
+    znp = ZNP(
+        CONFIG_SCHEMA(
+            {"znp_config": {"skip_bootloader": False}, "device": {"path": args.serial}}
+        )
+    )
+
+    # The bootloader handshake must be the very first command
+    await znp.connect(test_port=False)
+
+    await write_firmware(znp=znp, firmware=args.input.read(), reset_nvram=args.reset)
+
+    znp.close()
 
 
 if __name__ == "__main__":

zigpy_znp/tools/form_network.py

@@ -14,7 +14,7 @@ async def form_network(radio_path):
     config = ControllerApplication.SCHEMA({"device": {"path": radio_path}})
     app = ControllerApplication(config)
 
-    await app.startup(auto_form=True)
+    await app.startup(force_form=True)
     await app.shutdown()
 
 

zigpy_znp/tools/network_backup.py

@@ -18,10 +18,7 @@
 LOGGER = logging.getLogger(__name__)
 
 
-async def backup_network(radio_path: str) -> t.JSONType:
-    znp = ZNP(ControllerApplication.SCHEMA({"device": {"path": radio_path}}))
-    await znp.connect()
-
+async def backup_network(znp: ZNP) -> t.JSONType:
     try:
         await znp.load_network_info()
     except ValueError as e:
@@ -81,8 +78,6 @@ async def backup_network(radio_path: str) -> t.JSONType:
 
         obj["stack_specific"] = {"zstack": {"tclk_seed": tclk_seed.hex()}}
 
-    znp.close()
-
     # Ensure our generated backup is valid
     validate_backup_json(obj)
 
@@ -96,9 +91,11 @@ async def main(argv: list[str]) -> None:
     )
     args = parser.parse_args(argv)
 
-    backup_obj = await backup_network(
-        radio_path=args.serial,
-    )
+    znp = ZNP(ControllerApplication.SCHEMA({"device": {"path": args.serial}}))
+    await znp.connect()
+
+    backup_obj = await backup_network(znp)
+    znp.close()
 
     args.output.write(json.dumps(backup_obj, indent=4))
 

zigpy_znp/tools/nvram_read.py

@@ -14,10 +14,7 @@
 LOGGER = logging.getLogger(__name__)
 
 
-async def nvram_read(radio_path: str):
-    znp = ZNP(CONFIG_SCHEMA({"device": {"path": radio_path}}))
-    await znp.connect()
-
+async def nvram_read(znp: ZNP):
     data = {}
     data["LEGACY"] = {}
 
@@ -89,8 +86,12 @@ async def main(argv):
 
     args = parser.parse_args(argv)
 
-    obj = await nvram_read(args.serial)
+    znp = ZNP(CONFIG_SCHEMA({"device": {"path": args.serial}}))
+    await znp.connect()
+
+    obj = await nvram_read(znp)
     args.output.write(json.dumps(obj, indent=4) + "\n")
+    znp.close()
 
 
 if __name__ == "__main__":

zigpy_znp/tools/nvram_write.py

@@ -12,11 +12,7 @@
 LOGGER = logging.getLogger(__name__)
 
 
-async def nvram_write(radio_path, backup):
-    znp = ZNP(CONFIG_SCHEMA({"device": {"path": radio_path}}))
-
-    await znp.connect()
-
+async def nvram_write(znp: ZNP, backup):
     # First write the NVRAM items common to all radios
     for nwk_nvid, value in backup["LEGACY"].items():
         if "+" in nwk_nvid:
@@ -58,7 +54,11 @@ async def main(argv):
 
     args = parser.parse_args(argv)
     backup = json.load(args.input)
-    await nvram_write(args.serial, backup)
+
+    znp = ZNP(CONFIG_SCHEMA({"device": {"path": args.serial}}))
+    await znp.connect()
+
+    await nvram_write(znp, backup)
 
 
 if __name__ == "__main__":