1D Convolution & Pooling Quartus PyTests

Author: bo3z

test/pytest/test_conv1d.py

@@ -1,11 +1,10 @@
+from distutils.command.config import config
 from hls4ml.converters.keras_to_hls import keras_to_hls
 import pytest
 import hls4ml
 import numpy as np
 from sklearn.metrics import accuracy_score
-import tensorflow as tf
 from tensorflow.keras.models import model_from_json
-import yaml
 from pathlib import Path
 
 test_root_path = Path(__file__).parent
@@ -25,46 +24,58 @@ def keras_model():
     model.load_weights(example_model_path / 'keras/KERAS_conv1d_weights.h5')
     return model
 
-@pytest.fixture      
-@pytest.mark.parametrize('settings', [('io_parallel', 'latency'),
-                                      ('io_parallel', 'resource'),
-                                      ('io_stream', 'latency'),
-                                      ('io_stream', 'resource')])
-def hls_model(settings):
-    io_type = settings[0]
-    strategy = settings[1]
-    config = hls4ml.converters.create_config(output_dir = 'hls4mlprj_conv1d_{}_{}'.format(io_type, strategy))
-    config['KerasJson'] = str(example_model_path / 'keras/KERAS_conv1d.json')
-    config['KerasH5'] = str(example_model_path / 'keras/KERAS_conv1d_weights.h5')
-    config['OutputDir'] = str(test_root_path / 'hls4mlprj_conv1d_{}_{}'.format(io_type, strategy))
-    config['IOType'] = io_type
+@pytest.fixture   
+@pytest.mark.parametrize('backend, io_type, strategy', [
+                                      ('Quartus', 'io_parallel', 'resource'),
+                                      ('Vivado', 'io_parallel', 'resource'),
+
+                                      ('Vivado', 'io_parallel', 'latency'),
+                                      
+                                      ('Vivado', 'io_stream', 'latency'),
+                                      ('Vivado', 'io_stream', 'resource')
+                                    ])
+def hls_model(keras_model, backend, io_type, strategy):
+    default_precision = 'ap_fixed<16,3,AP_RND_CONV,AP_SAT>' if backend=='Vivado' else 'ac_fixed<16,3,true,AC_RND_CONV,AC_SAT>'
+    fc1_weight_precision = 'ap_fixed<16,3>' if backend=='Vivado' else 'ac_fixed<16,3,true>'
+    fc1_result_precision = 'ap_fixed<16,6,AP_RND_CONV,AP_SAT>' if backend=='Vivado' else 'ac_fixed<16,6,true,AC_RND_CONV,AC_SAT>'
+    output_softmax_weight_precision = 'ap_fixed<16,6>' if backend=='Vivado' else 'ac_fixed<16,6,true>'
+    output_softmax_result_precision = 'ap_fixed<16,6,AP_RND_CONV,AP_SAT>' if backend=='Vivado' else 'ac_fixed<16,6,true,AP_RND_CONV,AP_SAT>'
+
+    # Default config
+    hls_config = hls4ml.utils.config_from_keras_model(keras_model)
+    hls_config['Model']['Strategy'] = strategy
+    hls_config['Model']['ReuseFactor'] = 1
+    hls_config['Model']['Precision'] = default_precision
+
+    # Some model-specific precision tuning
+    hls_config['LayerName'] = {}
+    hls_config['LayerName']['fc1_relu'] = {'Precision':{'weight' : fc1_weight_precision, 'result' : fc1_result_precision}}
+    hls_config['LayerName']['output_softmax'] = {'Precision':{'weight' : output_softmax_weight_precision, 'result' : output_softmax_result_precision}}
+    hls_config['LayerName']['output_softmax_softmax'] = {'Strategy':'Stable'}
 
