apunkt/ai-privacy-toolkit
1
0
Fork 0
mirror of https://github.com/IBM/ai-privacy-toolkit.git synced 2026-04-24 20:36:21 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Keras regression support

Browse source
This commit is contained in:
abigailt 2022-07-20 14:17:25 +03:00 • committed by abigailgold
parent a7d156660e
commit 77a6e08c8e
4 changed files with 85 additions and 63 deletions
Download patch file Download diff file Expand all files Collapse all files

2
apt/utils/models/__init__.py
View file

@ -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

120
apt/utils/models/keras_model.py
View file

@ -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')

2
requirements.txt
View file

@ -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

24
tests/test_model.py
View file

@ -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()