diff --git a/docs/intro/setup.rst b/docs/intro/setup.rst index 4e3d192fcf..5e6913c8c5 100644 --- a/docs/intro/setup.rst +++ b/docs/intro/setup.rst @@ -54,6 +54,7 @@ The following Python packages are all optional and are only required if you inte * Quantization support * `QKeras `_: based on Keras v2. See `frontend/keras <../frontend/keras.html>`_ for more details + * `QKeras-v3 `_: based on Keras v3. See `frontend/keras <../frontend/keras.html>`_ for more details * `HGQ `_: Based on Keras v2. See `advanced/HGQ <../advanced/hgq.html>`_ for more details. * `HGQ2 `_: Based on Keras v3. See `advanced/HGQ2 <../advanced/hgq.html>`_ for more details. * `Brevitas `_: Based on PyTorch. See `frontend/pytorch <../frontend/pytorch.html>`_ for more details. @@ -197,6 +198,9 @@ Optional Dependencies # For QKeras frontend pip install hls4ml[qkeras] + # For QKeras-v3 frontend + pip install hls4ml[qkeras-v3] + # For Quartus report parsing pip install hls4ml[quartus-report] diff --git a/docs/intro/status.rst b/docs/intro/status.rst index 7526c3bec4..151778c16b 100644 --- a/docs/intro/status.rst +++ b/docs/intro/status.rst @@ -25,6 +25,7 @@ Frontend support: * HGQ * Keras v3 + * QKeras-v3 * HGQ2 * PyTorch * ONNX @@ -59,6 +60,8 @@ A summary of the on-going status of the ``hls4ml`` tool is in the table below. +-----------------------+-----+-----+--------------+--------+--------+-----+ | QKeras | ✅ | ✅ | ✅ | ✅ | N/A | N/A | +-----------------------+-----+-----+--------------+--------+--------+-----+ +| QKeras-v3 | ✅ | ✅ | ✅ | ✅ | ✅ | N/A | ++-----------------------+-----+-----+--------------+--------+--------+-----+ | HGQ | ✅ | ✅ | N/A | N/A | N/A | N/A | +-----------------------+-----+-----+--------------+--------+--------+-----+ | Keras v3 | ✅ | ✅ | ✅ | N/A | ✅ | ❌ | diff --git a/example-models b/example-models index e7a9dee394..ce390f9a80 160000 --- a/example-models +++ b/example-models @@ -1 +1 @@ -Subproject commit e7a9dee394b6c1f6e0eb23178d34e55f077297fe +Subproject commit ce390f9a806dac2c254736c6113896c540e7447b diff --git a/hls4ml/converters/keras_v3/__init__.py b/hls4ml/converters/keras_v3/__init__.py index 21950aea6c..807e14171a 100644 --- a/hls4ml/converters/keras_v3/__init__.py +++ b/hls4ml/converters/keras_v3/__init__.py @@ -5,6 +5,7 @@ hgq2, # noqa: F401 merge, # noqa: F401 pooling, # noqa: F401 + qkeras, # noqa: F401 recurrent, # noqa: F401 ) from ._base import registry as layer_handlers diff --git a/hls4ml/converters/keras_v3/qkeras/__init__.py b/hls4ml/converters/keras_v3/qkeras/__init__.py new file mode 100644 index 0000000000..bcbb5b40cc --- /dev/null +++ b/hls4ml/converters/keras_v3/qkeras/__init__.py @@ -0,0 +1 @@ +from . import activation, layer, utils diff --git a/hls4ml/converters/keras_v3/qkeras/activation.py b/hls4ml/converters/keras_v3/qkeras/activation.py new file mode 100644 index 0000000000..281da5cb30 --- /dev/null +++ b/hls4ml/converters/keras_v3/qkeras/activation.py @@ -0,0 +1,35 @@ +from typing import Any + +from hls4ml.converters.utils import IsolatedLayerReader + +from ..core import KerasV3LayerHandler +from .utils import set_default_config + + +class QKerasQActivationHandler(KerasV3LayerHandler): + handles = ('qkeras.qlayers.QActivation', 'QActivation') + + def handle( + self, + layer, + in_tensors, + out_tensors, + ) -> tuple[dict[str, Any], ...]: + + config = layer.get_config() + layer_dict = {'config': config, 'class_name': layer.__class__.__name__} + + reader = IsolatedLayerReader(layer) + input_shapes = [list(t.shape) for t in in_tensors] + input_names = [t.name for t in in_tensors] + + from hls4ml.converters.keras_v2_to_hls import layer_handlers as v2_layer_handlers + + v2_handler = v2_layer_handlers.get(layer.__class__.__name__) + if v2_handler is None: + raise ValueError(f'No v2 handler found for {layer.__class__.__name__}') + + hls_conf, _ = v2_handler(layer_dict, input_names, input_shapes, reader) + hls_conf = set_default_config(hls_conf, self.default_config) + + return (hls_conf,) diff --git a/hls4ml/converters/keras_v3/qkeras/layer.py b/hls4ml/converters/keras_v3/qkeras/layer.py new file mode 100644 index 0000000000..eb1cf5ccd9 --- /dev/null +++ b/hls4ml/converters/keras_v3/qkeras/layer.py @@ -0,0 +1,49 @@ +from hls4ml.converters.utils import IsolatedLayerReader + +from ..core import KerasV3LayerHandler +from .utils import set_default_config + + +class QKerasV3LayerHandler(KerasV3LayerHandler): + handles = ( + 'qkeras.qlayers.QDense', + 'qkeras.qconvolutional.QConv1D', + 'qkeras.qconvolutional.QConv2D', + 'qkeras.qconvolutional.QDepthwiseConv2D', + 'qkeras.qconv2d_batchnorm.QConv2DBatchnorm', + ) + + def handle(self, layer, in_tensors, out_tensors): + config = layer.get_config() + layer_dict = {'config': config, 'class_name': layer.__class__.__name__} + + reader = IsolatedLayerReader(layer) + input_shapes = [list(t.shape) for t in in_tensors] + input_names = [t.name for t in in_tensors] + + from hls4ml.converters.keras_v2_to_hls import layer_handlers as v2_layer_handlers + + v2_handler = v2_layer_handlers.get(layer.__class__.__name__) + if v2_handler is None: + raise ValueError(f'No v2 handler found for {layer.__class__.__name__}') + + ret, _ = v2_handler(layer_dict, input_names, input_shapes, reader) + ret = set_default_config(ret, self.default_config) + + activation = config.get('activation') + if activation not in (None, 'linear'): + from hls4ml.converters.keras.qkeras import get_activation_quantizer + + activation_config = get_activation_quantizer(layer_dict, input_names) + intermediate_tensor_name = f'{out_tensors[0].name}_activation' + ret['output_keras_tensor_names'] = [intermediate_tensor_name] + activation_config.update( + { + 'name': f'{layer.name}_activation', + 'input_keras_tensor_names': [intermediate_tensor_name], + 'output_keras_tensor_names': [out_tensors[0].name], + } + ) + return ret, activation_config + + return ret diff --git a/hls4ml/converters/keras_v3/qkeras/utils.py b/hls4ml/converters/keras_v3/qkeras/utils.py new file mode 100644 index 0000000000..e9d6fef8e8 --- /dev/null +++ b/hls4ml/converters/keras_v3/qkeras/utils.py @@ -0,0 +1,5 @@ +def set_default_config(hls_conf, default_config): + for key, value in default_config.items(): + if key not in hls_conf.keys(): + hls_conf[key] = value + return hls_conf diff --git a/hls4ml/converters/keras_v3_to_hls.py b/hls4ml/converters/keras_v3_to_hls.py index 359bc391d6..9abd031b56 100644 --- a/hls4ml/converters/keras_v3_to_hls.py +++ b/hls4ml/converters/keras_v3_to_hls.py @@ -4,8 +4,7 @@ from types import FunctionType from typing import Any -import numpy as np - +from hls4ml.converters.utils import IsolatedLayerReader from hls4ml.model import ModelGraph if typing.TYPE_CHECKING: @@ -238,15 +237,7 @@ def v2_call( config = layer.get_config() layer_dict = {'config': config, 'class_name': layer.__class__.__name__} - class IsolatedLayerReader: - def get_weights_data(self, layer_name, var_name): - assert layer_name == layer.name, f'Processing {layer.name}, but handler tried to read {layer_name}' - for w in layer.weights: - if var_name in w.name: - return np.array(w) - return None - - reader = IsolatedLayerReader() + reader = IsolatedLayerReader(layer) input_shapes = [list(t.shape) for t in inp_tensors] input_names = [t.name for t in inp_tensors] output_names = [t.name for t in out_tensors] diff --git a/hls4ml/converters/utils.py b/hls4ml/converters/utils.py index d35ec11fff..e4ddda2c50 100644 --- a/hls4ml/converters/utils.py +++ b/hls4ml/converters/utils.py @@ -1,5 +1,7 @@ import math +import numpy as np + def parse_data_format(input_shape, data_format='channels_last'): """Parses the given input shape according to the specified data format. @@ -287,3 +289,15 @@ def compute_padding_2d_pytorch( pad_left = pad_width return (out_height, out_width, pad_top, pad_bottom, pad_left, pad_right) + + +class IsolatedLayerReader: + def __init__(self, layer): + self.layer = layer + + def get_weights_data(self, layer_name, var_name): + assert layer_name == self.layer.name, f'Processing {self.layer.name}, but handler tried to read {layer_name}' + for w in self.layer.weights: + if var_name in w.name: + return np.array(w) + return None diff --git a/hls4ml/model/quantizers.py b/hls4ml/model/quantizers.py index 833d27a0d2..d306a4da75 100644 --- a/hls4ml/model/quantizers.py +++ b/hls4ml/model/quantizers.py @@ -127,7 +127,6 @@ def __init__(self, config): def __call__(self, data): data = np.array(data, dtype='float32') return self.quantizer_fn(data).numpy() - # return self.quantizer_fn(data) def _get_type(self, quantizer_config): width = quantizer_config['config']['bits'] @@ -170,7 +169,9 @@ def __init__(self, config, xnor=False): def __call__(self, data): data = np.array(data, dtype='float32') - y = self.quantizer_fn(data).numpy() + y = self.quantizer_fn(data) + if hasattr(y, 'numpy'): + y = y.numpy() return self.binary_quantizer(y) def serialize_state(self): diff --git a/hls4ml/model/types.py b/hls4ml/model/types.py index 5d434d8655..aa69130d9b 100644 --- a/hls4ml/model/types.py +++ b/hls4ml/model/types.py @@ -5,6 +5,7 @@ higher-dimensional tensors, which are defined as arrays or FIFO streams in the generated code. """ +import math from enum import Enum import numpy as np @@ -837,7 +838,7 @@ def _format(self): def __iter__(self): data = self._format() - self._iterator = iter(data.reshape((np.product(data.shape[:-1]), 2))) + self._iterator = iter(data.reshape((math.prod(data.shape[:-1]), 2))) return self def __next__(self): diff --git a/pyproject.toml b/pyproject.toml index a39c7cb362..799235cab7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,6 +49,7 @@ optional-dependencies.qkeras = [ "tensorflow>=2.8,<=2.14.1", "tensorflow-model-optimization<=0.7.5", ] +optional-dependencies.qkeras-v3 = [ "qkeras-v3" ] optional-dependencies.quartus-report = [ "calmjs-parse", "tabulate" ] optional-dependencies.sr = [ "sympy>=1.13.1" ] optional-dependencies.testing = [ @@ -72,6 +73,11 @@ optional-dependencies.testing-keras3 = [ "keras>=3.10", "tensorflow>=2.15", ] +optional-dependencies.testing-qkeras-v3 = [ + "keras==3.14.1", + "qkeras-v3", + "tensorflow>=2.21", +] urls.Homepage = "https://fastmachinelearning.org/hls4ml" scripts.hls4ml = "hls4ml.cli:main" entry-points.pytest_randomly.random_seeder = "hls4ml:reseed" diff --git a/test/pytest/ci-template.yml b/test/pytest/ci-template.yml index ebbcd8a21e..95447997ca 100644 --- a/test/pytest/ci-template.yml +++ b/test/pytest/ci-template.yml @@ -50,3 +50,9 @@ variables: CONDA_ENV: "hls4ml-testing-keras3" EXTRA_DEPS: "[da,testing,testing-keras3,sr]" + +.pytest-qkeras-v3-only: + extends: .pytest + variables: + CONDA_ENV: "hls4ml-testing-keras3" + EXTRA_DEPS: "[qkeras-v3]" diff --git a/test/pytest/generate_ci_yaml.py b/test/pytest/generate_ci_yaml.py index 684abc0511..a48bed9e7e 100644 --- a/test/pytest/generate_ci_yaml.py +++ b/test/pytest/generate_ci_yaml.py @@ -36,6 +36,7 @@ 'test_multiout_onnx', 'test_keras_v3_profiling', } +QKERAS3_LIST = {'test_qkerasV3'} # Test files to split by individual test cases # Value = chunk size per CI job @@ -79,7 +80,7 @@ def generate_test_yaml(test_root='.'): test_paths = [ path for path in test_root.glob('**/test_*.py') - if path.stem not in (BLACKLIST | LONGLIST | set(SPLIT_BY_TEST_CASE.keys()) | KERAS3_LIST) + if path.stem not in (BLACKLIST | LONGLIST | set(SPLIT_BY_TEST_CASE.keys()) | KERAS3_LIST | QKERAS3_LIST) ] need_example_models = [uses_example_model(path) for path in test_paths] @@ -140,6 +141,21 @@ def generate_test_yaml(test_root='.'): diff_yml = yaml.safe_load(template.format(name, '.pytest-keras3-only', test_files, batch_need_example_model)) yml.update(diff_yml) + qkeras3_paths = [path for path in test_root.glob('**/test_*.py') if path.stem in QKERAS3_LIST] + qkeras3_need_examples = [uses_example_model(path) for path in qkeras3_paths] + + qk3_idxs = list(range(len(qkeras3_need_examples))) + qk3_idxs = sorted(qk3_idxs, key=lambda i: f'{qkeras3_need_examples[i]}_{path_to_name(qkeras3_paths[i])}') + + for batch_idxs in batched(qk3_idxs, n_test_files_per_yml): + batch_paths: list[Path] = [qkeras3_paths[i] for i in batch_idxs] + names = [path_to_name(path) for path in batch_paths] + name = 'qkerasV3' + test_files = ' '.join([str(path.relative_to(test_root)) for path in batch_paths]) + batch_need_example_model = int(any([qkeras3_need_examples[i] for i in batch_idxs])) + diff_yml = yaml.safe_load(template.format(name, '.pytest-qkeras-v3-only', test_files, batch_need_example_model)) + yml.update(diff_yml) + return yml diff --git a/test/pytest/test_qkerasV3.py b/test/pytest/test_qkerasV3.py new file mode 100644 index 0000000000..cd9429108b --- /dev/null +++ b/test/pytest/test_qkerasV3.py @@ -0,0 +1,859 @@ +import warnings +from pathlib import Path + +import keras +import numpy as np +import pytest +from keras.layers import BatchNormalization, EinsumDense, Input +from keras.models import Model, Sequential +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 +from qkeras.qlayers import QActivation, QDense +from qkeras.quantizers import ( + binary, + quantized_bits, + quantized_po2, + quantized_relu, + quantized_sigmoid, + quantized_tanh, + ternary, +) +from qkeras.utils import _add_supported_quantized_objects +from sklearn.datasets import fetch_openml +from sklearn.model_selection import train_test_split +from sklearn.preprocessing import LabelEncoder, StandardScaler + +import hls4ml + +_original_init = QDense.__init__ + + +def patched_init(self, *args, **kwargs): + kwargs.pop('quantization_config', None) + _original_init(self, *args, **kwargs) + + +QDense.__init__ = patched_init + +co = {} +_add_supported_quantized_objects(co) + + +warnings.filterwarnings('ignore', message='numpy.dtype size changed') +warnings.filterwarnings('ignore', message='numpy.ufunc size changed') + +test_root_path = Path(__file__).parent +example_model_path = (test_root_path / '../../example-models').resolve() + + +@pytest.fixture(scope='module') +def get_jettagging_data(): + """ + Download the jet tagging dataset + """ + print('Fetching data from openml') + data = fetch_openml('hls4ml_lhc_jets_hlf') + X, y = data['data'], data['target'] + le = LabelEncoder() + y = le.fit_transform(y) + y = to_categorical(y, 5) + X_train_val, X_test, y_train_val, y_test = train_test_split(X, y, test_size=0.2, random_state=42) + scaler = StandardScaler() + X_train_val = scaler.fit_transform(X_train_val) + X_test = scaler.transform(X_test) + return X_train_val, X_test, y_train_val, y_test + + +@pytest.fixture(scope='module') +def load_jettagging_model(): + """ + Load the 3 hidden layer QKeras example model trained on the jet tagging dataset + """ + model_path = example_model_path / 'keras/qkeras-v3_3layer.keras' + model = keras.saving.load_model( + model_path, + custom_objects=co, + compile=False, + safe_mode=False, + ) + return model + + +# TODO - Paramaterize for Quartus (different strategies?) +@pytest.fixture +def convert(load_jettagging_model, request, test_case_id): + """ + Convert a QKeras model trained on the jet tagging dataset + """ + + strategy = request.param + model = load_jettagging_model + + config = hls4ml.utils.config_from_keras_model(model, granularity='name', backend='Vivado') + config['Model']['Strategy'] = strategy + config['LayerName']['softmax']['exp_table_t'] = 'ap_fixed<18,8>' + config['LayerName']['softmax']['inv_table_t'] = 'ap_fixed<18,4>' + hls_model = hls4ml.converters.convert_from_keras_model( + model, + hls_config=config, + output_dir=str(test_root_path / test_case_id), + part='xcu250-figd2104-2L-e', + ) + hls_model.compile() + return hls_model + + +@pytest.mark.parametrize('convert', ['latency', 'resource'], indirect=True, ids=['latency', 'resource']) +def test_accuracy(convert, load_jettagging_model, get_jettagging_data): + """ + Test the hls4ml-evaluated accuracy of a 3 hidden layer QKeras model trained on + the jet tagging dataset. QKeras model accuracy is required to be over 70%, and + hls4ml accuracy required to be within 1% of the QKeras model accuracy. + """ + print('Test accuracy') + from sklearn.metrics import accuracy_score + + X_train_val, X_test, y_train_val, y_test = get_jettagging_data + + hls_model = convert + model = load_jettagging_model + + y_qkeras = model.predict(np.ascontiguousarray(X_test)) + y_hls4ml = hls_model.predict(np.ascontiguousarray(X_test)) + + acc_qkeras = accuracy_score(np.argmax(y_test, axis=1), np.argmax(y_qkeras, axis=1)) + acc_hls4ml = accuracy_score(np.argmax(y_test, axis=1), np.argmax(y_hls4ml, axis=1)) + rel_diff = abs(acc_qkeras - acc_hls4ml) / acc_qkeras + + print(f'Accuracy qkeras: {acc_qkeras}') + print(f'Accuracy hls4ml: {acc_hls4ml}') + print(f'Relative difference: {rel_diff}') + + assert acc_qkeras > 0.7 and rel_diff < 0.01 + + +def randX(batch_size, N): + return np.random.rand(batch_size, N) + + +@pytest.fixture(scope='module') +def randX_100_16(): + return randX(100, 16) + + +# TODO: include wider bitwidths when that can be made to pass +# Note 4-bit test can still fail sometimes depending on random seed +# https://github.com/fastmachinelearning/hls4ml/issues/381 +# @pytest.mark.parametrize('bits', [4, 6, 8]) +@pytest.mark.parametrize('bits,alpha', [(4, 1), (4, 'auto_po2')]) +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis', 'Quartus', 'oneAPI']) +@pytest.mark.parametrize('io_type', ['io_parallel', 'io_stream']) +def test_single_dense_activation_exact(test_case_id, randX_100_16, bits, alpha, backend, io_type): + """ + Test a single Dense -> Activation layer topology for + bit exactness with number of bits parameter + """ + X = randX_100_16 + model = Sequential( + [ + QActivation(activation=quantized_bits(bits, 0, alpha=1), input_shape=(16,), name='inp_quant'), + QDense( + 16, + name='fc1', + kernel_quantizer=quantized_bits(bits, 0, alpha=alpha), + bias_quantizer=quantized_bits(bits, 0, alpha=1), + kernel_initializer='lecun_uniform', + ), + QActivation(activation=quantized_relu(bits, 0), name='relu1'), + ] + ) + model.compile() + + config = hls4ml.utils.config_from_keras_model(model, granularity='name', backend=backend) + output_dir = str(test_root_path / test_case_id) + + bit_exact = alpha == 1 + # alpha!=po2 case uses non-fixed-point data types, unsupported by the precision propagation flow + hls_model = hls4ml.converters.convert_from_keras_model( + model, hls_config=config, output_dir=output_dir, backend=backend, io_type=io_type, bit_exact=bit_exact + ) + hls_model.compile() + + y_qkeras = model.predict(X) + y_hls4ml = hls_model.predict(X) + + # alpha!=1 case for weights can be supported if weight conversion is done before writing + if bit_exact: + np.testing.assert_array_equal(y_qkeras, y_hls4ml) + else: + np.testing.assert_allclose(y_qkeras.ravel(), y_hls4ml.ravel(), atol=2**-bits, rtol=1.0) + + +@pytest.fixture +def make_btnn(test_no, N, kernel_quantizer, bias_quantizer, activation_quantizer, use_batchnorm, is_xnor): + shape = (N,) + model = Sequential() + model.add(QDense(10, input_shape=shape, kernel_quantizer=kernel_quantizer, bias_quantizer=bias_quantizer, name='dense')) + if use_batchnorm: + model.add(BatchNormalization(name='bn')) + model.add(QActivation(activation=activation_quantizer)) + model.compile() + return model, is_xnor, test_no + + +@pytest.fixture(scope='module') +def randX_100_10(): + return randX(100, 10) + + +@pytest.mark.parametrize( + 'quantizer', [(quantized_tanh(8)), (quantized_sigmoid(5)), (quantized_sigmoid(7, use_real_sigmoid=True))] +) +@pytest.mark.parametrize('backend', ['Vivado', 'Quartus', 'oneAPI']) +@pytest.mark.parametrize('io_type', ['io_parallel', 'io_stream']) +def test_quantizer_special(test_case_id, randX_1000_1, quantizer, backend, io_type): + """ + Test a single quantizer (tanh or sigmoid) as an Activation function. + Checks the type inference through the conversion is correct without just + using the same logic. + """ + X = randX_1000_1 + X = np.round(X * 2**10) * 2**-10 # make it an exact ap_fixed<16,6> + model = Sequential() + model.add(QActivation(input_shape=(1,), activation=quantizer, name='quantizer')) + model.compile() + + config = hls4ml.utils.config_from_keras_model(model, granularity='name', backend=backend) + output_dir = str(test_root_path / test_case_id) + hls_model = hls4ml.converters.convert_from_keras_model( + model, hls_config=config, output_dir=output_dir, backend=backend, io_type=io_type + ) + hls_model.compile() + + y_qkeras = model.predict(X) + y_hls4ml = hls_model.predict(X) + # Goal is to get it passing with all equal + np.testing.assert_allclose(y_qkeras, y_hls4ml, rtol=1e-2, atol=0.02) + + +@pytest.mark.parametrize( + 'test_no,N,kernel_quantizer,bias_quantizer,activation_quantizer,use_batchnorm,is_xnor', + [ + (1, 10, ternary(alpha=1), quantized_bits(5, 2), 'binary_tanh', False, False), + (2, 10, binary(), quantized_bits(5, 2), 'binary_tanh', False, True), + (3, 10, ternary(alpha='auto'), quantized_bits(5, 2), binary(), True, True), + (4, 10, ternary(alpha='auto'), quantized_bits(5, 2), 'ternary', True, False), + (5, 10, ternary(alpha='auto'), quantized_bits(5, 2), ternary(threshold=0.2), True, False), + (6, 10, ternary(alpha='auto'), quantized_bits(5, 2), ternary(threshold=0.8), True, False), + (7, 10, binary(), quantized_bits(5, 2), binary(), False, True), + ], +) +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis', 'Quartus', 'oneAPI']) +@pytest.mark.parametrize('io_type', ['io_parallel', 'io_stream']) +def test_btnn(test_case_id, make_btnn, randX_100_10, backend, io_type): + model, is_xnor, test_no = make_btnn + X = randX_100_10 + cfg = hls4ml.utils.config_from_keras_model(model, granularity='name', backend=backend) + output_dir = str(test_root_path / test_case_id) + hls_model = hls4ml.converters.convert_from_keras_model( + model, output_dir=output_dir, hls_config=cfg, backend=backend, io_type=io_type + ) + hls_model.compile() + y_hls = hls_model.predict(X) + # hls4ml may return XNOR binary + if is_xnor: + y_hls = np.where(y_hls == 0, -1, 1) + y_ker = model.predict(X) + wrong = (y_hls != y_ker).ravel() + assert sum(wrong) / len(wrong) < 0.005 + + +@pytest.fixture(scope='module') +def randX_1000_1(): + return randX(1000, 1) + + +# TODO: include quantized_relu tests when they are made to pass +# https://github.com/fastmachinelearning/hls4ml/issues/377 +@pytest.mark.parametrize( + 'quantizer', + [ + (quantized_bits(8, 0)), + (quantized_bits(8, 4)), + (quantized_bits(4, 2)), + (quantized_bits(4, 0)), + (quantized_bits(10, 0)), + (quantized_relu(4)), + (quantized_relu(4, 2)), + (quantized_relu(8)), + (quantized_relu(8, 4)), + (quantized_relu(10)), + (quantized_relu(10, 5)), + ], +) +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis', 'Quartus', 'oneAPI']) +@pytest.mark.parametrize('io_type', ['io_parallel', 'io_stream']) +def test_quantizer(test_case_id, randX_1000_1, quantizer, backend, io_type): + """ + Test a single quantizer as an Activation function. + Checks the type inference through the conversion is correct without just + using the same logic. + """ + X = randX_1000_1 + X = np.round(X * 2**10) * 2**-10 # make it an exact ap_fixed<16,6> + model = Sequential() + model.add(QActivation(input_shape=(1,), activation=quantizer, name='quantizer')) + model.compile() + + config = hls4ml.utils.config_from_keras_model(model, granularity='name', backend=backend) + output_dir = str(test_root_path / test_case_id) + hls_model = hls4ml.converters.convert_from_keras_model( + model, hls_config=config, output_dir=output_dir, backend=backend, io_type=io_type + ) + hls_model.compile() + + y_qkeras = model.predict(X) + y_hls4ml = hls_model.predict(X) + # Goal is to get it passing with all equal + np.testing.assert_array_equal(y_qkeras, y_hls4ml) + + +@pytest.mark.parametrize( + 'quantizer', + [ + (quantized_relu(4, negative_slope=0.5)), + (quantized_relu(8, 4, negative_slope=1.0)), + (quantized_relu(10, 2, negative_slope=0.25)), + ], +) +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis', 'Quartus', 'oneAPI']) +@pytest.mark.parametrize('io_type', ['io_parallel', 'io_stream']) +def test_relu_negative_slope(test_case_id, randX_1000_1, quantizer, backend, io_type): + """ + Test a a transformation of quantized_relu with negative_slope to leaky_relu activation layer. + """ + X = randX_1000_1 + X = -X # Make it negative so leaky relu does something + X = np.round(X * 2**10) * 2**-10 # make it an exact ap_fixed<16,6> + model = Sequential() + model.add(QActivation(input_shape=(1,), activation=quantizer, name='quantizer')) + model.compile() + + config = hls4ml.utils.config_from_keras_model(model, granularity='name', backend=backend) + output_dir = str(test_root_path / test_case_id) + hls_model = hls4ml.converters.convert_from_keras_model( + model, hls_config=config, output_dir=output_dir, backend=backend, io_type=io_type + ) + hls_model.compile() + + y_qkeras = model.predict(X) + y_hls4ml = hls_model.predict(X) + np.testing.assert_allclose(y_hls4ml, y_qkeras, rtol=1e-5, atol=0) + + +@pytest.mark.parametrize( + 'weight_quantizer,activation_quantizer,', + [ + ('binary', 'binary'), + ('ternary', 'ternary'), + ('quantized_bits(4, 0, alpha=1)', 'quantized_relu(2, 0)'), + ('quantized_bits(4, 0, alpha=1)', 'quantized_relu(4, 0)'), + ('quantized_bits(4, 0, alpha=1)', 'quantized_relu(8, 0)'), + ], +) +def test_qactivation_kwarg(test_case_id, randX_100_10, activation_quantizer, weight_quantizer): + + inputs = Input(shape=(10,)) + + outputs = QDense( + 10, + activation=activation_quantizer, + name='qdense', + kernel_quantizer=weight_quantizer, + bias_quantizer=weight_quantizer, + kernel_initializer='lecun_uniform', + )(inputs) + model = Model(inputs, outputs) + + config = hls4ml.utils.config_from_keras_model(model, granularity='name', backend='Vivado') + + out_dir = str(test_root_path / test_case_id) + + hls_model = hls4ml.converters.convert_from_keras_model(model, hls_config=config, output_dir=out_dir) + hls_model.compile() + + # Output tests + X = randX_100_10 + X = np.round(X * 2**10) * 2**-10 + y_qkeras = model.predict(X) + y_hls4ml = hls_model.predict(X) + if hasattr(eval(activation_quantizer), 'bits'): + np.testing.assert_allclose( + y_qkeras.ravel(), y_hls4ml.ravel(), atol=2 ** -(eval(activation_quantizer).bits - 1), rtol=1.0 + ) + else: + if activation_quantizer == 'binary': + y_hls4ml = np.where(y_hls4ml == 0, -1, 1) + wrong = (y_hls4ml != y_qkeras).ravel() + assert sum(wrong) / len(wrong) <= 0.005 + + +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis', 'oneAPI']) +@pytest.mark.parametrize('io_type', ['io_parallel']) +def test_qkeras_einsum_dense(test_case_id, randX_100_10, backend, io_type): + """ + Test a QKeras-quantized activation -> EinsumDense -> QKeras-quantized activation topology. + """ + if keras.__version__ < '3.0': + pytest.skip('EinsumDense conversion is only supported for Keras v3') + + X = randX_100_10[:, :4] + X = np.round(X * 2**10) * 2**-10 # make it an exact ap_fixed<16,6> + + inputs = Input(shape=(4,), name='input_layer') + x = QActivation(quantized_bits(6, 0, alpha=1), name='input_quant')(inputs) + x = EinsumDense( + 'bi,io->bo', + output_shape=3, + name='einsum_dense', + kernel_initializer=keras.initializers.Constant(0.25), + bias_axes=None, + )(x) + outputs = QActivation(quantized_relu(6, 0), name='output_quant')(x) + model = Model(inputs, outputs) + model.compile() + + config = hls4ml.utils.config_from_keras_model( + model, granularity='name', default_precision='fixed<24,8>', backend=backend + ) + output_dir = str(test_root_path / test_case_id) + hls_model = hls4ml.converters.convert_from_keras_model( + model, hls_config=config, output_dir=output_dir, backend=backend, io_type=io_type + ) + hls_model.compile() + + y_qkeras = model.predict(X) + y_hls4ml = hls_model.predict(X) + + np.testing.assert_array_equal(y_qkeras, y_hls4ml.reshape(y_qkeras.shape)) + + +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis', 'Quartus', 'oneAPI']) +@pytest.mark.parametrize('io_type', ['io_parallel', 'io_stream']) +def test_quantizer_parsing(test_case_id, randX_100_10, backend, io_type): + X = randX_100_10 + X = np.round(X * 2**10) * 2**-10 # make it an exact ap_fixed<16,6> + model = Sequential() + model.add( + QDense( + 8, + input_shape=(10,), + kernel_quantizer=None, # Incorrect usage, but shouldn't break hls4ml + kernel_initializer='ones', + bias_quantizer=None, + bias_initializer='zeros', + activation='quantized_relu(8, 0)', + ) + ) + model.compile() + + config = hls4ml.utils.config_from_keras_model( + model, granularity='name', default_precision='fixed<24,8>', backend=backend + ) + output_dir = str(test_root_path / test_case_id) + hls_model = hls4ml.converters.convert_from_keras_model( + model, hls_config=config, output_dir=output_dir, backend=backend, io_type=io_type + ) + hls_model.compile() + + y_qkeras = model.predict(X) + y_hls4ml = hls_model.predict(X) + + np.testing.assert_array_equal(y_qkeras, y_hls4ml.reshape(y_qkeras.shape)) + + +@pytest.fixture(scope='module') +def randX_100_8_8_1(): + return np.random.rand(100, 8, 8, 1) + + +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis', 'Quartus', 'oneAPI']) +@pytest.mark.parametrize('io_type', ['io_parallel', 'io_stream']) +@pytest.mark.parametrize( + 'qconv_layer,input_shape,input_data_shape,layer_kwargs', + [ + (QConv1D, (8, 2), (5, 8, 2), {'filters': 3, 'kernel_size': 3}), + (QConv2D, (8, 8, 1), (5, 8, 8, 1), {'filters': 2, 'kernel_size': (3, 3)}), + ], + ids=['qconv1d', 'qconv2d'], +) +def test_qconv_activation_kwarg(test_case_id, qconv_layer, input_shape, input_data_shape, layer_kwargs, backend, io_type): + """ + Test QConv1D and QConv2D handling with activation quantizers passed as layer kwargs. + """ + X = np.random.default_rng(12345).random(input_data_shape) + X = np.round(X * 2**10) * 2**-10 # make it an exact ap_fixed<16,6> + + inputs = Input(shape=input_shape, name='input_layer') + outputs = qconv_layer( + **layer_kwargs, + name='qconv', + kernel_quantizer='quantized_bits(4, 0, alpha=1)', + bias_quantizer='quantized_bits(4, 0, alpha=1)', + kernel_initializer='ones', + bias_initializer='zeros', + activation='quantized_relu(4, 0)', + )(inputs) + model = Model(inputs, outputs) + model.compile() + + config = hls4ml.utils.config_from_keras_model( + model, granularity='name', default_precision='fixed<24,8>', backend=backend + ) + assert 'qconv_activation' in config['LayerName'] + + output_dir = str(test_root_path / test_case_id) + hls_model = hls4ml.converters.convert_from_keras_model( + model, + hls_config=config, + output_dir=output_dir, + backend=backend, + io_type=io_type, + allow_da_fallback=False, + ) + hls_model.compile() + + y_qkeras = model.predict(X) + y_hls4ml = hls_model.predict(X) + + np.testing.assert_array_equal(y_qkeras, y_hls4ml.reshape(y_qkeras.shape)) + + +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis', 'Quartus', 'oneAPI']) +@pytest.mark.parametrize('io_type', ['io_parallel', 'io_stream']) +def test_qconv2dbn(test_case_id, randX_100_8_8_1, backend, io_type): + """ + Test proper handling of QConv2DBatchnorm. + """ + X = randX_100_8_8_1 + X = np.round(X * 2**10) * 2**-10 # make it an exact ap_fixed<16,6> + model = Sequential() + model.add( + QConv2DBatchnorm( + 4, + kernel_size=(3, 3), + input_shape=(8, 8, 1), + kernel_quantizer='quantized_bits(8, 0, alpha=1)', + kernel_initializer='ones', + bias_quantizer='quantized_bits(8, 0, alpha=1)', + bias_initializer='zeros', + activation='quantized_relu(8, 0)', + ) + ) + model.compile() + + config = hls4ml.utils.config_from_keras_model( + model, granularity='name', default_precision='fixed<24,8>', backend=backend + ) + output_dir = str(test_root_path / test_case_id) + hls_model = hls4ml.converters.convert_from_keras_model( + model, hls_config=config, output_dir=output_dir, backend=backend, io_type=io_type + ) + hls_model.compile() + + y_qkeras = model.predict(X) + y_hls4ml = hls_model.predict(X) + + np.testing.assert_array_equal(y_qkeras, y_hls4ml.reshape(y_qkeras.shape)) + + +@pytest.fixture(scope='module') +def randX_10_32_32_3(): + return np.random.rand(10, 32, 32, 3) + + +# Currently only Vivado and Vitis is supported for io_stream. +# Note, qkeras only supports 2d version of depthwise +def KQ(): + return quantized_bits(bits=8, integer=3, keep_negative=True, alpha=1.0, symmetric=False) + + +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis']) +@pytest.mark.parametrize('io_type', ['io_stream']) +@pytest.mark.parametrize('quantized_bits', ['quantized_bits(6, 0, alpha=1)', KQ()]) +@pytest.mark.parametrize('use_bias', [True, False]) +@pytest.mark.parametrize('padding', ['valid', 'same']) +def test_qdepthwiseconv2d(test_case_id, randX_10_32_32_3, backend, io_type, quantized_bits, use_bias, padding): + """ + Test proper handling of QDepthwiseConv2D. + """ + X = randX_10_32_32_3 + X = np.round(X * 2**10) * 2**-10 # make it an exact ap_fixed<16,6> + model = Sequential() + model.add( + QDepthwiseConv2D( + kernel_size=(3, 3), + input_shape=(32, 32, 3), + depthwise_quantizer=quantized_bits, + bias_quantizer=quantized_bits, + bias_initializer='he_normal', + use_bias=use_bias, + padding=padding, + activation='quantized_relu(3, 0)', + ) + ) + model.compile() + + config = hls4ml.utils.config_from_keras_model( + model, granularity='name', default_precision='fixed<24,8>', backend=backend + ) + output_dir = str(test_root_path / test_case_id) + hls_model = hls4ml.converters.convert_from_keras_model( + model, hls_config=config, output_dir=output_dir, backend=backend, io_type=io_type + ) + hls_model.compile() + + y_qkeras = model.predict(X) + y_hls4ml = hls_model.predict(X) + + np.testing.assert_allclose(y_qkeras, y_hls4ml.reshape(y_qkeras.shape), rtol=1e-2, atol=0.01) + + +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis', 'Quartus', 'oneAPI']) +@pytest.mark.parametrize('io_type', ['io_parallel', 'io_stream']) +@pytest.mark.parametrize('strategy', ['Latency', 'Resource']) +def test_quantised_po2_bit_width(test_case_id, backend, io_type, strategy): + input_shape = 26 + output_shape = 6 + X = np.random.rand(100, input_shape) + + # Set a high bit-width, so that we ensure HLS doesn't allocate 2**bits for the multiplication + # The biggest allowed bit-width in Vivado HLS is 65,536 (2**16) + keras_model = Sequential() + keras_model.add( + QDense(output_shape, input_shape=(input_shape,), name='dense', kernel_quantizer=quantized_po2(18, max_value=2**20)) + ) + + # Set weights to same high random number + weights = keras_model.layers[0].get_weights() + weights[0] = (2**18) * np.random.rand(input_shape, output_shape) + keras_model.layers[0].set_weights(weights) + + # Assert output is the same and bit-width is not over-allocated [it would throw a run-time error] + keras_model.compile() + y_keras = keras_model.predict(X) + + hls_config = hls4ml.utils.config_from_keras_model( + keras_model, granularity='name', default_precision='ap_fixed<64, 32>', default_reuse_factor=1, backend=backend + ) + hls_config['Model']['Strategy'] = strategy + output_dir = str(test_root_path / test_case_id) + hls_model = hls4ml.converters.convert_from_keras_model( + keras_model, hls_config=hls_config, output_dir=output_dir, backend=backend, io_type=io_type + ) + hls_model.compile() + y_hls = hls_model.predict(np.ascontiguousarray(X)) + + np.testing.assert_allclose(y_hls.flatten(), y_keras.flatten(), rtol=2e-2) + + +@pytest.mark.parametrize('backend', ['Quartus', 'oneAPI']) +def test_qsimplernn(test_case_id, backend): + """ + Test proper handling of QSimpleRNN. + """ + X = np.linspace(-0.25, 0.25, 5) + X = np.stack([X, X], axis=1).reshape(1, 5, 2) + + model = Sequential() + model.add(Input(shape=(5, 2))) + model.add( + QSimpleRNN( + 4, + kernel_quantizer='quantized_bits(16, 0, alpha=1)', + recurrent_quantizer='quantized_bits(16, 0, alpha=1)', + bias_quantizer='quantized_bits(16, 0, alpha=1)', + state_quantizer='quantized_bits(16, 0, alpha=1)', + activation='relu', + ) + ) + model.compile() + + config = hls4ml.utils.config_from_keras_model( + model, granularity='name', default_precision='ap_fixed<16,1>', backend=backend + ) + output_dir = str(test_root_path / test_case_id) + hls_model = hls4ml.converters.convert_from_keras_model(model, hls_config=config, output_dir=output_dir, backend=backend) + hls_model.compile() + + y_qkeras = model.predict(X) + y_hls4ml = hls_model.predict(X) + + np.testing.assert_allclose(y_qkeras, y_hls4ml.reshape(y_qkeras.shape), atol=0.1) + + +@pytest.mark.parametrize('backend', ['Vivado', 'Quartus', 'oneAPI']) +def test_qlstm(test_case_id, backend): + """ + Test proper handling of QLSTM. + """ + X = np.linspace(-0.5, 0.5, 5) + X = np.stack([X, X], axis=1).reshape(1, 5, 2) + + model = Sequential() + model.add(Input(shape=(5, 2))) + model.add( + QLSTM( + 4, + kernel_quantizer='quantized_bits(8, 0, alpha=1)', + recurrent_quantizer='quantized_bits(8, 0, alpha=1)', + bias_quantizer='quantized_bits(8, 0, alpha=1)', + state_quantizer='quantized_bits(8, 0, alpha=1)', + activation='tanh', + recurrent_activation='sigmoid', + ) + ) + model.compile() + + config = hls4ml.utils.config_from_keras_model( + model, granularity='name', default_precision='ap_fixed<8,1>', backend=backend + ) + output_dir = str(test_root_path / test_case_id) + hls_model = hls4ml.converters.convert_from_keras_model(model, hls_config=config, output_dir=output_dir, backend=backend) + hls_model.compile() + + y_qkeras = model.predict(X) + y_hls4ml = hls_model.predict(X) + + np.testing.assert_allclose(y_qkeras, y_hls4ml.reshape(y_qkeras.shape), atol=0.1) + + +@pytest.mark.parametrize('backend', ['Vivado', 'Quartus', 'oneAPI']) +def test_qgru(test_case_id, backend): + """ + Test proper handling of QGRU. + """ + X = np.linspace(-0.5, 0.5, 5) + X = np.stack([X, X], axis=1).reshape(1, 5, 2) + + model = Sequential() + model.add(Input(shape=(5, 2))) + model.add( + QGRU( + 4, + kernel_quantizer='quantized_bits(8, 0, alpha=1)', + recurrent_quantizer='quantized_bits(8, 0, alpha=1)', + bias_quantizer='quantized_bits(8, 0, alpha=1)', + state_quantizer='quantized_bits(8, 0, alpha=1)', + activation='tanh', + recurrent_activation='sigmoid', + reset_after='False', + ) + ) + model.compile() + + config = hls4ml.utils.config_from_keras_model( + model, granularity='name', default_precision='ap_fixed<8,1>', backend=backend + ) + output_dir = str(test_root_path / test_case_id) + hls_model = hls4ml.converters.convert_from_keras_model(model, hls_config=config, output_dir=output_dir, backend=backend) + hls_model.compile() + + y_qkeras = model.predict(X) + y_hls4ml = hls_model.predict(X) + + np.testing.assert_allclose(y_qkeras, y_hls4ml.reshape(y_qkeras.shape), atol=0.1) + + +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis']) +@pytest.mark.parametrize('io_type', ['io_stream']) +def test_qseparableconv1d(test_case_id, backend, io_type): + """ + Test proper handling of QSeparableConv1D. + """ + x_in = Input((13, 20), name='input_layer') + x = QSeparableConv1D( + 5, + 3, + depthwise_quantizer=quantized_bits(8, 3, alpha=1), + pointwise_quantizer=quantized_bits(8, 3, alpha=1), + bias_quantizer=quantized_bits(8, 3, alpha=1), + name='qsepconv_1', + )(x_in) + model = Model(inputs=x_in, outputs=x) + + config = hls4ml.utils.config_from_keras_model( + model, granularity='name', backend=backend, default_precision='fixed<23,7>' + ) + + # Use 8 bits for input + config['LayerName']['input_layer']['Precision']['result'] = 'fixed<8,1>' + # default_precision is will be used for accum_t and result_t of the conv layer, so we don't need to set them here + # We need <15,4> for the result of depthwise step + config['LayerName']['qsepconv_1']['Precision']['dw_output'] = 'fixed<15,4>' + + output_dir = str(test_root_path / test_case_id) + hls_model = hls4ml.converters.convert_from_keras_model( + model, + hls_config=config, + output_dir=output_dir, + io_type=io_type, + ) + hls_model.compile() + + data = np.random.rand(100, 13, 20) + input_quantizer = quantized_bits(8, 0, alpha=1) + dataq = input_quantizer(data).numpy() + + y_qkeras = model.predict(dataq) + y_hls4ml = hls_model.predict(dataq) + + np.testing.assert_allclose(y_qkeras, y_hls4ml.reshape(y_qkeras.shape), rtol=0, atol=0) + + +@pytest.mark.parametrize('backend', ['Vivado', 'Vitis']) +@pytest.mark.parametrize('io_type', ['io_stream']) +def test_qseparableconv2d(test_case_id, backend, io_type): + """ + Test proper handling of QSeparableConv2D. + """ + x_in = Input((13, 21, 20), name='input_layer') + x = QSeparableConv2D( + 5, + 3, + depthwise_quantizer=quantized_bits(8, 3, alpha=1), + pointwise_quantizer=quantized_bits(8, 3, alpha=1), + bias_quantizer=quantized_bits(8, 3, alpha=1), + name='qsepconv_1', + )(x_in) + model = Model(inputs=x_in, outputs=x) + + config = hls4ml.utils.config_from_keras_model( + model, granularity='name', backend=backend, default_precision='fixed<23,7>' + ) + + # Use 8 bits for input + config['LayerName']['input_layer']['Precision']['result'] = 'fixed<8,1>' + # default_precision is will be used for accum_t and result_t of the conv layer, so we don't need to set them here + # We need <15,4> for the result of depthwise step + config['LayerName']['qsepconv_1']['Precision']['dw_output'] = 'fixed<15,4>' + + output_dir = str(test_root_path / test_case_id) + hls_model = hls4ml.converters.convert_from_keras_model( + model, + hls_config=config, + output_dir=output_dir, + io_type=io_type, + ) + hls_model.compile() + + data = np.random.rand(100, 13, 21, 20) + input_quantizer = quantized_bits(8, 0, alpha=1) + dataq = input_quantizer(data).numpy() + + y_qkeras = model.predict(dataq) + y_hls4ml = hls_model.predict(dataq) + + np.testing.assert_allclose(y_qkeras, y_hls4ml.reshape(y_qkeras.shape), rtol=0, atol=0)