From 0c550c4d485e7c5a838147c9af50b0fd7109675c Mon Sep 17 00:00:00 2001 From: Chen Xinyang Date: Mon, 6 Jan 2025 10:16:24 +0200 Subject: [PATCH] Fix inline import and sync file io raising hass warnings --- custom_components/zha_toolkit/znp.py | 82 ++++++++++++++++------------ 1 file changed, 46 insertions(+), 36 deletions(-) diff --git a/custom_components/zha_toolkit/znp.py b/custom_components/zha_toolkit/znp.py index fd6544a..69a4cfe 100644 --- a/custom_components/zha_toolkit/znp.py +++ b/custom_components/zha_toolkit/znp.py @@ -1,11 +1,26 @@ import logging +import aiofiles +import json +from datetime import datetime 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.network_backup import backup_network + from zigpy_znp.tools.nvram_read import nvram_read + from zigpy_znp.tools.nvram_write import nvram_write + from zigpy_znp.tools.nvram_reset import nvram_reset + from zigpy_znp.tools.common import validate_backup_json + from zigpy_znp.tools.network_restore import json_backup_to_zigpy_state +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" + 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" + LOGGER.debug(msg) + raise RuntimeError(msg) + # Get/set parameters # command_data (data): @@ -70,8 +90,6 @@ 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 @@ -79,12 +97,6 @@ async def znp_restore( 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) @@ -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" + 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) @@ -168,9 +180,11 @@ async def znp_nvram_restore( msg = f"'{cmd}' is only available for ZNP" LOGGER.debug(msg) raise ValueError(msg) - - # Safety: backup current configuration - from datetime import datetime + + if nvram_write is None: + msg = "ZNP tools not available" + LOGGER.debug(msg) + raise RuntimeError(msg) current_datetime = datetime.now().strftime("_%Y%m%d_%H%M%S") await znp_nvram_backup( @@ -178,10 +192,6 @@ async def znp_nvram_restore( ) # 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.load(await f.read()) await nvram_write(app._znp, nvram_obj) LOGGER.info("Restored NVRAM from '%s'", fname) @@ -210,8 +220,11 @@ async def znp_nvram_reset( msg = f"'{cmd}' is only available for ZNP" LOGGER.debug(msg) raise ValueError(msg) - - from datetime import datetime + + if nvram_reset is None: + msg = "ZNP tools not available" + 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)