chendl/compositional_test

1# coding=utf-82# Copyright 2023 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 at7#8#     http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# 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 and14# limitations under the License.15 16 17import json18import os19import sys20import tempfile21import unittest22import unittest.mock as mock23from pathlib import Path24 25from huggingface_hub import HfFolder, delete_repo26from requests.exceptions import HTTPError27 28from transformers import AutoImageProcessor, ViTImageProcessor29from transformers.testing_utils import (30    TOKEN,31    USER,32    check_json_file_has_correct_format,33    get_tests_dir,34    is_staging_test,35    require_torch,36    require_vision,37)38from transformers.utils import is_torch_available, is_vision_available39 40 41sys.path.append(str(Path(__file__).parent.parent / "utils"))42 43from test_module.custom_image_processing import CustomImageProcessor  # noqa E40244 45 46if is_torch_available():47    import numpy as np48    import torch49 50if is_vision_available():51    from PIL import Image52 53 54SAMPLE_IMAGE_PROCESSING_CONFIG_DIR = get_tests_dir("fixtures")55 56 57def prepare_image_inputs(image_processor_tester, equal_resolution=False, numpify=False, torchify=False):58    """This function prepares a list of PIL images, or a list of numpy arrays if one specifies numpify=True,59    or a list of PyTorch tensors if one specifies torchify=True.60 61    One can specify whether the images are of the same resolution or not.62    """63 64    assert not (numpify and torchify), "You cannot specify both numpy and PyTorch tensors at the same time"65 66    image_inputs = []67    for i in range(image_processor_tester.batch_size):68        if equal_resolution:69            width = height = image_processor_tester.max_resolution70        else:71            # To avoid getting image width/height 072            min_resolution = image_processor_tester.min_resolution73            if getattr(image_processor_tester, "size_divisor", None):74                # If `size_divisor` is defined, the image needs to have width/size >= `size_divisor`75                min_resolution = max(image_processor_tester.size_divisor, min_resolution)76            width, height = np.random.choice(np.arange(min_resolution, image_processor_tester.max_resolution), 2)77        image_inputs.append(78            np.random.randint(255, size=(image_processor_tester.num_channels, width, height), dtype=np.uint8)79        )80 81    if not numpify and not torchify:82        # PIL expects the channel dimension as last dimension83        image_inputs = [Image.fromarray(np.moveaxis(image, 0, -1)) for image in image_inputs]84 85    if torchify:86        image_inputs = [torch.from_numpy(image) for image in image_inputs]87 88    return image_inputs89 90 91def prepare_video(image_processor_tester, width=10, height=10, numpify=False, torchify=False):92    """This function prepares a video as a list of PIL images/NumPy arrays/PyTorch tensors."""93 94    video = []95    for i in range(image_processor_tester.num_frames):96        video.append(np.random.randint(255, size=(image_processor_tester.num_channels, width, height), dtype=np.uint8))97 98    if not numpify and not torchify:99        # PIL expects the channel dimension as last dimension100        video = [Image.fromarray(np.moveaxis(frame, 0, -1)) for frame in video]101 102    if torchify:103        video = [torch.from_numpy(frame) for frame in video]104 105    return video106 107 108def prepare_video_inputs(image_processor_tester, equal_resolution=False, numpify=False, torchify=False):109    """This function prepares a batch of videos: a list of list of PIL images, or a list of list of numpy arrays if110    one specifies numpify=True, or a list of list of PyTorch tensors if one specifies torchify=True.111 112    One can specify whether the videos are of the same resolution or not.113    """114 115    assert not (numpify and torchify), "You cannot specify both numpy and PyTorch tensors at the same time"116 117    video_inputs = []118    for i in range(image_processor_tester.batch_size):119        if equal_resolution:120            width = height = image_processor_tester.max_resolution121        else:122            width, height = np.random.choice(123                np.arange(image_processor_tester.min_resolution, image_processor_tester.max_resolution), 2124            )125            video = prepare_video(126                image_processor_tester=image_processor_tester,127                width=width,128                height=height,129                numpify=numpify,130                torchify=torchify,131            )132        video_inputs.append(video)133 134    return video_inputs135 136 137class ImageProcessingSavingTestMixin:138    test_cast_dtype = None139 140    def test_image_processor_to_json_string(self):141        image_processor = self.image_processing_class(**self.image_processor_dict)142        obj = json.loads(image_processor.to_json_string())143        for key, value in self.image_processor_dict.items():144            self.assertEqual(obj[key], value)145 146    def test_image_processor_to_json_file(self):147        image_processor_first = self.image_processing_class(**self.image_processor_dict)148 149        with tempfile.TemporaryDirectory() as tmpdirname:150            json_file_path = os.path.join(tmpdirname, "image_processor.json")151            image_processor_first.to_json_file(json_file_path)152            image_processor_second = self.image_processing_class.from_json_file(json_file_path)153 154        self.assertEqual(image_processor_second.to_dict(), image_processor_first.to_dict())155 156    def test_image_processor_from_and_save_pretrained(self):157        image_processor_first = self.image_processing_class(**self.image_processor_dict)158 159        with tempfile.TemporaryDirectory() as tmpdirname:160            saved_file = image_processor_first.save_pretrained(tmpdirname)[0]161            check_json_file_has_correct_format(saved_file)162            image_processor_second = self.image_processing_class.from_pretrained(tmpdirname)163 164        self.assertEqual(image_processor_second.to_dict(), image_processor_first.to_dict())165 166    def test_init_without_params(self):167        image_processor = self.image_processing_class()168        self.assertIsNotNone(image_processor)169 170    @require_torch171    @require_vision172    def test_cast_dtype_device(self):173        if self.test_cast_dtype is not None:174            # Initialize image_processor175            image_processor = self.image_processing_class(**self.image_processor_dict)176 177            # create random PyTorch tensors178            image_inputs = prepare_image_inputs(self.image_processor_tester, equal_resolution=False, torchify=True)179 180            encoding = image_processor(image_inputs, return_tensors="pt")181            # for layoutLM compatiblity182            self.assertEqual(encoding.pixel_values.device, torch.device("cpu"))183            self.assertEqual(encoding.pixel_values.dtype, torch.float32)184 185            encoding = image_processor(image_inputs, return_tensors="pt").to(torch.float16)186            self.assertEqual(encoding.pixel_values.device, torch.device("cpu"))187            self.assertEqual(encoding.pixel_values.dtype, torch.float16)188 189            encoding = image_processor(image_inputs, return_tensors="pt").to("cpu", torch.bfloat16)190            self.assertEqual(encoding.pixel_values.device, torch.device("cpu"))191            self.assertEqual(encoding.pixel_values.dtype, torch.bfloat16)192 193            with self.assertRaises(TypeError):194                _ = image_processor(image_inputs, return_tensors="pt").to(torch.bfloat16, "cpu")195 196            # Try with text + image feature197            encoding = image_processor(image_inputs, return_tensors="pt")198            encoding.update({"input_ids": torch.LongTensor([[1, 2, 3], [4, 5, 6]])})199            encoding = encoding.to(torch.float16)200 201            self.assertEqual(encoding.pixel_values.device, torch.device("cpu"))202            self.assertEqual(encoding.pixel_values.dtype, torch.float16)203            self.assertEqual(encoding.input_ids.dtype, torch.long)204 205 206class ImageProcessorUtilTester(unittest.TestCase):207    def test_cached_files_are_used_when_internet_is_down(self):208        # A mock response for an HTTP head request to emulate server down209        response_mock = mock.Mock()210        response_mock.status_code = 500211        response_mock.headers = {}212        response_mock.raise_for_status.side_effect = HTTPError213        response_mock.json.return_value = {}214 215        # Download this model to make sure it's in the cache.216        _ = ViTImageProcessor.from_pretrained("hf-internal-testing/tiny-random-vit")217        # Under the mock environment we get a 500 error when trying to reach the model.218        with mock.patch("requests.request", return_value=response_mock) as mock_head:219            _ = ViTImageProcessor.from_pretrained("hf-internal-testing/tiny-random-vit")220            # This check we did call the fake head request221            mock_head.assert_called()222 223    def test_legacy_load_from_url(self):224        # This test is for deprecated behavior and can be removed in v5225        _ = ViTImageProcessor.from_pretrained(226            "https://huggingface.co/hf-internal-testing/tiny-random-vit/resolve/main/preprocessor_config.json"227        )228 229 230@is_staging_test231class ImageProcessorPushToHubTester(unittest.TestCase):232    @classmethod233    def setUpClass(cls):234        cls._token = TOKEN235        HfFolder.save_token(TOKEN)236 237    @classmethod238    def tearDownClass(cls):239        try:240            delete_repo(token=cls._token, repo_id="test-image-processor")241        except HTTPError:242            pass243 244        try:245            delete_repo(token=cls._token, repo_id="valid_org/test-image-processor-org")246        except HTTPError:247            pass248 249        try:250            delete_repo(token=cls._token, repo_id="test-dynamic-image-processor")251        except HTTPError:252            pass253 254    def test_push_to_hub(self):255        image_processor = ViTImageProcessor.from_pretrained(SAMPLE_IMAGE_PROCESSING_CONFIG_DIR)256        image_processor.push_to_hub("test-image-processor", use_auth_token=self._token)257 258        new_image_processor = ViTImageProcessor.from_pretrained(f"{USER}/test-image-processor")259        for k, v in image_processor.__dict__.items():260            self.assertEqual(v, getattr(new_image_processor, k))261 262        # Reset repo263        delete_repo(token=self._token, repo_id="test-image-processor")264 265        # Push to hub via save_pretrained266        with tempfile.TemporaryDirectory() as tmp_dir:267            image_processor.save_pretrained(268                tmp_dir, repo_id="test-image-processor", push_to_hub=True, use_auth_token=self._token269            )270 271        new_image_processor = ViTImageProcessor.from_pretrained(f"{USER}/test-image-processor")272        for k, v in image_processor.__dict__.items():273            self.assertEqual(v, getattr(new_image_processor, k))274 275    def test_push_to_hub_in_organization(self):276        image_processor = ViTImageProcessor.from_pretrained(SAMPLE_IMAGE_PROCESSING_CONFIG_DIR)277        image_processor.push_to_hub("valid_org/test-image-processor", use_auth_token=self._token)278 279        new_image_processor = ViTImageProcessor.from_pretrained("valid_org/test-image-processor")280        for k, v in image_processor.__dict__.items():281            self.assertEqual(v, getattr(new_image_processor, k))282 283        # Reset repo284        delete_repo(token=self._token, repo_id="valid_org/test-image-processor")285 286        # Push to hub via save_pretrained287        with tempfile.TemporaryDirectory() as tmp_dir:288            image_processor.save_pretrained(289                tmp_dir, repo_id="valid_org/test-image-processor-org", push_to_hub=True, use_auth_token=self._token290            )291 292        new_image_processor = ViTImageProcessor.from_pretrained("valid_org/test-image-processor-org")293        for k, v in image_processor.__dict__.items():294            self.assertEqual(v, getattr(new_image_processor, k))295 296    def test_push_to_hub_dynamic_image_processor(self):297        CustomImageProcessor.register_for_auto_class()298        image_processor = CustomImageProcessor.from_pretrained(SAMPLE_IMAGE_PROCESSING_CONFIG_DIR)299 300        image_processor.push_to_hub("test-dynamic-image-processor", use_auth_token=self._token)301 302        # This has added the proper auto_map field to the config303        self.assertDictEqual(304            image_processor.auto_map,305            {"ImageProcessor": "custom_image_processing.CustomImageProcessor"},306        )307 308        new_image_processor = AutoImageProcessor.from_pretrained(309            f"{USER}/test-dynamic-image-processor", trust_remote_code=True310        )311        # Can't make an isinstance check because the new_image_processor is from the CustomImageProcessor class of a dynamic module312        self.assertEqual(new_image_processor.__class__.__name__, "CustomImageProcessor")313 314    def test_image_processor_from_pretrained_subfolder(self):315        with self.assertRaises(OSError):316            # config is in subfolder, the following should not work without specifying the subfolder317            _ = AutoImageProcessor.from_pretrained("hf-internal-testing/stable-diffusion-all-variants")318 319        config = AutoImageProcessor.from_pretrained(320            "hf-internal-testing/stable-diffusion-all-variants", subfolder="feature_extractor"321        )322 323        self.assertIsNotNone(config)324