transformers

1
# coding=utf-8
2
# Copyright 2021 HuggingFace Inc.
3
#
4
# Licensed under the Apache License, Version 2.0 (the "License");
5
# you may not use this file except in compliance with the License.
6
# You may obtain a copy of the License at
7
#
8
#     http://www.apache.org/licenses/LICENSE-2.0
9
#
10
# Unless required by applicable law or agreed to in writing, software
11
# distributed under the License is distributed on an "AS IS" BASIS,
12
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13
# See the License for the specific language governing permissions and
14
# limitations under the License.
15

16

17
import json
18
import os
19
import tempfile
20
import unittest
21

22
import numpy as np
23
from datasets import load_dataset
24

25
from transformers.testing_utils import require_torch, require_vision, slow
26
from transformers.utils import is_torch_available, is_vision_available
27

28
from ...test_image_processing_common import ImageProcessingTestMixin, prepare_image_inputs
29

30

31
if is_torch_available():
32
    import torch
33

34
if is_vision_available():
35
    from PIL import Image
36

37
    from transformers import ImageGPTImageProcessor
38

39

40
class ImageGPTImageProcessingTester(unittest.TestCase):
41
    def __init__(
42
        self,
43
        parent,
44
        batch_size=7,
45
        num_channels=3,
46
        image_size=18,
47
        min_resolution=30,
48
        max_resolution=400,
49
        do_resize=True,
50
        size=None,
51
        do_normalize=True,
52
    ):
53
        size = size if size is not None else {"height": 18, "width": 18}
54
        self.parent = parent
55
        self.batch_size = batch_size
56
        self.num_channels = num_channels
57
        self.image_size = image_size
58
        self.min_resolution = min_resolution
59
        self.max_resolution = max_resolution
60
        self.do_resize = do_resize
61
        self.size = size
62
        self.do_normalize = do_normalize
63

64
    def prepare_image_processor_dict(self):
65
        return {
66
            # here we create 2 clusters for the sake of simplicity
67
            "clusters": np.asarray(
68
                [
69
                    [0.8866443634033203, 0.6618829369544983, 0.3891746401786804],
70
                    [-0.6042559146881104, -0.02295008860528469, 0.5423797369003296],
71
                ]
72
            ),
73
            "do_resize": self.do_resize,
74
            "size": self.size,
75
            "do_normalize": self.do_normalize,
76
        }
77

78
    def expected_output_image_shape(self, images):
79
        return (self.size["height"] * self.size["width"],)
80

81
    def prepare_image_inputs(self, equal_resolution=False, numpify=False, torchify=False):
82
        return prepare_image_inputs(
83
            batch_size=self.batch_size,
84
            num_channels=self.num_channels,
85
            min_resolution=self.min_resolution,
86
            max_resolution=self.max_resolution,
87
            equal_resolution=equal_resolution,
88
            numpify=numpify,
89
            torchify=torchify,
90
        )
91

92

93
@require_torch
94
@require_vision
95
class ImageGPTImageProcessingTest(ImageProcessingTestMixin, unittest.TestCase):
96
    image_processing_class = ImageGPTImageProcessor if is_vision_available() else None
97

98
    def setUp(self):
99
        self.image_processor_tester = ImageGPTImageProcessingTester(self)
100

101
    @property
102
    def image_processor_dict(self):
103
        return self.image_processor_tester.prepare_image_processor_dict()
104

105
    def test_image_processor_properties(self):
106
        image_processing = self.image_processing_class(**self.image_processor_dict)
107
        self.assertTrue(hasattr(image_processing, "clusters"))
108
        self.assertTrue(hasattr(image_processing, "do_resize"))
109
        self.assertTrue(hasattr(image_processing, "size"))
110
        self.assertTrue(hasattr(image_processing, "do_normalize"))
111

112
    def test_image_processor_from_dict_with_kwargs(self):
113
        image_processor = self.image_processing_class.from_dict(self.image_processor_dict)
114
        self.assertEqual(image_processor.size, {"height": 18, "width": 18})
115

116
        image_processor = self.image_processing_class.from_dict(self.image_processor_dict, size=42)
117
        self.assertEqual(image_processor.size, {"height": 42, "width": 42})
118

119
    def test_image_processor_to_json_string(self):
120
        image_processor = self.image_processing_class(**self.image_processor_dict)
121
        obj = json.loads(image_processor.to_json_string())
122
        for key, value in self.image_processor_dict.items():
123
            if key == "clusters":
124
                self.assertTrue(np.array_equal(value, obj[key]))
125
            else:
126
                self.assertEqual(obj[key], value)
127

128
    def test_image_processor_to_json_file(self):
129
        image_processor_first = self.image_processing_class(**self.image_processor_dict)
130

131
        with tempfile.TemporaryDirectory() as tmpdirname:
132
            json_file_path = os.path.join(tmpdirname, "image_processor.json")
133
            image_processor_first.to_json_file(json_file_path)
134
            image_processor_second = self.image_processing_class.from_json_file(json_file_path).to_dict()
135

136
        image_processor_first = image_processor_first.to_dict()
137
        for key, value in image_processor_first.items():
138
            if key == "clusters":
139
                self.assertTrue(np.array_equal(value, image_processor_second[key]))
140
            else:
141
                self.assertEqual(image_processor_first[key], value)
142

143
    def test_image_processor_from_and_save_pretrained(self):
144
        image_processor_first = self.image_processing_class(**self.image_processor_dict)
145

146
        with tempfile.TemporaryDirectory() as tmpdirname:
