feat: added support for abs using new Neutron flow

Author: novak-vaclav

backends/nxp/backend/ir/converter/node_converters/ops_converters/abs_converter.py

@@ -1,11 +1,14 @@
-# Copyright 2025 NXP
+# Copyright 2025-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 torch
+
 from executorch.backends.nxp.backend.ir.converter.node_converter import (
     CustomDelegationOptions,
+    NeutronTargetSpec,
     NodeConverter,
 )
 from executorch.backends.nxp.backend.ir.tflite_generator.builtin_options import (
@@ -25,6 +28,25 @@ def _is_supported_in_IR(
     ) -> bool:
         return True
 
+    @staticmethod
+    def _is_supported_on_target(
+        node: Node,
+        neutron_target_spec: NeutronTargetSpec,
+        parameters_mapping: dict[str, Parameter],
+        custom_delegation_options: CustomDelegationOptions,
+    ) -> bool:
+
+        if custom_delegation_options.use_new_flow_neutron_c:
+            # Requirements specified by the new Neutron flow documentation.
+
+            supported_types = [torch.int8, torch.uint8]
+            if not NodeConverter.uses_quantization_type_for_io(
+                node, supported_types, [0], None
+            ):
+                return False
+
+        return True
+
     def convert(self, node: Node):
         """Convert 'aten::abs' operator to TFLite 'Abs'."""
         self.assert_convertible(node)

backends/nxp/tests/dataset_creator.py

@@ -45,8 +45,10 @@ def generate_samples(
 class RandomDatasetCreator(DatasetCreator):
     """Dataset creator that generates random input samples."""
 
-    def __init__(self, num_samples=2):
+    def __init__(self, num_samples=2, low=0.0, high=1.0):
         self._num_samples = num_samples
+        self.low = low
+        self.high = high
 
     def generate_samples(
         self, dataset_dir: str, input_spec: list[ModelInputSpec]
@@ -103,9 +105,11 @@ def _gen_samples(
                     case _:
                         raise ValueError(f"Unsupported dim_order: {spec.dim_order}")
 
-                sample_vector = rng.random(
-                    np.prod(shape), torch_type_to_numpy_type(spec.dtype)
-                ).reshape(shape)
+                sample_vector = (
+                    rng.uniform(self.low, self.high, size=np.prod(shape))
+                    .astype(torch_type_to_numpy_type(spec.dtype))
+                    .reshape(shape)
+                )
                 file_name = (
                     f"{str(spec_idx).zfill(2)}.bin"
                     if len(input_spec) > 1

backends/nxp/tests/ir/converter/node_converter/test_abs_converter.py

@@ -1,12 +1,11 @@
-# Copyright 2025 NXP
+# Copyright 2025-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 numpy as np
 import pytest
 import torch
-
 from executorch.backends.nxp.backend.edge_program_converter import (
     EdgeProgramToIRConverter,
 )
@@ -17,6 +16,15 @@
     ToChannelFirstPreprocess,
     ToChannelLastPreprocess,
 )
+from executorch.backends.nxp.tests.graph_verifier import (
+    BaseGraphVerifier,
+    NonDelegatedNode,
+)
+
+from executorch.backends.nxp.tests.nsys_testing import (
+    lower_run_compare,
+    RandomDatasetCreator,
+)
 
 from executorch.exir.dialects._ops import ops as exir_ops
 from torch.export import ExportedProgram
@@ -29,7 +37,7 @@ def reseed_model_per_test_run():
     np.random.seed(23)
 
 
-class ConvBlocksWithAbs(torch.nn.Module):
+class ConvBlocksWithAbsModule(torch.nn.Module):
     def __init__(self, conv_in_channels: int = 3):
         super().__init__()
         self.block1 = torch.nn.Sequential(
@@ -56,36 +64,112 @@ def forward(self, x):
         return self.block2(x)
 
 
-class Abs(torch.nn.Module):
+class AbsModule(torch.nn.Module):
     def __init__(self):
         super().__init__()
 
     def forward(self, x):
         return x.abs()
 
 
-def test_conv_abs(mocker, use_qat, input_shape: tuple[int] = (1, 3, 112, 112)):
-    model = ConvBlocksWithAbs(conv_in_channels=input_shape[1])
+class TestAbsLegacyNeutronFlow:
+    def test_conv_abs(
+        self, mocker, use_qat, input_shape: tuple[int, ...] = (1, 3, 112, 112)
+    ):
+        model = ConvBlocksWithAbsModule(conv_in_channels=input_shape[1])
 
-    converter_spy = mocker.spy(EdgeProgramToIRConverter, "convert_program")
+        converter_spy = mocker.spy(EdgeProgramToIRConverter, "convert_program")
 
-    quantized_program = to_quantized_edge_program(
-        model, input_shape, use_qat=use_qat, use_neutron_for_format_conversion=False
-    ).exported_program()
+        quantized_program = to_quantized_edge_program(
+            model,
+            input_shape,
+            use_qat=use_qat,
+            use_neutron_for_format_conversion=False,
+            use_new_flow_neutron_c=False,
+        ).exported_program()
 
-    tflite_flatbuffers_model, io_formats = converter_spy.spy_return
-    exported_program: ExportedProgram = converter_spy.call_args.args[1]
+        tflite_flatbuffers_model, io_formats = converter_spy.spy_return
+        exported_program: ExportedProgram = converter_spy.call_args.args[1]
 
-    assert not graph_contains_any_of_ops(
-        graph=quantized_program.graph, ops=[exir_ops.edge.aten.abs.default]
-    )
+        assert not graph_contains_any_of_ops(
+            graph=quantized_program.graph, ops=[exir_ops.edge.aten.abs.default]
+        )
 
-    input_data = (np.random.random(input_shape) * 50).astype(np.int8)
-    convert_run_compare(
-        exported_program,
-        tfl_model=tflite_flatbuffers_model,
-        tflite_input_preprocess=ToChannelLastPreprocess(),
-        tflite_output_preprocess=ToChannelFirstPreprocess(),
-        input_data=input_data,
-        atol=1.0,
-    )
+        input_data = (np.random.random(input_shape) * 50).astype(np.int8)
+        convert_run_compare(
+            exported_program,
+            tfl_model=tflite_flatbuffers_model,
+            tflite_input_preprocess=ToChannelLastPreprocess(),
+            tflite_output_preprocess=ToChannelFirstPreprocess(),
+            input_data=input_data,
+            atol=1.0,
+        )
+
+
+class TestAbsNewNeutronFlow:
+    @staticmethod
+    def _get_dataset_creator():
+        # to test `abs` reliably, we need to include negative values
+        low = -255.0
+        high = 255.0
+
+        dataset = RandomDatasetCreator(low=low, high=high)
+        return dataset
+
+    def test__basic_nsys_inference(self):
+        input_shape = (2, 3, 6, 7)
+        model = AbsModule()
+        graph_verifier = BaseGraphVerifier(
+            exp_num_delegate_call_nodes=1,  # Delegated Abs.
+            exp_non_delegated_nodes=[],
+        )
+
+        dataset_creator = self._get_dataset_creator()
+        lower_run_compare(
+            model,
+            input_shape,
+            graph_verifier,
+            dataset_creator,
+            use_new_flow_neutron_c=True,
+        )
+
+    def test__basic_nsys_inference__big(self):
+        # some operators have delegation requirement that size must be < 4096
+        input_shape = (4097, 1)
+        model = AbsModule()
+        graph_verifier = BaseGraphVerifier(
+            exp_num_delegate_call_nodes=1,  # Delegated Abs.
+            exp_non_delegated_nodes=[],
+        )
+
+        dataset_creator = self._get_dataset_creator()
+        lower_run_compare(
+            model,
+            input_shape,
+            graph_verifier,
+            dataset_creator,
+            use_new_flow_neutron_c=True,
+        )
+
+    def test_basic_nsys_inference__with_conv(self):
+        input_shape = (2, 3, 6, 7)
+        in_channels = input_shape[1]
+        model = ConvBlocksWithAbsModule(conv_in_channels=in_channels)
+        graph_verifier = BaseGraphVerifier(
+            exp_num_delegate_call_nodes=1,  # Delegated `Abs` + `Relu` in one partition
+            exp_non_delegated_nodes=[
+                NonDelegatedNode("aten_convolution_default", 2),
+                NonDelegatedNode(
+                    "aten_relu_default", 1  # One `Relu` ends up in partition with `Abs`
+                ),
+            ],
+        )
+
+        dataset_creator = self._get_dataset_creator()
+        lower_run_compare(
+            model,
+            input_shape,
+            graph_verifier,
+            dataset_creator,
+            use_new_flow_neutron_c=True,
+        )