Merge branch 'main' into dev

Author: mdeweerd

custom_components/zha_toolkit/__init__.py

@@ -15,11 +15,9 @@
 from zigpy import types as t
 from zigpy.exceptions import DeliveryError
 
-from . import const as c
 from . import params as PARDEFS
 from . import utils as u
-
-DOMAIN = c.DOMAIN
+from .const import DOMAIN
 
 DEPENDENCIES = ["zha"]
 

custom_components/zha_toolkit/znp.py

@@ -1,11 +1,26 @@
+import json
 import logging
+from datetime import datetime
 
+import aiofiles
 from zigpy import types as t
 
 from . import utils as u
 
-# from zigpy.zcl import foundation
-# import zigpy.zcl as zcl
+try:
+    from zigpy_znp.tools.common import validate_backup_json
+    from zigpy_znp.tools.network_backup import backup_network
+    from zigpy_znp.tools.network_restore import json_backup_to_zigpy_state
+    from zigpy_znp.tools.nvram_read import nvram_read
+    from zigpy_znp.tools.nvram_reset import nvram_reset
+    from zigpy_znp.tools.nvram_write import nvram_write
+except ImportError:
+    backup_network = None
+    nvram_read = None
+    nvram_write = None
+    nvram_reset = None
+    validate_backup_json = None
+    json_backup_to_zigpy_state = None
 
 LOGGER = logging.getLogger(__name__)
 
@@ -22,10 +37,10 @@ async def znp_backup(
         LOGGER.debug(msg)
         raise ValueError(msg)
 
-    # Import stuff we need
-    import json
-
-    from zigpy_znp.tools.network_backup import backup_network
+    if backup_network is None:
+        msg = "ZNP tools not available (backup_network)"
+        LOGGER.debug(msg)
+        raise RuntimeError(msg)
 
     # Get backup information
     backup_obj = await backup_network(app._znp)
@@ -44,8 +59,8 @@ async def znp_backup(
     event_data["backup_file"] = fname
 
     LOGGER.debug("Writing to %s", fname)
-    with open(fname, "w", encoding="utf_8") as f:
-        f.write(json.dumps(backup_obj, indent=4))
+    async with aiofiles.open(fname, "w", encoding="utf_8") as f:
+        await f.write(json.dumps(backup_obj, indent=4))
 
 
 async def znp_restore(
@@ -58,6 +73,11 @@ async def znp_restore(
         LOGGER.debug(msg)
         raise ValueError(msg)
 
+    if validate_backup_json is None or json_backup_to_zigpy_state is None:
+        msg = "ZNP tools not available (validate_backup_json)"
+        LOGGER.debug(msg)
+        raise RuntimeError(msg)
+
     # Get/set parameters
 
     # command_data (data):
@@ -70,30 +90,22 @@ async def znp_restore(
 
     counter_increment = t.uint32_t(counter_increment)
 
-    from datetime import datetime
-
     current_datetime = datetime.now().strftime("_%Y%m%d_%H%M%S")
 
     # Safety: backup current configuration
     await znp_backup(
         app, listener, ieee, cmd, current_datetime, service, params, event_data
     )
 
-    # Import stuff we need for restoring
-    import json
-
-    from zigpy_znp.tools.common import validate_backup_json
-    from zigpy_znp.tools.network_restore import json_backup_to_zigpy_state
-
     # Set name with regards to local path
     fname = u.get_local_dir() + "nwk_backup.json"
     LOGGER.info("Restore from '%s'", fname)
 
     event_data["restore_file"] = fname
 
     # Read backup file
-    with open(fname, encoding="utf_8") as f:
-        backup = json.load(f)
+    async with aiofiles.open(fname, encoding="utf_8") as f:
+        backup = json.loads(await f.read())
 
     # validate the backup file
     LOGGER.info("Validating backup contents")
@@ -136,10 +148,10 @@ async def znp_nvram_backup(
         LOGGER.debug(msg)
         raise ValueError(msg)
 
-    # Store backup information to file
-    import json
-
-    from zigpy_znp.tools.nvram_read import nvram_read
+    if nvram_read is None:
+        msg = "ZNP tools not available (nvram_read)"
+        LOGGER.debug(msg)
+        raise RuntimeError(msg)
 
     # Set name with regards to local path
     out_dir = u.get_local_dir()
@@ -154,8 +166,8 @@ async def znp_nvram_backup(
     fname = out_dir + "nvram_backup" + str(data) + ".json"
 
     LOGGER.info("Saving NVRAM to '%s'", fname)
-    with open(fname, "w", encoding="utf_8") as f:
-        f.write(json.dumps(backup_obj, indent=4))
+    async with aiofiles.open(fname, "w", encoding="utf_8") as f:
+        await f.write(json.dumps(backup_obj, indent=4))
     LOGGER.info("NVRAM backup saved to '%s'", fname)
 
 
@@ -169,19 +181,17 @@ async def znp_nvram_restore(
         LOGGER.debug(msg)
         raise ValueError(msg)
 
-    # Safety: backup current configuration
-    from datetime import datetime
+    if nvram_write is None:
+        msg = "ZNP tools not available (nvram_write)"
+        LOGGER.debug(msg)
+        raise RuntimeError(msg)
 
     current_datetime = datetime.now().strftime("_%Y%m%d_%H%M%S")
     await znp_nvram_backup(
         app, listener, ieee, cmd, current_datetime, service, params, event_data
     )
 
     # Restore NVRAM backup from file
-    import json
-
-    from zigpy_znp.tools.nvram_write import nvram_write
-
     # Set name with regards to local path
     out_dir = u.get_local_dir()
 
@@ -192,8 +202,8 @@ async def znp_nvram_restore(
     fname = out_dir + "nvram_backup" + str(data) + ".json"
 
     LOGGER.info("Restoring NVRAM from '%s'", fname)
-    with open(fname, "w", encoding="utf_8") as f:
-        nvram_obj = json.load(f)
+    async with aiofiles.open(fname, "r", encoding="utf_8") as f:
+        nvram_obj = json.loads(await f.read())
 
     await nvram_write(app._znp, nvram_obj)
     LOGGER.info("Restored NVRAM from '%s'", fname)
@@ -211,7 +221,10 @@ async def znp_nvram_reset(
         LOGGER.debug(msg)
         raise ValueError(msg)
 
-    from datetime import datetime
+    if nvram_reset is None:
+        msg = "ZNP tools not available (nvram_reset)"
+        LOGGER.debug(msg)
+        raise RuntimeError(msg)
 
     current_datetime = datetime.now().strftime("_%Y%m%d_%H%M%S")
 
@@ -220,9 +233,6 @@ async def znp_nvram_reset(
         app, listener, ieee, cmd, current_datetime, service, params, event_data
     )
 
-    # Import stuff we need for resetting
-    from zigpy_znp.tools.nvram_reset import nvram_reset
-
     # Write back information from backup
     LOGGER.info("Reset NVRAM")
     await nvram_reset(app._znp)