From fa7d3e10d6256263c20b8b7678fe5366fce9708a Mon Sep 17 00:00:00 2001 From: jgaytanv Date: Tue, 20 Oct 2026 19:27:27 -0400 Subject: [PATCH 1/3] Apply QKeras weight quantizers to SimpleRNN, LSTM and GRU The QKeras parser reads the kernel, recurrent and bias quantizers of QSimpleRNN, QLSTM and QGRU into layer attributes, but the recurrent layers added their weights without them, so hls4ml wrote the float weights into the weight type while QKeras rounds them. Pass the quantizers when the weights are added, as Dense does. The Altera LSTM initializer, which splits the weights per gate, now uses the recurrent and bias quantizers for the recurrent kernel and the bias instead of the kernel quantizer for all three. Tests: the QKeras recurrent tests check that the weights equal the QKeras-quantized values and run on Vivado and Vitis too; a new test compares QSimpleRNN, QLSTM and QGRU with 8-bit and power-of-two weights against QKeras on both backends. Fixes #1579 --- hls4ml/backends/altera/altera_backend.py | 4 +- hls4ml/model/layers.py | 29 +++++++---- test/pytest/test_qkeras.py | 61 +++++++++++++++++++++--- test/pytest/test_qkerasV3.py | 53 ++++++++++++++++++-- 4 files changed, 125 insertions(+), 22 deletions(-) diff --git a/hls4ml/backends/altera/altera_backend.py b/hls4ml/backends/altera/altera_backend.py index d353847759..9354b85461 100644 --- a/hls4ml/backends/altera/altera_backend.py +++ b/hls4ml/backends/altera/altera_backend.py @@ -399,14 +399,14 @@ def init_lstm(self, layer): data=rec_weights_data[ 0 : layer.get_attr('n_out'), i * layer.get_attr('n_out') : (i + 1) * layer.get_attr('n_out') ], - quantizer=layer.get_attr('weight_quantizer'), + quantizer=layer.get_attr('recurrent_weight_quantizer'), compression=None, ) layer.add_weights_variable( name=f'bias_{weight_types[i]}', var_name=f'bias_{weight_types[i]}_{{index}}', data=bias_data[i * layer.get_attr('n_out') : (i + 1) * (layer.get_attr('n_out'))], - quantizer=layer.get_attr('weight_quantizer'), + quantizer=layer.get_attr('bias_quantizer'), compression=None, ) diff --git a/hls4ml/model/layers.py b/hls4ml/model/layers.py index 273477ed48..a254978b0b 100644 --- a/hls4ml/model/layers.py +++ b/hls4ml/model/layers.py @@ -1478,13 +1478,15 @@ def initialize(self): ) # weights - self.add_weights() + self.add_weights(quantizer=self.get_attr('weight_quantizer')) # recurrent weights - self.add_weights_variable(name='recurrent_weight', var_name='wr{index}') + self.add_weights_variable( + name='recurrent_weight', var_name='wr{index}', quantizer=self.get_attr('recurrent_weight_quantizer') + ) # biases - self.add_weights_variable(name='bias', var_name='b{index}') + self.add_weights_variable(name='bias', var_name='b{index}', quantizer=self.get_attr('bias_quantizer')) if 'pytorch' in self.attributes.keys(): self.add_weights_variable(name='recurrent_bias', var_name='br{index}') @@ -1527,14 +1529,19 @@ def initialize(self): ) # weights - self.add_weights() + self.add_weights(quantizer=self.get_attr('weight_quantizer')) # recurrent weights recurrent_weight = self.get_attr('recurrent_weight_data') - self.add_weights_variable(name='recurrent_weight', var_name='wr{index}', data=recurrent_weight) + self.add_weights_variable( + name='recurrent_weight', + var_name='wr{index}', + data=recurrent_weight, + quantizer=self.get_attr('recurrent_weight_quantizer'), + ) # biases - self.add_weights_variable(name='bias', var_name='b{index}') + self.add_weights_variable(name='bias', var_name='b{index}', quantizer=self.get_attr('bias_quantizer')) if 'pytorch' in self.attributes.keys(): self.add_weights_variable(name='recurrent_bias', var_name='br{index}') @@ -1582,14 +1589,16 @@ def initialize(self): ) # weights - self.add_weights() + self.add_weights(quantizer=self.get_attr('weight_quantizer')) # recurrent weights - self.add_weights_variable(name='recurrent_weight', var_name='wr{index}') + self.add_weights_variable( + name='recurrent_weight', var_name='wr{index}', quantizer=self.get_attr('recurrent_weight_quantizer') + ) # biases - self.add_weights_variable(name='bias', var_name='b{index}') - self.add_weights_variable(name='recurrent_bias', var_name='br{index}') + self.add_weights_variable(name='bias', var_name='b{index}', quantizer=self.get_attr('bias_quantizer')) + self.add_weights_variable(name='recurrent_bias', var_name='br{index}', quantizer=self.get_attr('bias_quantizer')) class TimeDistributed(Layer): diff --git a/test/pytest/test_qkeras.py b/test/pytest/test_qkeras.py index e726801414..413e44f3f2 100644 --- a/test/pytest/test_qkeras.py +++ b/test/pytest/test_qkeras.py @@ -5,7 +5,7 @@ import pytest from keras.layers import GRU, LSTM, BatchNormalization, Bidirectional, Input, SimpleRNN from keras.models import Model, Sequential, model_from_json -from keras.utils import to_categorical +from keras.utils import set_random_seed, to_categorical from qkeras import QGRU, QLSTM, QSimpleRNN from qkeras.qconv2d_batchnorm import QConv2DBatchnorm from qkeras.qconvolutional import QConv1D, QConv2D, QDepthwiseConv2D, QSeparableConv1D, QSeparableConv2D @@ -13,6 +13,7 @@ from qkeras.qnormalization import QBatchNormalization from qkeras.quantizers import ( binary, + get_quantizer, quantized_bits, quantized_po2, quantized_relu, @@ -414,11 +415,11 @@ def test_qactivation_kwarg(test_case_id, randX_100_10, activation_quantizer, wei 'QDepthwiseConv2D': ((8, 8, 3), lambda: QDepthwiseConv2D((3, 3)), ['Vivado', 'Vitis']), 'QSeparableConv1D': ((8, 3), lambda: QSeparableConv1D(4, 3), ['Vivado', 'Vitis']), 'QSeparableConv2D': ((8, 8, 3), lambda: QSeparableConv2D(4, (3, 3)), ['Vivado', 'Vitis']), - 'QSimpleRNN': ((5, 3), lambda: QSimpleRNN(4, activation='relu'), ['Quartus', 'Altera']), + 'QSimpleRNN': ((5, 3), lambda: QSimpleRNN(4, activation='relu'), ALL_BACKENDS), 'QLSTM': ( (5, 3), lambda: QLSTM(4, activation='tanh', recurrent_activation='sigmoid'), - ['Vivado', 'Quartus', 'Altera'], + ALL_BACKENDS, ), 'QGRU': ( (5, 3), @@ -431,7 +432,7 @@ def test_qactivation_kwarg(test_case_id, randX_100_10, activation_quantizer, wei reset_after=True, recurrent_quantizer='quantized_bits(24, 7, alpha=1)', ), - ['Vivado', 'Quartus', 'Altera'], + ALL_BACKENDS, ), 'QBatchNormalization': ( (10,), @@ -645,7 +646,14 @@ def test_quantised_po2_bit_width(test_case_id, backend, io_type, strategy): np.testing.assert_allclose(y_hls.flatten(), y_keras.flatten(), rtol=2e-2) -@pytest.mark.parametrize('backend', ['Quartus', 'Altera']) +def assert_rnn_weights_quantized(hls_layer, quantizer): + """Check that the weights of a recurrent layer are the QKeras-quantized values of the model's weights.""" + for name in ('weight', 'recurrent_weight', 'bias'): + w = hls_layer.weights[name] + np.testing.assert_array_equal(w.data, np.asarray(quantizer(w.data_unquantized))) + + +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis', 'Quartus', 'Altera']) def test_qsimplernn(test_case_id, backend): """ Test proper handling of QSimpleRNN. @@ -678,9 +686,10 @@ def test_qsimplernn(test_case_id, backend): y_hls4ml = hls_model.predict(X) np.testing.assert_allclose(y_qkeras, y_hls4ml.reshape(y_qkeras.shape), atol=0.1) + assert_rnn_weights_quantized(hls_model.graph[model.layers[0].name], quantized_bits(16, 0, alpha=1)) -@pytest.mark.parametrize('backend', ['Vivado', 'Quartus', 'Altera']) +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis', 'Quartus', 'Altera']) def test_qlstm(test_case_id, backend): """ Test proper handling of QLSTM. @@ -714,9 +723,10 @@ def test_qlstm(test_case_id, backend): y_hls4ml = hls_model.predict(X) np.testing.assert_allclose(y_qkeras, y_hls4ml.reshape(y_qkeras.shape), atol=0.1) + assert_rnn_weights_quantized(hls_model.graph[model.layers[0].name], quantized_bits(8, 0, alpha=1)) -@pytest.mark.parametrize('backend', ['Vivado', 'Quartus', 'Altera']) +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis', 'Quartus', 'Altera']) def test_qgru(test_case_id, backend): """ Test proper handling of QGRU. @@ -751,6 +761,43 @@ def test_qgru(test_case_id, backend): y_hls4ml = hls_model.predict(X) np.testing.assert_allclose(y_qkeras, y_hls4ml.reshape(y_qkeras.shape), atol=0.1) + assert_rnn_weights_quantized(hls_model.graph[model.layers[0].name], quantized_bits(8, 0, alpha=1)) + + +@pytest.mark.parametrize('layer', ['QSimpleRNN', 'QLSTM', 'QGRU']) +@pytest.mark.parametrize('quantizer', ['quantized_bits(8, 0, alpha=1)', 'quantized_po2(4)']) +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis']) +def test_qrnn_weight_quantizers(test_case_id, layer, quantizer, backend): + """The kernel, recurrent and bias quantizers of a QKeras recurrent layer are applied to its weights, so hls4ml + computes with the same weights as QKeras.""" + quantizers = dict( + kernel_quantizer=quantizer, recurrent_quantizer=quantizer, bias_quantizer='quantized_bits(8, 0, alpha=1)' + ) + layers = { + 'QSimpleRNN': lambda: QSimpleRNN(8, activation='relu', **quantizers), + 'QLSTM': lambda: QLSTM(8, activation='tanh', recurrent_activation='sigmoid', **quantizers), + 'QGRU': lambda: QGRU(8, activation='tanh', recurrent_activation='sigmoid', reset_after=True, **quantizers), + } + set_random_seed(0) + x_in = Input((6, 4)) + model = Model(x_in, layers[layer]()(x_in)) + X = np.round(np.random.default_rng(0).uniform(-0.5, 0.5, (50, 6, 4)) * 2**7) * 2**-7 + + config = hls4ml.utils.config_from_keras_model( + model, granularity='name', backend=backend, default_precision='ap_fixed<24,10>' + ) + output_dir = str(test_root_path / test_case_id) + hls_model = hls4ml.converters.convert_from_keras_model(model, hls_config=config, backend=backend, output_dir=output_dir) + hls_model.compile() + + hls_layer = hls_model.graph[model.layers[1].name] + for name, q in (('weight', quantizer), ('recurrent_weight', quantizer), ('bias', 'quantized_bits(8, 0, alpha=1)')): + w = hls_layer.weights[name] + np.testing.assert_array_equal(w.data, np.asarray(get_quantizer(q)(w.data_unquantized))) + + y_qkeras = model.predict(X, verbose=0) + y_hls4ml = hls_model.predict(X).reshape(y_qkeras.shape) + np.testing.assert_allclose(y_hls4ml, y_qkeras, rtol=0, atol=0.05) @pytest.mark.parametrize('rnn_layer', ['SimpleRNN', 'LSTM', 'GRU', 'Bidirectional']) diff --git a/test/pytest/test_qkerasV3.py b/test/pytest/test_qkerasV3.py index 82472949ca..7fcdc3d570 100644 --- a/test/pytest/test_qkerasV3.py +++ b/test/pytest/test_qkerasV3.py @@ -13,6 +13,7 @@ from qkeras.qlayers import QActivation, QDense from qkeras.quantizers import ( binary, + get_quantizer, quantized_bits, quantized_po2, quantized_relu, @@ -659,7 +660,14 @@ def test_quantised_po2_bit_width(test_case_id, backend, io_type, strategy): np.testing.assert_allclose(y_hls.flatten(), y_keras.flatten(), rtol=2e-2) -@pytest.mark.parametrize('backend', ['Quartus', 'Altera']) +def assert_rnn_weights_quantized(hls_layer, quantizer): + """Check that the weights of a recurrent layer are the QKeras-quantized values of the model's weights.""" + for name in ('weight', 'recurrent_weight', 'bias'): + w = hls_layer.weights[name] + np.testing.assert_array_equal(w.data, np.asarray(quantizer(w.data_unquantized))) + + +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis', 'Quartus', 'Altera']) def test_qsimplernn(test_case_id, backend): """ Test proper handling of QSimpleRNN. @@ -692,9 +700,10 @@ def test_qsimplernn(test_case_id, backend): y_hls4ml = hls_model.predict(X) np.testing.assert_allclose(y_qkeras, y_hls4ml.reshape(y_qkeras.shape), atol=0.1) + assert_rnn_weights_quantized(hls_model.graph[model.layers[0].name], quantized_bits(16, 0, alpha=1)) -@pytest.mark.parametrize('backend', ['Vivado', 'Quartus', 'Altera']) +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis', 'Quartus', 'Altera']) def test_qlstm(test_case_id, backend): """ Test proper handling of QLSTM. @@ -728,9 +737,10 @@ def test_qlstm(test_case_id, backend): y_hls4ml = hls_model.predict(X) np.testing.assert_allclose(y_qkeras, y_hls4ml.reshape(y_qkeras.shape), atol=0.1) + assert_rnn_weights_quantized(hls_model.graph[model.layers[0].name], quantized_bits(8, 0, alpha=1)) -@pytest.mark.parametrize('backend', ['Vivado', 'Quartus', 'Altera']) +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis', 'Quartus', 'Altera']) def test_qgru(test_case_id, backend): """ Test proper handling of QGRU. @@ -765,6 +775,43 @@ def test_qgru(test_case_id, backend): y_hls4ml = hls_model.predict(X) np.testing.assert_allclose(y_qkeras, y_hls4ml.reshape(y_qkeras.shape), atol=0.1) + assert_rnn_weights_quantized(hls_model.graph[model.layers[0].name], quantized_bits(8, 0, alpha=1)) + + +@pytest.mark.parametrize('layer', ['QSimpleRNN', 'QLSTM', 'QGRU']) +@pytest.mark.parametrize('quantizer', ['quantized_bits(8, 0, alpha=1)', 'quantized_po2(4)']) +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis']) +def test_qrnn_weight_quantizers(test_case_id, layer, quantizer, backend): + """The kernel, recurrent and bias quantizers of a QKeras recurrent layer are applied to its weights, so hls4ml + computes with the same weights as QKeras.""" + quantizers = dict( + kernel_quantizer=quantizer, recurrent_quantizer=quantizer, bias_quantizer='quantized_bits(8, 0, alpha=1)' + ) + layers = { + 'QSimpleRNN': lambda: QSimpleRNN(8, activation='relu', **quantizers), + 'QLSTM': lambda: QLSTM(8, activation='tanh', recurrent_activation='sigmoid', **quantizers), + 'QGRU': lambda: QGRU(8, activation='tanh', recurrent_activation='sigmoid', reset_after=True, **quantizers), + } + keras.utils.set_random_seed(0) + x_in = Input((6, 4)) + model = Model(x_in, layers[layer]()(x_in)) + X = np.round(np.random.default_rng(0).uniform(-0.5, 0.5, (50, 6, 4)) * 2**7) * 2**-7 + + config = hls4ml.utils.config_from_keras_model( + model, granularity='name', backend=backend, default_precision='ap_fixed<24,10>' + ) + output_dir = str(test_root_path / test_case_id) + hls_model = hls4ml.converters.convert_from_keras_model(model, hls_config=config, backend=backend, output_dir=output_dir) + hls_model.compile() + + hls_layer = hls_model.graph[model.layers[1].name] + for name, q in (('weight', quantizer), ('recurrent_weight', quantizer), ('bias', 'quantized_bits(8, 0, alpha=1)')): + w = hls_layer.weights[name] + np.testing.assert_array_equal(w.data, np.asarray(get_quantizer(q)(w.data_unquantized))) + + y_qkeras = model.predict(X, verbose=0) + y_hls4ml = hls_model.predict(X).reshape(y_qkeras.shape) + np.testing.assert_allclose(y_hls4ml, y_qkeras, rtol=0, atol=0.05) @pytest.mark.parametrize('backend', ['Vivado', 'Vitis']) From a94c7e30358bdb3a045b7ec8e92a0f9d2517025b Mon Sep 17 00:00:00 2001 From: jgaytanv Date: Thu, 29 Oct 2026 20:58:54 -0400 Subject: [PATCH 2/3] Keep the activation inside recurrent layers in the Keras 3 converter The Keras 3 converter has no handler for the qkeras-v3 recurrent layers, so they go through the Keras 2 fallback, which appended a separate Activation layer for the layer's activation. The recurrent layer applies its activation itself, so tanh was applied twice to the output of QLSTM and QGRU (relu is idempotent, so QSimpleRNN was unaffected). Skip the extra layer for recurrent layers, as the Keras 2 converter does. --- hls4ml/converters/keras_v3_to_hls.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/hls4ml/converters/keras_v3_to_hls.py b/hls4ml/converters/keras_v3_to_hls.py index d07125af4d..c4c927f4da 100644 --- a/hls4ml/converters/keras_v3_to_hls.py +++ b/hls4ml/converters/keras_v3_to_hls.py @@ -260,6 +260,8 @@ def v2_call( ret = (ret,) activation = getattr(layer, 'activation', None) + if key in ('SimpleRNN', 'LSTM', 'GRU', 'QSimpleRNN', 'QLSTM', 'QGRU', 'Bidirectional'): + activation = None # a recurrent layer applies its activation itself, as in the Keras v2 converter if activation not in (keras.activations.linear, None): assert isinstance(activation, FunctionType), f'Activation function for layer {layer.name} is not a function' intermediate_tensor_name = f'{output_names[0]}_activation' From 275f9c2148ccbdfcae04128cfee9ca0f9a1b07e4 Mon Sep 17 00:00:00 2001 From: jgaytanv Date: Thu, 29 Oct 2026 20:58:54 -0400 Subject: [PATCH 3/3] Check distinct quantizers per weight and handle zero-padded power-of-two weights Review follow-up for the QKeras recurrent quantizers. The tests use a different quantizer for the kernel, the recurrent kernel and the bias (8, 6 and 4 bits, or 4- and 5-bit power-of-two with an 8-bit bias), so a quantizer applied to the wrong weight is caught; GRU also checks the recurrent bias. The weight check compares against the quantized unquantized data and transposes the expectation on Quartus and Altera, which store the kernels transposed, so the three backend tests now check the weights on all four backends instead of skipping the check there. A conversion-only test checks the per-gate Altera LSTM weights and their types; the whole-array weights are quantized before the Altera initializer runs, so the per-gate pieces are now cut from the unquantized data and quantized once, as before. The Resource strategy of Vivado and Vitis zero-pads the recurrent kernels, and the power-of-two formatter took log2 of those zeros, writing the minimum int64 as the exponent. Zeros now get the smallest exponent of the type. The new test runs the Latency and Resource strategies, checks that every exponent is within the type, and compares the outputs against QKeras. --- hls4ml/backends/altera/altera_backend.py | 9 +- hls4ml/model/types.py | 11 +- test/pytest/test_qkeras.py | 123 +++++++++++++++++----- test/pytest/test_qkerasV3.py | 125 ++++++++++++++++++----- 4 files changed, 214 insertions(+), 54 deletions(-) diff --git a/hls4ml/backends/altera/altera_backend.py b/hls4ml/backends/altera/altera_backend.py index 9354b85461..a6d97ced78 100644 --- a/hls4ml/backends/altera/altera_backend.py +++ b/hls4ml/backends/altera/altera_backend.py @@ -377,10 +377,11 @@ def init_lstm(self, layer): self.set_closest_reuse_factor(layer, n_in_recr, n_out_recr, attribute='recurrent_reuse_factor') layer.set_attr('strategy', 'resource') - # Split weights for easier storage in on-chip memory and implementation in HLS - weights_data = layer.weights['weight'].data - rec_weights_data = layer.weights['recurrent_weight'].data - bias_data = layer.weights['bias'].data + # Split weights for easier storage in on-chip memory and implementation in HLS. The pieces are quantized + # from the unquantized weights, like the whole arrays were. + weights_data = layer.weights['weight'].data_unquantized + rec_weights_data = layer.weights['recurrent_weight'].data_unquantized + bias_data = layer.weights['bias'].data_unquantized weight_types = ['i', 'f', 'c', 'o'] for i in range(0, 4): diff --git a/hls4ml/model/types.py b/hls4ml/model/types.py index aa69130d9b..989aa05860 100644 --- a/hls4ml/model/types.py +++ b/hls4ml/model/types.py @@ -832,9 +832,14 @@ def _format(self): # Use an XnorBinary-like representation for the sign sign = np.where(y < 0, np.zeros_like(y), np.ones_like(y)) # Take the logarithm, since this is what we will write to the header - # for the optimized product using shifts - y = (np.log2(np.abs(y)) / np.log2(2.0)).astype('int') - return np.stack((sign, y), axis=-1) + # for the optimized product using shifts. A power of two is never zero, but the data can be padded with + # zeros (the Resource strategy of recurrent layers); those get the smallest exponent of the type, the value + # closest to zero it can represent. + precision = self.type.precision + exponent = np.full(y.shape, -(2 ** (precision.width - 1)) if precision.signed else 0, dtype='int') + nonzero = y != 0 + exponent[nonzero] = np.log2(np.abs(y[nonzero])).astype('int') + return np.stack((sign, exponent), axis=-1) def __iter__(self): data = self._format() diff --git a/test/pytest/test_qkeras.py b/test/pytest/test_qkeras.py index 413e44f3f2..4945882c7d 100644 --- a/test/pytest/test_qkeras.py +++ b/test/pytest/test_qkeras.py @@ -5,7 +5,7 @@ import pytest from keras.layers import GRU, LSTM, BatchNormalization, Bidirectional, Input, SimpleRNN from keras.models import Model, Sequential, model_from_json -from keras.utils import set_random_seed, to_categorical +from keras.utils import to_categorical from qkeras import QGRU, QLSTM, QSimpleRNN from qkeras.qconv2d_batchnorm import QConv2DBatchnorm from qkeras.qconvolutional import QConv1D, QConv2D, QDepthwiseConv2D, QSeparableConv1D, QSeparableConv2D @@ -27,6 +27,7 @@ from sklearn.preprocessing import LabelEncoder, StandardScaler import hls4ml +from hls4ml.model.types import ExponentWeightVariable co = {} _add_supported_quantized_objects(co) @@ -646,11 +647,17 @@ def test_quantised_po2_bit_width(test_case_id, backend, io_type, strategy): np.testing.assert_allclose(y_hls.flatten(), y_keras.flatten(), rtol=2e-2) -def assert_rnn_weights_quantized(hls_layer, quantizer): - """Check that the weights of a recurrent layer are the QKeras-quantized values of the model's weights.""" - for name in ('weight', 'recurrent_weight', 'bias'): +def assert_rnn_weights_quantized(hls_layer, quantizers, backend): + """Check that the weights of a recurrent layer are the QKeras-quantized values of the model's weights. + + ``quantizers`` maps weight names to quantizers. Quartus and Altera store the kernels transposed. + """ + for name, quantizer in quantizers.items(): w = hls_layer.weights[name] - np.testing.assert_array_equal(w.data, np.asarray(quantizer(w.data_unquantized))) + expected = np.asarray(quantizer(w.data_unquantized)) + if backend in ('Quartus', 'Altera') and expected.ndim == 2: + expected = expected.T + np.testing.assert_array_equal(w.data, expected) @pytest.mark.parametrize('backend', ['Vivado', 'Vitis', 'Quartus', 'Altera']) @@ -686,7 +693,10 @@ def test_qsimplernn(test_case_id, backend): y_hls4ml = hls_model.predict(X) np.testing.assert_allclose(y_qkeras, y_hls4ml.reshape(y_qkeras.shape), atol=0.1) - assert_rnn_weights_quantized(hls_model.graph[model.layers[0].name], quantized_bits(16, 0, alpha=1)) + q = quantized_bits(16, 0, alpha=1) + assert_rnn_weights_quantized( + hls_model.graph[model.layers[0].name], {'weight': q, 'recurrent_weight': q, 'bias': q}, backend + ) @pytest.mark.parametrize('backend', ['Vivado', 'Vitis', 'Quartus', 'Altera']) @@ -723,7 +733,10 @@ def test_qlstm(test_case_id, backend): y_hls4ml = hls_model.predict(X) np.testing.assert_allclose(y_qkeras, y_hls4ml.reshape(y_qkeras.shape), atol=0.1) - assert_rnn_weights_quantized(hls_model.graph[model.layers[0].name], quantized_bits(8, 0, alpha=1)) + q = quantized_bits(8, 0, alpha=1) + assert_rnn_weights_quantized( + hls_model.graph[model.layers[0].name], {'weight': q, 'recurrent_weight': q, 'bias': q}, backend + ) @pytest.mark.parametrize('backend', ['Vivado', 'Vitis', 'Quartus', 'Altera']) @@ -761,45 +774,109 @@ def test_qgru(test_case_id, backend): y_hls4ml = hls_model.predict(X) np.testing.assert_allclose(y_qkeras, y_hls4ml.reshape(y_qkeras.shape), atol=0.1) - assert_rnn_weights_quantized(hls_model.graph[model.layers[0].name], quantized_bits(8, 0, alpha=1)) + q = quantized_bits(8, 0, alpha=1) + assert_rnn_weights_quantized( + hls_model.graph[model.layers[0].name], + {'weight': q, 'recurrent_weight': q, 'bias': q, 'recurrent_bias': q}, + backend, + ) -@pytest.mark.parametrize('layer', ['QSimpleRNN', 'QLSTM', 'QGRU']) -@pytest.mark.parametrize('quantizer', ['quantized_bits(8, 0, alpha=1)', 'quantized_po2(4)']) -@pytest.mark.parametrize('backend', ['Vivado', 'Vitis']) -def test_qrnn_weight_quantizers(test_case_id, layer, quantizer, backend): - """The kernel, recurrent and bias quantizers of a QKeras recurrent layer are applied to its weights, so hls4ml - computes with the same weights as QKeras.""" - quantizers = dict( - kernel_quantizer=quantizer, recurrent_quantizer=quantizer, bias_quantizer='quantized_bits(8, 0, alpha=1)' +RNN_QUANTIZERS = { + 'quantized_bits': { + 'weight': 'quantized_bits(8, 0, alpha=1)', + 'recurrent_weight': 'quantized_bits(6, 0, alpha=1)', + 'bias': 'quantized_bits(4, 0, alpha=1)', + }, + 'quantized_po2': { + 'weight': 'quantized_po2(4)', + 'recurrent_weight': 'quantized_po2(5)', + 'bias': 'quantized_bits(8, 0, alpha=1)', + }, +} + + +def make_qrnn_model(layer, quantizers): + """The model of #1579: one QKeras recurrent layer of 8 units on a (6, 4) input, with the given kernel, recurrent + and bias quantizers. The weights are drawn from (-0.5, 0.5), so the bias is not zero and the model is the same + with every Keras version.""" + kwargs = dict( + kernel_quantizer=quantizers['weight'], + recurrent_quantizer=quantizers['recurrent_weight'], + bias_quantizer=quantizers['bias'], ) layers = { - 'QSimpleRNN': lambda: QSimpleRNN(8, activation='relu', **quantizers), - 'QLSTM': lambda: QLSTM(8, activation='tanh', recurrent_activation='sigmoid', **quantizers), - 'QGRU': lambda: QGRU(8, activation='tanh', recurrent_activation='sigmoid', reset_after=True, **quantizers), + 'QSimpleRNN': lambda: QSimpleRNN(8, activation='relu', **kwargs), + 'QLSTM': lambda: QLSTM(8, activation='tanh', recurrent_activation='sigmoid', **kwargs), + 'QGRU': lambda: QGRU(8, activation='tanh', recurrent_activation='sigmoid', reset_after=True, **kwargs), } - set_random_seed(0) x_in = Input((6, 4)) model = Model(x_in, layers[layer]()(x_in)) + rng = np.random.default_rng(1) + model.layers[1].set_weights([rng.uniform(-0.5, 0.5, w.shape).astype('float32') for w in model.layers[1].get_weights()]) + return model + + +@pytest.mark.parametrize('layer', ['QSimpleRNN', 'QLSTM', 'QGRU']) +@pytest.mark.parametrize('quantizers', list(RNN_QUANTIZERS)) +@pytest.mark.parametrize('strategy', ['Latency', 'Resource']) +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis']) +def test_qrnn_weight_quantizers(test_case_id, layer, quantizers, strategy, backend): + """The kernel, recurrent and bias quantizers of a QKeras recurrent layer are applied to its weights, so hls4ml + computes with the same weights as QKeras. The three quantizers differ, so each weight has to get its own.""" + quantizers = RNN_QUANTIZERS[quantizers] + model = make_qrnn_model(layer, quantizers) X = np.round(np.random.default_rng(0).uniform(-0.5, 0.5, (50, 6, 4)) * 2**7) * 2**-7 config = hls4ml.utils.config_from_keras_model( model, granularity='name', backend=backend, default_precision='ap_fixed<24,10>' ) + config['LayerName'][model.layers[1].name]['Strategy'] = strategy + if strategy == 'Resource': + # 4 inputs in 2 lanes over 3 cycles: the weights are padded with zeros + config['LayerName'][model.layers[1].name]['ReuseFactor'] = 3 output_dir = str(test_root_path / test_case_id) hls_model = hls4ml.converters.convert_from_keras_model(model, hls_config=config, backend=backend, output_dir=output_dir) hls_model.compile() hls_layer = hls_model.graph[model.layers[1].name] - for name, q in (('weight', quantizer), ('recurrent_weight', quantizer), ('bias', 'quantized_bits(8, 0, alpha=1)')): - w = hls_layer.weights[name] - np.testing.assert_array_equal(w.data, np.asarray(get_quantizer(q)(w.data_unquantized))) + if strategy == 'Latency': # the Resource strategy lays the weights out in the order its cell reads them + expected = {name: get_quantizer(q) for name, q in quantizers.items()} + if layer == 'QGRU': + expected['recurrent_bias'] = get_quantizer(quantizers['bias']) + assert_rnn_weights_quantized(hls_layer, expected, backend) + for w in hls_layer.weights.values(): + if isinstance(w, ExponentWeightVariable): # every exponent written to the project fits the type, zero padding too + assert w._format()[..., 1].min() >= -(2 ** (w.type.precision.width - 1)) y_qkeras = model.predict(X, verbose=0) y_hls4ml = hls_model.predict(X).reshape(y_qkeras.shape) np.testing.assert_allclose(y_hls4ml, y_qkeras, rtol=0, atol=0.05) +def test_qlstm_altera_gate_quantizers(test_case_id): + """On Altera the LSTM weights are split per gate, and each piece gets the quantizer and the type of the array it + comes from: the kernel, the recurrent kernel or the bias. Conversion only; compiling needs the Altera toolchain.""" + bits = {'weight': 8, 'recurrent_weight': 4, 'bias': 3} + quantizers = {name: f'quantized_bits({b}, 0, alpha=1)' for name, b in bits.items()} + model = make_qrnn_model('QLSTM', quantizers) + config = hls4ml.utils.config_from_keras_model( + model, granularity='name', backend='Altera', default_precision='ap_fixed<24,10>' + ) + output_dir = str(test_root_path / test_case_id) + hls_model = hls4ml.converters.convert_from_keras_model(model, hls_config=config, backend='Altera', output_dir=output_dir) + + hls_layer = hls_model.graph[model.layers[1].name] + assert_rnn_weights_quantized(hls_layer, {name: get_quantizer(q) for name, q in quantizers.items()}, 'Altera') + arrays = dict(zip(('weight', 'recurrent_weight', 'bias'), model.layers[1].get_weights())) + for i, gate in enumerate('ifco'): + for name, q in quantizers.items(): + w = hls_layer.weights[f'{name}_{gate}'] + expected = np.asarray(get_quantizer(q)(arrays[name][..., 8 * i : 8 * (i + 1)])) + np.testing.assert_array_equal(w.data, expected.T if expected.ndim == 2 else expected) + assert (w.type.precision.width, w.type.precision.integer) == (bits[name], 1) + + @pytest.mark.parametrize('rnn_layer', ['SimpleRNN', 'LSTM', 'GRU', 'Bidirectional']) @pytest.mark.parametrize('backend', ['Vivado', 'Vitis']) @pytest.mark.parametrize('io_type', ['io_parallel', 'io_stream']) diff --git a/test/pytest/test_qkerasV3.py b/test/pytest/test_qkerasV3.py index 7fcdc3d570..0eeb4442f4 100644 --- a/test/pytest/test_qkerasV3.py +++ b/test/pytest/test_qkerasV3.py @@ -27,6 +27,7 @@ from sklearn.preprocessing import LabelEncoder, StandardScaler import hls4ml +from hls4ml.model.types import ExponentWeightVariable _original_init = QDense.__init__ @@ -660,11 +661,17 @@ def test_quantised_po2_bit_width(test_case_id, backend, io_type, strategy): np.testing.assert_allclose(y_hls.flatten(), y_keras.flatten(), rtol=2e-2) -def assert_rnn_weights_quantized(hls_layer, quantizer): - """Check that the weights of a recurrent layer are the QKeras-quantized values of the model's weights.""" - for name in ('weight', 'recurrent_weight', 'bias'): +def assert_rnn_weights_quantized(hls_layer, quantizers, backend): + """Check that the weights of a recurrent layer are the QKeras-quantized values of the model's weights. + + ``quantizers`` maps weight names to quantizers. Quartus and Altera store the kernels transposed. + """ + for name, quantizer in quantizers.items(): w = hls_layer.weights[name] - np.testing.assert_array_equal(w.data, np.asarray(quantizer(w.data_unquantized))) + expected = np.asarray(quantizer(w.data_unquantized)) + if backend in ('Quartus', 'Altera') and expected.ndim == 2: + expected = expected.T + np.testing.assert_array_equal(w.data, expected) @pytest.mark.parametrize('backend', ['Vivado', 'Vitis', 'Quartus', 'Altera']) @@ -700,7 +707,10 @@ def test_qsimplernn(test_case_id, backend): y_hls4ml = hls_model.predict(X) np.testing.assert_allclose(y_qkeras, y_hls4ml.reshape(y_qkeras.shape), atol=0.1) - assert_rnn_weights_quantized(hls_model.graph[model.layers[0].name], quantized_bits(16, 0, alpha=1)) + q = quantized_bits(16, 0, alpha=1) + assert_rnn_weights_quantized( + hls_model.graph[model.layers[0].name], {'weight': q, 'recurrent_weight': q, 'bias': q}, backend + ) @pytest.mark.parametrize('backend', ['Vivado', 'Vitis', 'Quartus', 'Altera']) @@ -737,7 +747,10 @@ def test_qlstm(test_case_id, backend): y_hls4ml = hls_model.predict(X) np.testing.assert_allclose(y_qkeras, y_hls4ml.reshape(y_qkeras.shape), atol=0.1) - assert_rnn_weights_quantized(hls_model.graph[model.layers[0].name], quantized_bits(8, 0, alpha=1)) + q = quantized_bits(8, 0, alpha=1) + assert_rnn_weights_quantized( + hls_model.graph[model.layers[0].name], {'weight': q, 'recurrent_weight': q, 'bias': q}, backend + ) @pytest.mark.parametrize('backend', ['Vivado', 'Vitis', 'Quartus', 'Altera']) @@ -775,45 +788,109 @@ def test_qgru(test_case_id, backend): y_hls4ml = hls_model.predict(X) np.testing.assert_allclose(y_qkeras, y_hls4ml.reshape(y_qkeras.shape), atol=0.1) - assert_rnn_weights_quantized(hls_model.graph[model.layers[0].name], quantized_bits(8, 0, alpha=1)) - - -@pytest.mark.parametrize('layer', ['QSimpleRNN', 'QLSTM', 'QGRU']) -@pytest.mark.parametrize('quantizer', ['quantized_bits(8, 0, alpha=1)', 'quantized_po2(4)']) -@pytest.mark.parametrize('backend', ['Vivado', 'Vitis']) -def test_qrnn_weight_quantizers(test_case_id, layer, quantizer, backend): - """The kernel, recurrent and bias quantizers of a QKeras recurrent layer are applied to its weights, so hls4ml - computes with the same weights as QKeras.""" - quantizers = dict( - kernel_quantizer=quantizer, recurrent_quantizer=quantizer, bias_quantizer='quantized_bits(8, 0, alpha=1)' + q = quantized_bits(8, 0, alpha=1) + assert_rnn_weights_quantized( + hls_model.graph[model.layers[0].name], + {'weight': q, 'recurrent_weight': q, 'bias': q, 'recurrent_bias': q}, + backend, + ) + + +RNN_QUANTIZERS = { + 'quantized_bits': { + 'weight': 'quantized_bits(8, 0, alpha=1)', + 'recurrent_weight': 'quantized_bits(6, 0, alpha=1)', + 'bias': 'quantized_bits(4, 0, alpha=1)', + }, + 'quantized_po2': { + 'weight': 'quantized_po2(4)', + 'recurrent_weight': 'quantized_po2(5)', + 'bias': 'quantized_bits(8, 0, alpha=1)', + }, +} + + +def make_qrnn_model(layer, quantizers): + """The model of #1579: one QKeras recurrent layer of 8 units on a (6, 4) input, with the given kernel, recurrent + and bias quantizers. The weights are drawn from (-0.5, 0.5), so the bias is not zero and the model is the same + with every Keras version.""" + kwargs = dict( + kernel_quantizer=quantizers['weight'], + recurrent_quantizer=quantizers['recurrent_weight'], + bias_quantizer=quantizers['bias'], ) layers = { - 'QSimpleRNN': lambda: QSimpleRNN(8, activation='relu', **quantizers), - 'QLSTM': lambda: QLSTM(8, activation='tanh', recurrent_activation='sigmoid', **quantizers), - 'QGRU': lambda: QGRU(8, activation='tanh', recurrent_activation='sigmoid', reset_after=True, **quantizers), + 'QSimpleRNN': lambda: QSimpleRNN(8, activation='relu', **kwargs), + 'QLSTM': lambda: QLSTM(8, activation='tanh', recurrent_activation='sigmoid', **kwargs), + 'QGRU': lambda: QGRU(8, activation='tanh', recurrent_activation='sigmoid', reset_after=True, **kwargs), } - keras.utils.set_random_seed(0) x_in = Input((6, 4)) model = Model(x_in, layers[layer]()(x_in)) + rng = np.random.default_rng(1) + model.layers[1].set_weights([rng.uniform(-0.5, 0.5, w.shape).astype('float32') for w in model.layers[1].get_weights()]) + return model + + +@pytest.mark.parametrize('layer', ['QSimpleRNN', 'QLSTM', 'QGRU']) +@pytest.mark.parametrize('quantizers', list(RNN_QUANTIZERS)) +@pytest.mark.parametrize('strategy', ['Latency', 'Resource']) +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis']) +def test_qrnn_weight_quantizers(test_case_id, layer, quantizers, strategy, backend): + """The kernel, recurrent and bias quantizers of a QKeras recurrent layer are applied to its weights, so hls4ml + computes with the same weights as QKeras. The three quantizers differ, so each weight has to get its own.""" + quantizers = RNN_QUANTIZERS[quantizers] + model = make_qrnn_model(layer, quantizers) X = np.round(np.random.default_rng(0).uniform(-0.5, 0.5, (50, 6, 4)) * 2**7) * 2**-7 config = hls4ml.utils.config_from_keras_model( model, granularity='name', backend=backend, default_precision='ap_fixed<24,10>' ) + config['LayerName'][model.layers[1].name]['Strategy'] = strategy + if strategy == 'Resource': + # 4 inputs in 2 lanes over 3 cycles: the weights are padded with zeros + config['LayerName'][model.layers[1].name]['ReuseFactor'] = 3 output_dir = str(test_root_path / test_case_id) hls_model = hls4ml.converters.convert_from_keras_model(model, hls_config=config, backend=backend, output_dir=output_dir) hls_model.compile() hls_layer = hls_model.graph[model.layers[1].name] - for name, q in (('weight', quantizer), ('recurrent_weight', quantizer), ('bias', 'quantized_bits(8, 0, alpha=1)')): - w = hls_layer.weights[name] - np.testing.assert_array_equal(w.data, np.asarray(get_quantizer(q)(w.data_unquantized))) + if strategy == 'Latency': # the Resource strategy lays the weights out in the order its cell reads them + expected = {name: get_quantizer(q) for name, q in quantizers.items()} + if layer == 'QGRU': + expected['recurrent_bias'] = get_quantizer(quantizers['bias']) + assert_rnn_weights_quantized(hls_layer, expected, backend) + for w in hls_layer.weights.values(): + if isinstance(w, ExponentWeightVariable): # every exponent written to the project fits the type, zero padding too + assert w._format()[..., 1].min() >= -(2 ** (w.type.precision.width - 1)) y_qkeras = model.predict(X, verbose=0) y_hls4ml = hls_model.predict(X).reshape(y_qkeras.shape) np.testing.assert_allclose(y_hls4ml, y_qkeras, rtol=0, atol=0.05) +def test_qlstm_altera_gate_quantizers(test_case_id): + """On Altera the LSTM weights are split per gate, and each piece gets the quantizer and the type of the array it + comes from: the kernel, the recurrent kernel or the bias. Conversion only; compiling needs the Altera toolchain.""" + bits = {'weight': 8, 'recurrent_weight': 4, 'bias': 3} + quantizers = {name: f'quantized_bits({b}, 0, alpha=1)' for name, b in bits.items()} + model = make_qrnn_model('QLSTM', quantizers) + config = hls4ml.utils.config_from_keras_model( + model, granularity='name', backend='Altera', default_precision='ap_fixed<24,10>' + ) + output_dir = str(test_root_path / test_case_id) + hls_model = hls4ml.converters.convert_from_keras_model(model, hls_config=config, backend='Altera', output_dir=output_dir) + + hls_layer = hls_model.graph[model.layers[1].name] + assert_rnn_weights_quantized(hls_layer, {name: get_quantizer(q) for name, q in quantizers.items()}, 'Altera') + arrays = dict(zip(('weight', 'recurrent_weight', 'bias'), model.layers[1].get_weights())) + for i, gate in enumerate('ifco'): + for name, q in quantizers.items(): + w = hls_layer.weights[f'{name}_{gate}'] + expected = np.asarray(get_quantizer(q)(arrays[name][..., 8 * i : 8 * (i + 1)])) + np.testing.assert_array_equal(w.data, expected.T if expected.ndim == 2 else expected) + assert (w.type.precision.width, w.type.precision.integer) == (bits[name], 1) + + @pytest.mark.parametrize('backend', ['Vivado', 'Vitis']) @pytest.mark.parametrize('io_type', ['io_stream']) def test_qseparableconv1d(test_case_id, backend, io_type):