'
+
+ colspan = max(1, sum(int(x) for x in (show_dtype, show_shapes, show_trainable)))
+
+ # style for layer node, according to its class, as specified in layer_styles
+ layer_style_to_apply = {"color": "white", "bgcolor": "black"} # default style
+ layer_style_to_apply.update(layer_styles.get(type(layer), {}))
+
+ if show_layer_names:
+ table += (
+ f'| '
+ f''
+ f"{layer.name} ({class_name})"
+ " |
"
+ )
+ else:
+ table += (
+ f'| '
+ f''
+ f"{class_name}"
+ " |
"
+ )
+ if show_layer_activations and hasattr(layer, "activation") and layer.activation is not None:
+ table += (
+ f'| '
+ ''
+ f"Activation: {get_layer_activation_name(layer)}"
+ " |
"
+ )
+ for attr_name, attr_style in show_layer_attributes.items():
+ attribute = get_layer_attribute(layer, attr_name)
+ if attribute is not None:
+ default_style = {"bgcolor": "white", "color": "black"}
+ default_style.update(attr_style.get(None, {}))
+ style_to_apply = dict(default_style)
+ if attribute in attr_style:
+ style_to_apply.update(attr_style[attribute])
+ table += (
+ f'| '
+ f''
+ f"{attr_name.capitalize()}: {attribute}"
+ " |
"
+ )
+
+ cols = []
+ if show_shapes:
+ shape = None
+ try:
+ shape = layer.output.shape
+ except ValueError:
+ pass
+ cols.append(
+ ('' f'Output shape: {shape or "?"}' " | ")
+ )
+ if show_dtype:
+ dtype = None
+ try:
+ dtype = layer.output.dtype
+ except ValueError:
+ pass
+ cols.append(
+ ('' f'Output dtype: {dtype or "?"}' " | ")
+ )
+ if show_trainable and hasattr(layer, "trainable") and layer.weights:
+ if layer.trainable:
+ cols.append(
+ ('' '' "Trainable | ")
+ )
+ else:
+ cols.append(
+ ('' '' "Non-trainable | ")
+ )
+ if cols:
+ colspan = len(cols)
+ else:
+ colspan = 1
+
+ if cols:
+ table += "" + "".join(cols) + "
"
+ table += "
>"
+ return table
+
+
+def make_node(layer, **kwargs):
+ node = pydot.Node(str(id(layer)), label=make_layer_label(layer, **kwargs))
+ node.set("fontname", "Helvetica")
+ node.set("border", "0")
+ node.set("margin", "0")
+ return node
+
+
+def model_to_dot(
+ model,
+ show_shapes=False,
+ show_dtype=False,
+ show_layer_names=True,
+ rankdir="TB",
+ expand_nested=False,
+ dpi=200,
+ subgraph=False,
+ show_layer_activations=False,
+ show_trainable=False,
+ show_layer_attributes=None,
+ layer_styles=None,
+ **kwargs,
+):
+ """Convert a Keras model to dot format.
+
+ Args:
+ model: A Keras model instance.
+ show_shapes: whether to display shape information.
+ show_dtype: whether to display layer dtypes.
+ show_layer_names: whether to display layer names.
+ rankdir: `rankdir` argument passed to PyDot,
+ a string specifying the format of the plot: `"TB"`
+ creates a vertical plot; `"LR"` creates a horizontal plot.
+ expand_nested: whether to expand nested Functional models
+ into clusters.
+ dpi: Image resolution in dots per inch.
+ subgraph: whether to return a `pydot.Cluster` instance.
+ show_layer_activations: Display layer activations (only for layers that
+ have an `activation` property).
+ show_trainable: whether to display if a layer is trainable.
+
+ Returns:
+ A `pydot.Dot` instance representing the Keras model or
+ a `pydot.Cluster` instance representing nested model if
+ `subgraph=True`.
+ """
+ from keras.src.ops.function import make_node_key
+
+ if not model.built:
+ raise ValueError(
+ "This model has not yet been built. "
+ "Build the model first by calling `build()` or by calling "
+ "the model on a batch of data."
+ )
+
+ from keras.src.models import functional, sequential
+
+ # from keras.src.layers import Wrapper
+
+ if not check_pydot():
+ raise ImportError("You must install pydot (`pip install pydot`) for " "model_to_dot to work.")
+
+ if subgraph:
+ dot = pydot.Cluster(style="dashed", graph_name=model.name)
+ dot.set("label", model.name)
+ dot.set("labeljust", "l")
+ else:
+ dot = pydot.Dot()
+ dot.set("rankdir", rankdir)
+ dot.set("concentrate", True)
+ dot.set("dpi", dpi)
+ dot.set("splines", "ortho")
+ dot.set_node_defaults(shape="record")
+
+ if kwargs.pop("layer_range", None) is not None:
+ raise ValueError("Argument `layer_range` is no longer supported.")
+ if kwargs:
+ raise ValueError(f"Unrecognized keyword arguments: {kwargs}")
+
+ kwargs = {
+ "show_layer_names": show_layer_names,
+ "show_layer_activations": show_layer_activations,
+ "show_dtype": show_dtype,
+ "show_shapes": show_shapes,
+ "show_trainable": show_trainable,
+ "show_layer_attributes": show_layer_attributes,
+ "layer_styles": layer_styles,
+ }
+
+ if isinstance(model, sequential.Sequential):
+ # TODO
+ layers = model.layers
+ elif not isinstance(model, functional.Functional):
+ # We treat subclassed models as a single node.
+ node = make_node(model, **kwargs)
+ dot.add_node(node)
+ return dot
+ else:
+ layers = model._operations
+
+ # Create graph nodes.
+ sub_n_first_node = {}
+ sub_n_last_node = {}
+ for i, layer in enumerate(layers):
+ # Process nested functional models.
+ if expand_nested and isinstance(layer, functional.Functional):
+ submodel = model_to_dot(
+ layer,
+ show_shapes,
+ show_dtype,
+ show_layer_names,
+ rankdir,
+ expand_nested,
+ subgraph=True,
+ show_layer_activations=show_layer_activations,
+ show_trainable=show_trainable,
+ show_layer_attributes=show_layer_attributes,
+ layer_styles=layer_styles,
+ )
+ # sub_n : submodel
+ sub_n_nodes = submodel.get_nodes()
+ sub_n_first_node[layer.name] = sub_n_nodes[0]
+ sub_n_last_node[layer.name] = sub_n_nodes[-1]
+ dot.add_subgraph(submodel)
+
+ else:
+ node = make_node(layer, **kwargs)
+ dot.add_node(node)
+
+ # Connect nodes with edges.
+ # Sequential case.
+ if isinstance(model, sequential.Sequential):
+ for i in range(len(layers) - 1):
+ inbound_layer_id = str(id(layers[i]))
+ layer_id = str(id(layers[i + 1]))
+ add_edge(dot, inbound_layer_id, layer_id)
+ return dot
+
+ # Functional case.
+ for i, layer in enumerate(layers):
+ layer_id = str(id(layer))
+ for i, node in enumerate(layer._inbound_nodes):
+ node_key = make_node_key(layer, i)
+ if node_key in model._nodes:
+ for parent_node in node.parent_nodes:
+ inbound_layer = parent_node.operation
+ inbound_layer_id = str(id(inbound_layer))
+ if not expand_nested:
+ assert dot.get_node(inbound_layer_id)
+ assert dot.get_node(layer_id)
+ add_edge(dot, inbound_layer_id, layer_id)
+ else:
+ # if inbound_layer is not Functional
+ if not isinstance(inbound_layer, functional.Functional):
+ # if current layer is not Functional
+ if not isinstance(layer, functional.Functional):
+ assert dot.get_node(inbound_layer_id)
+ assert dot.get_node(layer_id)
+ add_edge(dot, inbound_layer_id, layer_id)
+ # if current layer is Functional
+ elif isinstance(layer, functional.Functional):
+ add_edge(
+ dot,
+ inbound_layer_id,
+ sub_n_first_node[layer.name].get_name(),
+ )
+ # if inbound_layer is Functional
+ elif isinstance(inbound_layer, functional.Functional):
+ name = sub_n_last_node[inbound_layer.name].get_name()
+ if isinstance(layer, functional.Functional):
+ output_name = sub_n_first_node[layer.name].get_name()
+ add_edge(dot, name, output_name)
+ else:
+ add_edge(dot, name, layer_id)
+ return dot
+
+
+def plot_model(
+ model,
+ to_file="model.png",
+ show_shapes=False,
+ show_dtype=False,
+ show_layer_names=True,
+ rankdir="TB",
+ expand_nested=False,
+ dpi=64,
+ show_layer_activations=False,
+ show_trainable=False,
+ show_layer_attributes=None,
+ layer_styles=None,
+ **kwargs,
+):
+ """Converts a Keras model to dot format and save to a file.
+
+ Example:
+
+ ```python
+ inputs = ...
+ outputs = ...
+ model = keras.Model(inputs=inputs, outputs=outputs)
+
+ dot_img_file = '/tmp/model_1.png'
+ keras.utils.plot_model(model, to_file=dot_img_file, show_shapes=True)
+ ```
+
+ Args:
+ model: A Keras model instance
+ to_file: File name of the plot image.
+ show_shapes: whether to display shape information.
+ show_dtype: whether to display layer dtypes.
+ show_layer_names: whether to display layer names.
+ rankdir: `rankdir` argument passed to PyDot,
+ a string specifying the format of the plot: `"TB"`
+ creates a vertical plot; `"LR"` creates a horizontal plot.
+ expand_nested: whether to expand nested Functional models
+ into clusters.
+ dpi: Image resolution in dots per inch.
+ show_layer_activations: Display layer activations (only for layers that
+ have an `activation` property).
+ show_trainable: whether to display if a layer is trainable.
+
+ Returns:
+ A Jupyter notebook Image object if Jupyter is installed.
+ This enables in-line display of the model plots in notebooks.
+ """
+
+ if not model.built:
+ raise ValueError(
+ "This model has not yet been built. "
+ "Build the model first by calling `build()` or by calling "
+ "the model on a batch of data."
+ )
+ if not check_pydot():
+ message = "You must install pydot (`pip install pydot`) " "for `plot_model` to work."
+ if "IPython.core.magics.namespace" in sys.modules:
+ # We don't raise an exception here in order to avoid crashing
+ # notebook tests where graphviz is not available.
+ io_utils.print_msg(message)
+ return
+ else:
+ raise ImportError(message)
+ if not check_graphviz():
+ message = (
+ "You must install graphviz "
+ "(see instructions at https://graphviz.gitlab.io/download/) "
+ "for `plot_model` to work."
+ )
+ if "IPython.core.magics.namespace" in sys.modules:
+ # We don't raise an exception here in order to avoid crashing
+ # notebook tests where graphviz is not available.
+ io_utils.print_msg(message)
+ return
+ else:
+ raise ImportError(message)
+
+ if kwargs.pop("layer_range", None) is not None:
+ raise ValueError("Argument `layer_range` is no longer supported.")
+ if kwargs:
+ raise ValueError(f"Unrecognized keyword arguments: {kwargs}")
+
+ if show_layer_attributes is None:
+ show_layer_attributes = show_decomon_layer_attributes
+ if layer_styles is None:
+ layer_styles = decomon_utilitary_layers_styles
+
+ dot = model_to_dot(
+ model,
+ show_shapes=show_shapes,
+ show_dtype=show_dtype,
+ show_layer_names=show_layer_names,
+ rankdir=rankdir,
+ expand_nested=expand_nested,
+ dpi=dpi,
+ show_layer_activations=show_layer_activations,
+ show_trainable=show_trainable,
+ show_layer_attributes=show_layer_attributes,
+ layer_styles=layer_styles,
+ )
+ to_file = str(to_file)
+ if dot is None:
+ return
+ _, extension = os.path.splitext(to_file)
+ if not extension:
+ extension = "png"
+ else:
+ extension = extension[1:]
+ # Save image to disk.
+ dot.write(to_file, format=extension)
+ # Return the image as a Jupyter Image object, to be displayed in-line.
+ # Note that we cannot easily detect whether the code is running in a
+ # notebook, and thus we always return the Image if Jupyter is available.
+ if extension != "pdf":
+ try:
+ from IPython import display
+
+ return display.Image(filename=to_file)
+ except ImportError:
+ pass
diff --git a/src/decomon/wrapper.py b/src/decomon/wrapper.py
index d3952e48..34c3c248 100644
--- a/src/decomon/wrapper.py
+++ b/src/decomon/wrapper.py
@@ -1,13 +1,14 @@
-from typing import Callable, List, Optional, Sequence, Tuple, Union
+from collections.abc import Callable, Sequence
+from typing import Optional, Union
import keras
import numpy as np
import numpy.typing as npt
-from decomon.core import BallDomain, BoxDomain, GridDomain
+from decomon.constants import ConvertMethod
from decomon.models.convert import clone
from decomon.models.models import DecomonModel
-from decomon.models.utils import ConvertMethod
+from decomon.perturbation_domain import BallDomain, BoxDomain, GridDomain
IntegerType = Union[int, np.int_]
"""Alias for integers types."""
@@ -61,8 +62,12 @@ def get_adv_box(
batch_size: for computational efficiency, one can split the
calls to minibatches
n_sub_boxes:
+
Returns:
numpy array, vector with upper bounds for adversarial attacks
+
+ Hypothesis: the underlying keras model has only 1 output
+
"""
if np.min(x_max - x_min) < 0:
raise UserWarning("Inconsistency Error: x_max < x_min")
@@ -75,6 +80,11 @@ def get_adv_box(
assert isinstance(model.perturbation_domain, (BoxDomain, GridDomain))
decomon_model = model
+ # if missing batch axis, add it
+ input_shape = decomon_model.model.inputs[0].shape[1:]
+ x_min = np.reshape(x_min, (-1,) + input_shape)
+ x_max = np.reshape(x_max, (-1,) + input_shape)
+
n_split = 1
n_batch = len(x_min)
@@ -86,24 +96,12 @@ def get_adv_box(
x_min = np.reshape(x_min, [-1] + shape)
x_max = np.reshape(x_max, [-1] + shape)
- # reshape x_mmin, x_max
- if decomon_model.backward_bounds:
- input_shape = list(decomon_model.input_shape[0][2:])
- else:
- input_shape = list(decomon_model.input_shape[2:])
- input_dim = np.prod(input_shape)
- x_reshaped = x_min + 0 * x_min
- x_reshaped = x_reshaped.reshape([-1] + input_shape)
-
- x_min = x_min.reshape((-1, 1, input_dim))
- x_max = x_max.reshape((-1, 1, input_dim))
-
- z = np.concatenate([x_min, x_max], 1)
+ # construct perturbation_domain_input
+ perturbation_domain_input = np.concatenate([x_min[:, None], x_max[:, None]], 1)
source_labels = prepare_labels(source_labels, n_batch)
if target_labels is not None:
target_labels = prepare_labels(target_labels, n_batch)
-
if n_split > 1:
shape = list(source_labels.shape[1:])
source_labels = np.reshape(np.concatenate([source_labels[:, None]] * n_split, 1), [-1] + shape)
@@ -113,16 +111,24 @@ def get_adv_box(
if batch_size > 0:
# split
r = 0
- if len(x_reshaped) % batch_size > 0:
+ if len(perturbation_domain_input) % batch_size > 0:
r += 1
- x_min_list = [x_min[batch_size * i : batch_size * (i + 1)] for i in range(len(x_reshaped) // batch_size + r)]
- x_max_list = [x_max[batch_size * i : batch_size * (i + 1)] for i in range(len(x_reshaped) // batch_size + r)]
+ x_min_list = [
+ x_min[batch_size * i : batch_size * (i + 1)]
+ for i in range(len(perturbation_domain_input) // batch_size + r)
+ ]
+ x_max_list = [
+ x_max[batch_size * i : batch_size * (i + 1)]
+ for i in range(len(perturbation_domain_input) // batch_size + r)
+ ]
source_labels_list = [
- source_labels[batch_size * i : batch_size * (i + 1)] for i in range(len(x_reshaped) // batch_size + r)
+ source_labels[batch_size * i : batch_size * (i + 1)]
+ for i in range(len(perturbation_domain_input) // batch_size + r)
]
if target_labels is not None:
target_labels_list = [
- target_labels[batch_size * i : batch_size * (i + 1)] for i in range(len(x_reshaped) // batch_size + r)
+ target_labels[batch_size * i : batch_size * (i + 1)]
+ for i in range(len(perturbation_domain_input) // batch_size + r)
]
adv_score = np.concatenate(
[
@@ -146,12 +152,13 @@ def get_adv_box(
n_label = source_labels.shape[-1]
# two possitible cases: the model improves the bound based on the knowledge of the labels
- output: List[npt.NDArray[np.float_]]
- if decomon_model.backward_bounds:
+ output: list[npt.NDArray[np.float_]]
+ needs_backward_bounds = len(decomon_model.inputs) > 1
+ if needs_backward_bounds: # backward bounds needed
C = np.diag([1] * n_label)[None] - source_labels[:, :, None]
- output = decomon_model.predict_on_single_batch_np([z, C]) # type: ignore
+ output = decomon_model.predict_on_single_batch_np([perturbation_domain_input, C]) # type: ignore
else:
- output = decomon_model.predict_on_single_batch_np(z) # type: ignore
+ output = decomon_model.predict_on_single_batch_np(perturbation_domain_input) # type: ignore
def get_ibp_score(
u_c: npt.NDArray[np.float_],
@@ -229,19 +236,26 @@ def get_affine_score(
return np.max(np.max(upper, -2), -1)
if ibp and affine:
- z, u_c, w_u_f, b_u_f, l_c, w_l_f, b_l_f = output[:7]
+ w_l_f, b_l_f, w_u_f, b_u_f, l_c, u_c = output
elif not ibp and affine:
- z, w_u_f, b_u_f, w_l_f, b_l_f = output[:5]
+ w_l_f, b_l_f, w_u_f, b_u_f = output
elif ibp and not affine:
- u_c, l_c = output[:2]
+ l_c, u_c = output
else:
raise NotImplementedError("not ibp and not affine not implemented")
if ibp:
- adv_ibp = get_ibp_score(u_c, l_c, source_labels, target_labels, backward=decomon_model.backward_bounds)
+ adv_ibp = get_ibp_score(u_c, l_c, source_labels, target_labels, backward=needs_backward_bounds)
if affine:
adv_f = get_affine_score(
- z, w_u_f, b_u_f, w_l_f, b_l_f, source_labels, target_labels, backward=decomon_model.backward_bounds
+ perturbation_domain_input,
+ w_u_f,
+ b_u_f,
+ w_l_f,
+ b_l_f,
+ source_labels,
+ target_labels,
+ backward=needs_backward_bounds,
)
if ibp and not affine:
@@ -295,21 +309,17 @@ def check_adv_box(
assert isinstance(model.perturbation_domain, (BoxDomain, GridDomain))
decomon_model = model
- ibp = decomon_model.ibp
- affine = decomon_model.affine
+ # if missing batch axis, add it
+ input_shape = decomon_model.model.inputs[0].shape[1:]
+ x_min = np.reshape(x_min, (-1,) + input_shape)
+ x_max = np.reshape(x_max, (-1,) + input_shape)
n_split = 1
n_batch = len(x_min)
- # reshape x_mmin, x_max
- input_shape = list(decomon_model.input_shape[2:])
- input_dim = np.prod(input_shape)
- x_reshaped = x_min + 0 * x_min
- x_reshaped = x_reshaped.reshape([-1] + input_shape)
- x_min = x_min.reshape((-1, 1, input_dim))
- x_max = x_max.reshape((-1, 1, input_dim))
+ # construct perturbation_domain_input
+ perturbation_domain_input = np.concatenate([x_min[:, None], x_max[:, None]], 1)
- z = np.concatenate([x_min, x_max], 1)
if isinstance(source_labels, (int, np.int_)):
source_labels = np.zeros((n_batch, 1)) + source_labels
@@ -332,24 +342,32 @@ def check_adv_box(
if batch_size > 0:
# split
r = 0
- if len(x_reshaped) % batch_size > 0:
+ if len(perturbation_domain_input) % batch_size > 0:
r += 1
- x_min_list = [x_min[batch_size * i : batch_size * (i + 1)] for i in range(len(x_reshaped) // batch_size + r)]
- x_max_list = [x_max[batch_size * i : batch_size * (i + 1)] for i in range(len(x_reshaped) // batch_size + r)]
+ x_min_list = [
+ x_min[batch_size * i : batch_size * (i + 1)]
+ for i in range(len(perturbation_domain_input) // batch_size + r)
+ ]
+ x_max_list = [
+ x_max[batch_size * i : batch_size * (i + 1)]
+ for i in range(len(perturbation_domain_input) // batch_size + r)
+ ]
source_labels_list = [
- source_labels[batch_size * i : batch_size * (i + 1)] for i in range(len(x_reshaped) // batch_size + r)
+ source_labels[batch_size * i : batch_size * (i + 1)]
+ for i in range(len(perturbation_domain_input) // batch_size + r)
]
- target_labels_list: List[Optional[npt.NDArray[np.int_]]]
+ target_labels_list: list[Optional[npt.NDArray[np.int_]]]
if (
(target_labels is not None)
and (not isinstance(target_labels, int))
and (str(target_labels.dtype)[:3] != "int")
):
target_labels_list = [
- target_labels[batch_size * i : batch_size * (i + 1)] for i in range(len(x_reshaped) // batch_size + r)
+ target_labels[batch_size * i : batch_size * (i + 1)]
+ for i in range(len(perturbation_domain_input) // batch_size + r)
]
else:
- target_labels_list = [target_labels] * (len(x_reshaped) // batch_size + r)
+ target_labels_list = [target_labels] * (len(perturbation_domain_input) // batch_size + r)
return np.concatenate(
[
@@ -361,15 +379,7 @@ def check_adv_box(
)
else:
- output = decomon_model.predict_on_single_batch_np(z)
-
- if not affine:
- # translate into affine information
- u_c = output[0]
- w_u = 0 * u_c[:, None] + np.zeros((1, input_dim, 1))
- output = [z, w_u, u_c, w_u, output[-1]]
- ibp = False
- affine = True
+ w_l_f, b_l_f, w_u_f, b_u_f = decomon_model.compute_affine_bounds_np([perturbation_domain_input])
def get_affine_sample(
z_tensor: npt.NDArray[np.float_],
@@ -409,12 +419,7 @@ def get_affine_sample(
return np.max(np.max(upper, -2), -1)
- if ibp:
- z, u_c, w_u_f, b_u_f, l_c, w_l_f, b_l_f = output[:7]
- else:
- z, w_u_f, b_u_f, w_l_f, b_l_f = output[:5]
-
- return get_affine_sample(z, w_l_f, b_l_f, w_u_f, b_u_f, source_labels)
+ return get_affine_sample(perturbation_domain_input, w_l_f, b_l_f, w_u_f, b_u_f, source_labels)
#### FORMAL BOUNDS ######
@@ -470,20 +475,24 @@ def get_range_box(
x_max: npt.NDArray[np.float_],
batch_size: int = -1,
n_sub_boxes: int = 1,
-) -> Tuple[npt.NDArray[np.float_], npt.NDArray[np.float_]]:
- """bounding the outputs of a model in a given box
- if the constant is negative, then it is a formal guarantee that there is no adversarial examples
+) -> tuple[npt.NDArray[np.float_], npt.NDArray[np.float_]]:
+ """Bound the outputs of a model in a given box
Args:
model: either a Keras model or a Decomon model
x_min: numpy array for the extremal lower corner of the boxes
x_max: numpy array for the extremal upper corner of the boxes
batch_size: for computational efficiency, one can split the
- calls to minibatches
+ calls into minibatches
Returns:
2 numpy array, vector with upper bounds and vector with lower
bounds
+
+ Hypotheses:
+ - the underlying keras model has only 1 output;
+ - if already cloned as a decomon model, non precomputed backward_bounds have been given.
+
"""
if np.min(x_max - x_min) < 0:
raise UserWarning("Inconsistency Error: x_max < x_min")
@@ -495,8 +504,10 @@ def get_range_box(
assert isinstance(model.perturbation_domain, (BoxDomain, GridDomain))
decomon_model = model
- n_split = 1
- n_batch = len(x_min)
+ # if missing batch axis, add it
+ input_shape = decomon_model.model.inputs[0].shape[1:]
+ x_min = np.reshape(x_min, (-1,) + input_shape)
+ x_max = np.reshape(x_max, (-1,) + input_shape)
if n_sub_boxes > 1:
x_min, x_max = refine_boxes(x_min, x_max, n_sub_boxes)
@@ -505,87 +516,33 @@ def get_range_box(
shape = list(x_min.shape[2:])
x_min = np.reshape(x_min, [-1] + shape)
x_max = np.reshape(x_max, [-1] + shape)
+ else:
+ n_split = 1
- # reshape x_mmin, x_max
- input_shape = list(decomon_model.input_shape[2:])
- input_dim = np.prod(input_shape)
- x_reshaped = x_min + 0 * x_min
- x_reshaped = x_reshaped.reshape([-1] + input_shape)
- x_min = x_min.reshape((-1, 1, input_dim))
- x_max = x_max.reshape((-1, 1, input_dim))
-
- z = np.concatenate([x_min, x_max], 1)
+ # construct perturbation_domain_input
+ perturbation_domain_input = np.concatenate([x_min[:, None], x_max[:, None]], 1)
if batch_size > 0:
- # split
+ # call by minibatch
r = 0
- if len(x_reshaped) % batch_size > 0:
+ if len(perturbation_domain_input) % batch_size > 0:
r += 1
- x_min_list = [x_min[batch_size * i : batch_size * (i + 1)] for i in range(len(x_reshaped) // batch_size + r)]
- x_max_list = [x_max[batch_size * i : batch_size * (i + 1)] for i in range(len(x_reshaped) // batch_size + r)]
+ x_min_list = [
+ x_min[batch_size * i : batch_size * (i + 1)]
+ for i in range(len(perturbation_domain_input) // batch_size + r)
+ ]
+ x_max_list = [
+ x_max[batch_size * i : batch_size * (i + 1)]
+ for i in range(len(perturbation_domain_input) // batch_size + r)
+ ]
results = [get_range_box(decomon_model, x_min_list[i], x_max_list[i], -1) for i in range(len(x_min_list))]
u_out = np.concatenate([r[0] for r in results])
l_out = np.concatenate([r[1] for r in results])
else:
- ibp = decomon_model.ibp
- affine = decomon_model.affine
-
- output = decomon_model.predict_on_single_batch_np(z)
- shape = list(output[-1].shape[1:])
- output_dim = np.prod(shape)
-
- if affine:
- if ibp:
- _, u_i, w_u_f, b_u_f, l_i, w_l_f, b_l_f = output[:7]
- if len(u_i.shape) > 2:
- u_i = np.reshape(u_i, (-1, output_dim))
- l_i = np.reshape(l_i, (-1, output_dim))
- else:
- _, w_u_f, b_u_f, w_l_f, b_l_f = output[:5]
-
- # reshape if necessary
- if len(w_u_f.shape) > 3:
- w_u_f = np.reshape(w_u_f, (-1, input_dim, output_dim))
- w_l_f = np.reshape(w_l_f, (-1, input_dim, output_dim))
- b_u_f = np.reshape(b_u_f, (-1, output_dim))
- b_l_f = np.reshape(b_l_f, (-1, output_dim))
-
- u_f = (
- np.sum(np.maximum(w_u_f, 0) * x_max[:, 0, :, None], 1)
- + np.sum(np.minimum(w_u_f, 0) * x_min[:, 0, :, None], 1)
- + b_u_f
- )
- l_f = (
- np.sum(np.maximum(w_l_f, 0) * x_min[:, 0, :, None], 1)
- + np.sum(np.minimum(w_l_f, 0) * x_max[:, 0, :, None], 1)
- + b_l_f
- )
-
- else:
- u_i = output[0]
- l_i = output[1]
- if len(u_i.shape) > 2:
- u_i = np.reshape(u_i, (-1, output_dim))
- l_i = np.reshape(l_i, (-1, output_dim))
-
- if ibp and affine:
- u_out = np.minimum(u_i, u_f)
- l_out = np.maximum(l_i, l_f)
- elif ibp and not affine:
- u_out = u_i
- l_out = l_i
- elif not ibp and affine:
- u_out = u_f
- l_out = l_f
- else:
- raise NotImplementedError("not ibp and not affine not implemented")
-
- #####
- if len(shape) > 1:
- u_out = np.reshape(u_out, [-1] + shape)
- l_out = np.reshape(l_out, [-1] + shape)
+ # call on all data at once
+ l_out, u_out = decomon_model.compute_constant_bounds_np([perturbation_domain_input])
if n_split > 1:
u_out = np.max(np.reshape(u_out, (-1, n_split)), -1)
@@ -652,8 +609,8 @@ def get_range_noise(
eps: float,
p: float = np.inf,
batch_size: int = -1,
-) -> Tuple[npt.NDArray[np.float_], npt.NDArray[np.float_]]:
- """Bounds the output of a model in an Lp Ball
+) -> tuple[npt.NDArray[np.float_], npt.NDArray[np.float_]]:
+ """Bound the output of a model on an Lp Ball
Args:
model: either a Keras model or a Decomon model
@@ -661,11 +618,15 @@ def get_range_noise(
eps: the radius of the ball
p: the type of Lp norm (p=2, 1, np.inf)
batch_size: for computational efficiency, one can split the
- calls to minibatches
+ calls into minibatches
Returns:
- 2 numpy arrays, vector with upper andlower bounds
- of the range of values taken by the model inside the ball
+ 2 numpy arrays, vector with upper and lower bounds of the range of values taken by the model inside the ball
+
+ Hypotheses:
+ - the underlying keras model has only 1 output;
+ - if already cloned as a decomon model, non precomputed backward_bounds have been given.
+
"""
# check that the model is a DecomonModel, else do the conversion
@@ -682,81 +643,34 @@ def get_range_noise(
if eps >= 0:
decomon_model.set_domain(perturbation_domain)
- # reshape x_mmin, x_max
- input_shape = list(decomon_model.input_shape[1:])
- input_dim = np.prod(input_shape)
- x_reshaped = x + 0 * x
- x_reshaped = x_reshaped.reshape([-1] + input_shape)
+ # if missing batch axis, add it
+ input_shape = decomon_model.model.inputs[0].shape[1:]
+ x = np.reshape(x, (-1,) + input_shape)
+
+ # perturbation_domain_input : x itself
+ perturbation_domain_input = x
if batch_size > 0:
- # split
+ # call by minibatch
r = 0
- if len(x_reshaped) % batch_size > 0:
+ if len(perturbation_domain_input) % batch_size > 0:
r += 1
- x_list = [x_reshaped[batch_size * i : batch_size * (i + 1)] for i in range(len(x_reshaped) // batch_size + r)]
+ x_list = [
+ x[batch_size * i : batch_size * (i + 1)] for i in range(len(perturbation_domain_input) // batch_size + r)
+ ]
results = [get_range_noise(decomon_model, x_list[i], eps=eps, p=p, batch_size=-1) for i in range(len(x_list))]
return np.concatenate([r[0] for r in results]), np.concatenate([r[1] for r in results])
- ibp = decomon_model.ibp
- affine = decomon_model.affine
-
- output = decomon_model.predict_on_single_batch_np(x_reshaped)
- shape = list(output[-1].shape[1:])
- output_dim = np.prod(shape)
-
- x_reshaped = x_reshaped.reshape((len(x_reshaped), -1))
- ord = _get_dual_ord(p)
-
- if affine:
- if ibp:
- _, u_i, w_u_f, b_u_f, l_i, w_l_f, b_l_f = output[:7]
- if len(u_i.shape) > 2:
- u_i = np.reshape(u_i, (-1, output_dim))
- l_i = np.reshape(l_i, (-1, output_dim))
- else:
- _, w_u_f, b_u_f, w_l_f, b_l_f = output[:5]
-
- # reshape if necessary
- if len(w_u_f.shape) > 3:
- w_u_f = np.reshape(w_u_f, (-1, input_dim, output_dim))
- b_u_f = np.reshape(b_u_f, (-1, output_dim))
- w_l_f = np.reshape(w_l_f, (-1, input_dim, output_dim))
- b_l_f = np.reshape(b_l_f, (-1, output_dim))
-
- u_f = eps * np.linalg.norm(w_u_f, ord=ord, axis=1) + np.sum(w_u_f * x_reshaped[:, :, None], 1) + b_u_f
- l_f = -eps * np.linalg.norm(w_l_f, ord=ord, axis=1) + np.sum(w_l_f * x_reshaped[:, :, None], 1) + b_l_f
-
- else:
- u_i = output[0]
- l_i = output[1]
- if len(u_i.shape) > 2:
- u_i = np.reshape(u_i, (-1, output_dim))
- l_i = np.reshape(l_i, (-1, output_dim))
- ######
-
- if ibp and affine:
- u_out = np.minimum(u_i, u_f)
- l_out = np.maximum(l_i, l_f)
- elif ibp and not affine:
- u_out = u_i
- l_out = l_i
- elif not ibp and affine:
- u_out = u_f
- l_out = l_f
else:
- raise NotImplementedError("not ibp and not affine not implemented")
-
- if len(shape) > 1:
- u_out = np.reshape(u_out, [-1] + shape)
- l_out = np.reshape(l_out, [-1] + shape)
-
- return u_out, l_out
+ # call on all data at once
+ l_out, u_out = decomon_model.compute_constant_bounds_np([perturbation_domain_input])
+ return u_out, l_out
def refine_boxes(
x_min: npt.NDArray[np.float_], x_max: npt.NDArray[np.float_], n_sub_boxes: int = 10
-) -> Tuple[npt.NDArray[np.float_], npt.NDArray[np.float_]]:
+) -> tuple[npt.NDArray[np.float_], npt.NDArray[np.float_]]:
# flatten x_min and x_max
shape = list(x_min.shape[1:])
output_dim = np.prod(shape)
@@ -771,7 +685,7 @@ def refine_boxes(
def split(
x_min: npt.NDArray[np.float_], x_max: npt.NDArray[np.float_], j: npt.NDArray[np.int_]
- ) -> Tuple[npt.NDArray[np.float_], npt.NDArray[np.float_]]:
+ ) -> tuple[npt.NDArray[np.float_], npt.NDArray[np.float_]]:
n_0 = len(x_min)
n_k = x_min.shape[1]
@@ -948,16 +862,12 @@ def get_adv_noise(
eps = model.perturbation_domain.eps
- # reshape x_mmin, x_max
- input_shape = list(decomon_model.input_shape[1:])
- x_reshaped = x + 0 * x
- x_reshaped = x_reshaped.reshape([-1] + input_shape)
- n_split = 1
- n_batch = len(x_reshaped)
+ # if missing batch axis, add it
+ input_shape = decomon_model.model.inputs[0].shape[1:]
+ x = np.reshape(x, (-1,) + input_shape)
- input_shape = list(decomon_model.input_shape[1:])
- x_reshaped = x + 0 * x
- x_reshaped = x_reshaped.reshape([-1] + input_shape)
+ n_split = 1
+ n_batch = len(x)
if isinstance(source_labels, (int, np.int_)):
source_labels = np.zeros((n_batch, 1), dtype=np.int_) + source_labels
@@ -971,23 +881,23 @@ def get_adv_noise(
if batch_size > 0:
# split
r = 0
- if len(x_reshaped) % batch_size > 0:
+ if len(x) % batch_size > 0:
r += 1
- x_list = [x_reshaped[batch_size * i : batch_size * (i + 1)] for i in range(len(x_reshaped) // batch_size + r)]
+ x_list = [x[batch_size * i : batch_size * (i + 1)] for i in range(len(x) // batch_size + r)]
source_labels_list = [
- source_labels[batch_size * i : batch_size * (i + 1)] for i in range(len(x_reshaped) // batch_size + r)
+ source_labels[batch_size * i : batch_size * (i + 1)] for i in range(len(x) // batch_size + r)
]
- target_labels_list: List[Optional[npt.NDArray[np.int_]]]
+ target_labels_list: list[Optional[npt.NDArray[np.int_]]]
if (
(target_labels is not None)
and (not isinstance(target_labels, int))
and (str(target_labels.dtype)[:3] != "int")
):
target_labels_list = [
- target_labels[batch_size * i : batch_size * (i + 1)] for i in range(len(x_reshaped) // batch_size + r)
+ target_labels[batch_size * i : batch_size * (i + 1)] for i in range(len(x) // batch_size + r)
]
else:
- target_labels_list = [target_labels] * (len(x_reshaped) // batch_size + r)
+ target_labels_list = [target_labels] * (len(x) // batch_size + r)
results = [
get_adv_noise(
@@ -1006,7 +916,7 @@ def get_adv_noise(
else:
ibp = decomon_model.ibp
affine = decomon_model.affine
- output = decomon_model.predict_on_single_batch_np(x_reshaped)
+ output = decomon_model.predict_on_single_batch_np(x)
def get_ibp_score(
u_c: npt.NDArray[np.float_],
@@ -1083,18 +993,18 @@ def get_affine_score(
return np.max(np.max(upper, -2), -1)
if ibp and affine:
- z, u_c, w_u_f, b_u_f, l_c, w_l_f, b_l_f = output[:7]
- if not ibp and affine:
- z, w_u_f, b_u_f, w_l_f, b_l_f = output[:5]
- if ibp and not affine:
- u_c, l_c = output[:2]
+ w_l_f, b_l_f, w_u_f, b_u_f, l_c, u_c = output
+ elif not ibp and affine:
+ w_l_f, b_l_f, w_u_f, b_u_f = output
+ elif ibp and not affine:
+ l_c, u_c = output
else:
raise NotImplementedError("not ibp and not affine not implemented")
if ibp:
adv_ibp = get_ibp_score(u_c, l_c, source_labels, target_labels)
if affine:
- adv_f = get_affine_score(z, w_u_f, b_u_f, w_l_f, b_l_f, source_labels, target_labels)
+ adv_f = get_affine_score(x, w_u_f, b_u_f, w_l_f, b_l_f, source_labels, target_labels)
if ibp and not affine:
adv_score = adv_ibp
diff --git a/src/decomon/wrapper_with_tuning.py b/src/decomon/wrapper_with_tuning.py
deleted file mode 100644
index c7af68a6..00000000
--- a/src/decomon/wrapper_with_tuning.py
+++ /dev/null
@@ -1,183 +0,0 @@
-from typing import Union
-
-import keras
-import numpy as np
-import numpy.typing as npt
-from keras.optimizers import Adam
-
-from decomon.metrics.loss import get_upper_loss
-from decomon.models.models import DecomonModel
-from decomon.wrapper import get_lower_box, get_upper_box, refine_boxes
-
-
-#### FORMAL BOUNDS ######
-def get_upper_box_tuning(
- model: Union[keras.Model, DecomonModel],
- decomon_model_concat: keras.Model,
- x_min: npt.NDArray[np.float_],
- x_max: npt.NDArray[np.float_],
- batch_size: int = 1,
- n_sub_boxes: int = 1,
- lr: float = 0.1,
- epochs: int = 100,
-) -> npt.NDArray[np.float_]:
- """upper bound the maximum of a model in a given box
-
- Args:
- model: either a Keras model or a Decomon model
- x_min: numpy array for the extremal lower corner of the boxes
- x_max: numpy array for the extremal upper corner of the boxes
- batch_size: for computational efficiency, one can split the
- calls to minibatches
- fast: useful in the forward-backward or in the hybrid-backward
- mode to optimize the scores
-
- Returns:
- numpy array, vector with upper bounds for adversarial attacks
- """
-
- if np.min(x_max - x_min) < 0:
- raise UserWarning("Inconsistency Error: x_max < x_min")
-
- # check that the model is a DecomonModel, else do the conversion
- decomon_model = model
-
- baseline_upper = get_upper_box(model, x_min, x_max, batch_size=batch_size, n_sub_boxes=n_sub_boxes)
-
- if n_sub_boxes > 1:
- x_min, x_max = refine_boxes(x_min, x_max, n_sub_boxes)
- # reshape
- shape = list(x_min.shape[2:])
- x_min = np.reshape(x_min, [-1] + shape)
- x_max = np.reshape(x_max, [-1] + shape)
-
- # reshape x_mmin, x_max
- input_shape = list(decomon_model.input_shape[2:])
- input_dim = np.prod(input_shape)
- x_reshaped = x_min + 0 * x_min
- x_reshaped = x_reshaped.reshape([-1] + input_shape)
- x_min = x_min.reshape((-1, 1, input_dim))
- x_max = x_max.reshape((-1, 1, input_dim))
-
- z = np.concatenate([x_min, x_max], 1)
-
- if batch_size > 1:
- # split
- r = 0
- if len(x_reshaped) % batch_size > 0:
- r += 1
- x_min_list = [x_min[batch_size * i : batch_size * (i + 1)] for i in range(len(x_reshaped) // batch_size + r)]
- x_max_list = [x_max[batch_size * i : batch_size * (i + 1)] for i in range(len(x_reshaped) // batch_size + r)]
-
- results = [
- get_upper_box_tuning(
- decomon_model, decomon_model_concat, x_min_list[i], x_max_list[i], -1, lr=lr, epochs=epochs
- )
- for i in range(len(x_min_list))
- ]
-
- return np.concatenate(results)
-
- else:
- # freeze_weights
- model.freeze_weights()
-
- # create loss
- loss_upper = get_upper_loss(model)
- decomon_model_concat.compile(Adam(lr=lr), loss_upper)
-
- decomon_model_concat.fit(z, baseline_upper, epochs=epochs, verbose=0)
-
- upper = get_upper_box(model, x_min, x_max, batch_size=batch_size, n_sub_boxes=n_sub_boxes)
-
- # reset alpha
- model.reset_finetuning()
-
- return np.minimum(baseline_upper, upper)
-
-
-#### FORMAL BOUNDS ######
-def get_lower_box_tuning(
- model: Union[keras.Model, DecomonModel],
- decomon_model_concat: keras.Model,
- x_min: npt.NDArray[np.float_],
- x_max: npt.NDArray[np.float_],
- batch_size: int = 1,
- n_sub_boxes: int = 1,
- lr: float = 0.1,
- epochs: int = 100,
-) -> npt.NDArray[np.float_]:
- """upper bound the maximum of a model in a given box
-
- Args:
- model: either a Keras model or a Decomon model
- x_min: numpy array for the extremal lower corner of the boxes
- x_max: numpy array for the extremal upper corner of the boxes
- batch_size: for computational efficiency, one can split the
- calls to minibatches
- fast: useful in the forward-backward or in the hybrid-backward
- mode to optimize the scores
-
- Returns:
- numpy array, vector with upper bounds for adversarial attacks
- """
-
- if np.min(x_max - x_min) < 0:
- raise UserWarning("Inconsistency Error: x_max < x_min")
-
- # check that the model is a DecomonModel, else do the conversion
- decomon_model = model
-
- baseline_upper = get_lower_box(model, x_min, x_max, batch_size=batch_size, n_sub_boxes=n_sub_boxes)
-
- if n_sub_boxes > 1:
- x_min, x_max = refine_boxes(x_min, x_max, n_sub_boxes)
- # reshape
- shape = list(x_min.shape[2:])
- x_min = np.reshape(x_min, [-1] + shape)
- x_max = np.reshape(x_max, [-1] + shape)
-
- # reshape x_mmin, x_max
- input_shape = list(decomon_model.input_shape[2:])
- input_dim = np.prod(input_shape)
- x_reshaped = x_min + 0 * x_min
- x_reshaped = x_reshaped.reshape([-1] + input_shape)
- x_min = x_min.reshape((-1, 1, input_dim))
- x_max = x_max.reshape((-1, 1, input_dim))
-
- z = np.concatenate([x_min, x_max], 1)
-
- if batch_size > 0 and batch_size != len(x_min):
- # split
- r = 0
- if len(x_reshaped) % batch_size > 0:
- r += 1
- x_min_list = [x_min[batch_size * i : batch_size * (i + 1)] for i in range(len(x_reshaped) // batch_size + r)]
- x_max_list = [x_max[batch_size * i : batch_size * (i + 1)] for i in range(len(x_reshaped) // batch_size + r)]
-
- results = [
- get_lower_box_tuning(
- decomon_model, decomon_model_concat, x_min_list[i], x_max_list[i], -1, lr=lr, epochs=epochs
- )
- for i in range(len(x_min_list))
- ]
-
- model.reset_finetuning()
-
- return np.concatenate(results)
- else:
- # freeze_weights
- model.freeze_weights()
-
- # create loss
- loss_upper = get_upper_loss(model)
- decomon_model_concat.compile(Adam(lr=lr), loss_upper)
-
- decomon_model_concat.fit(z, baseline_upper, epochs=epochs, verbose=0)
-
- upper = get_lower_box(model, x_min, x_max, batch_size=batch_size, n_sub_boxes=n_sub_boxes)
-
- # reset alpha
- model.reset_finetuning()
-
- return np.minimum(baseline_upper, upper)
diff --git a/tests/conftest.py b/tests/conftest.py
index 84cb3a1b..99b82805 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,50 +1,77 @@
-from typing import List, Optional, Union
+from typing import Optional, Union
import keras
import keras.config as keras_config
import keras.ops as K
import numpy as np
-import numpy.typing as npt
import pytest
-from keras import KerasTensor
-from keras.layers import (
- Activation,
- Add,
- Average,
- Conv2D,
- Dense,
- Flatten,
- Input,
- Reshape,
+from keras import KerasTensor, Model, Sequential
+from keras.layers import Activation, Add, Conv2D, Dense, Flatten, Input
+from pytest_cases import (
+ fixture,
+ fixture_union,
+ param_fixture,
+ param_fixtures,
+ unpack_fixture,
)
-from keras.models import Model, Sequential
-from numpy.testing import assert_almost_equal
-from pytest_cases import fixture, fixture_union, param_fixture
-from decomon.core import ForwardMode, Slope
+from decomon.constants import ConvertMethod, Propagation, Slope
from decomon.keras_utils import (
BACKEND_JAX,
BACKEND_NUMPY,
BACKEND_PYTORCH,
BACKEND_TENSORFLOW,
+ batch_multid_dot,
)
-from decomon.models.utils import ConvertMethod
-from decomon.types import Tensor
-
-
-@pytest.fixture(params=[m.value for m in ForwardMode])
-def mode(request):
- return request.param
+from decomon.layers.inputs_outputs_specs import InputsOutputsSpec
+from decomon.perturbation_domain import BoxDomain
+from decomon.types import BackendTensor, Tensor
+
+empty, diag, nobatch = param_fixtures(
+ "empty, diag, nobatch",
+ [
+ (True, True, True),
+ (False, True, True),
+ (False, True, False),
+ (False, False, True),
+ (False, False, False),
+ ],
+ ids=["identity", "diagonal-nobatch", "diagonal", "nobatch", "generic"],
+)
+ibp, affine, propagation = param_fixtures(
+ "ibp, affine, propagation",
+ [
+ (True, True, Propagation.FORWARD),
+ (False, True, Propagation.FORWARD),
+ (True, False, Propagation.FORWARD),
+ (True, True, Propagation.BACKWARD),
+ ],
+ ids=["forward-hybrid", "forward-affine", "forward-ibp", "backward"],
+)
+final_ibp, final_affine = param_fixtures(
+ "final_ibp, final_affine", [(True, False), (False, True), (True, True)], ids=["ibp", "affine", "hybrid"]
+)
+slope = param_fixture("slope", [s.value for s in Slope])
+n = param_fixture("n", list(range(10)))
+odd = param_fixture("odd", list(range(2)))
+use_bias = param_fixture("use_bias", [True, False])
+randomize = param_fixture("randomize", [True, False])
+padding = param_fixture("padding", ["same", "valid"])
+activation = param_fixture("activation", [None, "relu", "softsign"])
+data_format = param_fixture("data_format", ["channels_last", "channels_first"])
+method = param_fixture("method", [m.value for m in ConvertMethod])
+input_shape = param_fixture("input_shape", [(1,), (3,), (5, 6, 2)], ids=["0d", "1d", "multid"])
+equal_ibp = param_fixture("equal_ibp", [True, False])
-@pytest.fixture(params=[s.value for s in Slope])
-def slope(request):
- return request.param
+@pytest.fixture
+def batchsize():
+ return 10
-@pytest.fixture(params=list(range(10)))
-def n(request):
- return request.param
+@pytest.fixture
+def perturbation_domain():
+ return BoxDomain()
@pytest.fixture(params=[32, 64, 16])
@@ -71,96 +98,13 @@ def decimal(floatx):
return 4
-@pytest.fixture(params=[True, False])
-def dc_decomp(request):
- return request.param
-
-
-@pytest.fixture(params=[True, False])
-def use_bias(request):
- return request.param
-
-
-@pytest.fixture(params=[True, False])
-def shared(request):
- return request.param
-
-
-@pytest.fixture(params=["same", "valid"])
-def padding(request):
- return request.param
-
-
-@pytest.fixture(params=[None, "linear", "relu"])
-def activation(request):
- return request.param
-
-
-data_format = param_fixture(argname="data_format", argvalues=["channels_last", "channels_first"])
-
-
-@pytest.fixture(params=[0, 1])
-def odd(request):
- return request.param
-
-
-@pytest.fixture(params=[1, 2, 3, -1])
-def axis(request):
- return request.param
-
-
-@pytest.fixture(params=[1, 2, 11])
-def channels(request):
- return request.param
-
-
-@pytest.fixture(params=[1, 2, 3])
-def filter_size(request):
- return request.param
-
-
-@pytest.fixture(params=[1, 2])
-def strides(request):
- return request.param
-
-
-@pytest.fixture(params=[True, False])
-def flatten(request):
- return request.param
-
-
-@pytest.fixture(params=[0, 1, 2, 3, 4, 5, 6])
-def finetune_odd(request) -> Optional[np.ndarray]:
- # hard code several configuration of finetune for images odd=1 (6, 6, 2)
- finetune_params = np.zeros((6, 6, 2))
- if request.param == 1:
- finetune_params += 1
- elif request.param == 2:
- finetune_params[0] = 1
- elif request.param == 3:
- finetune_params[:, 0] = 1
- elif request.param == 4:
- finetune_params[:, :, 0] = 1
- elif request.param == 5:
- finetune_params[0, 0, 0] = 1
- else:
- return None
-
- return finetune_params
-
-
-@pytest.fixture(params=[m.value for m in ConvertMethod])
-def method(request):
- return request.param
-
-
class ModelNumpyFromKerasTensors:
- def __init__(self, inputs: List[KerasTensor], outputs: List[KerasTensor]):
+ def __init__(self, inputs: list[KerasTensor], outputs: list[KerasTensor]):
self.inputs = inputs
self.outputs = outputs
self._model = Model(inputs, outputs)
- def __call__(self, inputs_: List[np.ndarray]):
+ def __call__(self, inputs_: list[np.ndarray]):
output_tensors = self._model(inputs_)
if isinstance(output_tensors, list):
return [K.convert_to_numpy(output) for output in output_tensors]
@@ -169,6 +113,8 @@ def __call__(self, inputs_: List[np.ndarray]):
class Helpers:
+ function = ModelNumpyFromKerasTensors
+
@staticmethod
def in_GPU_mode() -> bool:
backend = keras.config.backend()
@@ -190,92 +136,347 @@ def in_GPU_mode() -> bool:
raise NotImplementedError(f"Not implemented for {backend} backend.")
@staticmethod
- def is_method_mode_compatible(method, mode):
- return not (
- ConvertMethod(method) in {ConvertMethod.CROWN_FORWARD_IBP, ConvertMethod.FORWARD_IBP}
- and ForwardMode(mode) != ForwardMode.IBP
+ def generate_random_tensor(shape_wo_batchsize, batchsize=10, dtype=keras_config.floatx(), nobatch=False):
+ if nobatch:
+ shape = shape_wo_batchsize
+ else:
+ shape = (batchsize,) + shape_wo_batchsize
+ return K.convert_to_tensor(2.0 * np.random.random(shape) - 1.0, dtype=dtype)
+
+ @staticmethod
+ def get_decomon_input_shapes(
+ model_input_shape,
+ model_output_shape,
+ layer_input_shape,
+ layer_output_shape,
+ ibp,
+ affine,
+ propagation,
+ perturbation_domain,
+ empty=False,
+ diag=False,
+ nobatch=False,
+ for_linear_layer=False,
+ remove_perturbation_domain_inputs=False,
+ add_perturbation_domain_inputs=False,
+ ):
+ inputs_outputs_spec = InputsOutputsSpec(
+ ibp=ibp,
+ affine=affine,
+ propagation=propagation,
+ layer_input_shape=layer_input_shape,
+ model_input_shape=model_input_shape,
+ model_output_shape=model_output_shape,
+ linear=for_linear_layer,
)
+ if (
+ inputs_outputs_spec.needs_perturbation_domain_inputs() and not remove_perturbation_domain_inputs
+ ) or add_perturbation_domain_inputs:
+ perturbation_domain_inputs_shape = [perturbation_domain.get_x_input_shape_wo_batchsize(model_input_shape)]
+ else:
+ perturbation_domain_inputs_shape = []
- function = ModelNumpyFromKerasTensors
+ if inputs_outputs_spec.needs_affine_bounds_inputs() and not empty:
+ if propagation == Propagation.FORWARD:
+ b_in_shape = layer_input_shape
+ w_in_shape = model_input_shape + layer_input_shape
+ else:
+ b_in_shape = model_output_shape
+ w_in_shape = layer_output_shape + model_output_shape
+ if diag:
+ w_in_shape = b_in_shape
+
+ affine_bounds_to_propagate_shape = [w_in_shape, b_in_shape, w_in_shape, b_in_shape]
+ else:
+ affine_bounds_to_propagate_shape = []
+
+ if inputs_outputs_spec.needs_constant_bounds_inputs():
+ constant_oracle_bounds_shape = [layer_input_shape, layer_input_shape]
+ else:
+ constant_oracle_bounds_shape = []
+
+ return affine_bounds_to_propagate_shape, constant_oracle_bounds_shape, perturbation_domain_inputs_shape
@staticmethod
- def predict_on_small_numpy(
- model: Model, x: Union[np.ndarray, List[np.ndarray]]
- ) -> Union[np.ndarray, List[np.ndarray]]:
- """Make predictions for model directly on small numpy arrays
+ def get_decomon_symbolic_inputs(
+ model_input_shape,
+ model_output_shape,
+ layer_input_shape,
+ layer_output_shape,
+ ibp,
+ affine,
+ propagation,
+ perturbation_domain,
+ empty=False,
+ diag=False,
+ nobatch=False,
+ for_linear_layer=False,
+ remove_perturbation_domain_inputs=False,
+ add_perturbation_domain_inputs=False,
+ dtype=keras_config.floatx(),
+ ):
+ """Generate decomon symbolic inputs for a decomon layer
- Avoid using `model.predict()` known to be not designed for small arrays,
- and leading to memory leaks when used in loops.
+ To be used as `decomon_layer(*decomon_inputs)`.
- See https://keras.io/api/models/model_training_apis/#predict-method and
- https://github.com/tensorflow/tensorflow/issues/44711
+ Args:
+ model_input_shape:
+ model_output_shape:
+ layer_input_shape:
+ layer_output_shape:
+ ibp:
+ affine:
+ propagation:
+ perturbation_domain:
+
+ Returns:
+
+ """
+ (
+ affine_bounds_to_propagate_shape,
+ constant_oracle_bounds_shape,
+ perturbation_domain_inputs_shape,
+ ) = Helpers.get_decomon_input_shapes(
+ model_input_shape,
+ model_output_shape,
+ layer_input_shape,
+ layer_output_shape,
+ ibp,
+ affine,
+ propagation,
+ perturbation_domain,
+ empty=empty,
+ diag=diag,
+ nobatch=nobatch,
+ for_linear_layer=for_linear_layer,
+ remove_perturbation_domain_inputs=remove_perturbation_domain_inputs,
+ add_perturbation_domain_inputs=add_perturbation_domain_inputs,
+ )
+ perturbation_domain_inputs = [Input(shape, dtype=dtype) for shape in perturbation_domain_inputs_shape]
+ constant_oracle_bounds = [Input(shape, dtype=dtype) for shape in constant_oracle_bounds_shape]
+ if nobatch:
+ affine_bounds_to_propagate = [
+ Input(batch_shape=shape, dtype=dtype) for shape in affine_bounds_to_propagate_shape
+ ]
+ else:
+ affine_bounds_to_propagate = [Input(shape=shape, dtype=dtype) for shape in affine_bounds_to_propagate_shape]
+ inputs_outputs_spec = InputsOutputsSpec(
+ ibp=ibp,
+ affine=affine,
+ propagation=propagation,
+ layer_input_shape=layer_input_shape,
+ model_input_shape=model_input_shape,
+ model_output_shape=model_output_shape,
+ )
+ return inputs_outputs_spec.flatten_inputs(
+ affine_bounds_to_propagate=affine_bounds_to_propagate,
+ constant_oracle_bounds=constant_oracle_bounds,
+ perturbation_domain_inputs=perturbation_domain_inputs,
+ )
+
+ @staticmethod
+ def generate_simple_decomon_layer_inputs_from_keras_input(
+ keras_input,
+ layer_output_shape,
+ ibp,
+ affine,
+ propagation,
+ perturbation_domain,
+ empty=False,
+ diag=False,
+ nobatch=False,
+ for_linear_layer=False,
+ dtype=keras_config.floatx(),
+ equal_ibp=True,
+ remove_perturbation_domain_inputs=False,
+ add_perturbation_domain_inputs=False,
+ ):
+ """Generate simple decomon inputs for a layer from the corresponding keras input
+
+ Hypothesis: single-layer model => model input/output = layer input/output
+
+ For affine bounds, weights= identity + bias = 0
+ For constant bounds, (keras_input, keras_input)
+
+ To be used as `decomon_layer(*decomon_inputs)`.
Args:
- model:
- x:
+ keras_input:
+ layer_output_shape:
+ ibp:
+ affine:
+ propagation:
+ perturbation_domain:
+ empty:
+ diag:
+ nobatch:
Returns:
"""
- output_tensors = model(x)
- if isinstance(output_tensors, list):
- return [K.convert_to_numpy(output) for output in output_tensors]
+ layer_input_shape = tuple(keras_input.shape[1:])
+ model_input_shape = layer_input_shape
+ model_output_shape = layer_output_shape
+ inputs_outputs_spec = InputsOutputsSpec(
+ ibp=ibp,
+ affine=affine,
+ propagation=propagation,
+ layer_input_shape=layer_input_shape,
+ model_input_shape=model_input_shape,
+ model_output_shape=model_output_shape,
+ linear=for_linear_layer,
+ )
+
+ if (
+ inputs_outputs_spec.needs_perturbation_domain_inputs() and not remove_perturbation_domain_inputs
+ ) or add_perturbation_domain_inputs:
+ x = Helpers.generate_simple_perturbation_domain_inputs_from_keras_input(
+ keras_input=keras_input, perturbation_domain=perturbation_domain
+ )
+ perturbation_domain_inputs = [x]
else:
- return K.convert_to_numpy(output_tensors)
+ perturbation_domain_inputs = []
+
+ if inputs_outputs_spec.needs_affine_bounds_inputs() and not empty:
+ batchsize = keras_input.shape[0]
+ if propagation == Propagation.FORWARD:
+ bias_shape = layer_input_shape
+ else:
+ bias_shape = layer_output_shape
+ flatten_bias_dim = int(np.prod(bias_shape))
+ if diag:
+ w_in = K.ones(bias_shape, dtype=dtype)
+ else:
+ w_in = K.reshape(K.eye(flatten_bias_dim, dtype=dtype), bias_shape + bias_shape)
+ b_in = K.zeros(bias_shape, dtype=dtype)
+ if not nobatch:
+ w_in = K.repeat(
+ w_in[None],
+ batchsize,
+ axis=0,
+ )
+ b_in = K.repeat(
+ b_in[None],
+ batchsize,
+ axis=0,
+ )
+ affine_bounds_to_propagate = [w_in, b_in, w_in, b_in]
+ else:
+ affine_bounds_to_propagate = []
+
+ if inputs_outputs_spec.needs_constant_bounds_inputs():
+ if equal_ibp:
+ constant_oracle_bounds = [keras_input, keras_input]
+ else:
+ batchsize = keras_input.shape[0]
+ lower = K.repeat(K.min(keras_input, axis=0, keepdims=True), batchsize, axis=0)
+ upper = K.repeat(K.max(keras_input, axis=0, keepdims=True), batchsize, axis=0)
+ constant_oracle_bounds = [lower, upper]
+ else:
+ constant_oracle_bounds = []
+
+ return inputs_outputs_spec.flatten_inputs(
+ affine_bounds_to_propagate=affine_bounds_to_propagate,
+ constant_oracle_bounds=constant_oracle_bounds,
+ perturbation_domain_inputs=perturbation_domain_inputs,
+ )
+
+ @staticmethod
+ def generate_simple_perturbation_domain_inputs_from_keras_input(
+ keras_input, perturbation_domain, equal_bounds=True
+ ):
+ if isinstance(perturbation_domain, BoxDomain):
+ if equal_bounds:
+ return K.concatenate([keras_input[:, None], keras_input[:, None]], axis=1)
+ else:
+ batchsize = keras_input.shape[0]
+ lower = K.repeat(K.min(keras_input, axis=0, keepdims=True), batchsize, axis=0)
+ upper = K.repeat(K.max(keras_input, axis=0, keepdims=True), batchsize, axis=0)
+ return K.concatenate([lower[:, None], upper[:, None]], axis=1)
+ else:
+ raise NotImplementedError
+
+ @staticmethod
+ def generate_merging_decomon_input_from_single_decomon_inputs(
+ decomon_inputs: list[list[Tensor]], ibp: bool, affine: bool, propagation: Propagation, linear: bool
+ ) -> list[Tensor]:
+ inputs_outputs_spec_single = InputsOutputsSpec(
+ ibp=ibp,
+ affine=affine,
+ propagation=propagation,
+ layer_input_shape=tuple(),
+ model_input_shape=tuple(),
+ model_output_shape=tuple(),
+ linear=linear,
+ )
+ affine_bounds_to_propagate, constant_oracle_bounds, perturbation_domain_inputs = [], [], []
+ for decomon_input in decomon_inputs:
+ (
+ affine_bounds_to_propagate_i,
+ constant_oracle_bounds_i,
+ perturbation_domain_inputs_i,
+ ) = inputs_outputs_spec_single.split_inputs(decomon_input)
+ perturbation_domain_inputs = perturbation_domain_inputs_i
+ if propagation == Propagation.FORWARD:
+ affine_bounds_to_propagate.append(affine_bounds_to_propagate_i)
+ else:
+ affine_bounds_to_propagate = affine_bounds_to_propagate_i
+ constant_oracle_bounds.append(constant_oracle_bounds_i)
+
+ inputs_outputs_spec_merging = InputsOutputsSpec(
+ ibp=ibp,
+ affine=affine,
+ propagation=propagation,
+ layer_input_shape=[tuple()],
+ model_input_shape=tuple(),
+ model_output_shape=tuple(),
+ is_merging_layer=True,
+ linear=linear,
+ )
+ return inputs_outputs_spec_merging.flatten_inputs(
+ affine_bounds_to_propagate=affine_bounds_to_propagate,
+ constant_oracle_bounds=constant_oracle_bounds,
+ perturbation_domain_inputs=perturbation_domain_inputs,
+ )
@staticmethod
- def get_standard_values_1d_box(n, dc_decomp=True, grad_bounds=False, nb=100):
+ def get_standard_values_0d_box(n, batchsize=10):
"""A set of functions with their monotonic decomposition for testing the activations"""
- w_u_ = np.ones(nb, dtype=keras_config.floatx())
- b_u_ = np.zeros(nb, dtype=keras_config.floatx())
- w_l_ = np.ones(nb, dtype=keras_config.floatx())
- b_l_ = np.zeros(nb, dtype=keras_config.floatx())
+ w_u_ = np.ones(batchsize)
+ b_u_ = np.zeros(batchsize)
+ w_l_ = np.ones(batchsize)
+ b_l_ = np.zeros(batchsize)
if n == 0:
# identity
- y_ = np.linspace(-2, -1, nb)
- x_ = np.linspace(-2, -1, nb)
- h_ = np.linspace(-2, -1, nb)
- g_ = np.zeros_like(x_)
+ y_ = np.linspace(-2, -1, batchsize)
+ x_ = np.linspace(-2, -1, batchsize)
elif n == 1:
- y_ = np.linspace(1, 2, nb)
- x_ = np.linspace(1, 2, nb)
- h_ = np.linspace(1, 2, nb)
- g_ = np.zeros_like(x_)
+ y_ = np.linspace(1, 2, batchsize)
+ x_ = np.linspace(1, 2, batchsize)
elif n == 2:
- y_ = np.linspace(-1, 1, nb)
- x_ = np.linspace(-1, 1, nb)
- h_ = np.linspace(-1, 1, nb)
- g_ = np.zeros_like(x_)
+ y_ = np.linspace(-1, 1, batchsize)
+ x_ = np.linspace(-1, 1, batchsize)
elif n == 3:
# identity
- y_ = np.linspace(-2, -1, nb)
- x_ = np.linspace(-2, -1, nb)
- h_ = 2 * np.linspace(-2, -1, nb)
- g_ = -np.linspace(-2, -1, nb)
+ y_ = np.linspace(-2, -1, batchsize)
+ x_ = np.linspace(-2, -1, batchsize)
elif n == 4:
- y_ = np.linspace(1, 2, nb)
- x_ = np.linspace(1, 2, nb)
- h_ = 2 * np.linspace(1, 2, nb)
- g_ = -np.linspace(1, 2, nb)
+ y_ = np.linspace(1, 2, batchsize)
+ x_ = np.linspace(1, 2, batchsize)
elif n == 5:
- y_ = np.linspace(-1, 1, nb)
- x_ = np.linspace(-1, 1, nb)
- h_ = 2 * np.linspace(-1, 1, nb)
- g_ = -np.linspace(-1, 1, nb)
+ y_ = np.linspace(-1, 1, batchsize)
+ x_ = np.linspace(-1, 1, batchsize)
elif n == 6:
- assert nb == 100, "expected nb=100 samples"
# cosine function
- x_ = np.linspace(-np.pi, np.pi, 100)
+ x_ = np.linspace(-np.pi, np.pi, batchsize)
y_ = np.cos(x_)
- h_ = np.concatenate([y_[:50], np.ones((50,))]) - 0.5
- g_ = np.concatenate([np.ones((50,)), y_[50:]]) - 0.5
w_u_ = np.zeros_like(x_)
w_l_ = np.zeros_like(x_)
b_u_ = np.ones_like(x_)
@@ -283,24 +484,24 @@ def get_standard_values_1d_box(n, dc_decomp=True, grad_bounds=False, nb=100):
elif n == 7:
# h and g >0
- h_ = np.linspace(0.5, 2, nb)
- g_ = np.linspace(1, 2, nb)[::-1]
+ h_ = np.linspace(0.5, 2, batchsize)
+ g_ = np.linspace(1, 2, batchsize)[::-1]
x_ = h_ + g_
y_ = h_ + g_
elif n == 8:
# h <0 and g <0
# h_max+g_max <=0
- h_ = np.linspace(-2, -1, nb)
- g_ = np.linspace(-2, -1, nb)[::-1]
+ h_ = np.linspace(-2, -1, batchsize)
+ g_ = np.linspace(-2, -1, batchsize)[::-1]
y_ = h_ + g_
x_ = h_ + g_
elif n == 9:
# h >0 and g <0
# h_min+g_min >=0
- h_ = np.linspace(4, 5, nb)
- g_ = np.linspace(-2, -1, nb)[::-1]
+ h_ = np.linspace(4, 5, batchsize)
+ g_ = np.linspace(-2, -1, batchsize)[::-1]
y_ = h_ + g_
x_ = h_ + g_
@@ -312,207 +513,26 @@ def get_standard_values_1d_box(n, dc_decomp=True, grad_bounds=False, nb=100):
x_0_ = np.concatenate([x_min_[:, None], x_max_[:, None]], 1)
- u_c_ = np.max(y_) * np.ones((nb,))
- l_c_ = np.min(y_) * np.ones((nb,))
-
- if dc_decomp:
- output = [
- x_[:, None],
- y_[:, None],
- x_0_[:, :, None],
- u_c_[:, None],
- w_u_[:, None, None],
- b_u_[:, None],
- l_c_[:, None],
- w_l_[:, None, None],
- b_l_[:, None],
- h_[:, None],
- g_[:, None],
- ]
- else:
- output = [
- x_[:, None],
- y_[:, None],
- x_0_[:, :, None],
- u_c_[:, None],
- w_u_[:, None, None],
- b_u_[:, None],
- l_c_[:, None],
- w_l_[:, None, None],
- b_l_[:, None],
- ]
+ u_c_ = np.max(y_) * np.ones((batchsize,))
+ l_c_ = np.min(y_) * np.ones((batchsize,))
+
+ output = [
+ x_[:, None],
+ y_[:, None],
+ x_0_[:, :, None],
+ u_c_[:, None],
+ w_u_[:, None, None],
+ b_u_[:, None],
+ l_c_[:, None],
+ w_l_[:, None, None],
+ b_l_[:, None],
+ ]
# cast element
return [e.astype(keras_config.floatx()) for e in output]
@staticmethod
- def get_inputs_for_mode_from_full_inputs(
- inputs: Union[List[Tensor], List[npt.NDArray[np.float_]]],
- mode: Union[str, ForwardMode] = ForwardMode.HYBRID,
- dc_decomp: bool = True,
- ) -> Union[List[Tensor], List[npt.NDArray[np.float_]]]:
- """Extract from full inputs the ones corresponding to the selected mode.
-
- Args:
- inputs: inputs from `get_standard_values_xxx()` or `get_tensor_decomposition_xxx()`
- mode:
- dc_decomp:
-
- Returns:
-
- """
- mode = ForwardMode(mode)
- if dc_decomp:
- x, y, z, u_c, W_u, b_u, l_c, W_l, b_l, h, g = inputs
- if mode == ForwardMode.HYBRID:
- return inputs[2:]
- elif mode == ForwardMode.AFFINE:
- return [z, W_u, b_u, W_l, b_l, h, g]
- elif mode == ForwardMode.IBP:
- return [u_c, l_c, h, g]
- else:
- raise ValueError("Unknown mode.")
- else:
- x, y, z, u_c, W_u, b_u, l_c, W_l, b_l = inputs
- if mode == ForwardMode.HYBRID:
- return inputs[2:]
- elif mode == ForwardMode.AFFINE:
- return [z, W_u, b_u, W_l, b_l]
- elif mode == ForwardMode.IBP:
- return [u_c, l_c]
- else:
- raise ValueError("Unknown mode.")
-
- @staticmethod
- def get_inputs_np_for_decomon_model_from_full_inputs(
- inputs: List[npt.NDArray[np.float_]],
- ) -> npt.NDArray[np.float_]:
- """Extract from full numpy inputs the ones for a decomon model prediction.
-
- Args:
- inputs: inputs from `get_standard_values_xxx()`
-
- Returns:
-
- """
- l_c_, u_c_ = Helpers.get_input_ref_bounds_from_full_inputs(inputs=inputs)
- return np.concatenate((l_c_[:, None], u_c_[:, None]), axis=1)
-
- @staticmethod
- def get_input_ref_bounds_from_full_inputs(
- inputs: Union[List[Tensor], List[npt.NDArray[np.float_]]],
- ) -> Union[List[Tensor], List[npt.NDArray[np.float_]]]:
- """Extract lower and upper bound for input ref from full inputs
-
- Args:
- inputs: inputs from `get_standard_values_xxx()` or `get_tensor_decomposition_xxx()`
-
- Returns:
-
- """
- u_c_, l_c_ = inputs[3], inputs[6]
- return [l_c_, u_c_]
-
- @staticmethod
- def prepare_full_np_inputs_for_convert_model(
- inputs: List[npt.NDArray[np.float_]],
- dc_decomp: bool = True,
- ) -> List[npt.NDArray[np.float_]]:
- """Prepare full numpy inputs for convert_forward or convert_backward.
-
- W_u and W_l will be idendity matrices, and b_u, b_l zeros vectors.
-
- Args:
- inputs: inputs from `get_standard_values_xxx()` or `get_tensor_decomposition_xxx()`
-
- Returns:
-
- """
- if dc_decomp:
- x, y, z, u_c, W_u, b_u, l_c, W_l, b_l, h, g = inputs
- else:
- x, y, z, u_c, W_u, b_u, l_c, W_l, b_l = inputs
-
- b_u = np.zeros_like(b_u)
- b_l = np.zeros_like(b_l)
- W_u = np.repeat(np.identity(n=W_u.shape[-1])[None, :, :], repeats=W_u.shape[0], axis=0)
- W_l = np.repeat(np.identity(n=W_l.shape[-1])[None, :, :], repeats=W_l.shape[0], axis=0)
-
- if dc_decomp:
- return [x, y, z, u_c, W_u, b_u, l_c, W_l, b_l, h, g]
- else:
- return [x, y, z, u_c, W_u, b_u, l_c, W_l, b_l]
-
- @staticmethod
- def get_input_tensors_for_decomon_convert_from_full_inputs(
- inputs: List[Tensor],
- mode: Union[str, ForwardMode] = ForwardMode.HYBRID,
- dc_decomp: bool = True,
- ) -> List[Tensor]:
- """Extract from full tensor inputs the ones for a conversion to decomon model.
-
- Args:
- inputs: inputs from `get_tensor_decomposition_xxx()`
- mode:
- dc_decomp:
-
- Returns:
-
- """
- mode = ForwardMode(mode)
- if dc_decomp:
- x, y, z, u_c, W_u, b_u, l_c, W_l, b_l, h, g = inputs
- input_box_tensor = K.concatenate([K.expand_dims(l_c, 1), K.expand_dims(u_c, 1)], 1)
- if mode == ForwardMode.HYBRID:
- return [input_box_tensor, u_c, W_u, b_u, l_c, W_l, b_l, h, g]
- elif mode == ForwardMode.AFFINE:
- return [input_box_tensor, W_u, b_u, W_l, b_l, h, g]
- elif mode == ForwardMode.IBP:
- return [u_c, l_c, h, g]
- else:
- raise ValueError("Unknown mode.")
- else:
- x, y, z, u_c, W_u, b_u, l_c, W_l, b_l = inputs
- input_box_tensor = K.concatenate([K.expand_dims(l_c, 1), K.expand_dims(u_c, 1)], 1)
- if mode == ForwardMode.HYBRID:
- return [input_box_tensor, u_c, W_u, b_u, l_c, W_l, b_l]
- elif mode == ForwardMode.AFFINE:
- return [input_box_tensor, W_u, b_u, W_l, b_l]
- elif mode == ForwardMode.IBP:
- return [u_c, l_c]
- else:
- raise ValueError("Unknown mode.")
-
- @staticmethod
- def get_input_ref_from_full_inputs(
- inputs: Union[List[Tensor], List[npt.NDArray[np.float_]]]
- ) -> Union[Tensor, npt.NDArray[np.float_]]:
- """Extract from full inputs the input of reference for the original Keras layer.
-
- Args:
- inputs: inputs from `get_standard_values_xxx()` or `get_tensor_decomposition_xxx()`
-
- Returns:
-
- """
- return inputs[1]
-
- @staticmethod
- def get_tensor_decomposition_1d_box(dc_decomp=True):
- if dc_decomp:
- return [
- Input((1,), dtype=keras_config.floatx()),
- Input((1,), dtype=keras_config.floatx()),
- Input((2, 1), dtype=keras_config.floatx()),
- Input((1,), dtype=keras_config.floatx()),
- Input((1, 1), dtype=keras_config.floatx()),
- Input((1,), dtype=keras_config.floatx()),
- Input((1,), dtype=keras_config.floatx()),
- Input((1, 1), dtype=keras_config.floatx()),
- Input((1,), dtype=keras_config.floatx()),
- Input((1,), dtype=keras_config.floatx()),
- Input((1,), dtype=keras_config.floatx()),
- ]
+ def get_tensor_decomposition_0d_box():
return [
Input((1,), dtype=keras_config.floatx()),
Input((1,), dtype=keras_config.floatx()),
@@ -526,154 +546,32 @@ def get_tensor_decomposition_1d_box(dc_decomp=True):
]
@staticmethod
- def get_full_outputs_from_outputs_for_mode(
- outputs_for_mode: Union[List[Tensor], List[npt.NDArray[np.float_]]],
- mode: Union[str, ForwardMode] = ForwardMode.HYBRID,
- dc_decomp: bool = True,
- full_inputs: Optional[Union[List[Tensor], List[npt.NDArray[np.float_]]]] = None,
- ) -> Union[List[Tensor], List[npt.NDArray[np.float_]]]:
- mode = ForwardMode(mode)
- if dc_decomp:
- if mode == ForwardMode.HYBRID:
- z_, u_c_, w_u_, b_u_, l_c_, w_l_, b_l_, h_, g_ = outputs_for_mode
- elif mode == ForwardMode.AFFINE:
- z_, w_u_, b_u_, w_l_, b_l_, h_, g_ = outputs_for_mode
- u_c_, l_c_ = None, None
- elif mode == ForwardMode.IBP:
- u_c_, l_c_, h_, g_ = outputs_for_mode
- z_, w_u_, b_u_, w_l_, b_l_ = None, None, None, None, None
- if full_inputs is not None:
- z_ = full_inputs[2]
- else:
- raise ValueError("Unknown mode.")
- else:
- if mode == ForwardMode.HYBRID:
- z_, u_c_, w_u_, b_u_, l_c_, w_l_, b_l_ = outputs_for_mode
- elif mode == ForwardMode.AFFINE:
- z_, w_u_, b_u_, w_l_, b_l_ = outputs_for_mode
- u_c_, l_c_ = None, None
- elif mode == ForwardMode.IBP:
- u_c_, l_c_ = outputs_for_mode
- z_, w_u_, b_u_, w_l_, b_l_ = None, None, None, None, None
- if full_inputs is not None:
- z_ = full_inputs[2]
- else:
- raise ValueError("Unknown mode.")
- h_, g_ = None, None
- return [z_, u_c_, w_u_, b_u_, l_c_, w_l_, b_l_, h_, g_]
-
- @staticmethod
- def get_input_dim_multid_box(odd):
- if odd:
- return 3
- else:
- return 2
-
- @staticmethod
- def get_input_dim_images_box(odd):
- if odd:
- return 7
- else:
- return 6
-
- @staticmethod
- def get_input_dim_from_full_inputs(inputs: Union[List[Tensor], List[npt.NDArray[np.float_]]]) -> int:
- """Get input_dim for to_decomon or to_backward from full inputs
-
- Args:
- inputs: inputs from `get_standard_values_xxx()` or `get_tensor_decomposition_xxx()`
-
- Returns:
-
- """
- return inputs[0].shape[-1]
-
- @staticmethod
- def get_tensor_decomposition_multid_box(odd=1, dc_decomp=True):
- n = Helpers.get_input_dim_multid_box(odd)
-
- if dc_decomp:
- # x, y, z, u, w_u, b_u, l, w_l, b_l, h, g
- return [
- Input((n,), dtype=keras_config.floatx()),
- Input((n,), dtype=keras_config.floatx()),
- Input((2, n), dtype=keras_config.floatx()),
- Input((n,), dtype=keras_config.floatx()),
- Input((n, n), dtype=keras_config.floatx()),
- Input((n,), dtype=keras_config.floatx()),
- Input((n,), dtype=keras_config.floatx()),
- Input((n, n), dtype=keras_config.floatx()),
- Input((n,), dtype=keras_config.floatx()),
- Input((n,), dtype=keras_config.floatx()),
- Input((n,), dtype=keras_config.floatx()),
- ]
- return [
- Input((n,), dtype=keras_config.floatx()),
- Input((n,), dtype=keras_config.floatx()),
- Input((2, n), dtype=keras_config.floatx()),
- Input((n,), dtype=keras_config.floatx()),
- Input((n, n), dtype=keras_config.floatx()),
- Input((n,), dtype=keras_config.floatx()),
- Input((n,), dtype=keras_config.floatx()),
- Input((n, n), dtype=keras_config.floatx()),
- Input((n,), dtype=keras_config.floatx()),
- ]
-
- @staticmethod
- def get_standard_values_multid_box(odd=1, dc_decomp=True):
- if dc_decomp:
- (
- x_0,
- y_0,
- z_0,
- u_c_0,
- w_u_0,
- b_u_0,
- l_c_0,
- w_l_0,
- b_l_0,
- h_0,
- g_0,
- ) = Helpers.get_standard_values_1d_box(0, dc_decomp)
- (
- x_1,
- y_1,
- z_1,
- u_c_1,
- w_u_1,
- b_u_1,
- l_c_1,
- w_l_1,
- b_l_1,
- h_1,
- g_1,
- ) = Helpers.get_standard_values_1d_box(1, dc_decomp)
- else:
- (
- x_0,
- y_0,
- z_0,
- u_c_0,
- w_u_0,
- b_u_0,
- l_c_0,
- w_l_0,
- b_l_0,
- ) = Helpers.get_standard_values_1d_box(0, dc_decomp)
- (
- x_1,
- y_1,
- z_1,
- u_c_1,
- w_u_1,
- b_u_1,
- l_c_1,
- w_l_1,
- b_l_1,
- ) = Helpers.get_standard_values_1d_box(1, dc_decomp)
+ def get_standard_values_1d_box(odd=1, batchsize=10):
+ (
+ x_0,
+ y_0,
+ z_0,
+ u_c_0,
+ w_u_0,
+ b_u_0,
+ l_c_0,
+ w_l_0,
+ b_l_0,
+ ) = Helpers.get_standard_values_0d_box(n=0, batchsize=batchsize)
+ (
+ x_1,
+ y_1,
+ z_1,
+ u_c_1,
+ w_u_1,
+ b_u_1,
+ l_c_1,
+ w_l_1,
+ b_l_1,
+ ) = Helpers.get_standard_values_0d_box(n=1, batchsize=batchsize)
if not odd:
- # output (x_0+x_1, x_0+2*x_0)
+ # output (x_0+x_1, x_0+2*x_1)
x_ = np.concatenate([x_0, x_1], -1)
z_min_ = np.concatenate([z_0[:, 0], z_1[:, 0]], -1)
z_max_ = np.concatenate([z_0[:, 1], z_1[:, 1]], -1)
@@ -684,10 +582,6 @@ def get_standard_values_multid_box(odd=1, dc_decomp=True):
b_l_ = np.concatenate([b_l_0 + b_l_1, b_l_0 + 2 * b_l_1], -1)
l_c_ = np.concatenate([l_c_0 + l_c_1, l_c_0 + 2 * l_c_1], -1)
- if dc_decomp:
- h_ = np.concatenate([h_0 + h_1, h_0 + 2 * h_1], -1)
- g_ = np.concatenate([g_0 + g_1, g_0 + 2 * g_1], -1)
-
w_u_ = np.zeros((len(x_), 2, 2))
w_u_[:, 0, 0] = w_u_0[:, 0, 0]
w_u_[:, 1, 0] = w_u_1[:, 0, 0]
@@ -711,9 +605,7 @@ def get_standard_values_multid_box(odd=1, dc_decomp=True):
l_c_2,
w_l_2,
b_l_2,
- h_2,
- g_2,
- ) = Helpers.get_standard_values_1d_box(2)
+ ) = Helpers.get_standard_values_0d_box(n=2, batchsize=batchsize)
# output (x_0+x_1, x_0+2*x_0, x_2)
x_ = np.concatenate([x_0, x_1, x_2], -1)
@@ -726,10 +618,6 @@ def get_standard_values_multid_box(odd=1, dc_decomp=True):
u_c_ = np.concatenate([u_c_0 + u_c_1, u_c_0 + 2 * u_c_1, u_c_2], -1)
l_c_ = np.concatenate([l_c_0 + l_c_1, l_c_0 + 2 * l_c_1, l_c_2], -1)
- if dc_decomp:
- h_ = np.concatenate([h_0 + h_1, h_0 + 2 * h_1, h_2], -1)
- g_ = np.concatenate([g_0 + g_1, g_0 + 2 * g_1, g_2], -1)
-
w_u_ = np.zeros((len(x_), 3, 3))
w_u_[:, 0, 0] = w_u_0[:, 0, 0]
w_u_[:, 1, 0] = w_u_1[:, 0, 0]
@@ -744,148 +632,36 @@ def get_standard_values_multid_box(odd=1, dc_decomp=True):
w_l_[:, 1, 1] = 2 * w_l_1[:, 0, 0]
w_l_[:, 2, 2] = w_l_2[:, 0, 0]
- if dc_decomp:
- return [x_, y_, z_, u_c_, w_u_, b_u_, l_c_, w_l_, b_l_, h_, g_]
return [x_, y_, z_, u_c_, w_u_, b_u_, l_c_, w_l_, b_l_]
@staticmethod
- def build_image_from_1D_box(odd=0, m=0, dc_decomp=True):
- n = Helpers.get_input_dim_images_box(odd)
-
- if dc_decomp:
- (
- x_,
- y_0,
- z_,
- u_c_0,
- w_u_0,
- b_u_0,
- l_c_0,
- w_l_0,
- b_l_0,
- h_0,
- g_0,
- ) = Helpers.get_standard_values_1d_box(m, dc_decomp=dc_decomp)
- else:
- (
- x_,
- y_0,
- z_,
- u_c_0,
- w_u_0,
- b_u_0,
- l_c_0,
- w_l_0,
- b_l_0,
- ) = Helpers.get_standard_values_1d_box(m, dc_decomp=dc_decomp)
-
- y_ = np.concatenate([(i + 1) * y_0 for i in range(n * n)], -1).reshape((-1, n, n))
- b_u_ = np.concatenate([(i + 1) * b_u_0 for i in range(n * n)], -1).reshape((-1, n, n))
- b_l_ = np.concatenate([(i + 1) * b_l_0 for i in range(n * n)], -1).reshape((-1, n, n))
-
- if dc_decomp:
- h_ = np.concatenate([(i + 1) * h_0 for i in range(n * n)], -1).reshape((-1, n, n))
- g_ = np.concatenate([(i + 1) * g_0 for i in range(n * n)], -1).reshape((-1, n, n))
+ def get_tensor_decomposition_1d_box(odd=1):
+ n = Helpers.get_input_dim_1d_box(odd)
- u_c_ = np.concatenate([(i + 1) * u_c_0 for i in range(n * n)], -1).reshape((-1, n, n))
- l_c_ = np.concatenate([(i + 1) * l_c_0 for i in range(n * n)], -1).reshape((-1, n, n))
-
- w_u_ = np.zeros((len(x_), 1, n * n))
- w_l_ = np.zeros((len(x_), 1, n * n))
-
- for i in range(n * n):
- w_u_[:, 0, i] = (i + 1) * w_u_0[:, 0, 0]
- w_l_[:, 0, i] = (i + 1) * w_l_0[:, 0, 0]
-
- w_u_ = w_u_.reshape((-1, 1, n, n))
- w_l_ = w_l_.reshape((-1, 1, n, n))
-
- if dc_decomp:
- return [x_, y_, z_, u_c_, w_u_, b_u_, l_c_, w_l_, b_l_, h_, g_]
- return [x_, y_, z_, u_c_, w_u_, b_u_, l_c_, w_l_, b_l_]
+ return [
+ Input((n,), dtype=keras_config.floatx()),
+ Input((n,), dtype=keras_config.floatx()),
+ Input((2, n), dtype=keras_config.floatx()),
+ Input((n,), dtype=keras_config.floatx()),
+ Input((n, n), dtype=keras_config.floatx()),
+ Input((n,), dtype=keras_config.floatx()),
+ Input((n,), dtype=keras_config.floatx()),
+ Input((n, n), dtype=keras_config.floatx()),
+ Input((n,), dtype=keras_config.floatx()),
+ ]
@staticmethod
- def build_image_from_2D_box(odd=0, m0=0, m1=1, dc_decomp=True):
- if dc_decomp:
- (
- x_0,
- y_0,
- z_0,
- u_c_0,
- w_u_0,
- b_u_0,
- l_c_0,
- w_l_0,
- b_l_0,
- h_0,
- g_0,
- ) = Helpers.build_image_from_1D_box(odd, m0, dc_decomp)
- (
- x_1,
- y_1,
- z_1,
- u_c_1,
- w_u_1,
- b_u_1,
- l_c_1,
- w_l_1,
- b_l_1,
- h_1,
- g_1,
- ) = Helpers.build_image_from_1D_box(odd, m1, dc_decomp)
+ def get_input_dim_1d_box(odd):
+ if odd:
+ return 3
else:
- (
- x_0,
- y_0,
- z_0,
- u_c_0,
- w_u_0,
- b_u_0,
- l_c_0,
- w_l_0,
- b_l_0,
- ) = Helpers.build_image_from_1D_box(odd, m0, dc_decomp)
- (
- x_1,
- y_1,
- z_1,
- u_c_1,
- w_u_1,
- b_u_1,
- l_c_1,
- w_l_1,
- b_l_1,
- ) = Helpers.build_image_from_1D_box(odd, m1, dc_decomp)
-
- x_ = np.concatenate([x_0, x_1], -1)
- z_min_ = np.concatenate([z_0[:, 0], z_1[:, 0]], -1)
- z_max_ = np.concatenate([z_0[:, 1], z_1[:, 1]], -1)
- z_ = np.concatenate([z_min_[:, None], z_max_[:, None]], 1)
- y_ = y_0 + y_1
- b_u_ = b_u_0 + b_u_1
- b_l_ = b_l_0 + b_l_1
-
- u_c_ = u_c_0 + u_c_1
- l_c_ = l_c_0 + l_c_1
-
- w_u_ = np.concatenate([w_u_0, w_u_1], 1)
- w_l_ = np.concatenate([w_l_0, w_l_1], 1)
-
- if dc_decomp:
- h_ = h_0 + h_1
- g_ = g_0 + g_1
-
- return [x_, y_, z_, u_c_, w_u_, b_u_, l_c_, w_l_, b_l_, h_, g_]
- return [x_, y_, z_, u_c_, w_u_, b_u_, l_c_, w_l_, b_l_]
+ return 2
@staticmethod
- def get_standard_values_images_box(data_format="channels_last", odd=0, m0=0, m1=1, dc_decomp=True):
+ def get_standard_values_images_box(data_format="channels_last", odd=0, m0=0, m1=1, batchsize=10):
if data_format == "channels_last":
- output = Helpers.build_image_from_2D_box(odd, m0, m1, dc_decomp)
- if dc_decomp:
- x_0, y_0, z_0, u_c_0, w_u_0, b_u_0, l_c_0, w_l_0, b_l_0, h_0, g_0 = output
- else:
- x_0, y_0, z_0, u_c_0, w_u_0, b_u_0, l_c_0, w_l_0, b_l_0 = output
+ output = Helpers.build_image_from_2D_box(odd=odd, m0=m0, m1=m1, batchsize=batchsize)
+ x_0, y_0, z_0, u_c_0, w_u_0, b_u_0, l_c_0, w_l_0, b_l_0 = output
x_ = x_0
z_ = z_0
@@ -905,22 +681,12 @@ def get_standard_values_images_box(data_format="channels_last", odd=0, m0=0, m1=
w_u_ = np.concatenate([w_u_0, w_u_0], -1)
w_l_ = np.concatenate([w_l_0, w_l_0], -1)
- if dc_decomp:
- h_0 = h_0[:, :, :, None]
- g_0 = g_0[:, :, :, None]
- h_ = np.concatenate([h_0, h_0], -1)
- g_ = np.concatenate([g_0, g_0], -1)
-
else:
output = Helpers.get_standard_values_images_box(
- data_format="channels_last", odd=odd, m0=m0, m1=m1, dc_decomp=dc_decomp
+ data_format="channels_last", odd=odd, m0=m0, m1=m1, batchsize=batchsize
)
- x_, y_, z_, u_c_, w_u_, b_u_, l_c_, w_l_, b_l_ = output[:9]
- if dc_decomp:
- h_, g_ = output[-2:]
- h_ = np.transpose(h_, (0, 3, 1, 2))
- g_ = np.transpose(g_, (0, 3, 1, 2))
+ x_, y_, z_, u_c_, w_u_, b_u_, l_c_, w_l_, b_l_ = output
y_ = np.transpose(y_, (0, 3, 1, 2))
u_c_ = np.transpose(u_c_, (0, 3, 1, 2))
l_c_ = np.transpose(l_c_, (0, 3, 1, 2))
@@ -929,13 +695,10 @@ def get_standard_values_images_box(data_format="channels_last", odd=0, m0=0, m1=
w_u_ = np.transpose(w_u_, (0, 1, 4, 2, 3))
w_l_ = np.transpose(w_l_, (0, 1, 4, 2, 3))
- if dc_decomp:
- return [x_, y_, z_, u_c_, w_u_, b_u_, l_c_, w_l_, b_l_, h_, g_]
- else:
- return [x_, y_, z_, u_c_, w_u_, b_u_, l_c_, w_l_, b_l_]
+ return [x_, y_, z_, u_c_, w_u_, b_u_, l_c_, w_l_, b_l_]
@staticmethod
- def get_tensor_decomposition_images_box(data_format, odd, dc_decomp=True):
+ def get_tensor_decomposition_images_box(data_format, odd):
n = Helpers.get_input_dim_images_box(odd)
if data_format == "channels_last":
@@ -953,8 +716,6 @@ def get_tensor_decomposition_images_box(data_format, odd, dc_decomp=True):
Input((n, n, 2), dtype=keras_config.floatx()),
]
- if dc_decomp:
- output += [Input((n, n, 2), dtype=keras_config.floatx()), Input((n, n, 2), dtype=keras_config.floatx())]
else:
output = [
Input((2,), dtype=keras_config.floatx()),
@@ -967,475 +728,454 @@ def get_tensor_decomposition_images_box(data_format, odd, dc_decomp=True):
Input((2, 2, n, n), dtype=keras_config.floatx()),
Input((2, n, n), dtype=keras_config.floatx()),
]
- if dc_decomp:
- output += [Input((2, n, n), dtype=keras_config.floatx()), Input((2, n, n), dtype=keras_config.floatx())]
return output
@staticmethod
- def assert_output_properties_box(x_, y_, h_, g_, x_min_, x_max_, u_c_, w_u_, b_u_, l_c_, w_l_, b_l_, decimal=4):
- if y_ is None:
- y_ = h_ + g_
- if h_ is not None:
- assert_almost_equal(h_ + g_, y_, decimal=decimal, err_msg="decomposition error")
-
- assert np.min(x_min_ <= x_max_), "x_min >x_max"
-
- assert_almost_equal(np.clip(x_min_ - x_, 0, np.inf), 0.0, decimal=decimal, err_msg="x_min >x_")
- assert_almost_equal(np.clip(x_ - x_max_, 0, np.inf), 0.0, decimal=decimal, err_msg="x_max < x_")
- if w_u_ is not None or w_l_ is not None:
- x_expand = x_ + np.zeros_like(x_)
- n_expand = len(w_u_.shape) - len(x_expand.shape)
- for i in range(n_expand):
- x_expand = np.expand_dims(x_expand, -1)
-
- if w_l_ is not None:
- lower_ = np.sum(w_l_ * x_expand, 1) + b_l_
- if w_u_ is not None:
- upper_ = np.sum(w_u_ * x_expand, 1) + b_u_
-
- # check that the functions h_ and g_ remains monotonic
- if h_ is not None:
- assert_almost_equal(
- np.clip(h_[:-1] - h_[1:], 0, np.inf),
- np.zeros_like(h_[1:]),
- decimal=decimal,
- err_msg="h is not increasing",
- )
- assert_almost_equal(
- np.clip(g_[1:] - g_[:-1], 0, np.inf),
- np.zeros_like(g_[1:]),
- decimal=decimal,
- err_msg="g is not increasing",
- )
+ def get_input_dim_images_box(odd):
+ if odd:
+ return 7
+ else:
+ return 6
- #
- if w_u_ is not None:
- if keras_config.floatx() == "float32":
- assert_almost_equal(
- np.clip(y_ - upper_, 0.0, 1e6),
- np.zeros_like(y_),
- decimal=decimal,
- err_msg="upper ![\"Decomon!\"](\"data/decomon.jpg\")
\n",
+ "