Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
import warnings

Check failure on line 1 in bindings/pyroot/pythonizations/test/generate_keras_functional.py

View workflow job for this annotation

GitHub Actions / ruff

ruff (I001)

bindings/pyroot/pythonizations/test/generate_keras_functional.py:1:1: I001 Import block is un-sorted or un-formatted help: Organize imports

def generate_keras_functional(dst_dir):

import numpy as np
Expand All @@ -16,8 +18,18 @@

model.compile(optimizer='adam', loss='mean_squared_error', metrics=['mae'])
model.summary()
model.fit(x_train, y_train, epochs=1, verbose=0)
model.save(f"{dst_dir}/Functional_{name}_test.keras")
if len(model.trainable_weights) > 0:
model.fit(x_train, y_train, epochs=1, verbose=0)

with warnings.catch_warnings():
# Some object inside TensorFlow/Keras has an outdated __array__ implementation
warnings.filterwarnings(
"ignore",
category=DeprecationWarning,
message=".*__array__.*copy keyword.*"
)
model.save(f"{dst_dir}/Functional_{name}_test.keras")

print("generated and saved functional model",name)


Expand Down Expand Up @@ -211,7 +223,7 @@
sub = layers.Subtract()([d1, d2])
mul = layers.Multiply()([d1, d2])
merged = layers.Concatenate()([add, sub, mul])
merged = layers.LeakyReLU(alpha=0.1)(merged)
merged = layers.LeakyReLU(negative_slope=0.1)(merged)
out = layers.Dense(4, activation="softmax")(merged)
model = models.Model([inp1, inp2], out)
train_and_save(model, "Layer_Combination_3")
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
import warnings

Check failure on line 1 in bindings/pyroot/pythonizations/test/generate_keras_sequential.py

View workflow job for this annotation

GitHub Actions / ruff

ruff (I001)

bindings/pyroot/pythonizations/test/generate_keras_sequential.py:1:1: I001 Import block is un-sorted or un-formatted help: Organize imports

def generate_keras_sequential(dst_dir):

import numpy as np
Expand All @@ -9,10 +11,19 @@
x_train = np.random.rand(32, *model.input_shape[1:])
y_train = np.random.rand(32, *model.output_shape[1:])
model.compile(optimizer='adam', loss='mean_squared_error', metrics=['mae'])
model.fit(x_train, y_train, epochs=1, verbose=0)
if len(model.trainable_weights) > 0:
model.fit(x_train, y_train, epochs=1, verbose=0)
model.summary()
print("fitting sequential model",name)
model.save(f"{dst_dir}/Sequential_{name}_test.keras")

with warnings.catch_warnings():
# Some object inside TensorFlow/Keras has an outdated __array__ implementation
warnings.filterwarnings(
"ignore",
category=DeprecationWarning,
message=".*__array__.*copy keyword.*"
)
model.save(f"{dst_dir}/Sequential_{name}_test.keras")


# Binary Ops: Add, Subtract, Multiply are not typical in Sequential - skipping those
Expand Down Expand Up @@ -193,7 +204,7 @@
layers.Permute((2, 1)),
layers.Flatten(),
layers.Dense(32),
layers.LeakyReLU(alpha=0.1),
layers.LeakyReLU(negative_slope=0.1),
layers.Dense(10, activation='softmax'),
])
train_and_save(modelB, "Layer_Combination_2")
Expand All @@ -210,4 +221,4 @@
layers.Dense(8, activation='swish'),
layers.Dense(3, activation='softmax'),
])
train_and_save(modelC, "Layer_Combination_3")
train_and_save(modelC, "Layer_Combination_3")
35 changes: 13 additions & 22 deletions bindings/pyroot/pythonizations/test/parser_test_function.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,6 @@

'''
The test file contains two types of functions:
is_accurate:
- This function checks whether the inference results from SOFIE and Keras are accurate within a specified
tolerance. Since the inference result from Keras is not flattened, the function flattens both tensors before
performing the comparison.

generate_and_test_inference:
- This function accepts the following inputs:
Expand All @@ -29,7 +25,7 @@
shape from the model object.
- Convert the inference results to NumPy arrays:
The SOFIE result is of type vector<float>, and the Keras result is a TensorFlow tensor. Both are converted to
NumPy arrays before being passed to the is_accurate function for comparison.
NumPy arrays before being passed to the np.testing.assert_allclose function for comparison.

'''
def is_channels_first_supported() :
Expand All @@ -42,16 +38,6 @@ def is_channels_first_supported() :

return True

def is_accurate(tensor_a, tensor_b, tolerance=1e-2):
tensor_a = tensor_a.flatten()
tensor_b = tensor_b.flatten()
for i in range(len(tensor_a)):
difference = abs(tensor_a[i] - tensor_b[i])
if difference > tolerance:
print(tensor_a[i], tensor_b[i])
return False
return True

def generate_and_test_inference(model_file_path: str, generated_header_file_dir: str = None, batch_size=1):

import keras
Expand Down Expand Up @@ -81,7 +67,6 @@ def generate_and_test_inference(model_file_path: str, generated_header_file_dir:
sofie_model_namespace = getattr(ROOT, "TMVA_SOFIE_" + model_name)
inference_session = sofie_model_namespace.Session(generated_header_file_path.removesuffix(".hxx") + ".dat")
keras_model = keras.models.load_model(model_file_path)
keras_model.load_weights(model_file_path)

input_tensors = []
for model_input in keras_model.inputs:
Expand All @@ -91,11 +76,17 @@ def generate_and_test_inference(model_file_path: str, generated_header_file_dir:
sofie_inference_result = inference_session.infer(*input_tensors)
sofie_output_tensor_shape = list(rmodel.GetTensorShape(rmodel.GetOutputTensorNames()[0])) # get output shape
# from SOFIE
keras_inference_result = keras_model(input_tensors)
# Keras explicitly forbids input tensor lists of size 1
if len(keras_model.inputs) == 1:
keras_inference_result = keras_model(input_tensors[0])
else:
keras_inference_result = keras_model(input_tensors)
if sofie_output_tensor_shape != list(keras_inference_result.shape):
raise AssertionError("Output tensor dimensions from SOFIE and Keras do not match")
sofie_inference_result = np.asarray(sofie_inference_result)
keras_inference_result = np.asarray(keras_inference_result)
is_inference_accurate = is_accurate(sofie_inference_result, keras_inference_result)
if not is_inference_accurate:
raise AssertionError("Inference results from SOFIE and Keras do not match")

np.testing.assert_allclose(
np.asarray(sofie_inference_result).flatten(),
np.asarray(keras_inference_result).flatten(),
atol=1e-2,
rtol=0. # explicitly disable relative tolerance (NumPy uses |a - b| <= atol + rtol * |b|)
)
Loading