Apply fusinginfo quantization config to Activation quantization config. (#1467)

Apply fusinginfo quantization config to Activation quantization config. (#1467)
---------
Co-authored-by: KazunoriSumiya <Sumiya.kazunori@jp.panasonic.com>
Co-authored-by: kawasaki.kenta <kawasaki.kenta@miraxia.com>

Author: 3 people

model_compression_toolkit/core/common/graph/base_graph.py

@@ -32,13 +32,13 @@
 from model_compression_toolkit.core.common.collectors.statistics_collector import scale_statistics, shift_statistics
 from model_compression_toolkit.core.common.pruning.pruning_section import PruningSection
 from model_compression_toolkit.core.common.user_info import UserInformation
-from model_compression_toolkit.core.common.quantization.node_quantization_config import ActivationQuantizationMode
+from model_compression_toolkit.core.common.quantization.node_quantization_config import \
+    NodeActivationQuantizationConfig, ActivationQuantizationMode
 from model_compression_toolkit.logger import Logger
 from model_compression_toolkit.target_platform_capabilities.targetplatform2framework import LayerFilterParams
 from model_compression_toolkit.target_platform_capabilities.targetplatform2framework.framework_quantization_capabilities import \
     FrameworkQuantizationCapabilities
 
-
 def validate_graph_after_change(method: Callable) -> Callable:
     """
     Decorator for graph-mutating methods. After the decorated method executes,
@@ -876,15 +876,32 @@ def _find_intermediate_and_exit_nodes(self, entry_node: BaseNode, fw_impl: Any)
 
         return intermediate_nodes, next_node
 
-    def disable_fused_nodes_activation_quantization(self):
+    def override_fused_node_activation_quantization_candidates(self):
         """
-        Disable activation quantization for all nodes in fused operations,
+        Override fused node activation quantization candidates for all nodes in fused operations,
         except for the last node in each fused group.
-        """
-        nodes_to_disable = self.fusing_info.get_inner_fln_nodes()
-        for node in nodes_to_disable:
-            for qc in node.candidates_quantization_cfg:
-                qc.activation_quantization_cfg.quant_mode = ActivationQuantizationMode.FLN_QUANT
+        Update the value of quantization_config with the value of op_quaitization_cfg from FusingInfo.
+        """
+        from model_compression_toolkit.core.common.quantization.candidate_node_quantization_config import CandidateNodeQuantizationConfig
+
+        nodes_in_fln = self.fusing_info.get_inner_fln_nodes()
+        for node in nodes_in_fln:
+            fused_node_op_id = self.fusing_info.get_fused_op_id_for_node(node.name)
+            fusiong_op_quaitization_cfg = self.fusing_info.get_fused_op_quantization_config(fused_node_op_id)             
+            org_candidate = node.candidates_quantization_cfg[0]
+            if fusiong_op_quaitization_cfg is not None and fusiong_op_quaitization_cfg.enable_activation_quantization:
+                # Set ActivationQuantizationMode to FLN_QUANT and update the value of quantization_config
+                activation_quantization_cfg = NodeActivationQuantizationConfig(qc=org_candidate,
+                                                                               op_cfg=fusiong_op_quaitization_cfg,
+                                                                               activation_quantization_fn=org_candidate.activation_quantization_cfg.activation_quantization_fn,
+                                                                               activation_quantization_params_fn=org_candidate.activation_quantization_cfg.activation_quantization_params_fn)
+                activation_quantization_cfg.quant_mode = ActivationQuantizationMode.FLN_QUANT
+                for qc in node.candidates_quantization_cfg:
+                    qc.activation_quantization_cfg = activation_quantization_cfg
+            else:
+                # Set ActivationQuantizationMode to FLN_NO_QUANT
+                for qc in node.candidates_quantization_cfg:
+                    qc.activation_quantization_cfg.quant_mode = ActivationQuantizationMode.FLN_NO_QUANT
 
     def validate(self):
         """
@@ -908,4 +925,4 @@ def remove_edge(self, *args, **kwargs):
         """
         Wrap networkx functions (that modifies the graph) with our validate decorator.
         """
-        return super().remove_edge(*args, **kwargs)
+        return super().remove_edge(*args, **kwargs)

model_compression_toolkit/core/common/graph/base_node.py

@@ -216,19 +216,31 @@ def is_activation_quantization_enabled(self) -> bool:
         Returns: Whether node activation quantization is enabled or not.
         """
         return self._is_single_quant_mode(ActivationQuantizationMode.QUANT)
-
-    def is_fln_quantization(self) -> bool:
+    
+    def is_fln_no_quantization(self) -> bool:
         """
-        Returns: Whether the node's activation quantization is FLN
+        Returns: Whether node is FLN no quantization.
         """
-        return self._is_single_quant_mode(ActivationQuantizationMode.FLN_QUANT)
-
+        return self._is_single_quant_mode(ActivationQuantizationMode.FLN_NO_QUANT)
+    
     def is_quantization_preserving(self) -> bool:
         """
         Returns: Whether node activation quantization information is preserved from its inputs.
         """
         return self._is_single_quant_mode(ActivationQuantizationMode.PRESERVE_QUANT)
 
+    def is_no_quantization(self) -> bool:
+        """
+        Returns: Whether node is no quantization.
+        """
+        return self._is_single_quant_mode(ActivationQuantizationMode.NO_QUANT)
+
+    def is_fln_quantization(self) -> bool:
+        """
+        Returns: Whether the node's activation quantization is FLN
+        """
+        return self._is_single_quant_mode(ActivationQuantizationMode.FLN_QUANT)
+
     def is_weights_quantization_enabled(self, attr_name: str) -> bool:
         """
         Checks whether a node's weights attribute quantization is enabled.

model_compression_toolkit/core/common/quantization/filter_nodes_candidates.py

@@ -21,7 +21,6 @@
 from model_compression_toolkit.core.common.quantization.candidate_node_quantization_config import \
     CandidateNodeQuantizationConfig
 
-
 def filter_nodes_candidates(graph: Graph):
     """
     Filters the graph's nodes candidates configuration list.
@@ -87,7 +86,7 @@ def filter_node_candidates(node: BaseNode) -> List[CandidateNodeQuantizationConf
     filtered_candidates = copy.deepcopy(node.candidates_quantization_cfg)
     final_candidates = copy.deepcopy(node.candidates_quantization_cfg)
 
-    if (node.kernel_attr is None or not node.is_weights_quantization_enabled(node.kernel_attr)) and not node.is_activation_quantization_enabled():
+    if (node.kernel_attr is None or not node.is_weights_quantization_enabled(node.kernel_attr)) and node.is_no_quantization():
         # If activation quantization is disabled and the node doesn't have a kernel or doesn't quantize the kernel,
         # but for some reason the node has multiple candidates then replace it with a single dummy candidate with
         # default bit-width values.
@@ -102,9 +101,10 @@ def filter_node_candidates(node: BaseNode) -> List[CandidateNodeQuantizationConf
 
         final_candidates = [single_dummy_candidate]
 
-    elif not node.is_activation_quantization_enabled():
+    elif node.is_no_quantization():
         # Remove candidates that have duplicated weights candidates for node with disabled activation quantization.
         # Replacing the activation n_bits in the remained configurations with default value to prevent confusion.
+        # Set the config of the non-quantized FLN node to POWER_OF_TWO.
         seen_candidates = set()
         filtered_candidates = [candidate for candidate in filtered_candidates if
                                candidate.weights_quantization_cfg not in seen_candidates
@@ -116,6 +116,14 @@ def filter_node_candidates(node: BaseNode) -> List[CandidateNodeQuantizationConf
 
         final_candidates = _filter_bit_method_dups(filtered_candidates, node.kernel_attr)
 
+    elif node.is_fln_no_quantization() or node.is_fln_quantization():
+        # Remove candidates that have duplicated weights candidates for node with disabled activation quantization.
+        seen_candidates = set()
+        filtered_candidates = [candidate for candidate in filtered_candidates if
+                               candidate.weights_quantization_cfg not in seen_candidates
+                               and not seen_candidates.add(candidate.weights_quantization_cfg)]
+        final_candidates = _filter_bit_method_dups(filtered_candidates, node.kernel_attr)
+
     elif node.kernel_attr is None or not node.is_weights_quantization_enabled(node.kernel_attr):
         # TODO:
         #  To allow MP on positional weights we need to modify this to consider all weights not only kernel.

model_compression_toolkit/core/common/quantization/node_quantization_config.py

@@ -47,7 +47,7 @@ class ActivationQuantizationMode(Enum):
     FLN_QUANT = auto()
     PRESERVE_QUANT = auto()
     NO_QUANT = auto()
-
+    FLN_NO_QUANT = auto()
 
 class BaseNodeQuantizationConfig(object):
     """

model_compression_toolkit/core/graph_prep_runner.py

@@ -155,7 +155,7 @@ def get_finalized_graph(initial_graph: Graph,
     ######################################
     fusing_info = FusingInfoGenerator(fqc.get_fusing_patterns()).generate_fusing_info(transformed_graph)
     transformed_graph.fusing_info = fusing_info
-    transformed_graph.disable_fused_nodes_activation_quantization()
+    transformed_graph.override_fused_node_activation_quantization_candidates()
 
     ######################################
     # Channel equalization

tests/keras_tests/function_tests/test_activation_weights_composition_substitution.py

@@ -124,7 +124,7 @@ def prepare_graph(in_model, keras_impl, mixed_precision_candidates_list, base_co
 
     fusing_info = FusingInfoGenerator(fqc.get_fusing_patterns()).generate_fusing_info(graph)
     graph.fusing_info = fusing_info
-    graph.disable_fused_nodes_activation_quantization()
+    graph.override_fused_node_activation_quantization_candidates()
 
     graph = filter_nodes_candidates(graph)
 
@@ -230,7 +230,7 @@ def test_two_conv_net_compose_after_split_activation_only(self):
 
         graph.skip_validation_check = False
 
-        self._verify_two_conv_with_split_test(graph, v_graph, 3, 3)
+        self._verify_two_conv_with_split_test(graph, v_graph, 9, 3)
 
     def test_all_weights_layers_composition(self):
         in_model = multiple_weights_nodes_model()

tests/keras_tests/function_tests/test_cfg_candidates_filter.py

@@ -60,7 +60,7 @@ def prepare_graph(in_model, base_config, default_config, bitwidth_candidates):
 
     fusing_info = FusingInfoGenerator(fqc.get_fusing_patterns()).generate_fusing_info(graph)
     graph.fusing_info = fusing_info
-    graph.disable_fused_nodes_activation_quantization()
+    graph.override_fused_node_activation_quantization_candidates()
 
     return graph
 

tests_pytest/_fw_tests_common_base/fusing/base_graph_with_fusing_metadata_test.py

@@ -117,7 +117,7 @@ def test_disable_act_quantization(self, graph_with_fusion_metadata: Graph):
             for qc in node.candidates_quantization_cfg:
                 qc.activation_quantization_cfg.quant_mode = ActivationQuantizationMode.QUANT
 
-        graph_with_fusion_metadata.disable_fused_nodes_activation_quantization()
+        graph_with_fusion_metadata.override_fused_node_activation_quantization_candidates()
         disabled_nodes = [
             node.name for node in graph_with_fusion_metadata.nodes
             if all(not qc.activation_quantization_cfg.enable_activation_quantization

tests_pytest/common_tests/unit_tests/core/graph/test_base_graph.py

@@ -0,0 +1,133 @@
+# Copyright 2025 Sony Semiconductor Israel, Inc. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ==============================================================================
+from copy import deepcopy 
+
+import pytest
+from unittest.mock import Mock
+
+from mct_quantizers import QuantizationMethod
+from model_compression_toolkit.core.common import Graph
+from model_compression_toolkit.core.common.graph.base_node import BaseNode
+from model_compression_toolkit.core.common.fusion.fusing_info import FusingInfo
+from model_compression_toolkit.target_platform_capabilities.schema.mct_current_schema import Signedness
+from tests.common_tests.helpers.generate_test_tpc import generate_test_attr_configs, generate_test_op_qc
+from model_compression_toolkit.core.common.quantization.node_quantization_config import ActivationQuantizationMode, NodeActivationQuantizationConfig
+from model_compression_toolkit.core.common.quantization.candidate_node_quantization_config import CandidateNodeQuantizationConfig
+from model_compression_toolkit.core.common.quantization.quantization_params_generation.power_of_two_selection import power_of_two_selection_histogram
+from model_compression_toolkit.core.common.quantization.quantization_params_generation.symmetric_selection import symmetric_selection_histogram
+from model_compression_toolkit.core import QuantizationErrorMethod
+
+def build_mock_fusing_info(nodes, idx):
+    """
+    Creates a mock FusingInfo object that simulates the behavior of fusing information in a graph.
+    """
+
+    OpQCfg = Mock(spec=NodeActivationQuantizationConfig)
+    OpQCfg.activation_n_bits = 16
+    OpQCfg.signedness = Signedness.AUTO
+    OpQCfg.activation_quantization_method = QuantizationMethod.POWER_OF_TWO
+    OpQCfg.activation_quantization_params_fn = power_of_two_selection_histogram
+    OpQCfg.quantization_preserving = False
+
+    fusing_info = Mock(spec=FusingInfo)
+    fusing_info.get_inner_fln_nodes.return_value = [nodes[0], nodes[1]]
+    
+    if idx == 1:
+        OpQCfg.enable_activation_quantization = True
+        fusing_info.get_fused_op_quantization_config.side_effect = [OpQCfg, None]
+    elif idx == 2:
+        fusing_info.get_fused_op_quantization_config.side_effect = [None, None]
+    else:
+        OpQCfg.enable_activation_quantization = False
+        fusing_info.get_fused_op_quantization_config.side_effect = [OpQCfg, None]
+
+    return fusing_info
+
+def build_mock_node(name, layer_class):
+    """
+    Creates mock nodes representing a simple neural network structure.
+    """
+    node = Mock(spec=BaseNode)
+    node.name = name
+    node.layer_class = layer_class
+
+    activation_quantization_cfg = Mock(spec=NodeActivationQuantizationConfig)
+    activation_quantization_cfg.quant_mode = Mock()
+    activation_quantization_cfg.activation_quantization_fn = symmetric_selection_histogram
+    activation_quantization_cfg.activation_quantization_params_fn = power_of_two_selection_histogram
+   
+    candidate_quantization_config = Mock(spec=CandidateNodeQuantizationConfig)
+    candidate_quantization_config.activation_quantization_cfg = activation_quantization_cfg
+    candidate_quantization_config.activation_error_method = QuantizationErrorMethod.MSE
+    candidate_quantization_config.relu_bound_to_power_of_2 = 0
+    candidate_quantization_config.activation_channel_equalization = False
+    candidate_quantization_config.input_scaling = False
+    candidate_quantization_config.min_threshold = 0
+    candidate_quantization_config.l_p_value = 0
+    candidate_quantization_config.shift_negative_activation_correction = 0
+    candidate_quantization_config.z_threshold = 0
+    candidate_quantization_config.shift_negative_ratio = 0
+    candidate_quantization_config.shift_negative_threshold_recalculation = 0
+    candidate_quantization_config.concat_threshold_update = 0
+    candidate_quantization_config.weights_quantization_cfg = 0
+
+    node.candidates_quantization_cfg = [candidate_quantization_config]
+
+    return node
+
+
+class TestGraph:
+    
+    @pytest.mark.parametrize(("idx"), [
+        1,
+        2,
+        3,
+    ])
+    def test_override_fused_node_activation_quantization_candidates(self, idx):
+        """
+        Test the override_fused_node_activation_quantization_candidates function for a graph with multiple nodes and configurations.
+        """
+        ### Create Test Nodes
+        mock_nodes = []
+        mock_nodes.append(build_mock_node(name='conv', layer_class='Conv2d'))
+        mock_nodes.append(build_mock_node(name='fc', layer_class='Linear'))
+
+        ### Create a mock graph
+        ### Note: Generate the graph first because fusing_info cannot be set without it.
+        ###       In the following Mock, use wraps to mock everything except fusing_info.
+        real_graph = Graph("dummy", [], [], [], [])
+        real_graph.fusing_info = build_mock_fusing_info(mock_nodes, idx) 
+        
+        graph = Mock(spec=Graph, wraps=real_graph)
+        graph.nodes = mock_nodes
+
+        ### call override_fused_node_activation_quantization_candidates
+        graph.override_fused_node_activation_quantization_candidates()
+
+        ### Check if the ActivationQuantization settings set on the graph nodes match the expected values
+        nodes = list(graph.nodes)
+        
+        if idx == 1:
+            ### Check if the first node ActivationQuantization settings match the expected values
+            assert nodes[0].candidates_quantization_cfg[0].activation_quantization_cfg.quant_mode == ActivationQuantizationMode.FLN_QUANT
+            assert nodes[0].candidates_quantization_cfg[0].activation_quantization_cfg.activation_n_bits == 16
+            assert nodes[0].candidates_quantization_cfg[0].activation_quantization_cfg.signedness == Signedness.AUTO
+            assert nodes[0].candidates_quantization_cfg[0].activation_quantization_cfg.activation_quantization_method == QuantizationMethod.POWER_OF_TWO
+            assert nodes[0].candidates_quantization_cfg[0].activation_quantization_cfg.activation_quantization_params_fn == power_of_two_selection_histogram
+            ### Check if the second node ActivationQuantization settings match the expected values
+            assert nodes[1].candidates_quantization_cfg[0].activation_quantization_cfg.quant_mode == ActivationQuantizationMode.FLN_NO_QUANT
+        else:
+            ### Check if the first node ActivationQuantization settings match the expected values
+            assert nodes[0].candidates_quantization_cfg[0].activation_quantization_cfg.quant_mode == ActivationQuantizationMode.FLN_NO_QUANT

tests_pytest/common_tests/unit_tests/core/mixed_precision/resource_utilization_tools/test_resource_utilization_calculator.py

@@ -591,7 +591,7 @@ def test_compute_cuts_random_fusion_valid_utilization(self, seed, disable_quanti
         graph.fusing_info = fusing_info
 
         if disable_quantization:
-            graph.disable_fused_nodes_activation_quantization()
+            graph.override_fused_node_activation_quantization_candidates()
 
         graph.find_node_by_name = MethodType(Graph.find_node_by_name, graph)