[nrf noup] Add zcl_generate to west command

scripts/west/west-commands.yml

@@ -10,3 +10,8 @@ west-commands:
       - name: zap-gui
         class: ZapGui
         help: Run Matter ZCL Advanced Platform (ZAP) GUI
+  - file: scripts/west/zap_append.py
+    commands:
+      - name: zap-append
+        class: ZapAppend
+        help: Append a new custom cluster to the ZCL database

scripts/west/zap_append.py

@@ -0,0 +1,108 @@
+# Copyright (c) 2025 Nordic Semiconductor ASA
+#
+# SPDX-License-Identifier: LicenseRef-Nordic-5-Clause
+
+import argparse
+from pathlib import Path
+import json
+
+from west.commands import WestCommand
+from west import log
+
+from zap_common import DEFAULT_MATTER_PATH, DEFAULT_ZCL_JSON_RELATIVE_PATH
+
+
+def add_cluster_to_zcl(zcl_base: Path, cluster_xml_paths: list, output: Path):
+    """
+    Add the cluster to the ZCL file.
+    """
+
+    try:
+        with open(zcl_base, "r") as zcl_json_base:
+            zcl_json = json.load(zcl_json_base)
+    except IOError as e:
+        raise RuntimeError(f"No such ZCL file: {zcl_base}")
+
+    # If the output file is provided, we would like to generate a new ZCL file, so we must set
+    # the relative paths from the xml.file to the data model directories and manufacturers.xml file.
+    # It is because the base zcl.json file contains the relative paths to itself inside as the "xmlRoot",
+    # and if we create a new zcl file outside the data model directory it will not work properly.
+    if output:
+        roots_replaced = list()
+        replace = False
+
+        # Replace existing paths with the relative to output ones
+        for path in zcl_json.get("xmlRoot"):
+            path = zcl_base.parent.joinpath(Path(path))
+            if not path == "./" and not path == "." and not path.is_relative_to(output.parent):
+                roots_replaced.append(str(path.relative_to(output.parent, walk_up=True)))
+                replace = True
+        if replace:
+            zcl_json["xmlRoot"] = roots_replaced
+
+        # Add the relative path to manufacturers XML
+        manufacturers_xml = zcl_base.parent.joinpath(Path(zcl_json.get("manufacturersXml"))).resolve()
+        if not manufacturers_xml.parent.is_relative_to(output.parent):
+            zcl_json.update({"manufacturersXml": str(manufacturers_xml.relative_to(output.parent, walk_up=True))})
+
+    # Add the new clusters to the ZCL file
+    for cluster in cluster_xml_paths:
+        if not Path(cluster).exists():
+            raise RuntimeError(f"No such cluster file: {cluster}")
+
+        # Get cluster file name
+        file = Path(cluster).name
+        # Get relative path from the cluster file to the output file.
+        relative_path = Path(cluster).absolute().parent.relative_to(output.parent, walk_up=True)
+
+        # We need to add two things:
+        # 1. The absolute path to the directory where the new xml file exists to the xmlRoot array.
+        # 2. The new xml file name to the xmlFile array.
+        if not str(relative_path) in zcl_json.get("xmlRoot"):
+            zcl_json.get("xmlRoot").append(str(relative_path))
+        if not file in zcl_json.get("xmlFile"):
+            zcl_json.get("xmlFile").append(file)
+            log.dbg(f"Successfully added {file}")
+
+    # If output file is not provided, we will edit the existing ZCL file
+    file_to_write = output if output else zcl_base
+
+    # Save the dumped JSON to the output file
+    with open(file_to_write, "w+") as zcl_output:
+        zcl_output.write(json.dumps(zcl_json, indent=4))
+
+
+class ZapAppend(WestCommand):
+    def __init__(self):
+        super().__init__(
+            'zap-append',
+            'Add a new custom cluster to the ZCL Matter data model file',
+            'A tool for adding a custom cluster to the ZCL Matter Data Model file according to the base ZCL file and custom clusters definitions.')
+
+    def do_add_parser(self, parser_adder) -> argparse.ArgumentParser:
+        parser = parser_adder.add_parser(self.name,
+                                         help=self.help,
+                                         formatter_class=argparse.RawDescriptionHelpFormatter,
+                                         description=self.description)
+        parser.add_argument("-b", "--base", type=Path,
+                            help=f"An absolute path to the base zcl.json file. If not provided the path will be set to MATTER/{DEFAULT_ZCL_JSON_RELATIVE_PATH}.")
+        parser.add_argument("-m", "--matter", type=Path, default=DEFAULT_MATTER_PATH,
+                            help=f"An absolute path to the Matter directory. If not set the path with be set to the {DEFAULT_MATTER_PATH}")
+        parser.add_argument("-o", "--output", type=Path,
+                            help=f"Output path to store the generated zcl.json file. If not provided the path will be set to the base zcl.json file (MATTER/{DEFAULT_ZCL_JSON_RELATIVE_PATH}).")
+        parser.add_argument("new_clusters", nargs='+',
+                            help="Paths to the XML files that contain the custom cluster definitions")
+        return parser
+
+    def do_run(self, args, unknown_args) -> None:
+        if not args.base:
+            args.base = args.matter.joinpath(DEFAULT_ZCL_JSON_RELATIVE_PATH)
+        if not args.output:
+            args.output = args.matter.joinpath(DEFAULT_ZCL_JSON_RELATIVE_PATH)
+
+        for cluster in args.new_clusters:
+            if not Path(cluster).exists():
+                log.err(f"No such cluster file: {cluster}")
+                return
+
+        add_cluster_to_zcl(args.base.absolute(), args.new_clusters, args.output.absolute())

