blob: e135c9c1b6984cc36ecaf23a9ec858c9487df98b [file] [log] [blame]
# Copyright 2021 The TensorFlow Authors. 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.
# ==============================================================================
"""Tests for QuantizationDebugger."""
import csv
import io
import re
from unittest import mock
from absl.testing import parameterized
import numpy as np
import tensorflow as tf
from tensorflow.lite.python import convert
from tensorflow.lite.python import lite
from tensorflow.lite.tools.optimize.debugging.python import debugger
from tensorflow.python.framework import test_util
from tensorflow.python.platform import test
from tensorflow.python.training.tracking import tracking
# pylint: disable=g-import-not-at-top
try:
from tensorflow.lite.python import metrics_portable as metrics
except ImportError:
from tensorflow.lite.python import metrics_nonportable as metrics
# pylint: enable=g-import-not-at-top
def _get_model():
"""Returns somple model with Conv2D and representative dataset gen."""
root = tracking.AutoTrackable()
kernel_in = np.array([-2, -1, 1, 2], dtype=np.float32).reshape((2, 2, 1, 1))
@tf.function(
input_signature=[tf.TensorSpec(shape=[1, 3, 3, 1], dtype=tf.float32)])
def func(inp):
kernel = tf.constant(kernel_in, dtype=tf.float32)
conv = tf.nn.conv2d(inp, kernel, strides=1, padding='SAME')
output = tf.nn.relu(conv, name='output')
return output
root.f = func
to_save = root.f.get_concrete_function()
return (root, to_save)
def _calibration_gen():
for i in range(5):
yield [np.arange(9).reshape((1, 3, 3, 1)).astype(np.float32) * i]
def _convert_model(model, func):
"""Converts TF model to TFLite float model."""
converter = lite.TFLiteConverterV2.from_concrete_functions([func], model)
# TODO(b/191205988): Explicitly disable saved model lowering in conversion.
converter.experimental_lower_to_saved_model = False
return converter.convert()
def _quantize_converter(model, func, calibration_gen, debug=True):
"""Returns a converter appropriate for the function and debug configs."""
converter = lite.TFLiteConverterV2.from_concrete_functions([func], model)
converter.target_spec.supported_ops = [lite.OpsSet.TFLITE_BUILTINS_INT8]
converter.representative_dataset = calibration_gen
# TODO(b/191205988): Explicitly disable saved model lowering in conversion.
converter.experimental_lower_to_saved_model = False
# Create a TFLite model with new quantizer and numeric verify ops.
converter.optimizations = [lite.Optimize.DEFAULT]
converter.experimental_new_quantizer = True
if debug:
converter._experimental_calibrate_only = True
return converter
def _quantize_model(model,
func,
calibration_gen,
quantized_io=False,
debug=True):
"""Quantizes model, in debug or normal mode."""
converter = _quantize_converter(model, func, calibration_gen, debug)
if debug:
calibrated = converter.convert()
return convert.mlir_quantize(
calibrated, enable_numeric_verify=True, fully_quantize=quantized_io)
else:
return converter.convert()
def _dummy_fn(*unused_args):
return 0.0
class QuantizationDebugOptionsTest(test_util.TensorFlowTestCase,
parameterized.TestCase):
@test_util.run_v2_only
def test_init_duplicate_keys_raises_ValueError(self):
with self.assertRaises(ValueError):
debugger.QuantizationDebugOptions(
layer_debug_metrics={
'a': _dummy_fn,
'b': _dummy_fn
},
model_debug_metrics={
'c': _dummy_fn,
'd': _dummy_fn
},
layer_direct_compare_metrics={
'a': _dummy_fn,
'e': _dummy_fn
})
with self.assertRaises(ValueError):
debugger.QuantizationDebugOptions(
layer_debug_metrics={
'a': _dummy_fn,
'b': _dummy_fn
},
layer_direct_compare_metrics={
'a': _dummy_fn,
'e': _dummy_fn
})
class QuantizationDebuggerTest(test_util.TensorFlowTestCase,
parameterized.TestCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.tf_model_root, cls.tf_model = _get_model()
cls.float_model = _convert_model(cls.tf_model_root, cls.tf_model)
cls.debug_model_float = _quantize_model(
cls.tf_model_root, cls.tf_model, _calibration_gen, quantized_io=False)
cls.debug_model_int8 = _quantize_model(
cls.tf_model_root, cls.tf_model, _calibration_gen, quantized_io=True)
@parameterized.named_parameters(
('float_io', False, False),
('quantized_io', True, False),
('float_io_from_converter', False, True),
('quantized_io_from_converter', True, True),
)
@test_util.run_v2_only
def test_layer_metrics(self, quantized_io, from_converter):
options = debugger.QuantizationDebugOptions(
layer_debug_metrics={'l1_norm': lambda diffs: np.mean(np.abs(diffs))})
if not from_converter:
if quantized_io:
debug_model = QuantizationDebuggerTest.debug_model_int8
else:
debug_model = QuantizationDebuggerTest.debug_model_float
quant_debugger = debugger.QuantizationDebugger(
quant_debug_model_content=debug_model,
debug_dataset=_calibration_gen,
debug_options=options)
else:
options.fully_quantize = quantized_io
quant_debugger = debugger.QuantizationDebugger(
converter=_quantize_converter(self.tf_model_root, self.tf_model,
_calibration_gen),
debug_dataset=_calibration_gen,
debug_options=options)
quant_debugger.run()
expected_metrics = {
'num_elements': 9,
'stddev': 0.03850026,
'mean_error': 0.01673192,
'max_abs_error': 0.10039272,
'mean_squared_error': 0.0027558778,
'l1_norm': 0.023704167,
}
self.assertLen(quant_debugger.layer_statistics, 1)
actual_metrics = next(iter(quant_debugger.layer_statistics.values()))
self.assertCountEqual(expected_metrics.keys(), actual_metrics.keys())
for key, value in expected_metrics.items():
self.assertAlmostEqual(value, actual_metrics[key], places=5)
buffer = io.StringIO()
quant_debugger.layer_statistics_dump(buffer)
reader = csv.DictReader(buffer.getvalue().split())
actual_values = next(iter(reader))
expected_values = expected_metrics.copy()
expected_values.update({
'op_name': 'CONV_2D',
'tensor_idx': 7 if quantized_io else 8,
'scale': 0.15686275,
'zero_point': -128,
'tensor_name': r'Identity[1-9]?$'
})
for key, value in expected_values.items():
if isinstance(value, str):
self.assertIsNotNone(
re.match(value, actual_values[key]),
'String is different from expected string. Please fix test code if'
" it's being affected by graph manipulation changes.")
elif isinstance(value, list):
self.assertAlmostEqual(
value[0], float(actual_values[key][1:-1]), places=5)
else:
self.assertAlmostEqual(value, float(actual_values[key]), places=5)
@parameterized.named_parameters(
('float_io', False),
('quantized_io', True),
)
@test_util.run_v2_only
def test_model_metrics(self, quantized_io):
if quantized_io:
debug_model = QuantizationDebuggerTest.debug_model_int8
else:
debug_model = QuantizationDebuggerTest.debug_model_float
options = debugger.QuantizationDebugOptions(
model_debug_metrics={'stdev': lambda x, y: np.std(x[0] - y[0])})
quant_debugger = debugger.QuantizationDebugger(
quant_debug_model_content=debug_model,
float_model_content=QuantizationDebuggerTest.float_model,
debug_dataset=_calibration_gen,
debug_options=options)
quant_debugger.run()
expected_metrics = {'stdev': 0.050998904}
actual_metrics = quant_debugger.model_statistics
self.assertCountEqual(expected_metrics.keys(), actual_metrics.keys())
for key, value in expected_metrics.items():
self.assertAlmostEqual(value, actual_metrics[key], places=5)
@parameterized.named_parameters(
('float_io', False),
('quantized_io', True),
)
@test_util.run_v2_only
def test_layer_direct_compare_metrics(self, quantized_io):
def _corr(float_values, quant_values, scale, zero_point):
dequant_values = (quant_values.astype(np.int32) - zero_point) * scale
return np.corrcoef(float_values.flatten(), dequant_values.flatten())[0, 1]
if quantized_io:
debug_model = QuantizationDebuggerTest.debug_model_int8
else:
debug_model = QuantizationDebuggerTest.debug_model_float
options = debugger.QuantizationDebugOptions(
layer_direct_compare_metrics={'corr': _corr})
quant_debugger = debugger.QuantizationDebugger(
quant_debug_model_content=debug_model,
debug_dataset=_calibration_gen,
debug_options=options)
quant_debugger.run()
expected_metrics = {
'corr': 0.99999,
}
self.assertLen(quant_debugger.layer_statistics, 1)
actual_metrics = next(iter(quant_debugger.layer_statistics.values()))
for key, value in expected_metrics.items():
self.assertAlmostEqual(value, actual_metrics[key], places=5)
@test_util.run_v2_only
def test_wrong_input_raises_ValueError(self):
def wrong_calibration_gen():
for _ in range(5):
yield [
np.ones((1, 3, 3, 1), dtype=np.float32),
np.ones((1, 3, 3, 1), dtype=np.float32)
]
quant_debugger = debugger.QuantizationDebugger(
quant_debug_model_content=QuantizationDebuggerTest.debug_model_float,
debug_dataset=wrong_calibration_gen)
with self.assertRaisesRegex(
ValueError, r'inputs provided \(2\).+inputs to the model \(1\)'):
quant_debugger.run()
@test_util.run_v2_only
def test_non_debug_model_raises_ValueError(self):
normal_quant_model = _quantize_model(
QuantizationDebuggerTest.tf_model_root,
QuantizationDebuggerTest.tf_model,
_calibration_gen,
debug=False)
with self.assertRaisesRegex(
ValueError, 'Please check if the quantized model is in debug mode'):
debugger.QuantizationDebugger(
quant_debug_model_content=normal_quant_model,
debug_dataset=_calibration_gen)
@parameterized.named_parameters(
('empty quantization parameter', {
'quantization_parameters': {}
}, None),
('empty scales/zero points', {
'quantization_parameters': {
'scales': [],
'zero_points': []
}
}, None),
('invalid scales/zero points', {
'quantization_parameters': {
'scales': [1.0],
'zero_points': []
}
}, None),
('correct case', {
'quantization_parameters': {
'scales': [0.5, 1.0],
'zero_points': [42, 7]
}
}, (0.5, 42)),
)
def test_get_quant_params(self, tensor_detail, expected_value):
self.assertEqual(debugger._get_quant_params(tensor_detail), expected_value)
@parameterized.named_parameters(
('float_io', False),
('quantized_io', True))
@test_util.run_v2_only
def test_denylisted_ops(self, quantized_io):
options = debugger.QuantizationDebugOptions(
layer_debug_metrics={'l1_norm': lambda diffs: np.mean(np.abs(diffs))})
options.fully_quantize = quantized_io
quant_debugger = debugger.QuantizationDebugger(
converter=_quantize_converter(self.tf_model_root, self.tf_model,
_calibration_gen),
debug_dataset=_calibration_gen,
debug_options=options)
options.denylisted_ops = ['CONV_2D']
with self.assertRaisesRegex(
ValueError, 'Please check if the quantized model is in debug mode'):
quant_debugger.options = options
@parameterized.named_parameters(
('float_io', False),
('quantized_io', True))
@test_util.run_v2_only
def test_denylisted_nodes(self, quantized_io):
options = debugger.QuantizationDebugOptions(
layer_debug_metrics={'l1_norm': lambda diffs: np.mean(np.abs(diffs))})
options.fully_quantize = quantized_io
options.fully_quantize = quantized_io
quant_debugger = debugger.QuantizationDebugger(
converter=_quantize_converter(self.tf_model_root, self.tf_model,
_calibration_gen),
debug_dataset=_calibration_gen,
debug_options=options)
options.denylisted_nodes = ['Identity']
with self.assertRaisesRegex(
ValueError, 'Please check if the quantized model is in debug mode'):
quant_debugger.options = options
@mock.patch.object(metrics.TFLiteMetrics,
'increase_counter_debugger_creation')
def test_creation_counter(self, increase_call):
debug_model = QuantizationDebuggerTest.debug_model_float
debugger.QuantizationDebugger(
quant_debug_model_content=debug_model, debug_dataset=_calibration_gen)
increase_call.assert_called_once()
if __name__ == '__main__':
test.main()