From 740ac9fc3def542f1ae8c7014ec09c7b1ebddba9 Mon Sep 17 00:00:00 2001 From: roman-janik-nxp Date: Thu, 27 Aug 2026 19:04:48 +0200 Subject: [PATCH] NXP backend: Add support for Cortex-M backend benchmarking --- backends/nxp/tests/cortex_m_benchmarking.py | 79 +++++ backends/nxp/tests/executorch_pipeline.py | 6 +- backends/nxp/tests/model_output_comparator.py | 10 +- backends/nxp/tests/nsys_testing.py | 270 ++++++++---------- backends/nxp/tests/utils.py | 166 +++++++++++ 5 files changed, 381 insertions(+), 150 deletions(-) create mode 100644 backends/nxp/tests/cortex_m_benchmarking.py diff --git a/backends/nxp/tests/cortex_m_benchmarking.py b/backends/nxp/tests/cortex_m_benchmarking.py new file mode 100644 index 00000000000..7859e9e2c3b --- /dev/null +++ b/backends/nxp/tests/cortex_m_benchmarking.py @@ -0,0 +1,79 @@ +# Copyright 2025-2026 Arm Limited and/or its affiliates. +# Copyright 2026 NXP +# +# This source code is licensed under the BSD-style license found in the +# LICENSE file in the root directory of this source tree. + +from executorch.backends.cortex_m.target_config import CortexM, CortexMTargetConfig +from executorch.backends.cortex_m.test.tester import CortexMQuantize, CortexMTester +from executorch.backends.nxp.tests.utils import ( + process_input_sample, + process_output_sample, + read_prepared_samples, + store_results, +) +from executorch.backends.test.harness.stages import StageType + + +class CortexMNXPBenchmarkTester(CortexMTester): + def __init__( + self, + module, + example_inputs, + target_config: CortexMTargetConfig | None = None, + timeout: int = 120, + ): + target_config = target_config or CortexMTargetConfig( + cpu=CortexM.M33 + ) # set default to M33 for NXP boards + super().__init__(module, example_inputs, target_config, timeout) + + def run_benchmark( + self, + calibration_samples, + input_spec, + output_spec, + testing_dataset_dir, + cpu_results_dir, + npu_results_dir, + ): + quantization_stage = CortexMQuantize(calibration_samples=calibration_samples) + + self.quantize(quantization_stage) + self.export() + self.to_edge() + self.run_passes() + self.to_executorch() + self.serialize() + self.run_program( + input_spec, + output_spec, + testing_dataset_dir, + cpu_results_dir, + npu_results_dir, + ) + + return self.stages[StageType.SERIALIZE].executorch_program_manager + + def run_program( + self, + input_spec, + output_spec, + testing_dataset_dir, + cpu_results_dir, + npu_results_dir, + ): + all_outputs = [] + + for input_samples in read_prepared_samples(testing_dataset_dir, input_spec): + current_input_samples = process_input_sample(input_spec, input_samples) + + # Run the model. + output = self.stages[StageType.SERIALIZE].run_artifact( + *current_input_samples + ) + current_outputs = process_output_sample(output, output_spec) + all_outputs.append(current_outputs) + + # Store all the results. + store_results(all_outputs, cpu_results_dir, npu_results_dir) diff --git a/backends/nxp/tests/executorch_pipeline.py b/backends/nxp/tests/executorch_pipeline.py index 964b8de159c..73b8194636e 100644 --- a/backends/nxp/tests/executorch_pipeline.py +++ b/backends/nxp/tests/executorch_pipeline.py @@ -154,7 +154,7 @@ def _nested( return _nested -def _get_example_input( +def get_example_input( input_spec: tuple[ModelInputSpec, ...], ) -> tuple[torch.Tensor, ...]: example_input = [] @@ -200,7 +200,7 @@ def to_quantized_edge_program( ) input_spec = to_model_input_spec(input_spec) calibration_inputs = get_calibration_inputs_fn(input_spec) - example_input = _get_example_input(input_spec) + example_input = get_example_input(input_spec) # Make sure the model is in the evaluation mode. model.eval() @@ -318,7 +318,7 @@ def to_edge_program( model: nn.Module, input_spec: Iterable[ModelInputSpec] | tuple[int, ...] | list[tuple[int, ...]], ) -> EdgeProgramManager: - example_input = _get_example_input(to_model_input_spec(input_spec)) + example_input = get_example_input(to_model_input_spec(input_spec)) # Make sure the model is in the evaluation mode. model.eval() diff --git a/backends/nxp/tests/model_output_comparator.py b/backends/nxp/tests/model_output_comparator.py index 17adbcf7265..552e73a8b93 100644 --- a/backends/nxp/tests/model_output_comparator.py +++ b/backends/nxp/tests/model_output_comparator.py @@ -89,11 +89,19 @@ def compare_results(self, cpu_results_dir, npu_results_dir, output_tensor_spec): store_txt_input_tensor(npu_tensor_path, tensor_spec) store_txt_input_tensor(diff_cpu_npu_tensor_path, tensor_spec) + try: + self.compare_sample(sample_dir, cpu_output_tensors, npu_output_tensors) + except Exception as e: + # We need to archive the test_dir if comparison fails + test_dir = os.path.dirname(cpu_results_dir) + if logging.root.isEnabledFor(logging.DEBUG): + archive_test_dir(test_dir) + raise e + # We need to archive the test_dir before comparison, as comparison can cause AssertionError exception test_dir = os.path.dirname(cpu_results_dir) if logging.root.isEnabledFor(logging.DEBUG): archive_test_dir(test_dir) - self.compare_sample(sample_dir, cpu_output_tensors, npu_output_tensors) @abstractmethod def compare_sample( diff --git a/backends/nxp/tests/nsys_testing.py b/backends/nxp/tests/nsys_testing.py index 78a8f473024..07b58f06f26 100644 --- a/backends/nxp/tests/nsys_testing.py +++ b/backends/nxp/tests/nsys_testing.py @@ -10,22 +10,20 @@ import re import shutil import subprocess +import sys from copy import deepcopy from enum import Enum from importlib.metadata import version from os import environ, mkdir from typing import Callable, Iterable -import numpy as np import torch import yaml -from executorch.backends.nxp.backend.edge_helper import is_channels_last_dim_order -from executorch.backends.nxp.backend.ir.converter.conversion import translator -from executorch.backends.nxp.backend.ir.converter.conversion.translator import ( - torch_type_to_numpy_type, -) from executorch.backends.nxp.neutron_partitioner import NeutronPartitioner from executorch.backends.nxp.tests.config_importer import test_config +from executorch.backends.nxp.tests.cortex_m_benchmarking import ( + CortexMNXPBenchmarkTester, +) from executorch.backends.nxp.tests.dataset_creator import ( create_quantized_variant_of_dataset, InputQuantizationSpec, @@ -33,6 +31,7 @@ ) from executorch.backends.nxp.tests.executorch_pipeline import ( get_calibration_inputs_fn_from_dataset_dir, + get_example_input, ModelInputSpec, to_model_input_spec, to_quantized_edge_program, @@ -43,7 +42,14 @@ AllCloseOutputComparator, ) from executorch.backends.nxp.tests.outputs_dir_importer import outputs_dir -from executorch.backends.nxp.tests.utils import save_pte_program, store_txt_input_tensor +from executorch.backends.nxp.tests.utils import ( + process_input_sample, + process_output_sample, + read_prepared_samples, + save_pte_program, + store_results, + store_txt_input_tensor, +) from executorch.devtools.visualization.visualization_utils import ( visualize_with_clusters, ) @@ -68,6 +74,7 @@ class ReferenceModel(Enum): # QUANTIZED_ATEN_PYTHON = 2 # Not implemented. # FLOAT_ATEN_PYTHON = 3 # Not implemented. FLOAT_PYTORCH_PYTHON = 4 + QUANTIZED_CORTEX_M = 5 def _get_dataset_cli_args(input_spec: list[ModelInputSpec], testing_dataset_dir): @@ -232,102 +239,6 @@ def _run_non_delegated_executorch_program( return non_delegated_program.exported_program() -def read_prepared_samples( - dataset_dir: str, input_spec: list[ModelInputSpec] -) -> list[tuple[np.ndarray, ...]]: - """Read numpy arrays generated by a `DatasetCreator`. - - :param dataset_dir: Directory containing the generated samples - :param input_spec: List of ModelInputSpec defining the shape and type of each input - - :return: List of tuples, where each tuple contains numpy arrays for one sample - """ - all_samples = [] - - # Multi-input: samples are in numbered subdirectories - if len(input_spec) > 1: - sample_dirs = sorted( - [ - d - for d in os.listdir(dataset_dir) - if os.path.isdir(os.path.join(dataset_dir, d)) - ] - ) - - for sample_name in sample_dirs: - sample_dir = os.path.join(dataset_dir, sample_name) - current_samples = [] - - for spec_idx, spec in enumerate(input_spec): - bin_file_path = os.path.join( - sample_dir, f"{str(spec_idx).zfill(2)}.bin" - ) - sample_vector = np.fromfile( - bin_file_path, dtype=torch_type_to_numpy_type(spec.dtype) - ).reshape(spec.shape) - current_samples.append(sample_vector) - - all_samples.append(tuple(current_samples)) - - # Single-input: binary files are directly in dataset_dir - else: - bin_files = sorted([f for f in os.listdir(dataset_dir) if f.endswith(".bin")]) - - for bin_file in bin_files: - bin_file_path = os.path.join(dataset_dir, bin_file) - sample_vector = np.fromfile( - bin_file_path, dtype=torch_type_to_numpy_type(input_spec[0].dtype) - ).reshape(input_spec[0].shape) - all_samples.append((sample_vector,)) - - return all_samples - - -def store_results( - results: list[tuple[np.ndarray, ...]], output_dir: str, reference_dir: str -): - """Store a list of output arrays in the directory structure matching the reference directory. - - :param results: List of tuples, where each tuple contains numpy arrays (outputs for one sample) - :param output_dir: Directory where results will be stored - - Directory structure created matches reference_dir: - output_dir/ - ├── sample_0/ - │ ├── 0000.bin - │ └── 0001.bin - ├── some_other_sample/ - │ ├── 0000.bin - │ └── 0001.bin - """ - os.makedirs(output_dir, exist_ok=True) - - # Get subdirectories from reference directory - sample_dirs = sorted( - [ - d - for d in os.listdir(reference_dir) - if os.path.isdir(os.path.join(reference_dir, d)) - ] - ) - - assert len(sample_dirs) == len( - results - ), f"Number of samples ({len(results)}) must match number of subdirectories in reference_dir ({len(sample_dirs)})" - - for _sample_idx, (sample_name, sample_outputs) in enumerate( - zip(sample_dirs, results) - ): - sample_dir = os.path.join(output_dir, sample_name) - os.makedirs(sample_dir, exist_ok=True) - - # Store each output tensor - for output_idx, output_array in enumerate(sample_outputs): - bin_file_name = f"{str(output_idx).zfill(4)}.bin" - bin_file_path = os.path.join(sample_dir, bin_file_name) - output_array.tofile(bin_file_path) - - def _run_python_program( model: torch.nn.Module | GraphModule, testing_dataset_dir, @@ -352,53 +263,94 @@ def _run_python_program( all_outputs = [] for input_samples in read_prepared_samples(testing_dataset_dir, input_spec): - current_input_samples = [] - for spec, sample in zip(input_spec, input_samples, strict=True): - match spec.dim_order: - case torch.contiguous_format: - # Use the data as is, just turn it into a PyTorch tensor. - sample = torch.tensor(sample) - - case torch.channels_last: - # The tensor data was stored by the DatasetCreator as channels last (NHWC), but it was now - # incorrectly parsed as contiguous/channels first (NCHW). Transpose it to channels last to preserve - # the semantics. - channels_last_shape = translator.dims_to_channels_last( - list(spec.shape) - ) - sample = np.moveaxis(sample.reshape(channels_last_shape), -1, 1) - sample = torch.tensor(sample).to(memory_format=torch.channels_last) - - case _: - raise ValueError(f"Unsupported dim_order: {spec.dim_order}") - - current_input_samples.append(sample) + current_input_samples = process_input_sample(input_spec, input_samples) # Run the model. output = model(*current_input_samples) - if isinstance(output, torch.Tensor): - output = (output,) + current_outputs = process_output_sample(output, output_spec) + all_outputs.append(current_outputs) + + # Store all the results. + store_results(all_outputs, cpu_results_dir, npu_results_dir) + + +def _run_cortex_m_program( + model: torch.nn.Module | GraphModule, + test_dir, + test_name, + testing_dataset_dir, + input_spec: list[ModelInputSpec], + output_spec: list[torch.Tensor], + cpu_results_dir, + npu_results_dir, +): + """Run a model with Cortex-M backend with channels last inputs. - current_outputs = [] + :param model: Any PyTorch/ExecuTorch model runnable with Cortex-M with channels last inputs. + :param test_dir: Directory for saving test artifacts. + :param test_name: Name of the test. + :param testing_dataset_dir: Directory containing testing data. The samples have to be channels last (NHWC). + The format must match the input_spec.dim_order. + :param input_spec: List of ModelInputSpec defining the shape, type, and dimension order of each input. + :param output_spec: List of output tensor specifications. + :param cpu_results_dir: Directory where CPU results will be stored. The structure will match the existing structure + of `npu_results_dir`. + :param npu_results_dir: Directory where NPU results are already stored, to serve as reference directory structure + for `cpu_results_dir`. + """ + # Assert Cortex-M dependencies are available + assert_cortex_m() + + # Use testing dataset for calibration as testing dataset is always in channel last format for Cortex-M backend. + numpy_samples = read_prepared_samples(testing_dataset_dir, input_spec) + calibration_samples = [ + process_input_sample(input_spec, sample) for sample in numpy_samples + ] - for o, o_spec in zip(output, output_spec, strict=True): - dim_order = list(o_spec.dim_order()) # ExecuTorch dim order. - rank = len(o_spec.shape) - if dim_order == list(range(rank)): # Contiguous dim order. - current_outputs.append(o.detach().numpy()) + example_inputs = get_example_input(input_spec) - elif is_channels_last_dim_order(dim_order): # Channels last dim order. - # The NPU variant outputs channels last (NHWC). We need to convert the CPU output to match. - o = o.detach().numpy().reshape(o_spec.shape) - current_outputs.append(np.moveaxis(o, 1, -1)) + tester = CortexMNXPBenchmarkTester( + model, + example_inputs, + ) + cortex_m_delegated_program = tester.run_benchmark( + calibration_samples, + input_spec, + output_spec, + testing_dataset_dir, + cpu_results_dir, + npu_results_dir, + ) + save_pte_program( + cortex_m_delegated_program, test_name + "_cortex_m_delegated", test_dir + ) - else: - raise ValueError(f"Unsupported dim_order: {o_spec.dim_order}") - all_outputs.append(current_outputs) +def assert_cortex_m(): + # Follow backends/cortex_m/README.md to install the required dependencies. + # Build Arm executor runner with target="cortex-m33": + # ./backends/cortex_m/test/build_test_runner.sh --target="cortex-m33" + # FVP Corstone-300 simulator needs to be added to PATH. + python_version = "".join(map(str, sys.version_info[:2])) + cmsis_nn_lib_path = os.path.join( + PROJECT_DIR, + "backends/cortex_m/library/_cmsis_nn", + f"cmsis_nn.cpython-{python_version}-x86_64-linux-gnu.so", + ) - # Store all the results. - store_results(all_outputs, cpu_results_dir, npu_results_dir) + assert os.path.exists( + cmsis_nn_lib_path + ), "CMSIS-NN lib is not available, check if ET is built correctly." + fvp_simulator_path = os.path.join( + PROJECT_DIR, + "examples/arm/arm-scratch/FVP-corstone300/models/Linux64_GCC-9.3/FVP_Corstone_SSE-300_Ethos-U55", + ) + assert os.path.exists(fvp_simulator_path), "Arm FVP Corstone-300 is not installed." + arm_executor_runner_path = os.path.join( + PROJECT_DIR, + "arm_test/arm_semihosting_executor_runner_corstone-300_cortex-m33/arm_executor_runner", + ) + assert os.path.exists(arm_executor_runner_path), "Arm executor is not installed." def assert_NSYS(): @@ -563,11 +515,39 @@ def lower_run_compare( npu_results_dir, ) + case ReferenceModel.QUANTIZED_CORTEX_M: + if use_qat: + raise ValueError( + "Flag use_qat is not applicable to QUANTIZED_CORTEX_M reference model" + "as it doesn't support QAT. Run with use_qat=False." + ) + if remove_quant_io_ops: + raise ValueError( + "Flag remove_quant_io_ops is not applicable to QUANTIZED_CORTEX_M reference model" + "as it works with float data only. Run with remove_quant_io_ops=False." + ) + if any(spec.dim_order != torch.channels_last for spec in input_spec): + raise ValueError( + "Cortex-M backend supports only channel last dim order inputs." + ) + + model_to_delegate_cortex_m = deepcopy(model) + + # Lower to quantized Cortex-M program and run on Arm simulator. + _run_cortex_m_program( + model_to_delegate_cortex_m, + test_dir, + test_name, + testing_dataset_dir, + input_spec, + output_spec, + cpu_results_dir, + npu_results_dir, + ) + case _: raise ValueError(f"Unsupported reference model: `{reference_model}`.") - output_tensor_spec = _get_program_output_spec(delegated_program) - if logging.root.isEnabledFor(logging.DEBUG): _generate_txt_test_data( calibration_dataset_dir, testing_dataset_dir, list(input_spec) @@ -575,9 +555,7 @@ def lower_run_compare( dump_debug_test_summary(test_name, test_dir) npu_results_dir = os.path.join(test_dir, "results_npu") cpu_results_dir = os.path.join(test_dir, "results_cpu") - output_comparator.compare_results( - cpu_results_dir, npu_results_dir, output_tensor_spec - ) + output_comparator.compare_results(cpu_results_dir, npu_results_dir, output_spec) def lower_run_compare_ptq_qat( diff --git a/backends/nxp/tests/utils.py b/backends/nxp/tests/utils.py index 00b7c364a31..00e58340259 100644 --- a/backends/nxp/tests/utils.py +++ b/backends/nxp/tests/utils.py @@ -11,7 +11,11 @@ import numpy as np +import torch +from executorch.backends.nxp.backend.edge_helper import is_channels_last_dim_order + from executorch.backends.nxp.backend.ir.converter.conversion.translator import ( + dims_to_channels_last, torch_type_to_numpy_type, ) from executorch.backends.nxp.tests.executorch_pipeline import ModelInputSpec @@ -64,3 +68,165 @@ def store_txt_input_tensor( def archive_test_dir(test_dir: str): shutil.make_archive(test_dir, "zip", test_dir) + + +def read_prepared_samples( + dataset_dir: str, input_spec: list[ModelInputSpec] +) -> list[tuple[np.ndarray, ...]]: + """Read numpy arrays generated by a `DatasetCreator`. + + :param dataset_dir: Directory containing the generated samples + :param input_spec: List of ModelInputSpec defining the shape and type of each input + + :return: List of tuples, where each tuple contains numpy arrays for one sample + """ + all_samples = [] + + # Multi-input: samples are in numbered subdirectories + if len(input_spec) > 1: + sample_dirs = sorted( + [ + d + for d in os.listdir(dataset_dir) + if os.path.isdir(os.path.join(dataset_dir, d)) + ] + ) + + for sample_name in sample_dirs: + sample_dir = os.path.join(dataset_dir, sample_name) + current_samples = [] + + for spec_idx, spec in enumerate(input_spec): + bin_file_path = os.path.join( + sample_dir, f"{str(spec_idx).zfill(2)}.bin" + ) + sample_vector = np.fromfile( + bin_file_path, dtype=torch_type_to_numpy_type(spec.dtype) + ).reshape(spec.shape) + current_samples.append(sample_vector) + + all_samples.append(tuple(current_samples)) + + # Single-input: binary files are directly in dataset_dir + else: + bin_files = sorted([f for f in os.listdir(dataset_dir) if f.endswith(".bin")]) + + for bin_file in bin_files: + bin_file_path = os.path.join(dataset_dir, bin_file) + sample_vector = np.fromfile( + bin_file_path, dtype=torch_type_to_numpy_type(input_spec[0].dtype) + ).reshape(input_spec[0].shape) + all_samples.append((sample_vector,)) + + return all_samples + + +def store_results( + results: list[tuple[np.ndarray, ...]], output_dir: str, reference_dir: str +): + """Store a list of output arrays in the directory structure matching the reference directory. + + :param results: List of tuples, where each tuple contains numpy arrays (outputs for one sample) + :param output_dir: Directory where results will be stored + + Directory structure created matches reference_dir: + output_dir/ + ├── sample_0/ + │ ├── 0000.bin + │ └── 0001.bin + ├── some_other_sample/ + │ ├── 0000.bin + │ └── 0001.bin + """ + os.makedirs(output_dir, exist_ok=True) + + # Get subdirectories from reference directory + sample_dirs = sorted( + [ + d + for d in os.listdir(reference_dir) + if os.path.isdir(os.path.join(reference_dir, d)) + ] + ) + + assert len(sample_dirs) == len( + results + ), f"Number of samples ({len(results)}) must match number of subdirectories in reference_dir ({len(sample_dirs)})" + + for _sample_idx, (sample_name, sample_outputs) in enumerate( + zip(sample_dirs, results) + ): + sample_dir = os.path.join(output_dir, sample_name) + os.makedirs(sample_dir, exist_ok=True) + + # Store each output tensor + for output_idx, output_array in enumerate(sample_outputs): + bin_file_name = f"{str(output_idx).zfill(4)}.bin" + bin_file_path = os.path.join(sample_dir, bin_file_name) + output_array.tofile(bin_file_path) + + +def process_input_sample( + input_spec: list[ModelInputSpec], input_samples: tuple[np.ndarray, ...] +) -> list[torch.Tensor]: + """Process input samples by converting them to PyTorch tensors with correct dimension order. + + :param input_spec: List of ModelInputSpec defining the shape, type, and dimension order of each input + :param input_samples: Tuple of numpy arrays representing one sample + + :return: List of PyTorch tensors with correct dimension order + """ + current_input_samples = [] + for spec, sample in zip(input_spec, input_samples, strict=True): + match spec.dim_order: + case torch.contiguous_format: + # Use the data as is, just turn it into a PyTorch tensor. + sample = torch.tensor(sample) + + case torch.channels_last: + # The tensor data was stored by the DatasetCreator as channels last (NHWC), but it was now + # incorrectly parsed as contiguous/channels first (NCHW). Transpose it to channels last to preserve + # the semantics. + channels_last_shape = dims_to_channels_last(list(spec.shape)) + sample = np.moveaxis(sample.reshape(channels_last_shape), -1, 1) + sample = torch.tensor(sample).to(memory_format=torch.channels_last) + + case _: + raise ValueError(f"Unsupported dim_order: {spec.dim_order}") + + current_input_samples.append(sample) + + return current_input_samples + + +def process_output_sample( + output: tuple[torch.Tensor, ...] | torch.Tensor, output_spec: list[torch.Tensor] +) -> list[np.ndarray]: + """Process output tensors by converting them to numpy arrays with correct dimension order. + + :param output: Model output - either a single tensor or tuple of tensors + :param output_spec: List of output tensor specifications + + :return: List of numpy arrays with correct dimension order matching NPU output format + """ + + if isinstance(output, torch.Tensor): + output = (output,) + + current_outputs = [] + + for o, o_spec in zip(output, output_spec, strict=True): + dim_order = list(o_spec.dim_order()) # ExecuTorch dim order. + rank = len(o_spec.shape) + if dim_order == list(range(rank)): # Contiguous dim order. + current_outputs.append(o.detach().numpy()) + + elif is_channels_last_dim_order(dim_order): # Channels last dim order. + # The NPU variant outputs channels last (NHWC). We need to convert the CPU output to match. + o = o.detach().numpy().reshape(o_spec.shape) + current_outputs.append(np.moveaxis(o, 1, -1)) + + else: + raise ValueError(f"Unsupported dim_order: {o_spec.dim_order}") + + return current_outputs