scripts/west/zap_common.py

@@ -11,6 +11,8 @@
 import subprocess
 import tempfile
 import wget
+import json
+import signal
 
 from collections import deque
 from pathlib import Path
@@ -20,6 +22,8 @@
 from west import log
 
 DEFAULT_MATTER_PATH = Path(__file__).parents[2]
+DEFAULT_ZCL_JSON_RELATIVE_PATH = Path('src/app/zap-templates/zcl/zcl.json')
+DEFAULT_APP_TEMPLATES_RELATIVE_PATH = Path('src/app/zap-templates/app-templates.json')
 
 
 def find_zap(root: Path = Path.cwd(), max_depth: int = 2):
@@ -77,13 +81,43 @@ def existing_dir_path(arg: str) -> Path:
     raise argparse.ArgumentTypeError(f'invalid directory path: \'{arg}\'')
 
 
+def update_zcl_in_zap(zap_file: Path, zcl_json: Path, app_templates: Path) -> bool:
+    """
+    In the .zap file, there is a relative path to the zcl.json file.
+    Use this function to update zcl.json path if needed.
+    Functions returns True if the path was updated, False otherwise.
+    """
+    updated = False
+
+    with open(zap_file, 'r+') as file:
+        data = json.load(file)
+        packages = data.get("package")
+
+        for package in packages:
+            if package.get("type") == "zcl-properties":
+                if not zcl_json.parent.absolute().is_relative_to(zap_file.parent.absolute()):
+                    package.update({"path": str(zcl_json.absolute().relative_to(zap_file.parent.absolute(), walk_up=True))})
+                    updated = True
+            if package.get("type") == "gen-templates-json":
+                if not app_templates.parent.absolute().is_relative_to(zap_file.parent.absolute()):
+                    package.update({"path": str(app_templates.absolute().relative_to(zap_file.parent.absolute(), walk_up=True))})
+                    updated = True
+
+        file.seek(0)
+        json.dump(data, file, indent=2)
+        file.truncate()
+
+    return updated
+
+
 class ZapInstaller:
     INSTALL_DIR = Path('.zap-install')
     ZAP_URL_PATTERN = 'https://github.com/project-chip/zap/releases/download/v%04d.%02d.%02d-nightly/%s.zip'
 
     def __init__(self, matter_path: Path):
         self.matter_path = matter_path
         self.install_path = matter_path / ZapInstaller.INSTALL_DIR
