Merge pull request #1444 from nilakshdas/fix-1442

Set _device parameter automatically in PreprocessorPyTorch (fixes #1442)

Author: beat-buesser

art/defences/preprocessor/mp3_compression_pytorch.py

@@ -62,22 +62,19 @@ def __init__(
         :param device_type: Type of device on which the classifier is run, either `gpu` or `cpu`.
         :param verbose: Show progress bars.
         """
-        import torch  # lgtm [py/repeated-import]
         from torch.autograd import Function
 
-        super().__init__(is_fitted=True, apply_fit=apply_fit, apply_predict=apply_predict)
+        super().__init__(
+            device_type=device_type,
+            is_fitted=True,
+            apply_fit=apply_fit,
+            apply_predict=apply_predict,
+        )
         self.channels_first = channels_first
         self.sample_rate = sample_rate
         self.verbose = verbose
         self._check_params()
 
-        # Set device
-        if device_type == "cpu" or not torch.cuda.is_available():
-            self._device = torch.device("cpu")
-        else:  # pragma: no cover
-            cuda_idx = torch.cuda.current_device()
-            self._device = torch.device("cuda:{}".format(cuda_idx))
-
         self.compression_numpy = Mp3Compression(
             sample_rate=sample_rate,
             channels_first=channels_first,

art/defences/preprocessor/preprocessor.py

@@ -140,6 +140,18 @@ class PreprocessorPyTorch(Preprocessor):
     Abstract base class for preprocessing defences implemented in PyTorch that support efficient preprocessor-chaining.
     """
 
+    def __init__(self, device_type: str = "gpu", **kwargs):
+        import torch  # lgtm [py/repeated-import]
+
+        super().__init__(**kwargs)
+
+        # Set device
+        if device_type == "cpu" or not torch.cuda.is_available():
+            self._device = torch.device("cpu")
+        else:  # pragma: no cover
+            cuda_idx = torch.cuda.current_device()
+            self._device = torch.device("cuda:{}".format(cuda_idx))
+
     @abc.abstractmethod
     def forward(
         self, x: "torch.Tensor", y: Optional["torch.Tensor"] = None

art/defences/preprocessor/spatial_smoothing_pytorch.py

@@ -72,22 +72,18 @@ def __init__(
         :param apply_predict: True if applied during predicting.
         :param device_type: Type of device on which the classifier is run, either `gpu` or `cpu`.
         """
-        import torch  # lgtm [py/repeated-import]
 
-        super().__init__(apply_fit=apply_fit, apply_predict=apply_predict)
+        super().__init__(
+            device_type=device_type,
+            apply_fit=apply_fit,
+            apply_predict=apply_predict,
+        )
 
         self.channels_first = channels_first
         self.window_size = window_size
         self.clip_values = clip_values
         self._check_params()
 
-        # Set device
-        if device_type == "cpu" or not torch.cuda.is_available():
-            self._device = torch.device("cpu")
-        else:  # pragma: no cover
-            cuda_idx = torch.cuda.current_device()
-            self._device = torch.device("cuda:{}".format(cuda_idx))
-
         from kornia.filters import MedianBlur
 
         class MedianBlurCustom(MedianBlur):

art/defences/preprocessor/video_compression_pytorch.py

@@ -68,23 +68,20 @@ def __init__(
         :param device_type: Type of device on which the classifier is run, either `gpu` or `cpu`.
         :param verbose: Show progress bars.
         """
-        import torch  # lgtm [py/repeated-import]
         from torch.autograd import Function
 
-        super().__init__(is_fitted=True, apply_fit=apply_fit, apply_predict=apply_predict)
+        super().__init__(
+            device_type=device_type,
+            is_fitted=True,
+            apply_fit=apply_fit,
+            apply_predict=apply_predict,
+        )
         self.video_format = video_format
         self.constant_rate_factor = constant_rate_factor
         self.channels_first = channels_first
         self.verbose = verbose
         self._check_params()
 
-        # Set device
-        if device_type == "cpu" or not torch.cuda.is_available():
-            self._device = torch.device("cpu")
-        else:  # pragma: no cover
-            cuda_idx = torch.cuda.current_device()
-            self._device = torch.device("cuda:{}".format(cuda_idx))
-
         self.compression_numpy = VideoCompression(
             video_format=video_format,
             constant_rate_factor=constant_rate_factor,

art/preprocessing/audio/l_filter/pytorch.py

@@ -71,23 +71,20 @@ def __init__(
         :param verbose: Show progress bars.
         :param device_type: Type of device on which the classifier is run, either `gpu` or `cpu`.
         """
-        import torch  # lgtm [py/repeated-import]
 
-        super().__init__(is_fitted=True, apply_fit=apply_fit, apply_predict=apply_predict)
+        super().__init__(
+            device_type=device_type,
+            is_fitted=True,
+            apply_fit=apply_fit,
+            apply_predict=apply_predict,
+        )
 
         self.numerator_coef = numerator_coef
         self.denominator_coef = denominator_coef
         self.clip_values = clip_values
         self.verbose = verbose
         self._check_params()
 
-        # Set device
-        if device_type == "cpu" or not torch.cuda.is_available():
-            self._device = torch.device("cpu")
-        else:  # pragma: no cover
-            cuda_idx = torch.cuda.current_device()
-            self._device = torch.device("cuda:{}".format(cuda_idx))
-
     def forward(
         self, x: "torch.Tensor", y: Optional["torch.Tensor"] = None
     ) -> Tuple["torch.Tensor", Optional["torch.Tensor"]]:

art/preprocessing/standardisation_mean_std/pytorch.py

@@ -54,9 +54,13 @@ def __init__(
         :param mean: Mean.
         :param std: Standard Deviation.
         """
-        import torch  # lgtm [py/repeated-import]
 
-        super().__init__(is_fitted=True, apply_fit=apply_fit, apply_predict=apply_predict)
+        super().__init__(
+            device_type=device_type,
+            is_fitted=True,
+            apply_fit=apply_fit,
+            apply_predict=apply_predict,
+        )
         self.mean = np.asarray(mean, dtype=ART_NUMPY_DTYPE)
         self.std = np.asarray(std, dtype=ART_NUMPY_DTYPE)
         self._check_params()
@@ -65,13 +69,6 @@ def __init__(
         self._broadcastable_mean = None
         self._broadcastable_std = None
 
-        # Set device
-        if device_type == "cpu" or not torch.cuda.is_available():
-            self._device = torch.device("cpu")
-        else:  # pragma: no cover
-            cuda_idx = torch.cuda.current_device()
-            self._device = torch.device("cuda:{}".format(cuda_idx))
-
     def forward(
         self, x: "torch.Tensor", y: Optional["torch.Tensor"] = None
     ) -> Tuple["torch.Tensor", Optional["torch.Tensor"]]:

tests/defences/preprocessor/test_preprocessor.py

@@ -0,0 +1,33 @@
+import pytest
+
+from art.defences.preprocessor.preprocessor import PreprocessorPyTorch
+from tests.utils import ARTTestException
+
+
+class DummyPreprocessorPyTorch(PreprocessorPyTorch):
+    def forward(self, x, y):
+        return x, y
+
+
+@pytest.mark.parametrize("is_fitted", [True, False])
+@pytest.mark.parametrize("apply_fit", [True, False])
+@pytest.mark.parametrize("apply_predict", [True, False])
+@pytest.mark.only_with_platform("pytorch")
+def test_preprocessor_pytorch_init(art_warning, is_fitted, apply_fit, apply_predict):
+    try:
+        import torch
+
+        preprocessor = DummyPreprocessorPyTorch(
+            device_type="cpu",
+            is_fitted=is_fitted,
+            apply_fit=apply_fit,
+            apply_predict=apply_predict,
+        )
+
+        assert preprocessor.device == torch.device("cpu")
+        assert preprocessor.is_fitted == is_fitted
+        assert preprocessor.apply_fit == apply_fit
+        assert preprocessor.apply_predict == apply_predict
+
+    except ARTTestException as e:
+        art_warning(e)