SUMIN/HuggingFace_transformer
1
0
Fork 0
Code Issues Pull Requests Actions 7 Packages Projects Releases Wiki Activity

Fast image processor for VitMatte added and bug in slow version fixed (#37616)

Browse Source 
* added fast image processor for VitMatte including updated and new tests, fixed a bug in the slow image processor that processed images incorrectly for input format ChannelDimension.FIRST in which case the trimaps were not added in the correct dimension, this bug was also reflected in the tests through incorretly shaped trimaps being passed

* final edits for fast vitmatte image processor and tests

* final edits for fast vitmatte image processor and tests

---------

Co-authored-by: Yoni Gozlan <74535834+yonigozlan@users.noreply.github.com>
This commit is contained in:
Henrik Matthiesen
2025-04-28 20:51:50 +02:00
committed by GitHub
parent 65e940208c
commit a847d4aa6b
6 changed files with 413 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

206
tests/models/vitmatte/test_image_processing_vitmatte.py
View File

@@ -13,13 +13,15 @@
# limitations under the License.
import time
import unittest
import warnings
import numpy as np
import requests
from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available
from transformers.testing_utils import is_flaky, require_torch, require_vision
from transformers.utils import is_torch_available, is_torchvision_available, is_vision_available
from ...test_image_processing_common import ImageProcessingTestMixin, prepare_image_inputs
@@ -33,6 +35,9 @@ if is_vision_available():
from transformers import VitMatteImageProcessor
if is_torchvision_available():
from transformers import VitMatteImageProcessorFast
class VitMatteImageProcessingTester:
def __init__(
@@ -92,6 +97,7 @@ class VitMatteImageProcessingTester:
@require_vision
class VitMatteImageProcessingTest(ImageProcessingTestMixin, unittest.TestCase):
image_processing_class = VitMatteImageProcessor if is_vision_available() else None
fast_image_processing_class = VitMatteImageProcessorFast if is_torchvision_available() else None
def setUp(self):
super().setUp()
@@ -102,18 +108,17 @@ class VitMatteImageProcessingTest(ImageProcessingTestMixin, unittest.TestCase):
return self.image_processor_tester.prepare_image_processor_dict()
def test_image_processor_properties(self):
image_processing = self.image_processing_class(**self.image_processor_dict)
self.assertTrue(hasattr(image_processing, "image_mean"))
self.assertTrue(hasattr(image_processing, "image_std"))
self.assertTrue(hasattr(image_processing, "do_normalize"))
self.assertTrue(hasattr(image_processing, "do_rescale"))
self.assertTrue(hasattr(image_processing, "rescale_factor"))
self.assertTrue(hasattr(image_processing, "do_pad"))
self.assertTrue(hasattr(image_processing, "size_divisibility"))
for image_processing_class in self.image_processor_list:
image_processing = image_processing_class(**self.image_processor_dict)
self.assertTrue(hasattr(image_processing, "image_mean"))
self.assertTrue(hasattr(image_processing, "image_std"))
self.assertTrue(hasattr(image_processing, "do_normalize"))
self.assertTrue(hasattr(image_processing, "do_rescale"))
self.assertTrue(hasattr(image_processing, "rescale_factor"))
self.assertTrue(hasattr(image_processing, "do_pad"))
self.assertTrue(hasattr(image_processing, "size_divisibility"))
def test_call_numpy(self):
# Initialize image_processing
image_processing = self.image_processing_class(**self.image_processor_dict)
# create random numpy tensors
image_inputs = self.image_processor_tester.prepare_image_inputs(equal_resolution=False, numpify=True)
for image in image_inputs:
@@ -122,15 +127,16 @@ class VitMatteImageProcessingTest(ImageProcessingTestMixin, unittest.TestCase):
# Test not batched input (image processor does not support batched inputs)
image = image_inputs[0]
trimap = np.random.randint(0, 3, size=image.shape[:2])
encoded_images = image_processing(images=image, trimaps=trimap, return_tensors="pt").pixel_values
for image_processing_class in self.image_processor_list:
image_processing = image_processing_class(**self.image_processor_dict)
encoded_images = image_processing(images=image, trimaps=trimap, return_tensors="pt").pixel_values
# Verify that width and height can be divided by size_divisibility
self.assertTrue(encoded_images.shape[-1] % self.image_processor_tester.size_divisibility == 0)
self.assertTrue(encoded_images.shape[-2] % self.image_processor_tester.size_divisibility == 0)
# Verify that width and height can be divided by size_divisibility and that correct dimensions got merged
self.assertTrue(encoded_images.shape[-1] % self.image_processor_tester.size_divisibility == 0)
self.assertTrue(encoded_images.shape[-2] % self.image_processor_tester.size_divisibility == 0)
self.assertTrue(encoded_images.shape[-3] == 4)
def test_call_pytorch(self):
# Initialize image_processing
image_processing = self.image_processing_class(**self.image_processor_dict)
# create random PyTorch tensors
image_inputs = self.image_processor_tester.prepare_image_inputs(equal_resolution=False, torchify=True)
@@ -139,16 +145,37 @@ class VitMatteImageProcessingTest(ImageProcessingTestMixin, unittest.TestCase):
# Test not batched input (image processor does not support batched inputs)
image = image_inputs[0]
trimap = np.random.randint(0, 3, size=image.shape[:2])
encoded_images = image_processing(images=image, trimaps=trimap, return_tensors="pt").pixel_values
trimap = np.random.randint(0, 3, size=image.shape[1:])
for image_processing_class in self.image_processor_list:
image_processing = image_processing_class(**self.image_processor_dict)
encoded_images = image_processing(images=image, trimaps=trimap, return_tensors="pt").pixel_values
# Verify that width and height can be divided by size_divisibility
self.assertTrue(encoded_images.shape[-1] % self.image_processor_tester.size_divisibility == 0)
self.assertTrue(encoded_images.shape[-2] % self.image_processor_tester.size_divisibility == 0)
# Verify that width and height can be divided by size_divisibility and that correct dimensions got merged
self.assertTrue(encoded_images.shape[-1] % self.image_processor_tester.size_divisibility == 0)
self.assertTrue(encoded_images.shape[-2] % self.image_processor_tester.size_divisibility == 0)
self.assertTrue(encoded_images.shape[-3] == 4)
# create batched tensors
image_inputs = self.image_processor_tester.prepare_image_inputs(equal_resolution=True, torchify=True)
image_input = torch.stack(image_inputs, dim=0)
self.assertIsInstance(image_input, torch.Tensor)
self.assertTrue(image_input.shape[1] == 3)
trimap_shape = [image_input.shape[0]] + [1] + list(image_input.shape)[2:]
trimap_input = torch.randint(0, 3, trimap_shape, dtype=torch.uint8)
self.assertIsInstance(trimap_input, torch.Tensor)
self.assertTrue(trimap_input.shape[1] == 1)
for image_processing_class in self.image_processor_list:
image_processing = image_processing_class(**self.image_processor_dict)
encoded_images = image_processing(images=image, trimaps=trimap, return_tensors="pt").pixel_values
# Verify that width and height can be divided by size_divisibility and that correct dimensions got merged
self.assertTrue(encoded_images.shape[-1] % self.image_processor_tester.size_divisibility == 0)
self.assertTrue(encoded_images.shape[-2] % self.image_processor_tester.size_divisibility == 0)
self.assertTrue(encoded_images.shape[-3] == 4)
def test_call_pil(self):
# Initialize image_processing
image_processing = self.image_processing_class(**self.image_processor_dict)
# create random PIL images
image_inputs = self.image_processor_tester.prepare_image_inputs(equal_resolution=False)
for image in image_inputs:
@@ -157,16 +184,17 @@ class VitMatteImageProcessingTest(ImageProcessingTestMixin, unittest.TestCase):
# Test not batched input (image processor does not support batched inputs)
image = image_inputs[0]
trimap = np.random.randint(0, 3, size=image.size[::-1])
encoded_images = image_processing(images=image, trimaps=trimap, return_tensors="pt").pixel_values
for image_processing_class in self.image_processor_list:
image_processing = image_processing_class(**self.image_processor_dict)
encoded_images = image_processing(images=image, trimaps=trimap, return_tensors="pt").pixel_values
# Verify that width and height can be divided by size_divisibility
self.assertTrue(encoded_images.shape[-1] % self.image_processor_tester.size_divisibility == 0)
self.assertTrue(encoded_images.shape[-2] % self.image_processor_tester.size_divisibility == 0)
# Verify that width and height can be divided by size_divisibility and that correct dimensions got merged
self.assertTrue(encoded_images.shape[-1] % self.image_processor_tester.size_divisibility == 0)
self.assertTrue(encoded_images.shape[-2] % self.image_processor_tester.size_divisibility == 0)
self.assertTrue(encoded_images.shape[-3] == 4)
def test_call_numpy_4_channels(self):
# Test that can process images which have an arbitrary number of channels
# Initialize image_processing
image_processor = self.image_processing_class(**self.image_processor_dict)
# create random numpy tensors
self.image_processor_tester.num_channels = 4
@@ -175,20 +203,23 @@ class VitMatteImageProcessingTest(ImageProcessingTestMixin, unittest.TestCase):
# Test not batched input (image processor does not support batched inputs)
image = image_inputs[0]
trimap = np.random.randint(0, 3, size=image.shape[:2])
encoded_images = image_processor(
images=image,
trimaps=trimap,
input_data_format="channels_first",
image_mean=0,
image_std=1,
return_tensors="pt",
).pixel_values
for image_processing_class in self.image_processor_list:
image_processor = image_processing_class(**self.image_processor_dict)
encoded_images = image_processor(
images=image,
trimaps=trimap,
input_data_format="channels_last",
image_mean=0,
image_std=1,
return_tensors="pt",
).pixel_values
# Verify that width and height can be divided by size_divisibility
self.assertTrue(encoded_images.shape[-1] % self.image_processor_tester.size_divisibility == 0)
self.assertTrue(encoded_images.shape[-2] % self.image_processor_tester.size_divisibility == 0)
# Verify that width and height can be divided by size_divisibility and that correct dimensions got merged
self.assertTrue(encoded_images.shape[-1] % self.image_processor_tester.size_divisibility == 0)
self.assertTrue(encoded_images.shape[-2] % self.image_processor_tester.size_divisibility == 0)
self.assertTrue(encoded_images.shape[-3] == 5)
def test_padding(self):
def test_padding_slow(self):
image_processing = self.image_processing_class(**self.image_processor_dict)
image = np.random.randn(3, 249, 491)
images = image_processing.pad_image(image)
@@ -198,6 +229,17 @@ class VitMatteImageProcessingTest(ImageProcessingTestMixin, unittest.TestCase):
images = image_processing.pad_image(image)
assert images.shape == (3, 256, 512)
def test_padding_fast(self):
# extra test because name is different for fast image processor
image_processing = self.fast_image_processing_class(**self.image_processor_dict)
image = torch.rand(3, 249, 491)
images = image_processing._pad_image(image)
assert images.shape == (3, 256, 512)
image = torch.rand(3, 249, 512)
images = image_processing._pad_image(image)
assert images.shape == (3, 256, 512)
def test_image_processor_preprocess_arguments(self):
# vitmatte require additional trimap input for image_processor
# that is why we override original common test
@@ -214,3 +256,81 @@ class VitMatteImageProcessingTest(ImageProcessingTestMixin, unittest.TestCase):
messages = " ".join([str(w.message) for w in raised_warnings])
self.assertGreaterEqual(len(raised_warnings), 1)
self.assertIn("extra_argument", messages)
@is_flaky()
def test_fast_is_faster_than_slow(self):
if not self.test_slow_image_processor or not self.test_fast_image_processor:
self.skipTest(reason="Skipping speed test")
if self.image_processing_class is None or self.fast_image_processing_class is None:
self.skipTest(reason="Skipping speed test as one of the image processors is not defined")
def measure_time(image_processor, images, trimaps):
# Warmup
for _ in range(5):
_ = image_processor(images, trimaps=trimaps, return_tensors="pt")
all_times = []
for _ in range(10):
start = time.time()
_ = image_processor(images, trimaps=trimaps, return_tensors="pt")
all_times.append(time.time() - start)
# Take the average of the fastest 3 runs
avg_time = sum(sorted(all_times[:3])) / 3.0
return avg_time
dummy_images = torch.randint(0, 255, (4, 3, 400, 800), dtype=torch.uint8)
dummy_trimaps = torch.randint(0, 3, (4, 400, 800), dtype=torch.uint8)
image_processor_slow = self.image_processing_class(**self.image_processor_dict)
image_processor_fast = self.fast_image_processing_class(**self.image_processor_dict)
fast_time = measure_time(image_processor_fast, dummy_images, dummy_trimaps)
slow_time = measure_time(image_processor_slow, dummy_images, dummy_trimaps)
self.assertLessEqual(fast_time, slow_time)
def test_slow_fast_equivalence(self):
if not self.test_slow_image_processor or not self.test_fast_image_processor:
self.skipTest(reason="Skipping slow/fast equivalence test")
if self.image_processing_class is None or self.fast_image_processing_class is None:
self.skipTest(reason="Skipping slow/fast equivalence test as one of the image processors is not defined")
dummy_image = Image.open(
requests.get("http://images.cocodataset.org/val2017/000000039769.jpg", stream=True).raw
)
dummy_trimap = np.random.randint(0, 3, size=dummy_image.size[::-1])
image_processor_slow = self.image_processing_class(**self.image_processor_dict)
image_processor_fast = self.fast_image_processing_class(**self.image_processor_dict)
encoding_slow = image_processor_slow(dummy_image, trimaps=dummy_trimap, return_tensors="pt")
encoding_fast = image_processor_fast(dummy_image, trimaps=dummy_trimap, return_tensors="pt")
self.assertTrue(torch.allclose(encoding_slow.pixel_values, encoding_fast.pixel_values, atol=1e-1))
self.assertLessEqual(
torch.mean(torch.abs(encoding_slow.pixel_values - encoding_fast.pixel_values)).item(), 1e-3
)
def test_slow_fast_equivalence_batched(self):
# this only checks on equal resolution, since the slow processor doesn't work otherwise
if not self.test_slow_image_processor or not self.test_fast_image_processor:
self.skipTest(reason="Skipping slow/fast equivalence test")
if self.image_processing_class is None or self.fast_image_processing_class is None:
self.skipTest(reason="Skipping slow/fast equivalence test as one of the image processors is not defined")
if hasattr(self.image_processor_tester, "do_center_crop") and self.image_processor_tester.do_center_crop:
self.skipTest(
reason="Skipping as do_center_crop is True and center_crop functions are not equivalent for fast and slow processors"
)
dummy_images = self.image_processor_tester.prepare_image_inputs(equal_resolution=True, torchify=True)
dummy_trimaps = [np.random.randint(0, 3, size=image.shape[1:]) for image in dummy_images]
image_processor_slow = self.image_processing_class(**self.image_processor_dict)
image_processor_fast = self.fast_image_processing_class(**self.image_processor_dict)
encoding_slow = image_processor_slow(dummy_images, trimaps=dummy_trimaps, return_tensors="pt")
encoding_fast = image_processor_fast(dummy_images, trimaps=dummy_trimaps, return_tensors="pt")
self.assertTrue(torch.allclose(encoding_slow.pixel_values, encoding_fast.pixel_values, atol=1e-1))
self.assertLessEqual(
torch.mean(torch.abs(encoding_slow.pixel_values - encoding_fast.pixel_values)).item(), 1e-3
)