diff --git a/backends/nxp/tests/generic_tests/test_aot_example.py b/backends/nxp/tests/generic_tests/test_aot_example.py index b75d3605d34..672ddc351a1 100644 --- a/backends/nxp/tests/generic_tests/test_aot_example.py +++ b/backends/nxp/tests/generic_tests/test_aot_example.py @@ -334,3 +334,77 @@ def test_aot_example__mlperf_tiny_kws__profiling(): with _cleanup_generated_files(pte_file, etrecord_file): result = _run_compile(cmd) _assert_profiling(result, pte_file, etrecord_file) + + +def test_aot_example__mlperf_tiny_ad(): + """Test that the MLPerf Tiny Anomaly detection (DeepAutoEncoder) can be lowered to Neutron backend via + `aot_neutron_compile.py` and all ops are delegated.""" + + # Number of random samples to generate, must be divisible by number of classes + num_random_samples = 60 + + # Run the compilation script as a module (like run_aot_example.sh does). + cmd = [ + sys.executable, + "-m", + "examples.nxp.aot_neutron_compile", + "--model_name", + "mlperf_tiny_anomaly_detection", + "--delegate", + "--quantize", + "--target", + "imxrt700", + "--use_random_dataset", + "--num_random_samples", + str(num_random_samples), + ] + + # Output file will be created in executorch_root + pte_file = Path( + os.path.join(EXECUTORCH_ROOT, "mlperf_tiny_anomaly_detection_nxp_delegate.pte") + ) + + with _cleanup_generated_files(pte_file): + result = _run_compile(cmd) + _assert_delegation(result, pte_file) + + +def test_aot_example__mlperf_tiny_ad__profiling(): + """Test that the MLPerf Tiny Anomaly detection (DeepAutoEncoder) can be lowered to Neutron backend via + `aot_neutron_compile.py` and profiling works as intended.""" + + # Number of random samples to generate, must be divisible by number of classes + num_random_samples = 60 + + # Run the compilation script as a module (like run_aot_example.sh does) + cmd = [ + sys.executable, + "-m", + "examples.nxp.aot_neutron_compile", + "--model_name", + "mlperf_tiny_anomaly_detection", + "--delegate", + "--quantize", + "--target", + "imxrt700", + "--remove-quant-io-ops", + "--use_profiling", # Generate profilable model and create ETRecord + "--use_random_dataset", + "--num_random_samples", + str(num_random_samples), + ] + + pte_file = Path( + os.path.join( + EXECUTORCH_ROOT, "mlperf_tiny_anomaly_detection_nxp_delegate_profile.pte" + ) + ) + etrecord_file = Path( + os.path.join( + EXECUTORCH_ROOT, "etrecord", "mlperf_tiny_anomaly_detection_etrecord.bin" + ) + ) + + with _cleanup_generated_files(pte_file, etrecord_file): + result = _run_compile(cmd) + _assert_profiling(result, pte_file, etrecord_file) diff --git a/backends/nxp/tests/models/test_mlperf_tiny_anomaly_detection.py b/backends/nxp/tests/models/test_mlperf_tiny_anomaly_detection.py new file mode 100644 index 00000000000..b259443e79e --- /dev/null +++ b/backends/nxp/tests/models/test_mlperf_tiny_anomaly_detection.py @@ -0,0 +1,127 @@ +# 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. + +import os +from functools import partial + +import numpy as np + +# noinspection PyUnusedImports +import pytest +import torch + +from executorch.backends.nxp.tests.dataset_creator import ( + FromCalibrationDataDatasetCreator, +) +from executorch.backends.nxp.tests.executorch_pipeline import ModelInputSpec +from executorch.backends.nxp.tests.graph_verifier import BaseGraphVerifier +from executorch.backends.nxp.tests.model_output_comparator import ( + ClassificationAccuracyOutputComparator, + NumericalStatsOutputComparator, +) +from executorch.backends.nxp.tests.nsys_testing import ( + get_test_name, + lower_run_compare, + lower_run_compare_ptq_qat, + OUTPUTS_DIR, +) +from executorch.backends.nxp.tests.use_qat import * # noqa F403 +from executorch.examples.nxp.models.mlperf_tiny.anomaly_detection.mlperf_tiny_anomaly_detection import ( + MLPerfTinyAnomalyDetection, +) + +BOUNDS_MSE = { + "PTQ": 1.4e-08, + "QAT": 5.205e-06, +} + + +@pytest.fixture(autouse=True) +def reseed_model_per_test_run(): + torch.manual_seed(23) + np.random.seed(23) + + +def test_mlperf_tiny_anomaly_detection_mse_cpu_vs_npu( + mocker, + request, + use_qat, +): + num_samples = 60 + + anomaly_detection = MLPerfTinyAnomalyDetection( + num_samples=num_samples, use_random_dataset=True + ) + model = anomaly_detection.get_eager_model() + dataset = anomaly_detection.dataset + labels = anomaly_detection.labels + + dataset_creator = FromCalibrationDataDatasetCreator( + dataset, num_examples=num_samples, idx_to_label=labels + ) + + input_spec = ModelInputSpec(anomaly_detection.input_shape) + quant_type_key = "QAT" if use_qat else "PTQ" + + mse = BOUNDS_MSE[quant_type_key] + comparator = NumericalStatsOutputComparator(max_mse_error=mse) + model_verifier = BaseGraphVerifier(1, []) + train_fn = anomaly_detection.train_model_fn if use_qat else None + + lower_run_compare( + model, + [input_spec], + model_verifier, + request, + dataset_creator=dataset_creator, + output_comparator=comparator, + mocker=mocker, + use_qat=use_qat, + train_fn=train_fn, + ) + + +def test_mlperf_tiny_anomaly_detection_ptq_qat_equivalence(request): + num_samples = 60 + + anomaly_detection = MLPerfTinyAnomalyDetection( + num_samples=num_samples, use_random_dataset=True + ) + + model = anomaly_detection.get_eager_model() + dataset = anomaly_detection.dataset + labels = anomaly_detection.labels + + dataset_creator = FromCalibrationDataDatasetCreator( + dataset, num_examples=num_samples, idx_to_label=labels + ) + + test_name = get_test_name(request) + input_parent_path = os.path.join( + OUTPUTS_DIR, + test_name, + "dataset/calibration/", + ) + + comparator = ClassificationAccuracyOutputComparator( + class_dict=labels, + postprocess_fn=partial( + anomaly_detection.get_class_from_reconstruction_error, + input_parent_path=input_parent_path, + ), + ) + + input_spec = ModelInputSpec(anomaly_detection.input_shape) + model_verifier = BaseGraphVerifier(1, []) + + lower_run_compare_ptq_qat( + model, + [input_spec], + model_verifier, + request, + train_fn=anomaly_detection.train_model_fn, + dataset_creator=dataset_creator, + output_comparator=comparator, + ) diff --git a/backends/nxp/tests/models/test_mlperf_tiny_keyword_spotting.py b/backends/nxp/tests/models/test_mlperf_tiny_keyword_spotting.py index ab56849e970..84584855826 100644 --- a/backends/nxp/tests/models/test_mlperf_tiny_keyword_spotting.py +++ b/backends/nxp/tests/models/test_mlperf_tiny_keyword_spotting.py @@ -6,6 +6,8 @@ from functools import partial import numpy as np + +# noinspection PyUnusedImports import pytest import torch from executorch.backends.nxp.tests.dataset_creator import ( diff --git a/examples/nxp/aot_neutron_compile.py b/examples/nxp/aot_neutron_compile.py index b9e3298c26d..5d57837870a 100644 --- a/examples/nxp/aot_neutron_compile.py +++ b/examples/nxp/aot_neutron_compile.py @@ -43,6 +43,9 @@ train_cifarnet_model, verify_cifarnet_model, ) +from executorch.examples.nxp.models.mlperf_tiny.anomaly_detection.mlperf_tiny_anomaly_detection import ( + MLPerfTinyAnomalyDetection, +) from executorch.examples.nxp.models.mlperf_tiny.image_classification.mlperf_tiny_image_classification import ( MLPerfTinyImageClassification, ) @@ -67,6 +70,7 @@ MODELS = { "cifar10": CifarNet, "mobilenetv2": MobilenetV2, + "mlperf_tiny_anomaly_detection": MLPerfTinyAnomalyDetection, "mlperf_tiny_image_classification": MLPerfTinyImageClassification, "mlperf_tiny_keyword_spotting": MLPerfTinyKeywordSpotting, } @@ -126,7 +130,11 @@ def _get_model_info_from_name( ) model_cls_inst = model_cls() - elif model_cls in (MLPerfTinyImageClassification, MLPerfTinyKeywordSpotting): + elif model_cls in ( + MLPerfTinyImageClassification, + MLPerfTinyKeywordSpotting, + MLPerfTinyAnomalyDetection, + ): model_cls_inst = model_cls( dataset_path=dataset_path, use_random_dataset=use_random_dataset, @@ -337,7 +345,12 @@ def _get_arg_parser(): if args.use_qat: if not isinstance( model_cls_inst, - (CifarNet, MLPerfTinyImageClassification, MLPerfTinyKeywordSpotting), + ( + CifarNet, + MLPerfTinyImageClassification, + MLPerfTinyKeywordSpotting, + MLPerfTinyAnomalyDetection, + ), ): raise ValueError( f"QAT training is not supported for model '{args.model_name}'" diff --git a/examples/nxp/models/mlperf_tiny/anomaly_detection/__init__.py b/examples/nxp/models/mlperf_tiny/anomaly_detection/__init__.py new file mode 100644 index 00000000000..55dc5fccf45 --- /dev/null +++ b/examples/nxp/models/mlperf_tiny/anomaly_detection/__init__.py @@ -0,0 +1,4 @@ +# 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. diff --git a/examples/nxp/models/mlperf_tiny/anomaly_detection/mlperf_tiny_anomaly_detection.py b/examples/nxp/models/mlperf_tiny/anomaly_detection/mlperf_tiny_anomaly_detection.py new file mode 100644 index 00000000000..6c92c1a528f --- /dev/null +++ b/examples/nxp/models/mlperf_tiny/anomaly_detection/mlperf_tiny_anomaly_detection.py @@ -0,0 +1,180 @@ +# 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. + +import logging +import os +from pathlib import Path +from typing import Iterator + +import numpy as np + +import torch + +from executorch.backends.nxp.tests.calibration_dataset import ( + CalibrationDataset, + RandomCalibrationDataset, +) +from executorch.examples.models.mlperf_tiny import DeepAutoEncoderModel +from executorch.examples.nxp.models.mlperf_tiny.mlperf_tiny_model import MLPerfTinyModel +from torch.utils.data import Dataset +from torchao.quantization.pt2e import disable_observer +from tqdm import tqdm + +log = logging.getLogger(__name__) + + +class MLPerfTinyAnomalyDetection(MLPerfTinyModel): + """MLPerf Tiny Anomaly Detection model (DeepAutoEncoder). + + The input shape is set to (98, 640) as the reference internal model was trained with this shape. + The dataset is generated for this shape and thus for calibration the data needs to be flattened/unbatched + first. The model is not a classification model so a different loss function than the other MLPerf Tiny models + is used. For class interpretation in output comparison a get_class_from_reconstruction_error() post-processing + function is used. + """ + + INPUT_SHAPE = (98, 640) + IDX_TO_LABEL = {0: "normal", 1: "anomaly"} + CLASS_THRESHOLD = 17.0 # Empirically chosen + _batch_size = INPUT_SHAPE[0] + + # DeepAutoEncoder specific QAT training hyperparameters. + TRAIN_HYPERPARAMETERS = { + "num_epochs": 15, + "batch_size": _batch_size, + "lr": 5e-6, + "eps": 1e-7, + } + + def __init__( + self, + dataset_path: Path | str | None = None, + use_random_dataset: bool = False, + num_samples: int | None = None, + num_workers: int = 4, + ): + self._dataset_flattened = False + super().__init__( + dataset_path=dataset_path, + use_random_dataset=use_random_dataset, + num_samples=num_samples, + num_workers=num_workers, + ) + + @property + def input_shape(self): + return self.INPUT_SHAPE + + @property + def labels(self): + return self.IDX_TO_LABEL + + def _flatten_dataset(self): + flat_xs = [] + flat_ys = [] + + for sample in self.dataset.examples: + input_batch, single_label = sample + batch_size = input_batch.shape[0] + + if batch_size != self._batch_size: + logging.warning( + f"The dataset was exported with `batch_size={batch_size}` " + f"which is different than `self.batch_size={self._batch_size}`. " + "This may produce unexpected behavior. " + "Note: The `self._batch_size` should match " + "the one from `prepare_calibration_data.py` in MLPerfTiny." + ) + + flat_xs.extend(list(input_batch)) + flat_ys.extend([single_label] * batch_size) + + self._dataset_flattened = True + self.dataset.examples = list(zip(flat_xs, flat_ys)) + + def get_class_from_reconstruction_error( + self, + preds: np.ndarray, + pred_path: str, + input_parent_path: str, + ) -> np.ndarray: + sample_name = pred_path.split("/")[-2] + input_path = os.path.join(input_parent_path, sample_name) + inps = np.fromfile(input_path, dtype=preds.dtype).reshape(preds.shape) + + # high error -> anomaly (above threshold) + return np.mean(np.square(inps - preds), axis=1) > self.CLASS_THRESHOLD + + def _init_dataset(self) -> Dataset: + if self._use_random_dataset: + num_classes = len(self.labels) + sample_shape = tuple(self.input_shape) + return RandomCalibrationDataset( + self._num_samples, sample_shape, num_classes + ) + else: + return CalibrationDataset(self._dataset_path) + + # noinspection PyMethodMayBeStatic + def _init_eager_model(self) -> torch.nn.Module: + return DeepAutoEncoderModel().get_eager_model() + + def get_calibration_inputs( + self, batch_size: int = 1 + ) -> Iterator[tuple[torch.Tensor]]: + if not self._dataset_flattened: + self._flatten_dataset() # For Anomaly detection data have to flattened/unbatched first + return super().get_calibration_inputs(batch_size) + + def get_qat_train_inputs( + self, batch_size: int = 5, dataset_portion: float = 0.1 + ) -> Iterator[tuple[torch.Tensor]]: + if not self._dataset_flattened: + self._flatten_dataset() # For Anomaly detection data have to flattened/unbatched first for calibration + return super().get_qat_train_inputs( + batch_size=batch_size, dataset_portion=dataset_portion + ) + + def train_model_fn( + self, model, num_epochs=None, batch_size=None, channels_last=False + ): + assert not channels_last, "This model does not support channels last." + hyperparameters = self.TRAIN_HYPERPARAMETERS + num_epochs = ( + num_epochs if num_epochs is not None else hyperparameters["num_epochs"] + ) + batch_size = ( + batch_size if batch_size is not None else hyperparameters["batch_size"] + ) + + torch.manual_seed(42) + torch.use_deterministic_algorithms(True) + + optimizer = torch.optim.Adam( + params=model.parameters(), + lr=hyperparameters["lr"], + eps=hyperparameters["eps"], + ) + loss_fn = torch.nn.MSELoss() + + log.warning("Starting training...") + + data = self.get_qat_train_inputs(batch_size=batch_size) + for nepoch in range(num_epochs): + for samples, labels in tqdm(data): + # Skip anomaly samples to evaluate using reconstruction error + if sum(labels) > 0: + continue + + optimizer.zero_grad() + outputs = model(samples) + loss = loss_fn(outputs, samples) + loss.backward() + optimizer.step() + + if nepoch >= num_epochs / 3: + model.apply(disable_observer) + + return model