-    hls_config = {'Model' : {'Strategy' : strategy,
-                             'ReuseFactor' : 1,
-                             'Precision' : 'ap_fixed<16,3,AP_RND_CONV,AP_SAT>'}}
-    # Some model specific precision tuning
-    config['LayerName'] = {}
-    config['LayerName']['fc1_relu'] = {'Precision':{'weight' : 'ap_fixed<16,3>', 'result' : 'ap_fixed<16,6,AP_RND_CONV,AP_SAT>'}}
-    config['LayerName']['output_softmax'] = {'Precision':{'weight' : 'ap_fixed<16,6>', 'result' : 'ap_fixed<16,6,AP_RND_CONV,AP_SAT>'}}
-    config['LayerName']['output_softmax_softmax'] = {'Strategy':'Stable'}
-    config['HLSConfig'] = hls_config
-    hls_model = keras_to_hls(config)
+    output_dir = str(test_root_path / 'hls4mlprj_conv1d_{}_{}_{}'.format(backend, io_type, strategy))
+    hls_model = hls4ml.converters.convert_from_keras_model(keras_model, hls_config=hls_config, backend=backend, io_type=io_type, output_dir=output_dir)
     hls_model.compile()
     return hls_model
 
-@pytest.mark.parametrize('settings', [('io_parallel', 'latency'),
-                                      ('io_parallel', 'resource'),
-                                      ('io_stream', 'latency'),
-                                      ('io_stream', 'resource')])
+@pytest.mark.parametrize('backend, io_type, strategy', [
+                                      ('Quartus', 'io_parallel', 'resource'),
+                                      ('Vivado', 'io_parallel', 'resource'),
+
+                                      ('Vivado', 'io_parallel', 'latency'),
+                                      
+                                      ('Vivado', 'io_stream', 'latency'),
+                                      ('Vivado', 'io_stream', 'resource')
+                                    ])
 def test_accuracy(data, keras_model, hls_model):
     X = data
     model = keras_model
-    # model under test predictions and accuracy
+
+    # Model under test predictions and accuracy
     y_keras = model.predict(X)
     y_hls4ml   = hls_model.predict(X)
-    # "accuracy" of hls4ml predictions vs keras
+    
+    # "Accuracy" of hls4ml predictions vs keras
     rel_acc = accuracy_score(np.argmax(y_keras, axis=1), np.argmax(y_hls4ml, axis=1))
-
     print('hls4ml accuracy relative to keras: {}'.format(rel_acc))
-
     assert rel_acc > 0.98

test/pytest/test_keras_api.py

@@ -90,15 +90,13 @@ def test_activations(activation_function, backend):
 
     assert list(hls_model.get_layers())[2].attributes['class_name'] == activation_function.__class__.__name__
 
-
-keras_conv1d = [Conv1D]
 padds_options = ['same', 'valid']
-@pytest.mark.parametrize("conv1d", keras_conv1d)
-@pytest.mark.parametrize("padds", padds_options)
-def test_conv1d(conv1d, padds):
+@pytest.mark.parametrize('padds', padds_options)
+@pytest.mark.parametrize('backend', ['Vivado', 'Quartus'])
+def test_conv1d(padds, backend):
     model = tf.keras.models.Sequential()
     input_shape = (10, 128, 4)