147
            image_processor_first.save_pretrained(tmpdirname)
148
            image_processor_second = self.image_processing_class.from_pretrained(tmpdirname).to_dict()
149

150
        image_processor_first = image_processor_first.to_dict()
151
        for key, value in image_processor_first.items():
152
            if key == "clusters":
153
                self.assertTrue(np.array_equal(value, image_processor_second[key]))
154
            else:
155
                self.assertEqual(image_processor_first[key], value)
156

157
    @unittest.skip("ImageGPT requires clusters at initialization")
158
    def test_init_without_params(self):
159
        pass
160

161
    # Override the test from ImageProcessingTestMixin as ImageGPT model takes input_ids as input
162
    def test_call_pil(self):
163
        # Initialize image_processing
164
        image_processing = self.image_processing_class(**self.image_processor_dict)
165
        # create random PIL images
166
        image_inputs = self.image_processor_tester.prepare_image_inputs(equal_resolution=False)
167
        for image in image_inputs:
168
            self.assertIsInstance(image, Image.Image)
169

170
        # Test not batched input
171
        encoded_images = image_processing(image_inputs[0], return_tensors="pt").input_ids
172
        expected_output_image_shape = self.image_processor_tester.expected_output_image_shape(encoded_images)
173
        self.assertEqual(tuple(encoded_images.shape), (1, *expected_output_image_shape))
174

175
        # Test batched
176
        encoded_images = image_processing(image_inputs, return_tensors="pt").input_ids
177
        self.assertEqual(
178
            tuple(encoded_images.shape), (self.image_processor_tester.batch_size, *expected_output_image_shape)
179
        )
180

181
    # Override the test from ImageProcessingTestMixin as ImageGPT model takes input_ids as input
182
    def test_call_numpy(self):
183
        # Initialize image_processing
184
        image_processing = self.image_processing_class(**self.image_processor_dict)
185
        # create random numpy tensors
186
        image_inputs = self.image_processor_tester.prepare_image_inputs(equal_resolution=False, numpify=True)
187
        for image in image_inputs:
188
            self.assertIsInstance(image, np.ndarray)
189

190
        # Test not batched input
191
        encoded_images = image_processing(image_inputs[0], return_tensors="pt").input_ids
192
        expected_output_image_shape = self.image_processor_tester.expected_output_image_shape(encoded_images)
193
        self.assertEqual(tuple(encoded_images.shape), (1, *expected_output_image_shape))
194

195
        # Test batched
196
        encoded_images = image_processing(image_inputs, return_tensors="pt").input_ids
197
        self.assertEqual(
198
            tuple(encoded_images.shape), (self.image_processor_tester.batch_size, *expected_output_image_shape)
199
        )
200

201
    @unittest.skip("ImageGPT assumes clusters for 3 channels")
202
    def test_call_numpy_4_channels(self):
203
        pass
204

205
    # Override the test from ImageProcessingTestMixin as ImageGPT model takes input_ids as input
206
    def test_call_pytorch(self):
207
        # Initialize image_processing
208
        image_processing = self.image_processing_class(**self.image_processor_dict)
209
        # create random PyTorch tensors
210
        image_inputs = self.image_processor_tester.prepare_image_inputs(equal_resolution=False, torchify=True)
211
        expected_output_image_shape = self.image_processor_tester.expected_output_image_shape(image_inputs)
212

213
        for image in image_inputs:
214
            self.assertIsInstance(image, torch.Tensor)
215

216
        # Test not batched input
217
        encoded_images = image_processing(image_inputs[0], return_tensors="pt").input_ids
218
        self.assertEqual(tuple(encoded_images.shape), (1, *expected_output_image_shape))
219

220
        # Test batched
221
        encoded_images = image_processing(image_inputs, return_tensors="pt").input_ids
222
        self.assertEqual(
223
            tuple(encoded_images.shape),
224
            (self.image_processor_tester.batch_size, *expected_output_image_shape),
225
        )
226

227

228
def prepare_images():
229
    # we use revision="refs/pr/1" until the PR is merged
230
    # https://hf.co/datasets/hf-internal-testing/fixtures_image_utils/discussions/1
231
    dataset = load_dataset("hf-internal-testing/fixtures_image_utils", split="test", revision="refs/pr/1")
232

233
    image1 = dataset[4]["image"]
234
    image2 = dataset[5]["image"]
235

236
    images = [image1, image2]
237

238
    return images
239

240

241
@require_vision
242
@require_torch
243
class ImageGPTImageProcessorIntegrationTest(unittest.TestCase):
244
    @slow
245
    def test_image(self):
246
        image_processing = ImageGPTImageProcessor.from_pretrained("openai/imagegpt-small")
247

248
        images = prepare_images()
249

250
        # test non-batched
251
        encoding = image_processing(images[0], return_tensors="pt")
252

253
        self.assertIsInstance(encoding.input_ids, torch.LongTensor)
254
        self.assertEqual(encoding.input_ids.shape, (1, 1024))
255

256
        expected_slice = [306, 191, 191]
257
        self.assertEqual(encoding.input_ids[0, :3].tolist(), expected_slice)
258

259
        # test batched
260
        encoding = image_processing(images, return_tensors="pt")
261

262
        self.assertIsInstance(encoding.input_ids, torch.LongTensor)
263
        self.assertEqual(encoding.input_ids.shape, (2, 1024))
264

265
        expected_slice = [303, 13, 13]
266
        self.assertEqual(encoding.input_ids[1, -3:].tolist(), expected_slice)
267