diff --git a/apt/utils/models/__init__.py b/apt/utils/models/__init__.py index 65861f3..e458139 100644 --- a/apt/utils/models/__init__.py +++ b/apt/utils/models/__init__.py @@ -2,4 +2,4 @@ from apt.utils.models.model import Model, BlackboxClassifier, ModelOutputType, S BlackboxClassifierPredictions, BlackboxClassifierPredictFunction, get_nb_classes, is_one_hot, \ check_correct_model_output from apt.utils.models.sklearn_model import SklearnModel, SklearnClassifier, SklearnRegressor -from apt.utils.models.keras_model import KerasClassifier +from apt.utils.models.keras_model import KerasClassifier, KerasRegressor diff --git a/apt/utils/models/keras_model.py b/apt/utils/models/keras_model.py index 6bce043..2f01385 100644 --- a/apt/utils/models/keras_model.py +++ b/apt/utils/models/keras_model.py @@ -7,12 +7,14 @@ import tensorflow as tf from tensorflow import keras tf.compat.v1.disable_eager_execution() +from sklearn.metrics import mean_squared_error + from apt.utils.models import Model, ModelOutputType, ScoringMethod, check_correct_model_output from apt.utils.datasets import Dataset, OUTPUT_DATA_ARRAY_TYPE from art.utils import check_and_transform_label_format from art.estimators.classification.keras import KerasClassifier as ArtKerasClassifier -# from art.estimators.regression.keras import KerasRegressor as ArtKerasRegressor +from art.estimators.regression.keras import KerasRegressor as ArtKerasRegressor class KerasModel(Model): @@ -90,62 +92,60 @@ class KerasClassifier(KerasModel): raise NotImplementedError -# class KerasRegressor(KerasModel): -# """ -# Wrapper class for keras regression models. -# -# :param model: The original keras model object. -# :type model: `keras.models.Model` -# :param black_box_access: Boolean describing the type of deployment of the model (when in production). -# Set to True if the model is only available via query (API) access, i.e., -# only the outputs of the model are exposed, and False if the model internals -# are also available. Default is True. -# :type black_box_access: boolean, optional -# :param unlimited_queries: If black_box_access is True, this boolean indicates whether a user can perform -# unlimited queries to the model API or whether there is a limit to the number of -# queries that can be submitted. Default is True. -# :type unlimited_queries: boolean, optional -# """ -# def __init__(self, model: keras.models.Model, black_box_access: Optional[bool] = True, -# unlimited_queries: Optional[bool] = True, **kwargs): -# super().__init__(model, ModelOutputType.REGRESSOR_SCALAR, black_box_access, unlimited_queries, **kwargs) -# self._art_model = ArtKerasRegressor(model) -# -# def fit(self, train_data: Dataset, **kwargs) -> None: -# """ -# Fit the model using the training data. -# -# :param train_data: Training data. -# :type train_data: `Dataset` -# :return: None -# """ -# self._art_model.fit(train_data.get_samples(), train_data.get_labels(), **kwargs) -# -# def predict(self, x: Dataset, **kwargs) -> OUTPUT_DATA_ARRAY_TYPE: -# """ -# Perform predictions using the model for input `x`. -# -# :param x: Input samples. -# :type x: `Dataset` -# :return: Predictions from the model as numpy array. -# """ -# return self._art_model.predict(x.get_samples(), **kwargs) -# -# def score(self, test_data: Dataset, scoring_method: Optional[ScoringMethod] = ScoringMethod.MEAN_SQUARED_ERROR, -# **kwargs): -# """ -# Score the model using test data. -# -# :param test_data: Test data. -# :type train_data: `Dataset` -# :param scoring_method: The method for scoring predictions. Default is ACCURACY. -# :type scoring_method: `ScoringMethod`, optional -# :return: the score as float -# """ -# y = check_and_transform_label_format(test_data.get_labels(), self._art_model.nb_classes) -# predicted = self.predict(test_data) -# if scoring_method == ScoringMethod.MEAN_SQUARED_ERROR: -# mse = keras.losses.MeanSquaredError(reduction=keras.losses.Reduction.SUM) -# return mse(y, predicted).numpy() -# else: -# raise NotImplementedError('Only MEAN_SQUARED_ERROR supported as scoring method') +class KerasRegressor(KerasModel): + """ + Wrapper class for keras regression models. + + :param model: The original keras model object. + :type model: `keras.models.Model` + :param black_box_access: Boolean describing the type of deployment of the model (when in production). + Set to True if the model is only available via query (API) access, i.e., + only the outputs of the model are exposed, and False if the model internals + are also available. Default is True. + :type black_box_access: boolean, optional + :param unlimited_queries: If black_box_access is True, this boolean indicates whether a user can perform + unlimited queries to the model API or whether there is a limit to the number of + queries that can be submitted. Default is True. + :type unlimited_queries: boolean, optional + """ + def __init__(self, model: keras.models.Model, black_box_access: Optional[bool] = True, + unlimited_queries: Optional[bool] = True, **kwargs): + super().__init__(model, ModelOutputType.REGRESSOR_SCALAR, black_box_access, unlimited_queries, **kwargs) + self._art_model = ArtKerasRegressor(model) + + def fit(self, train_data: Dataset, **kwargs) -> None: + """ + Fit the model using the training data. + + :param train_data: Training data. + :type train_data: `Dataset` + :return: None + """ + self._art_model.fit(train_data.get_samples(), train_data.get_labels(), **kwargs) + + def predict(self, x: Dataset, **kwargs) -> OUTPUT_DATA_ARRAY_TYPE: + """ + Perform predictions using the model for input `x`. + + :param x: Input samples. + :type x: `Dataset` + :return: Predictions from the model as numpy array. + """ + return self._art_model.predict(x.get_samples(), **kwargs) + + def score(self, test_data: Dataset, scoring_method: Optional[ScoringMethod] = ScoringMethod.MEAN_SQUARED_ERROR, + **kwargs): + """ + Score the model using test data. + + :param test_data: Test data. + :type train_data: `Dataset` + :param scoring_method: The method for scoring predictions. Default is ACCURACY. + :type scoring_method: `ScoringMethod`, optional + :return: the score as float + """ + predicted = self.predict(test_data) + if scoring_method == ScoringMethod.MEAN_SQUARED_ERROR: + return mean_squared_error(test_data.get_labels(), predicted) + else: + raise NotImplementedError('Only MEAN_SQUARED_ERROR supported as scoring method') diff --git a/requirements.txt b/requirements.txt index 4ce8d46..849ca96 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,7 +3,7 @@ pandas==1.1.0 scipy==1.4.1 scikit-learn==0.22.2 torch>=1.8.0 -adversarial-robustness-toolbox>=1.10.1 +adversarial-robustness-toolbox>=1.11.0 # testing pytest==5.4.2 diff --git a/tests/test_model.py b/tests/test_model.py index 7ad260d..d5f4216 100644 --- a/tests/test_model.py +++ b/tests/test_model.py @@ -1,7 +1,7 @@ import pytest import numpy as np -from apt.utils.models import SklearnClassifier, SklearnRegressor, ModelOutputType, KerasClassifier, \ +from apt.utils.models import SklearnClassifier, SklearnRegressor, ModelOutputType, KerasClassifier, KerasRegressor, \ BlackboxClassifierPredictions, BlackboxClassifierPredictFunction, is_one_hot, get_nb_classes from apt.utils.datasets import ArrayDataset, Data from apt.utils import dataset_utils @@ -66,6 +66,28 @@ def test_keras_classifier(): assert(0.0 <= score <= 1.0) +def test_keras_regressor(): + (x_train, y_train), (x_test, y_test) = dataset_utils.get_diabetes_dataset_np() + + underlying_model = Sequential() + underlying_model.add(Input(shape=(10,))) + underlying_model.add(Dense(100, activation="relu")) + underlying_model.add(Dense(10, activation="relu")) + underlying_model.add(Dense(1)) + + underlying_model.compile(loss="mean_squared_error", optimizer="adam", metrics=["accuracy"]) + + model = KerasRegressor(underlying_model) + + train = ArrayDataset(x_train, y_train) + test = ArrayDataset(x_test, y_test) + model.fit(train) + pred = model.predict(test) + assert (pred.shape[0] == x_test.shape[0]) + + score = model.score(test) + + def test_blackbox_classifier(): (x_train, y_train), (x_test, y_test) = dataset_utils.get_iris_dataset_np()