-    model.add(conv1d(filters=32,
+    model.add(Conv1D(filters=32,
                      kernel_size=3,
                      strides=1,
                      padding=padds,
@@ -108,22 +106,23 @@ def test_conv1d(conv1d, padds):
                      use_bias=False,
                      data_format='channels_last'))
     model.add(Activation(activation='relu'))
-
     model.compile(optimizer='adam', loss='mse')
+    
     X_input = np.random.rand(10,128,4)
     keras_prediction = model.predict(X_input)
+    
     config = hls4ml.utils.config_from_keras_model(model)
-    output_dir = str(test_root_path / 'hls4mlprj_keras_api_conv1d_{}'.format(padds))
-    hls_model = hls4ml.converters.convert_from_keras_model(model, hls_config=config, output_dir=output_dir)
+    output_dir = str(test_root_path / 'hls4mlprj_keras_api_conv1d_{}_{}'.format(padds, backend))
+    hls_model = hls4ml.converters.convert_from_keras_model(model, hls_config=config, output_dir=output_dir, backend=backend)
     hls_model.compile()
-    hls_prediction = hls_model.predict(X_input)
+    hls_prediction = hls_model.predict(X_input).reshape(keras_prediction.shape)
+
+     # 5e-2 might be too high
+    np.testing.assert_allclose(hls_prediction, keras_prediction, rtol=0, atol=5e-2)
 
     assert len(model.layers) + 2 == len(hls_model.get_layers())
     assert list(hls_model.get_layers())[1].attributes['name'] == model.layers[0]._name
-
-    print(list(hls_model.get_layers())[1].attributes)
-    if conv1d == 'Conv1D':
-      assert list(hls_model.get_layers())[1].attributes['class_name'] == 'Conv1D'
+    assert list(hls_model.get_layers())[1].attributes['class_name'] == 'Conv1D'
     assert list(hls_model.get_layers())[1].attributes['activation'] == str(model.layers[0].activation).split()[1]
     assert list(hls_model.get_layers())[1].attributes["in_width"] == model.layers[0]._batch_input_shape[1]
     assert list(hls_model.get_layers())[1].attributes['filt_width'] == model.layers[0].kernel_size[0]
@@ -139,12 +138,9 @@ def test_conv1d(conv1d, padds):
     pad_left = pad_along_width // 2
     pad_right = pad_along_width - pad_left
 
-    out_valid	= math.ceil(float(model.layers[0]._batch_input_shape[1] - model.layers[0].kernel_size[0] + 1) / float(model.layers[0].strides[0]))
-
     if model.layers[0].padding == 'same':
         assert list(hls_model.get_layers())[1].attributes['pad_left'] == pad_left
         assert list(hls_model.get_layers())[1].attributes['pad_right'] == pad_right
-
     elif model.layers[0].padding == 'valid':
         assert list(hls_model.get_layers())[1].attributes['pad_left'] == 0
         assert list(hls_model.get_layers())[1].attributes['pad_right'] == 0
@@ -229,11 +225,13 @@ def test_conv2d(chans, padds, backend):
       assert list(hls_model.get_layers())[1].attributes['pad_right'] == 0
 
 pooling_layers = [MaxPooling1D, MaxPooling2D, AveragePooling1D, AveragePooling2D]
-@pytest.mark.parametrize("pooling", pooling_layers)
-@pytest.mark.parametrize("padds", padds_options)
-@pytest.mark.parametrize("chans", chans_options)
-def test_pooling(pooling, padds, chans):
+@pytest.mark.parametrize('pooling', pooling_layers)
+@pytest.mark.parametrize('padds', padds_options)
+@pytest.mark.parametrize('chans', chans_options)
+@pytest.mark.parametrize('backend', ['Vivado', 'Quartus'])
+def test_pooling(pooling, padds, chans, backend):
     assert '1D' in pooling.__name__ or '2D' in pooling.__name__
+    
     input_shape = (8,8,3) if '2D' in pooling.__name__ else (64,3)
     model = tf.keras.models.Sequential()
     if '2D' in pooling.__name__:
@@ -243,20 +241,22 @@ def test_pooling(pooling, padds, chans):
         model.add(Conv1D(filters=32, kernel_size=3, strides=1, padding=padds,
                       input_shape=input_shape, kernel_initializer='normal', use_bias=False,
                       data_format=chans))  
-    
     pool_size = (2, 2) if '2D' in pooling.__name__ else 2
     model.add(pooling(pool_size=pool_size, strides=None, padding=padds, data_format=None))
     model.compile(optimizer='adam', loss='mse')
 
     X_input = np.random.rand(100, *input_shape)
     keras_prediction = model.predict(X_input)
     config = hls4ml.utils.config_from_keras_model(model)
-    output_dir = str(test_root_path / 'hls4mlprj_keras_api_pooling_{}_{}_{}'.format(pooling.__name__, chans, padds))
+    output_dir = str(test_root_path / 'hls4mlprj_keras_api_pooling_{}_{}_{}_{}'.format(pooling.__name__, chans, padds, backend))
 
-    hls_model = hls4ml.converters.convert_from_keras_model(model, hls_config=config, output_dir=output_dir)
+    hls_model = hls4ml.converters.convert_from_keras_model(model, hls_config=config, output_dir=output_dir, backend=backend)
     hls_model.compile()
     hls_prediction = hls_model.predict(X_input).reshape(keras_prediction.shape)
 
+     # 5e-2 might be too high
+    np.testing.assert_allclose(hls_prediction, keras_prediction, rtol=0, atol=5e-2)
+
     hls_pool = list(hls_model.get_layers())[-1]
     ker_pool = model.layers[-1]
     if '2D' in pooling.__name__: