Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
79 changes: 79 additions & 0 deletions backends/nxp/tests/cortex_m_benchmarking.py
Original file line number Diff line number Diff line change
@@ -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)
6 changes: 3 additions & 3 deletions backends/nxp/tests/executorch_pipeline.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 = []
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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()
Expand Down
10 changes: 9 additions & 1 deletion backends/nxp/tests/model_output_comparator.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
Loading
Loading