QDQ debugger - Adding Error Calculator (#12632)

QDQ debugger - Adding Error Calculator

onnxruntime/python/tools/quantization/qdq_loss_debug.py

@@ -40,9 +40,10 @@ is a list of tensors, one from each model run
 """
 
 import logging
+import math
 import time
 from pathlib import Path
-from typing import Dict, List, Optional, Sequence, Union
+from typing import Callable, Dict, List, Optional, Sequence, Union
 
 import numpy
 import onnx
@@ -276,8 +277,8 @@ def create_weight_matching(float_model_path: str, qdq_model_path: str) -> Dict[s
         Dict for comparing weight tensors. E.g.
         ```
         qdq_weight_cmp = create_weight_matching(float_model, qdq_model)
-        print(qdq_weight_cmp['activation1']['float'][0])
-        print(qdq_weight_cmp['activation1']['dequantized'][0])
+        print(qdq_weight_cmp['activation1']['float'])
+        print(qdq_weight_cmp['activation1']['dequantized'])
         ```
     """
     float_onnx_model = ONNXModel(load_model(Path(float_model_path), need_optimize=False))
@@ -323,3 +324,52 @@ def create_weight_matching(float_model_path: str, qdq_model_path: str) -> Dict[s
         matched_weights[weight_name] = {"float": weight_float, "dequantized": weight_quant}
 
     return matched_weights
+
+
+def compute_signal_to_quantization_noice_ratio(
+    x: Union[Sequence[numpy.ndarray], numpy.ndarray], y: Union[Sequence[numpy.ndarray], numpy.ndarray]
+) -> float:
+    if isinstance(x, numpy.ndarray):
+        xlist = [x]
+    else:
+        xlist = x
+    if isinstance(y, numpy.ndarray):
+        ylist = [y]
+    else:
+        ylist = y
+    if len(xlist) != len(ylist):
+        raise RuntimeError("Unequal number of tensors to compare!")
+
+    left = numpy.concatenate(xlist).flatten()
+    right = numpy.concatenate(ylist).flatten()
+
+    Ps = numpy.linalg.norm(left)
+    Pn = numpy.linalg.norm(left - right)
+    return 20 * math.log10(Ps / Pn)
+
+
+def compute_weight_error(
+    weights_match: Dict[str, Dict[str, numpy.ndarray]],
+    err_func: Callable[[numpy.ndarray, numpy.ndarray], float] = compute_signal_to_quantization_noice_ratio,
+) -> Dict[str, float]:
+    result: Dict[str, float] = {}
+    for weight_name, weight_match in weights_match.items():
+        result[weight_name] = err_func(weight_match["float"], weight_match["dequantized"])
+    return result
+
+
+def compute_activation_error(
+    activations_match: Dict[str, Dict[str, Sequence[numpy.ndarray]]],
+    err_func: Callable[
+        [Sequence[numpy.ndarray], Sequence[numpy.ndarray]], float
+    ] = compute_signal_to_quantization_noice_ratio,
+) -> Dict[str, Dict[str, float]]:
+    result: Dict[str, Dict[str, float]] = {}
+    for name, match in activations_match.items():
+        err_result: Dict[str, float] = {}
+        err_result["qdq_err"] = err_func(match["pre_qdq"], match["post_qdq"])
+        float_activation = match["float"]
+        if float_activation:
+            err_result["xmodel_err"] = err_func(float_activation, match["post_qdq"])
+        result[name] = err_result
+    return result

onnxruntime/test/python/quantization/test_qdq_loss_debug.py

@@ -21,6 +21,8 @@ from onnxruntime.quantization import QuantFormat, QuantType, quantize_static
 from onnxruntime.quantization.calibrate import CalibrationDataReader
 from onnxruntime.quantization.qdq_loss_debug import (
     collect_activations,
+    compute_activation_error,
+    compute_weight_error,
     create_activation_matching,
     create_weight_matching,
     modify_model_output_intermediate_tensors,
@@ -204,6 +206,19 @@ class TestSaveActivations(unittest.TestCase):
 
         self.assertFalse(compare_dict.get("Conv1Out"))
 
+        activations_error = compute_activation_error(compare_dict)
+        for tensor_name in tensor_names:
+            self.assertGreater(
+                activations_error[tensor_name]["xmodel_err"],
+                0.1,
+                f"{tensor_name} cross model error {activations_error[tensor_name]['xmodel_err']} exceeds threashold.",
+            )
+            self.assertGreater(
+                activations_error[tensor_name]["qdq_err"],
+                0.1,
+                f"{tensor_name} qdq error {activations_error[tensor_name]['qdq_err']} exceeds threashold.",
+            )
+
     def test_create_weight_matching(self):
         # Setup: create float model:
         float_model_path = str(Path(self._tmp_model_dir.name) / "float_model3.onnx")
@@ -231,6 +246,14 @@ class TestSaveActivations(unittest.TestCase):
             dq_array = matched_weights[weight_name]["dequantized"]
             self.assertEqual(float_array.shape, dq_array.shape)
 
+        weights_error = compute_weight_error(matched_weights)
+        for weight_name in weight_names:
+            self.assertGreater(
+                weights_error[weight_name],
+                0.1,
+                f"{weight_name} quantization error {weights_error[weight_name]} too big!",
+            )
+
     def test_create_weight_matching_per_channel(self):
 
         # float model