+        self.current_os = platform.system()
 
         def unzip_darwin(zip: Path, out: Path):
             subprocess.check_call(['unzip', zip, '-d', out])
@@ -93,24 +127,23 @@ def unzip(zip: Path, out: Path):
             f.extractall(out)
             f.close()
 
-        current_os = platform.system()
-        if current_os == 'Linux':
+        if self.current_os == 'Linux':
             self.package = 'zap-linux-x64'
             self.zap_exe = 'zap'
             self.zap_cli_exe = 'zap-cli'
             self.unzip = unzip
-        elif current_os == 'Windows':
+        elif self.current_os == 'Windows':
             self.package = 'zap-win-x64'
             self.zap_exe = 'zap.exe'
             self.zap_cli_exe = 'zap-cli.exe'
             self.unzip = unzip
-        elif current_os == 'Darwin':
+        elif self.current_os == 'Darwin':
             self.package = 'zap-mac-x64'
             self.zap_exe = 'zap.app/Contents/MacOS/zap'
             self.zap_cli_exe = 'zap-cli'
             self.unzip = unzip_darwin
         else:
-            raise RuntimeError(f"Unsupported platform: {current_os}")
+            raise RuntimeError(f"Unsupported platform: {self.current_os}")
 
     def get_install_path(self) -> Path:
         """
@@ -172,9 +205,20 @@ def install_zap(self, version: Tuple[int, int, int]) -> None:
         with tempfile.TemporaryDirectory() as temp_dir:
             url = ZapInstaller.ZAP_URL_PATTERN % (*version, self.package)
             log.inf(f'Downloading {url}...')
+            zip_file_path = str(Path(temp_dir).joinpath(f'{self.package}.zip'))
+
+            # Handle SIGINT and SIGTERM to clean up broken files if the user cancels
+            # the installation
+            def handle_signal(signum, frame):
+                log.inf(f'\nCancelled by user, cleaning up...')
+                shutil.rmtree(self.install_path, ignore_errors=True)
+                exit()
+
+            signal.signal(signal.SIGINT, handle_signal)
+            signal.signal(signal.SIGTERM, handle_signal)
 
             try:
-                wget.download(url, out=temp_dir)
+                wget.download(url, out=zip_file_path)
             except Exception as e:
                 raise RuntimeError(f'Failed to download ZAP package from {url}: {e}')
 
@@ -184,7 +228,7 @@ def install_zap(self, version: Tuple[int, int, int]) -> None:
             log.inf(f'Unzipping ZAP package to {self.install_path}...')
 
             try:
-                self.unzip(Path(temp_dir) / f'{self.package}.zip', self.install_path)
+                self.unzip(zip_file_path, self.install_path)
             except Exception as e:
                 raise RuntimeError(f'Failed to unzip ZAP package: {e}')
 

scripts/west/zap_gui.py

@@ -3,12 +3,15 @@
 # SPDX-License-Identifier: LicenseRef-Nordic-5-Clause
 
 import argparse
+from pathlib import Path
 
 from textwrap import dedent
 
 from west.commands import WestCommand
+from west import log
 
-from zap_common import existing_file_path, existing_dir_path, find_zap, ZapInstaller, DEFAULT_MATTER_PATH
+from zap_common import existing_file_path, existing_dir_path, find_zap, ZapInstaller, DEFAULT_MATTER_PATH, DEFAULT_ZCL_JSON_RELATIVE_PATH, DEFAULT_APP_TEMPLATES_RELATIVE_PATH, update_zcl_in_zap
+from zap_append import add_cluster_to_zcl
 
 
 class ZapGui(WestCommand):
@@ -31,33 +34,55 @@ def do_add_parser(self, parser_adder):
                                          description=self.description)
         parser.add_argument('-z', '--zap-file', type=existing_file_path,
                             help='Path to data model configuration file (*.zap)')
-        parser.add_argument('-j', '--zcl-json', type=existing_file_path,
-                            help='Path to data model definition file (zcl.json)')
+        parser.add_argument('-j', '--zcl-json', type=str,
+                            help='Path to data model definition file (zcl.json). If new clusters are added using --clusters, the new zcl.json file will be created and used.')
         parser.add_argument('-m', '--matter-path', type=existing_dir_path,
-                            default=DEFAULT_MATTER_PATH, help='Path to Matter SDK')
+                            default=DEFAULT_MATTER_PATH, help=f'Path to Matter SDK. Default is set to {DEFAULT_MATTER_PATH}')
+        parser.add_argument('--clusters', nargs='+',
+                            help="Paths to the XML files that contain the external cluster definitions")
+        parser.add_argument('-c', '--cache', type=Path,
+                            help='Path to the custom cache directory. If not provided a temporary directory will be used and cleared after the usage.')
         return parser
 
     def do_run(self, args, unknown_args):
-        if args.zap_file:
-            zap_file_path = args.zap_file
-        else:
-            zap_file_path = find_zap()
+        default_zcl_path = args.matter_path.joinpath(DEFAULT_ZCL_JSON_RELATIVE_PATH)
 
-        if args.zcl_json:
-            zcl_json_path = args.zcl_json.absolute()
-        else:
-            zcl_json_path = args.matter_path / 'src/app/zap-templates/zcl/zcl.json'
+        zap_file_path = args.zap_file or find_zap()
+        zcl_json_path = Path(args.zcl_json).absolute() if args.zcl_json else default_zcl_path
+
+        if args.clusters:
+            # If the user provided the clusters and the zcl.json file provided by -j argument does not exist
+            # we will create a new zcl.json file according to the base zcl.json file in default_zcl_path.
+            # If the provided zcl.json file exists, we will use it as a base and update with a new cluster.
+            base_zcl = zcl_json_path if zcl_json_path.exists() else default_zcl_path
+            add_cluster_to_zcl(base_zcl, args.clusters, zcl_json_path)
+        elif not zcl_json_path.exists():
+            # If clusters are not provided, but user provided a zcl.json file we need to check whether the file exists.
+            log.err(f"ZCL file not found: {zcl_json_path}")
+            return
+
+        app_templates_path = args.matter_path.joinpath(DEFAULT_APP_TEMPLATES_RELATIVE_PATH)
 
-        app_templates_path = args.matter_path / 'src/app/zap-templates/app-templates.json'
+        log.inf(f"Using ZAP file: {zap_file_path}")
+        log.inf(f"Using ZCL file: {zcl_json_path}")
+        log.inf(f"Using app templates: {app_templates_path.absolute()}")
 
         zap_installer = ZapInstaller(args.matter_path)
         zap_installer.update_zap_if_needed()
-        zap_cache_path = zap_installer.get_install_path() / ".zap"
+
+        # The zcl.json path in the .zap file must be the same as the one provided by the user
+        # If not, update the .zap file with the new relative path to the zcl.json file.
+        # After that we must clear the ZAP cache.
+        was_updated = update_zcl_in_zap(zap_file_path, zcl_json_path, app_templates_path)
+        if args.cache and was_updated:
+            log.wrn("ZCL file path in the ZAP file has been updated. The ZAP cache must be cleared to use it.")
 
         cmd = [zap_installer.get_zap_path()]
         cmd += [zap_file_path] if zap_file_path else []
-        cmd += ["--zcl", zcl_json_path]
-        cmd += ["--gen", app_templates_path]
-        cmd += ["--stateDirectory", zap_cache_path]
-
+        cmd += ["--zcl", zcl_json_path.absolute()]
+        cmd += ["--gen", app_templates_path.absolute()]
+        if args.cache:
+            cmd += ["--stateDirectory", args.cache.absolute()]
+        else:
+            cmd += ["--tempState"]
         self.check_call([str(x) for x in cmd])