Add OneFormer Model (#20577)
* Add Oneformer Model * Add OneFormer Tests * Add UNIVERSAL_SEGMENTATION_MAPPING * Fix config * 🐛 Fix error encountered while writing tests * 🔨 Fix instance segmentation post processing * Format Files and Add Documentation * Add Documentation mdx file * Run make fixup * Run make fix-copies * Remove unnecessary code * Format modeling_oneformer.py * Add OneFormer to ImageSegmentationPipeline * Format files * Add Demo link to Readme * Fix fomatting errors * Fix test failures * Update Table in index.mdx * Fix version * Fix style * Remove OneFormer from TF * Fix Imports * Fix dummy objects * Fix tests * Add newline * Remove OneFormerFeatureExtractor * Remove CUDA Kernels * Use AutoBackbone for Swin * Fix description * Use Image Processor * Fix copies * Fix formatting * Fix import order * Fix flake8 errors * Fix doc errors * Add Hindi Readme entry * Update supported backbones * Update supported backbones * Undo Changes * Fix type of config * Fix isort * Fix auto.mdx * Fix swin config * Replace DinatBackbone with AutoBackbone * Use SwinBackbone * Use SwinBackbone * Fix conversion script * Fix arguments * Add argument description * Fix style * Add OneFormerProcessor * Fix OneFormerProcessor Tests * Fix mapping * Fix imports * Fix inits * Fix style * Fix comment * Fix docstring * Move OneFormer to MultiModal * Fix Copies * Remove size divisor * Fix check_repo.py * Fix copies * Add Processor for Testing Pipeline * Fix padding for tokens * Fix variables * Fix formatting with correct black version * Add Image Processor Test * Apply suggestions * Revert common modeling * Add check for task * Fix conversion script * Fix initialization order * Fix tests * Undo Pipeline Changes * Fix layers in MLP * Fix copies * Update image paths * Fix copies * Apply suggestions
This commit is contained in:
Jitesh Jain
0
tests/models/oneformer/__init__.py
Normal file
0
tests/models/oneformer/__init__.py
Normal file
456
tests/models/oneformer/test_image_processing_oneformer.py
Normal file
456
tests/models/oneformer/test_image_processing_oneformer.py
Normal file
@@ -0,0 +1,456 @@
|
||||
# coding=utf-8
|
||||
# Copyright 2022 HuggingFace Inc.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
|
||||
import json
|
||||
import unittest
|
||||
|
||||
import numpy as np
|
||||
|
||||
from huggingface_hub import hf_hub_download
|
||||
from transformers.testing_utils import require_torch, require_vision
|
||||
from transformers.utils import is_torch_available, is_vision_available
|
||||
|
||||
from ...test_feature_extraction_common import FeatureExtractionSavingTestMixin, prepare_image_inputs
|
||||
|
||||
|
||||
if is_torch_available():
|
||||
import torch
|
||||
|
||||
if is_vision_available():
|
||||
from transformers import OneFormerImageProcessor
|
||||
from transformers.models.oneformer.image_processing_oneformer import binary_mask_to_rle
|
||||
from transformers.models.oneformer.modeling_oneformer import OneFormerForUniversalSegmentationOutput
|
||||
|
||||
if is_vision_available():
|
||||
from PIL import Image
|
||||
|
||||
|
||||
def prepare_metadata(class_info_file, repo_path="shi-labs/oneformer_demo"):
|
||||
with open(hf_hub_download(repo_path, class_info_file, repo_type="dataset"), "r") as f:
|
||||
class_info = json.load(f)
|
||||
metadata = {}
|
||||
class_names = []
|
||||
thing_ids = []
|
||||
for key, info in class_info.items():
|
||||
metadata[key] = info["name"]
|
||||
class_names.append(info["name"])
|
||||
if info["isthing"]:
|
||||
thing_ids.append(int(key))
|
||||
metadata["thing_ids"] = thing_ids
|
||||
metadata["class_names"] = class_names
|
||||
return metadata
|
||||
|
||||
|
||||
class OneFormerImageProcessorTester(unittest.TestCase):
|
||||
def __init__(
|
||||
self,
|
||||
parent,
|
||||
batch_size=7,
|
||||
num_channels=3,
|
||||
min_resolution=30,
|
||||
max_resolution=400,
|
||||
size=None,
|
||||
do_resize=True,
|
||||
do_normalize=True,
|
||||
image_mean=[0.5, 0.5, 0.5],
|
||||
image_std=[0.5, 0.5, 0.5],
|
||||
num_labels=10,
|
||||
reduce_labels=False,
|
||||
ignore_index=255,
|
||||
repo_path="shi-labs/oneformer_demo",
|
||||
class_info_file="ade20k_panoptic.json",
|
||||
num_text=10,
|
||||
):
|
||||
self.parent = parent
|
||||
self.batch_size = batch_size
|
||||
self.num_channels = num_channels
|
||||
self.min_resolution = min_resolution
|
||||
self.max_resolution = max_resolution
|
||||
self.do_resize = do_resize
|
||||
self.size = {"shortest_edge": 32, "longest_edge": 1333} if size is None else size
|
||||
self.do_normalize = do_normalize
|
||||
self.image_mean = image_mean
|
||||
self.image_std = image_std
|
||||
self.class_info_file = class_info_file
|
||||
self.metadata = prepare_metadata(class_info_file, repo_path)
|
||||
self.num_text = num_text
|
||||
self.repo_path = repo_path
|
||||
|
||||
# for the post_process_functions
|
||||
self.batch_size = 2
|
||||
self.num_queries = 10
|
||||
self.num_classes = 10
|
||||
self.height = 3
|
||||
self.width = 4
|
||||
self.num_labels = num_labels
|
||||
self.reduce_labels = reduce_labels
|
||||
self.ignore_index = ignore_index
|
||||
|
||||
def prepare_feat_extract_dict(self):
|
||||
return {
|
||||
"do_resize": self.do_resize,
|
||||
"size": self.size,
|
||||
"do_normalize": self.do_normalize,
|
||||
"image_mean": self.image_mean,
|
||||
"image_std": self.image_std,
|
||||
"num_labels": self.num_labels,
|
||||
"reduce_labels": self.reduce_labels,
|
||||
"ignore_index": self.ignore_index,
|
||||
"class_info_file": self.class_info_file,
|
||||
"metadata": self.metadata,
|
||||
"num_text": self.num_text,
|
||||
}
|
||||
|
||||
def get_expected_values(self, image_inputs, batched=False):
|
||||
"""
|
||||
This function computes the expected height and width when providing images to OneFormerImageProcessor,
|
||||
assuming do_resize is set to True with a scalar size.
|
||||
"""
|
||||
if not batched:
|
||||
image = image_inputs[0]
|
||||
if isinstance(image, Image.Image):
|
||||
w, h = image.size
|
||||
else:
|
||||
h, w = image.shape[1], image.shape[2]
|
||||
if w < h:
|
||||
expected_height = int(self.size["shortest_edge"] * h / w)
|
||||
expected_width = self.size["shortest_edge"]
|
||||
elif w > h:
|
||||
expected_height = self.size["shortest_edge"]
|
||||
expected_width = int(self.size["shortest_edge"] * w / h)
|
||||
else:
|
||||
expected_height = self.size["shortest_edge"]
|
||||
expected_width = self.size["shortest_edge"]
|
||||
|
||||
else:
|
||||
expected_values = []
|
||||
for image in image_inputs:
|
||||
expected_height, expected_width = self.get_expected_values([image])
|
||||
expected_values.append((expected_height, expected_width))
|
||||
expected_height = max(expected_values, key=lambda item: item[0])[0]
|
||||
expected_width = max(expected_values, key=lambda item: item[1])[1]
|
||||
|
||||
return expected_height, expected_width
|
||||
|
||||
def get_fake_oneformer_outputs(self):
|
||||
return OneFormerForUniversalSegmentationOutput(
|
||||
# +1 for null class
|
||||
class_queries_logits=torch.randn((self.batch_size, self.num_queries, self.num_classes + 1)),
|
||||
masks_queries_logits=torch.randn((self.batch_size, self.num_queries, self.height, self.width)),
|
||||
)
|
||||
|
||||
|
||||
@require_torch
|
||||
@require_vision
|
||||
class OneFormerImageProcessingTest(FeatureExtractionSavingTestMixin, unittest.TestCase):
|
||||
image_processing_class = OneFormerImageProcessor if (is_vision_available() and is_torch_available()) else None
|
||||
# only for test_feat_extracttion_common.test_feat_extract_to_json_string
|
||||
feature_extraction_class = image_processing_class
|
||||
|
||||
def setUp(self):
|
||||
self.image_processing_tester = OneFormerImageProcessorTester(self)
|
||||
|
||||
@property
|
||||
def feat_extract_dict(self):
|
||||
return self.image_processing_tester.prepare_feat_extract_dict()
|
||||
|
||||
def test_feat_extract_properties(self):
|
||||
image_processor = self.image_processing_class(**self.feat_extract_dict)
|
||||
self.assertTrue(hasattr(image_processor, "image_mean"))
|
||||
self.assertTrue(hasattr(image_processor, "image_std"))
|
||||
self.assertTrue(hasattr(image_processor, "do_normalize"))
|
||||
self.assertTrue(hasattr(image_processor, "do_resize"))
|
||||
self.assertTrue(hasattr(image_processor, "size"))
|
||||
self.assertTrue(hasattr(image_processor, "ignore_index"))
|
||||
self.assertTrue(hasattr(image_processor, "class_info_file"))
|
||||
self.assertTrue(hasattr(image_processor, "num_text"))
|
||||
self.assertTrue(hasattr(image_processor, "repo_path"))
|
||||
self.assertTrue(hasattr(image_processor, "metadata"))
|
||||
self.assertTrue(hasattr(image_processor, "reduce_labels"))
|
||||
|
||||
def test_batch_feature(self):
|
||||
pass
|
||||
|
||||
def test_call_pil(self):
|
||||
# Initialize image_processor
|
||||
image_processor = self.image_processing_class(**self.feat_extract_dict)
|
||||
# create random PIL images
|
||||
image_inputs = prepare_image_inputs(self.image_processing_tester, equal_resolution=False)
|
||||
for image in image_inputs:
|
||||
self.assertIsInstance(image, Image.Image)
|
||||
|
||||
# Test not batched input
|
||||
encoded_images = image_processor(image_inputs[0], ["semantic"], return_tensors="pt").pixel_values
|
||||
|
||||
expected_height, expected_width = self.image_processing_tester.get_expected_values(image_inputs)
|
||||
|
||||
self.assertEqual(
|
||||
encoded_images.shape,
|
||||
(1, self.image_processing_tester.num_channels, expected_height, expected_width),
|
||||
)
|
||||
|
||||
# Test batched
|
||||
expected_height, expected_width = self.image_processing_tester.get_expected_values(image_inputs, batched=True)
|
||||
|
||||
encoded_images = image_processor(
|
||||
image_inputs, ["semantic"] * len(image_inputs), return_tensors="pt"
|
||||
).pixel_values
|
||||
self.assertEqual(
|
||||
encoded_images.shape,
|
||||
(
|
||||
self.image_processing_tester.batch_size,
|
||||
self.image_processing_tester.num_channels,
|
||||
expected_height,
|
||||
expected_width,
|
||||
),
|
||||
)
|
||||
|
||||
def test_call_numpy(self):
|
||||
# Initialize image_processor
|
||||
image_processor = self.image_processing_class(**self.feat_extract_dict)
|
||||
# create random numpy tensors
|
||||
image_inputs = prepare_image_inputs(self.image_processing_tester, equal_resolution=False, numpify=True)
|
||||
for image in image_inputs:
|
||||
self.assertIsInstance(image, np.ndarray)
|
||||
|
||||
# Test not batched input
|
||||
encoded_images = image_processor(image_inputs[0], ["semantic"], return_tensors="pt").pixel_values
|
||||
|
||||
expected_height, expected_width = self.image_processing_tester.get_expected_values(image_inputs)
|
||||
|
||||
self.assertEqual(
|
||||
encoded_images.shape,
|
||||
(1, self.image_processing_tester.num_channels, expected_height, expected_width),
|
||||
)
|
||||
|
||||
# Test batched
|
||||
expected_height, expected_width = self.image_processing_tester.get_expected_values(image_inputs, batched=True)
|
||||
|
||||
encoded_images = image_processor(
|
||||
image_inputs, ["semantic"] * len(image_inputs), return_tensors="pt"
|
||||
).pixel_values
|
||||
self.assertEqual(
|
||||
encoded_images.shape,
|
||||
(
|
||||
self.image_processing_tester.batch_size,
|
||||
self.image_processing_tester.num_channels,
|
||||
expected_height,
|
||||
expected_width,
|
||||
),
|
||||
)
|
||||
|
||||
def test_call_pytorch(self):
|
||||
# Initialize image_processor
|
||||
image_processor = self.image_processing_class(**self.feat_extract_dict)
|
||||
# create random PyTorch tensors
|
||||
image_inputs = prepare_image_inputs(self.image_processing_tester, equal_resolution=False, torchify=True)
|
||||
for image in image_inputs:
|
||||
self.assertIsInstance(image, torch.Tensor)
|
||||
|
||||
# Test not batched input
|
||||
encoded_images = image_processor(image_inputs[0], ["semantic"], return_tensors="pt").pixel_values
|
||||
|
||||
expected_height, expected_width = self.image_processing_tester.get_expected_values(image_inputs)
|
||||
|
||||
self.assertEqual(
|
||||
encoded_images.shape,
|
||||
(1, self.image_processing_tester.num_channels, expected_height, expected_width),
|
||||
)
|
||||
|
||||
# Test batched
|
||||
expected_height, expected_width = self.image_processing_tester.get_expected_values(image_inputs, batched=True)
|
||||
|
||||
encoded_images = image_processor(
|
||||
image_inputs, ["semantic"] * len(image_inputs), return_tensors="pt"
|
||||
).pixel_values
|
||||
self.assertEqual(
|
||||
encoded_images.shape,
|
||||
(
|
||||
self.image_processing_tester.batch_size,
|
||||
self.image_processing_tester.num_channels,
|
||||
expected_height,
|
||||
expected_width,
|
||||
),
|
||||
)
|
||||
|
||||
def test_equivalence_pad_and_create_pixel_mask(self):
|
||||
# Initialize image_processors
|
||||
image_processor_1 = self.image_processing_class(**self.feat_extract_dict)
|
||||
image_processor_2 = self.image_processing_class(
|
||||
do_resize=False,
|
||||
do_normalize=False,
|
||||
do_rescale=False,
|
||||
num_labels=self.image_processing_tester.num_classes,
|
||||
class_info_file="ade20k_panoptic.json",
|
||||
num_text=self.image_processing_tester.num_text,
|
||||
repo_path="shi-labs/oneformer_demo",
|
||||
)
|
||||
# create random PyTorch tensors
|
||||
image_inputs = prepare_image_inputs(self.image_processing_tester, equal_resolution=False, torchify=True)
|
||||
for image in image_inputs:
|
||||
self.assertIsInstance(image, torch.Tensor)
|
||||
|
||||
# Test whether the method "pad_and_return_pixel_mask" and calling the image processor return the same tensors
|
||||
encoded_images_with_method = image_processor_1.encode_inputs(
|
||||
image_inputs, ["semantic"] * len(image_inputs), return_tensors="pt"
|
||||
)
|
||||
encoded_images = image_processor_2(image_inputs, ["semantic"] * len(image_inputs), return_tensors="pt")
|
||||
|
||||
self.assertTrue(
|
||||
torch.allclose(encoded_images_with_method["pixel_values"], encoded_images["pixel_values"], atol=1e-4)
|
||||
)
|
||||
self.assertTrue(
|
||||
torch.allclose(encoded_images_with_method["pixel_mask"], encoded_images["pixel_mask"], atol=1e-4)
|
||||
)
|
||||
|
||||
def comm_get_image_processor_inputs(
|
||||
self, with_segmentation_maps=False, is_instance_map=False, segmentation_type="np"
|
||||
):
|
||||
image_processor = self.image_processing_class(**self.feat_extract_dict)
|
||||
# prepare image and target
|
||||
num_labels = self.image_processing_tester.num_labels
|
||||
annotations = None
|
||||
instance_id_to_semantic_id = None
|
||||
image_inputs = prepare_image_inputs(self.image_processing_tester, equal_resolution=False)
|
||||
if with_segmentation_maps:
|
||||
high = num_labels
|
||||
if is_instance_map:
|
||||
labels_expanded = list(range(num_labels)) * 2
|
||||
instance_id_to_semantic_id = {
|
||||
instance_id: label_id for instance_id, label_id in enumerate(labels_expanded)
|
||||
}
|
||||
annotations = [
|
||||
np.random.randint(0, high * 2, (img.size[1], img.size[0])).astype(np.uint8) for img in image_inputs
|
||||
]
|
||||
if segmentation_type == "pil":
|
||||
annotations = [Image.fromarray(annotation) for annotation in annotations]
|
||||
|
||||
inputs = image_processor(
|
||||
image_inputs,
|
||||
["semantic"] * len(image_inputs),
|
||||
annotations,
|
||||
return_tensors="pt",
|
||||
instance_id_to_semantic_id=instance_id_to_semantic_id,
|
||||
pad_and_return_pixel_mask=True,
|
||||
)
|
||||
|
||||
return inputs
|
||||
|
||||
def test_init_without_params(self):
|
||||
pass
|
||||
|
||||
def test_call_with_segmentation_maps(self):
|
||||
def common(is_instance_map=False, segmentation_type=None):
|
||||
inputs = self.comm_get_image_processor_inputs(
|
||||
with_segmentation_maps=True, is_instance_map=is_instance_map, segmentation_type=segmentation_type
|
||||
)
|
||||
|
||||
mask_labels = inputs["mask_labels"]
|
||||
class_labels = inputs["class_labels"]
|
||||
pixel_values = inputs["pixel_values"]
|
||||
text_inputs = inputs["text_inputs"]
|
||||
|
||||
# check the batch_size
|
||||
for mask_label, class_label, text_input in zip(mask_labels, class_labels, text_inputs):
|
||||
self.assertEqual(mask_label.shape[0], class_label.shape[0])
|
||||
# this ensure padding has happened
|
||||
self.assertEqual(mask_label.shape[1:], pixel_values.shape[2:])
|
||||
self.assertEqual(len(text_input), self.image_processing_tester.num_text)
|
||||
|
||||
common()
|
||||
common(is_instance_map=True)
|
||||
common(is_instance_map=False, segmentation_type="pil")
|
||||
common(is_instance_map=True, segmentation_type="pil")
|
||||
|
||||
def test_binary_mask_to_rle(self):
|
||||
fake_binary_mask = np.zeros((20, 50))
|
||||
fake_binary_mask[0, 20:] = 1
|
||||
fake_binary_mask[1, :15] = 1
|
||||
fake_binary_mask[5, :10] = 1
|
||||
|
||||
rle = binary_mask_to_rle(fake_binary_mask)
|
||||
self.assertEqual(len(rle), 4)
|
||||
self.assertEqual(rle[0], 21)
|
||||
self.assertEqual(rle[1], 45)
|
||||
|
||||
def test_post_process_semantic_segmentation(self):
|
||||
fature_extractor = self.image_processing_class(
|
||||
num_labels=self.image_processing_tester.num_classes,
|
||||
max_seq_length=77,
|
||||
task_seq_length=77,
|
||||
class_info_file="ade20k_panoptic.json",
|
||||
num_text=self.image_processing_tester.num_text,
|
||||
repo_path="shi-labs/oneformer_demo",
|
||||
)
|
||||
outputs = self.image_processing_tester.get_fake_oneformer_outputs()
|
||||
|
||||
segmentation = fature_extractor.post_process_semantic_segmentation(outputs)
|
||||
|
||||
self.assertEqual(len(segmentation), self.image_processing_tester.batch_size)
|
||||
self.assertEqual(
|
||||
segmentation[0].shape,
|
||||
(
|
||||
self.image_processing_tester.height,
|
||||
self.image_processing_tester.width,
|
||||
),
|
||||
)
|
||||
|
||||
target_sizes = [(1, 4) for i in range(self.image_processing_tester.batch_size)]
|
||||
segmentation = fature_extractor.post_process_semantic_segmentation(outputs, target_sizes=target_sizes)
|
||||
|
||||
self.assertEqual(segmentation[0].shape, target_sizes[0])
|
||||
|
||||
def test_post_process_instance_segmentation(self):
|
||||
image_processor = self.image_processing_class(
|
||||
num_labels=self.image_processing_tester.num_classes,
|
||||
max_seq_length=77,
|
||||
task_seq_length=77,
|
||||
class_info_file="ade20k_panoptic.json",
|
||||
num_text=self.image_processing_tester.num_text,
|
||||
repo_path="shi-labs/oneformer_demo",
|
||||
)
|
||||
outputs = self.image_processing_tester.get_fake_oneformer_outputs()
|
||||
segmentation = image_processor.post_process_instance_segmentation(outputs, threshold=0)
|
||||
|
||||
self.assertTrue(len(segmentation) == self.image_processing_tester.batch_size)
|
||||
for el in segmentation:
|
||||
self.assertTrue("segmentation" in el)
|
||||
self.assertTrue("segments_info" in el)
|
||||
self.assertEqual(type(el["segments_info"]), list)
|
||||
self.assertEqual(
|
||||
el["segmentation"].shape, (self.image_processing_tester.height, self.image_processing_tester.width)
|
||||
)
|
||||
|
||||
def test_post_process_panoptic_segmentation(self):
|
||||
image_processor = self.image_processing_class(
|
||||
num_labels=self.image_processing_tester.num_classes,
|
||||
max_seq_length=77,
|
||||
task_seq_length=77,
|
||||
class_info_file="ade20k_panoptic.json",
|
||||
num_text=self.image_processing_tester.num_text,
|
||||
repo_path="shi-labs/oneformer_demo",
|
||||
)
|
||||
outputs = self.image_processing_tester.get_fake_oneformer_outputs()
|
||||
segmentation = image_processor.post_process_panoptic_segmentation(outputs, threshold=0)
|
||||
|
||||
self.assertTrue(len(segmentation) == self.image_processing_tester.batch_size)
|
||||
for el in segmentation:
|
||||
self.assertTrue("segmentation" in el)
|
||||
self.assertTrue("segments_info" in el)
|
||||
self.assertEqual(type(el["segments_info"]), list)
|
||||
self.assertEqual(
|
||||
el["segmentation"].shape, (self.image_processing_tester.height, self.image_processing_tester.width)
|
||||
)
|
||||
536
tests/models/oneformer/test_modeling_oneformer.py
Normal file
536
tests/models/oneformer/test_modeling_oneformer.py
Normal file
@@ -0,0 +1,536 @@
|
||||
# coding=utf-8
|
||||
# Copyright 2022 The HuggingFace Inc. team. All rights reserved.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
""" Testing suite for the PyTorch OneFormer model. """
|
||||
|
||||
import copy
|
||||
import inspect
|
||||
import unittest
|
||||
|
||||
import numpy as np
|
||||
|
||||
from tests.test_modeling_common import floats_tensor
|
||||
from transformers import OneFormerConfig, is_torch_available, is_vision_available
|
||||
from transformers.testing_utils import require_torch, require_torch_multi_gpu, require_vision, slow, torch_device
|
||||
from transformers.utils import cached_property
|
||||
|
||||
from ...test_configuration_common import ConfigTester
|
||||
from ...test_modeling_common import ModelTesterMixin
|
||||
|
||||
|
||||
if is_torch_available():
|
||||
import torch
|
||||
|
||||
from transformers import OneFormerForUniversalSegmentation, OneFormerModel
|
||||
|
||||
if is_vision_available():
|
||||
from transformers import OneFormerProcessor
|
||||
|
||||
if is_vision_available():
|
||||
from PIL import Image
|
||||
|
||||
|
||||
def _config_zero_init(config):
|
||||
configs_no_init = copy.deepcopy(config)
|
||||
for key in configs_no_init.__dict__.keys():
|
||||
if "_range" in key or "_std" in key or "initializer_factor" in key or "layer_scale" in key:
|
||||
setattr(configs_no_init, key, 1e-10)
|
||||
return configs_no_init
|
||||
|
||||
|
||||
class OneFormerModelTester:
|
||||
def __init__(
|
||||
self,
|
||||
parent,
|
||||
batch_size=2,
|
||||
is_training=True,
|
||||
use_auxiliary_loss=False,
|
||||
num_queries=10,
|
||||
num_channels=3,
|
||||
min_size=32 * 8,
|
||||
max_size=32 * 8,
|
||||
num_labels=4,
|
||||
hidden_dim=64,
|
||||
sequence_length=77,
|
||||
n_ctx=4,
|
||||
):
|
||||
self.parent = parent
|
||||
self.batch_size = batch_size
|
||||
self.is_training = is_training
|
||||
self.use_auxiliary_loss = use_auxiliary_loss
|
||||
self.num_queries = num_queries
|
||||
self.num_channels = num_channels
|
||||
self.min_size = min_size
|
||||
self.max_size = max_size
|
||||
self.num_labels = num_labels
|
||||
self.hidden_dim = hidden_dim
|
||||
self.sequence_length = sequence_length
|
||||
self.n_ctx = n_ctx
|
||||
|
||||
def prepare_config_and_inputs(self):
|
||||
pixel_values = floats_tensor([self.batch_size, self.num_channels, self.min_size, self.max_size]).to(
|
||||
torch_device
|
||||
)
|
||||
|
||||
task_inputs = torch.randint(high=49408, size=(self.batch_size, self.sequence_length)).to(torch_device).long()
|
||||
|
||||
pixel_mask = torch.ones([self.batch_size, self.min_size, self.max_size], device=torch_device)
|
||||
|
||||
text_inputs = (
|
||||
torch.randint(high=49408, size=(self.batch_size, self.num_queries - self.n_ctx, self.sequence_length))
|
||||
.to(torch_device)
|
||||
.long()
|
||||
)
|
||||
|
||||
mask_labels = (
|
||||
torch.rand([self.batch_size, self.num_labels, self.min_size, self.max_size], device=torch_device) > 0.5
|
||||
).float()
|
||||
class_labels = (torch.rand((self.batch_size, self.num_labels), device=torch_device) > 0.5).long()
|
||||
|
||||
config = self.get_config()
|
||||
return config, pixel_values, task_inputs, text_inputs, pixel_mask, mask_labels, class_labels
|
||||
|
||||
def get_config(self):
|
||||
config = OneFormerConfig(
|
||||
hidden_size=self.hidden_dim,
|
||||
)
|
||||
|
||||
config.num_queries = self.num_queries
|
||||
config.num_labels = self.num_labels
|
||||
|
||||
config.backbone_config.depths = [1, 1, 1, 1]
|
||||
config.backbone_config.num_channels = self.num_channels
|
||||
|
||||
config.encoder_feedforward_dim = 64
|
||||
config.dim_feedforward = 128
|
||||
config.hidden_dim = self.hidden_dim
|
||||
config.mask_dim = self.hidden_dim
|
||||
config.conv_dim = self.hidden_dim
|
||||
|
||||
config.text_encoder_width = self.hidden_dim
|
||||
config.task_seq_len = self.sequence_length
|
||||
config.max_seq_len = self.sequence_length
|
||||
config.text_encoder_context_length = self.sequence_length
|
||||
config.text_encoder_n_ctx = self.n_ctx
|
||||
|
||||
return config
|
||||
|
||||
def prepare_config_and_inputs_for_common(self):
|
||||
config, pixel_values, task_inputs, pixel_mask, _, _, _ = self.prepare_config_and_inputs()
|
||||
inputs_dict = {"pixel_values": pixel_values, "pixel_mask": pixel_mask, "task_inputs": task_inputs}
|
||||
return config, inputs_dict
|
||||
|
||||
def check_output_hidden_state(self, output, config):
|
||||
encoder_hidden_states = output.encoder_hidden_states
|
||||
pixel_decoder_hidden_states = output.pixel_decoder_hidden_states
|
||||
transformer_decoder_hidden_states = output.transformer_decoder_hidden_states
|
||||
|
||||
self.parent.assertTrue(len(encoder_hidden_states), len(config.backbone_config.depths))
|
||||
self.parent.assertTrue(len(pixel_decoder_hidden_states), config.encoder_layers)
|
||||
self.parent.assertTrue(len(transformer_decoder_hidden_states), config.decoder_layers - 1)
|
||||
|
||||
def create_and_check_oneformer_model(
|
||||
self, config, pixel_values, task_inputs, pixel_mask, output_hidden_states=False
|
||||
):
|
||||
with torch.no_grad():
|
||||
model = OneFormerModel(config=config)
|
||||
model.to(torch_device)
|
||||
model.eval()
|
||||
|
||||
output = model(pixel_values=pixel_values, task_inputs=task_inputs, pixel_mask=pixel_mask)
|
||||
output = model(pixel_values, task_inputs=task_inputs, output_hidden_states=True)
|
||||
# the correct shape of output.transformer_decoder_hidden_states ensure the correcteness of the
|
||||
# encoder and pixel decoder
|
||||
self.parent.assertEqual(
|
||||
output.transformer_decoder_object_queries.shape,
|
||||
(self.batch_size, self.num_queries, self.hidden_dim),
|
||||
)
|
||||
# let's ensure the other two hidden state exists
|
||||
self.parent.assertTrue(output.pixel_decoder_hidden_states is not None)
|
||||
self.parent.assertTrue(output.encoder_hidden_states is not None)
|
||||
|
||||
if output_hidden_states:
|
||||
self.check_output_hidden_state(output, config)
|
||||
|
||||
def create_and_check_oneformer_universal_segmentation_head_model(
|
||||
self, config, pixel_values, task_inputs, text_inputs, pixel_mask, mask_labels, class_labels
|
||||
):
|
||||
model = OneFormerForUniversalSegmentation(config=config)
|
||||
model.to(torch_device)
|
||||
model.eval()
|
||||
|
||||
def comm_check_on_output(result):
|
||||
# let's still check that all the required stuff is there
|
||||
self.parent.assertTrue(result.transformer_decoder_hidden_states is not None)
|
||||
self.parent.assertTrue(result.pixel_decoder_hidden_states is not None)
|
||||
self.parent.assertTrue(result.encoder_hidden_states is not None)
|
||||
# okay, now we need to check the logits shape
|
||||
# due to the encoder compression, masks have a //4 spatial size
|
||||
self.parent.assertEqual(
|
||||
result.masks_queries_logits.shape,
|
||||
(self.batch_size, self.num_queries, self.min_size // 4, self.max_size // 4),
|
||||
)
|
||||
# + 1 for null class
|
||||
self.parent.assertEqual(
|
||||
result.class_queries_logits.shape, (self.batch_size, self.num_queries, self.num_labels + 1)
|
||||
)
|
||||
|
||||
with torch.no_grad():
|
||||
result = model(pixel_values=pixel_values, task_inputs=task_inputs, pixel_mask=pixel_mask)
|
||||
result = model(pixel_values, task_inputs)
|
||||
|
||||
comm_check_on_output(result)
|
||||
|
||||
config.is_training = True
|
||||
model = OneFormerForUniversalSegmentation(config=config)
|
||||
model.to(torch_device)
|
||||
model.eval()
|
||||
|
||||
with torch.no_grad():
|
||||
result = model(
|
||||
pixel_values=pixel_values,
|
||||
task_inputs=task_inputs,
|
||||
pixel_mask=pixel_mask,
|
||||
mask_labels=mask_labels,
|
||||
class_labels=class_labels,
|
||||
text_inputs=text_inputs,
|
||||
)
|
||||
|
||||
comm_check_on_output(result)
|
||||
|
||||
self.parent.assertTrue(result.loss is not None)
|
||||
self.parent.assertEqual(result.loss.shape, torch.Size([1]))
|
||||
|
||||
|
||||
@require_torch
|
||||
class OneFormerModelTest(ModelTesterMixin, unittest.TestCase):
|
||||
all_model_classes = (OneFormerModel, OneFormerForUniversalSegmentation) if is_torch_available() else ()
|
||||
|
||||
is_encoder_decoder = False
|
||||
test_pruning = False
|
||||
test_head_masking = False
|
||||
test_missing_keys = False
|
||||
|
||||
def setUp(self):
|
||||
self.model_tester = OneFormerModelTester(self)
|
||||
self.config_tester = ConfigTester(self, config_class=OneFormerConfig, has_text_modality=False)
|
||||
|
||||
def test_config(self):
|
||||
self.config_tester.run_common_tests()
|
||||
|
||||
def test_oneformer_model(self):
|
||||
config, inputs = self.model_tester.prepare_config_and_inputs_for_common()
|
||||
self.model_tester.create_and_check_oneformer_model(config, **inputs, output_hidden_states=False)
|
||||
|
||||
def test_oneformer_universal_segmentation_head_model(self):
|
||||
config_and_inputs = self.model_tester.prepare_config_and_inputs()
|
||||
self.model_tester.create_and_check_oneformer_universal_segmentation_head_model(*config_and_inputs)
|
||||
|
||||
def test_model_main_input_name(self):
|
||||
for model_class in self.all_model_classes:
|
||||
model_signature = inspect.signature(getattr(model_class, "forward"))
|
||||
# The main input is the name of the argument after `self`
|
||||
observed_main_input_name = list(model_signature.parameters.keys())[1:3]
|
||||
self.assertEqual(model_class.main_input_name, observed_main_input_name)
|
||||
|
||||
@unittest.skip(reason="OneFormer uses two main inputs")
|
||||
def test_torchscript_simple(self):
|
||||
pass
|
||||
|
||||
@unittest.skip(reason="OneFormer uses two main inputs")
|
||||
def test_torchscript_output_attentions(self):
|
||||
pass
|
||||
|
||||
@unittest.skip(reason="OneFormer uses two main inputs")
|
||||
def test_torchscript_output_hidden_state(self):
|
||||
pass
|
||||
|
||||
@unittest.skip(reason="OneFormer does not use inputs_embeds")
|
||||
def test_inputs_embeds(self):
|
||||
pass
|
||||
|
||||
@unittest.skip(reason="OneFormer does not have a get_input_embeddings method")
|
||||
def test_model_common_attributes(self):
|
||||
pass
|
||||
|
||||
@unittest.skip(reason="OneFormer is not a generative model")
|
||||
def test_generate_without_input_ids(self):
|
||||
pass
|
||||
|
||||
@unittest.skip(reason="OneFormer does not use token embeddings")
|
||||
def test_resize_tokens_embeddings(self):
|
||||
pass
|
||||
|
||||
@require_torch_multi_gpu
|
||||
@unittest.skip(
|
||||
reason="OneFormer has some layers using `add_module` which doesn't work well with `nn.DataParallel`"
|
||||
)
|
||||
def test_multi_gpu_data_parallel_forward(self):
|
||||
pass
|
||||
|
||||
def test_forward_signature(self):
|
||||
config, _ = self.model_tester.prepare_config_and_inputs_for_common()
|
||||
|
||||
for model_class in self.all_model_classes:
|
||||
model = model_class(config)
|
||||
signature = inspect.signature(model.forward)
|
||||
# signature.parameters is an OrderedDict => so arg_names order is deterministic
|
||||
arg_names = [*signature.parameters.keys()]
|
||||
|
||||
expected_arg_names = ["pixel_values", "task_inputs"]
|
||||
self.assertListEqual(arg_names[:2], expected_arg_names)
|
||||
|
||||
@slow
|
||||
def test_model_from_pretrained(self):
|
||||
for model_name in ["shi-labs/oneformer_ade20k_swin_tiny"]:
|
||||
model = OneFormerModel.from_pretrained(model_name)
|
||||
self.assertIsNotNone(model)
|
||||
|
||||
def test_model_with_labels(self):
|
||||
size = (self.model_tester.min_size,) * 2
|
||||
inputs = {
|
||||
"pixel_values": torch.randn((2, 3, *size), device=torch_device),
|
||||
"task_inputs": torch.randint(high=49408, size=(2, 77), device=torch_device).long(),
|
||||
"text_inputs": torch.randint(high=49408, size=(2, 134, 77), device=torch_device).long(),
|
||||
"mask_labels": torch.randn((2, 150, *size), device=torch_device),
|
||||
"class_labels": torch.zeros(2, 150, device=torch_device).long(),
|
||||
}
|
||||
|
||||
config = OneFormerConfig()
|
||||
config.is_training = True
|
||||
|
||||
model = OneFormerForUniversalSegmentation(config).to(torch_device)
|
||||
outputs = model(**inputs)
|
||||
self.assertTrue(outputs.loss is not None)
|
||||
|
||||
def test_hidden_states_output(self):
|
||||
config, inputs = self.model_tester.prepare_config_and_inputs_for_common()
|
||||
self.model_tester.create_and_check_oneformer_model(config, **inputs, output_hidden_states=True)
|
||||
|
||||
def test_attention_outputs(self):
|
||||
config, inputs = self.model_tester.prepare_config_and_inputs_for_common()
|
||||
|
||||
for model_class in self.all_model_classes:
|
||||
model = model_class(config).to(torch_device)
|
||||
outputs = model(**inputs, output_attentions=True)
|
||||
self.assertTrue(outputs.attentions is not None)
|
||||
|
||||
def test_initialization(self):
|
||||
config, inputs_dict = self.model_tester.prepare_config_and_inputs_for_common()
|
||||
config.contrastive_temperature = 1
|
||||
|
||||
configs_no_init = _config_zero_init(config)
|
||||
for model_class in self.all_model_classes:
|
||||
model = model_class(config=configs_no_init)
|
||||
for name, param in model.named_parameters():
|
||||
if param.requires_grad:
|
||||
self.assertIn(
|
||||
((param.data.mean() * 1e9).round() / 1e9).item(),
|
||||
[0.0, 1.0],
|
||||
msg=f"Parameter {name} of model {model_class} seems not properly initialized",
|
||||
)
|
||||
|
||||
def test_training(self):
|
||||
if not self.model_tester.is_training:
|
||||
return
|
||||
# only OneFormerForUniversalSegmentation has the loss
|
||||
model_class = self.all_model_classes[1]
|
||||
(
|
||||
config,
|
||||
pixel_values,
|
||||
task_inputs,
|
||||
text_inputs,
|
||||
pixel_mask,
|
||||
mask_labels,
|
||||
class_labels,
|
||||
) = self.model_tester.prepare_config_and_inputs()
|
||||
config.is_training = True
|
||||
|
||||
model = model_class(config)
|
||||
model.to(torch_device)
|
||||
model.train()
|
||||
|
||||
loss = model(
|
||||
pixel_values, task_inputs, text_inputs=text_inputs, mask_labels=mask_labels, class_labels=class_labels
|
||||
).loss
|
||||
loss.backward()
|
||||
|
||||
def test_retain_grad_hidden_states_attentions(self):
|
||||
# only OneFormerForUniversalSegmentation has the loss
|
||||
model_class = self.all_model_classes[1]
|
||||
(
|
||||
config,
|
||||
pixel_values,
|
||||
task_inputs,
|
||||
text_inputs,
|
||||
pixel_mask,
|
||||
mask_labels,
|
||||
class_labels,
|
||||
) = self.model_tester.prepare_config_and_inputs()
|
||||
config.output_hidden_states = True
|
||||
config.output_attentions = True
|
||||
config.is_training = True
|
||||
|
||||
model = model_class(config)
|
||||
model.to(torch_device)
|
||||
model.train()
|
||||
|
||||
outputs = model(
|
||||
pixel_values, task_inputs, text_inputs=text_inputs, mask_labels=mask_labels, class_labels=class_labels
|
||||
)
|
||||
|
||||
encoder_hidden_states = outputs.encoder_hidden_states[0]
|
||||
encoder_hidden_states.retain_grad()
|
||||
|
||||
pixel_decoder_hidden_states = outputs.pixel_decoder_hidden_states[0]
|
||||
pixel_decoder_hidden_states.retain_grad()
|
||||
|
||||
transformer_decoder_class_predictions = outputs.transformer_decoder_class_predictions
|
||||
transformer_decoder_class_predictions.retain_grad()
|
||||
|
||||
transformer_decoder_mask_predictions = outputs.transformer_decoder_mask_predictions
|
||||
transformer_decoder_mask_predictions.retain_grad()
|
||||
|
||||
attentions = outputs.attentions[0][0]
|
||||
attentions.retain_grad()
|
||||
|
||||
outputs.loss.backward(retain_graph=True)
|
||||
|
||||
self.assertIsNotNone(encoder_hidden_states.grad)
|
||||
self.assertIsNotNone(pixel_decoder_hidden_states.grad)
|
||||
self.assertIsNotNone(transformer_decoder_class_predictions.grad)
|
||||
self.assertIsNotNone(transformer_decoder_mask_predictions.grad)
|
||||
self.assertIsNotNone(attentions.grad)
|
||||
|
||||
|
||||
TOLERANCE = 1e-4
|
||||
|
||||
|
||||
# We will verify our results on an image of cute cats
|
||||
def prepare_img():
|
||||
image = Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png")
|
||||
return image
|
||||
|
||||
|
||||
@require_vision
|
||||
@slow
|
||||
class OneFormerModelIntegrationTest(unittest.TestCase):
|
||||
@cached_property
|
||||
def model_checkpoints(self):
|
||||
return "shi-labs/oneformer_ade20k_swin_tiny"
|
||||
|
||||
@cached_property
|
||||
def default_processor(self):
|
||||
return OneFormerProcessor.from_pretrained(self.model_checkpoints) if is_vision_available() else None
|
||||
|
||||
def test_inference_no_head(self):
|
||||
model = OneFormerModel.from_pretrained(self.model_checkpoints).to(torch_device)
|
||||
processor = self.default_processor
|
||||
image = prepare_img()
|
||||
inputs = processor(image, ["semantic"], return_tensors="pt").to(torch_device)
|
||||
inputs_shape = inputs["pixel_values"].shape
|
||||
# check size
|
||||
self.assertEqual(inputs_shape, (1, 3, 512, 682))
|
||||
|
||||
task_inputs_shape = inputs["task_inputs"].shape
|
||||
# check size
|
||||
self.assertEqual(task_inputs_shape, (1, 77))
|
||||
|
||||
with torch.no_grad():
|
||||
outputs = model(**inputs)
|
||||
|
||||
expected_slice_hidden_state = torch.tensor(
|
||||
[[0.2724, 0.8287, 0.6025], [1.2706, 1.1252, 1.1445], [1.1357, 0.6150, 0.4185]]
|
||||
).to(torch_device)
|
||||
self.assertTrue(
|
||||
torch.allclose(
|
||||
outputs.encoder_hidden_states[-1][0, 0, :3, :3], expected_slice_hidden_state, atol=TOLERANCE
|
||||
)
|
||||
)
|
||||
|
||||
expected_slice_hidden_state = torch.tensor(
|
||||
[[1.0581, 1.2275, 1.2000], [1.1901, 1.2925, 1.2861], [1.1578, 1.2558, 1.3212]]
|
||||
).to(torch_device)
|
||||
self.assertTrue(
|
||||
torch.allclose(
|
||||
outputs.pixel_decoder_hidden_states[0][0, 0, :3, :3], expected_slice_hidden_state, atol=TOLERANCE
|
||||
)
|
||||
)
|
||||
|
||||
expected_slice_hidden_state = torch.tensor(
|
||||
[[3.0711, -1.1855, -5.1095], [3.5536, -3.2710, -5.2052], [2.6020, -4.3605, -4.1422]]
|
||||
).to(torch_device)
|
||||
self.assertTrue(
|
||||
torch.allclose(
|
||||
outputs.transformer_decoder_class_predictions[0, :3, :3], expected_slice_hidden_state, atol=TOLERANCE
|
||||
)
|
||||
)
|
||||
|
||||
def test_inference_universal_segmentation_head(self):
|
||||
model = OneFormerForUniversalSegmentation.from_pretrained(self.model_checkpoints).to(torch_device).eval()
|
||||
processor = self.default_processor
|
||||
image = prepare_img()
|
||||
inputs = processor(image, ["semantic"], return_tensors="pt").to(torch_device)
|
||||
inputs_shape = inputs["pixel_values"].shape
|
||||
# check size
|
||||
self.assertEqual(inputs_shape, (1, 3, 512, 682))
|
||||
|
||||
with torch.no_grad():
|
||||
outputs = model(**inputs)
|
||||
|
||||
# masks_queries_logits
|
||||
masks_queries_logits = outputs.masks_queries_logits
|
||||
self.assertEqual(
|
||||
masks_queries_logits.shape,
|
||||
(1, model.config.num_queries, inputs_shape[-2] // 4, (inputs_shape[-1] + 2) // 4),
|
||||
)
|
||||
expected_slice = [[[3.1215, 4.1250, 4.1106], [2.8183, 3.4623, 3.5512], [2.4550, 2.9841, 3.5081]]]
|
||||
expected_slice = torch.tensor(expected_slice).to(torch_device)
|
||||
self.assertTrue(torch.allclose(masks_queries_logits[0, 0, :3, :3], expected_slice, atol=TOLERANCE))
|
||||
# class_queries_logits
|
||||
class_queries_logits = outputs.class_queries_logits
|
||||
self.assertEqual(
|
||||
class_queries_logits.shape,
|
||||
(1, model.config.num_queries, model.config.num_labels + 1),
|
||||
)
|
||||
expected_slice = torch.tensor(
|
||||
[[3.0711, -1.1855, -5.1095], [3.5536, -3.2710, -5.2052], [2.6020, -4.3605, -4.1422]]
|
||||
).to(torch_device)
|
||||
self.assertTrue(torch.allclose(class_queries_logits[0, :3, :3], expected_slice, atol=TOLERANCE))
|
||||
|
||||
def test_with_segmentation_maps_and_loss(self):
|
||||
dummy_model = OneFormerForUniversalSegmentation.from_pretrained(self.model_checkpoints)
|
||||
processor = self.default_processor
|
||||
processor.image_processor.num_text = dummy_model.config.num_queries - dummy_model.config.text_encoder_n_ctx
|
||||
dummy_model.config.is_training = True
|
||||
model = OneFormerForUniversalSegmentation(dummy_model.config).to(torch_device).eval()
|
||||
del dummy_model
|
||||
|
||||
inputs = processor(
|
||||
[np.zeros((3, 512, 640)), np.zeros((3, 512, 640))],
|
||||
["semantic", "semantic"],
|
||||
segmentation_maps=[np.zeros((384, 384)).astype(np.float32), np.zeros((384, 384)).astype(np.float32)],
|
||||
return_tensors="pt",
|
||||
)
|
||||
|
||||
inputs["pixel_values"] = inputs["pixel_values"].to(torch_device)
|
||||
inputs["task_inputs"] = inputs["task_inputs"].to(torch_device)
|
||||
inputs["text_inputs"] = inputs["text_inputs"].to(torch_device)
|
||||
inputs["mask_labels"] = [el.to(torch_device) for el in inputs["mask_labels"]]
|
||||
inputs["class_labels"] = [el.to(torch_device) for el in inputs["class_labels"]]
|
||||
|
||||
with torch.no_grad():
|
||||
outputs = model(**inputs)
|
||||
|
||||
self.assertTrue(outputs.loss is not None)
|
||||
833
tests/models/oneformer/test_processor_oneformer.py
Normal file
833
tests/models/oneformer/test_processor_oneformer.py
Normal file
@@ -0,0 +1,833 @@
|
||||
# coding=utf-8
|
||||
# Copyright 2022 HuggingFace Inc.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
import numpy as np
|
||||
from datasets import load_dataset
|
||||
|
||||
from huggingface_hub import hf_hub_download
|
||||
from transformers.testing_utils import check_json_file_has_correct_format, require_torch, require_vision
|
||||
from transformers.utils import is_torch_available, is_vision_available
|
||||
|
||||
from ...test_feature_extraction_common import prepare_image_inputs
|
||||
|
||||
|
||||
if is_torch_available():
|
||||
import torch
|
||||
|
||||
if is_vision_available():
|
||||
from transformers import CLIPTokenizer, OneFormerImageProcessor, OneFormerProcessor
|
||||
from transformers.models.oneformer.image_processing_oneformer import binary_mask_to_rle
|
||||
from transformers.models.oneformer.modeling_oneformer import OneFormerForUniversalSegmentationOutput
|
||||
|
||||
if is_vision_available():
|
||||
from PIL import Image
|
||||
|
||||
|
||||
def prepare_metadata(class_info_file, repo_path="shi-labs/oneformer_demo"):
|
||||
with open(hf_hub_download(repo_path, class_info_file, repo_type="dataset"), "r") as f:
|
||||
class_info = json.load(f)
|
||||
metadata = {}
|
||||
class_names = []
|
||||
thing_ids = []
|
||||
|
||||
for key, info in class_info.items():
|
||||
metadata[key] = info["name"]
|
||||
class_names.append(info["name"])
|
||||
if info["isthing"]:
|
||||
thing_ids.append(int(key))
|
||||
|
||||
metadata["thing_ids"] = thing_ids
|
||||
metadata["class_names"] = class_names
|
||||
return metadata
|
||||
|
||||
|
||||
class OneFormerProcessorTester(unittest.TestCase):
|
||||
def __init__(
|
||||
self,
|
||||
parent,
|
||||
batch_size=7,
|
||||
num_channels=3,
|
||||
min_resolution=30,
|
||||
max_resolution=400,
|
||||
size=None,
|
||||
do_resize=True,
|
||||
do_normalize=True,
|
||||
image_mean=[0.5, 0.5, 0.5],
|
||||
image_std=[0.5, 0.5, 0.5],
|
||||
num_labels=10,
|
||||
reduce_labels=False,
|
||||
ignore_index=255,
|
||||
max_seq_length=77,
|
||||
task_seq_length=77,
|
||||
model_repo="shi-labs/oneformer_ade20k_swin_tiny",
|
||||
class_info_file="ade20k_panoptic.json",
|
||||
num_text=10,
|
||||
):
|
||||
self.parent = parent
|
||||
self.batch_size = batch_size
|
||||
self.num_channels = num_channels
|
||||
self.min_resolution = min_resolution
|
||||
self.max_resolution = max_resolution
|
||||
self.do_resize = do_resize
|
||||
self.size = {"shortest_edge": 32, "longest_edge": 1333} if size is None else size
|
||||
self.do_normalize = do_normalize
|
||||
self.image_mean = image_mean
|
||||
self.image_std = image_std
|
||||
self.max_seq_length = max_seq_length
|
||||
self.task_seq_length = task_seq_length
|
||||
self.class_info_file = class_info_file
|
||||
self.metadata = prepare_metadata(class_info_file)
|
||||
self.num_text = num_text
|
||||
self.model_repo = model_repo
|
||||
|
||||
# for the post_process_functions
|
||||
self.batch_size = 2
|
||||
self.num_queries = 10
|
||||
self.num_classes = 10
|
||||
self.height = 3
|
||||
self.width = 4
|
||||
self.num_labels = num_labels
|
||||
self.reduce_labels = reduce_labels
|
||||
self.ignore_index = ignore_index
|
||||
|
||||
def prepare_processor_dict(self):
|
||||
image_processor_dict = {
|
||||
"do_resize": self.do_resize,
|
||||
"size": self.size,
|
||||
"do_normalize": self.do_normalize,
|
||||
"image_mean": self.image_mean,
|
||||
"image_std": self.image_std,
|
||||
"num_labels": self.num_labels,
|
||||
"reduce_labels": self.reduce_labels,
|
||||
"ignore_index": self.ignore_index,
|
||||
"class_info_file": self.class_info_file,
|
||||
"metadata": self.metadata,
|
||||
"num_text": self.num_text,
|
||||
}
|
||||
|
||||
image_processor = OneFormerImageProcessor(**image_processor_dict)
|
||||
tokenizer = CLIPTokenizer.from_pretrained(self.model_repo)
|
||||
|
||||
return {
|
||||
"image_processor": image_processor,
|
||||
"tokenizer": tokenizer,
|
||||
"max_seq_length": self.max_seq_length,
|
||||
"task_seq_length": self.task_seq_length,
|
||||
}
|
||||
|
||||
def get_expected_values(self, image_inputs, batched=False):
|
||||
"""
|
||||
This function computes the expected height and width when providing images to OneFormerProcessor,
|
||||
assuming do_resize is set to True with a scalar size. It also provides the expected sequence length
|
||||
for the task_inputs and text_list_input.
|
||||
"""
|
||||
if not batched:
|
||||
image = image_inputs[0]
|
||||
if isinstance(image, Image.Image):
|
||||
w, h = image.size
|
||||
else:
|
||||
h, w = image.shape[1], image.shape[2]
|
||||
if w < h:
|
||||
expected_height = int(self.size["shortest_edge"] * h / w)
|
||||
expected_width = self.size["shortest_edge"]
|
||||
elif w > h:
|
||||
expected_height = self.size["shortest_edge"]
|
||||
expected_width = int(self.size["shortest_edge"] * w / h)
|
||||
else:
|
||||
expected_height = self.size["shortest_edge"]
|
||||
expected_width = self.size["shortest_edge"]
|
||||
|
||||
else:
|
||||
expected_values = []
|
||||
for image in image_inputs:
|
||||
expected_height, expected_width, expected_sequence_length = self.get_expected_values([image])
|
||||
expected_values.append((expected_height, expected_width, expected_sequence_length))
|
||||
expected_height = max(expected_values, key=lambda item: item[0])[0]
|
||||
expected_width = max(expected_values, key=lambda item: item[1])[1]
|
||||
|
||||
expected_sequence_length = self.max_seq_length
|
||||
|
||||
return expected_height, expected_width, expected_sequence_length
|
||||
|
||||
def get_fake_oneformer_outputs(self):
|
||||
return OneFormerForUniversalSegmentationOutput(
|
||||
# +1 for null class
|
||||
class_queries_logits=torch.randn((self.batch_size, self.num_queries, self.num_classes + 1)),
|
||||
masks_queries_logits=torch.randn((self.batch_size, self.num_queries, self.height, self.width)),
|
||||
)
|
||||
|
||||
|
||||
@require_torch
|
||||
@require_vision
|
||||
class OneFormerProcessingTest(unittest.TestCase):
|
||||
processing_class = OneFormerProcessor if (is_vision_available() and is_torch_available()) else None
|
||||
# only for test_feat_extracttion_common.test_feat_extract_to_json_string
|
||||
feature_extraction_class = processing_class
|
||||
|
||||
def setUp(self):
|
||||
self.processing_tester = OneFormerProcessorTester(self)
|
||||
|
||||
@property
|
||||
def processor_dict(self):
|
||||
return self.processing_tester.prepare_processor_dict()
|
||||
|
||||
def test_feat_extract_properties(self):
|
||||
processor = self.processing_class(**self.processor_dict)
|
||||
self.assertTrue(hasattr(processor, "image_processor"))
|
||||
self.assertTrue(hasattr(processor, "tokenizer"))
|
||||
self.assertTrue(hasattr(processor, "max_seq_length"))
|
||||
self.assertTrue(hasattr(processor, "task_seq_length"))
|
||||
|
||||
def test_batch_feature(self):
|
||||
pass
|
||||
|
||||
def test_call_pil(self):
|
||||
# Initialize processor
|
||||
processor = self.processing_class(**self.processor_dict)
|
||||
# create random PIL images
|
||||
image_inputs = prepare_image_inputs(self.processing_tester, equal_resolution=False)
|
||||
for image in image_inputs:
|
||||
self.assertIsInstance(image, Image.Image)
|
||||
|
||||
# Test not batched input
|
||||
encoded_images = processor(image_inputs[0], ["semantic"], return_tensors="pt").pixel_values
|
||||
|
||||
expected_height, expected_width, expected_sequence_length = self.processing_tester.get_expected_values(
|
||||
image_inputs
|
||||
)
|
||||
|
||||
self.assertEqual(
|
||||
encoded_images.shape,
|
||||
(1, self.processing_tester.num_channels, expected_height, expected_width),
|
||||
)
|
||||
|
||||
tokenized_task_inputs = processor(image_inputs[0], ["semantic"], return_tensors="pt").task_inputs
|
||||
|
||||
self.assertEqual(
|
||||
tokenized_task_inputs.shape,
|
||||
(1, expected_sequence_length),
|
||||
)
|
||||
|
||||
# Test batched
|
||||
expected_height, expected_width, expected_sequence_length = self.processing_tester.get_expected_values(
|
||||
image_inputs, batched=True
|
||||
)
|
||||
|
||||
encoded_images = processor(image_inputs, ["semantic"] * len(image_inputs), return_tensors="pt").pixel_values
|
||||
self.assertEqual(
|
||||
encoded_images.shape,
|
||||
(
|
||||
self.processing_tester.batch_size,
|
||||
self.processing_tester.num_channels,
|
||||
expected_height,
|
||||
expected_width,
|
||||
),
|
||||
)
|
||||
|
||||
tokenized_task_inputs = processor(
|
||||
image_inputs, ["semantic"] * len(image_inputs), return_tensors="pt"
|
||||
).task_inputs
|
||||
|
||||
self.assertEqual(
|
||||
tokenized_task_inputs.shape,
|
||||
(self.processing_tester.batch_size, expected_sequence_length),
|
||||
)
|
||||
|
||||
def test_call_numpy(self):
|
||||
# Initialize processor
|
||||
processor = self.processing_class(**self.processor_dict)
|
||||
# create random numpy tensors
|
||||
image_inputs = prepare_image_inputs(self.processing_tester, equal_resolution=False, numpify=True)
|
||||
for image in image_inputs:
|
||||
self.assertIsInstance(image, np.ndarray)
|
||||
|
||||
# Test not batched input
|
||||
encoded_images = processor(image_inputs[0], ["semantic"], return_tensors="pt").pixel_values
|
||||
|
||||
expected_height, expected_width, expected_sequence_length = self.processing_tester.get_expected_values(
|
||||
image_inputs
|
||||
)
|
||||
|
||||
self.assertEqual(
|
||||
encoded_images.shape,
|
||||
(1, self.processing_tester.num_channels, expected_height, expected_width),
|
||||
)
|
||||
|
||||
tokenized_task_inputs = processor(image_inputs[0], ["semantic"], return_tensors="pt").task_inputs
|
||||
|
||||
self.assertEqual(
|
||||
tokenized_task_inputs.shape,
|
||||
(1, expected_sequence_length),
|
||||
)
|
||||
|
||||
# Test batched
|
||||
expected_height, expected_width, expected_sequence_length = self.processing_tester.get_expected_values(
|
||||
image_inputs, batched=True
|
||||
)
|
||||
|
||||
encoded_images = processor(image_inputs, ["semantic"] * len(image_inputs), return_tensors="pt").pixel_values
|
||||
self.assertEqual(
|
||||
encoded_images.shape,
|
||||
(
|
||||
self.processing_tester.batch_size,
|
||||
self.processing_tester.num_channels,
|
||||
expected_height,
|
||||
expected_width,
|
||||
),
|
||||
)
|
||||
|
||||
tokenized_task_inputs = processor(
|
||||
image_inputs, ["semantic"] * len(image_inputs), return_tensors="pt"
|
||||
).task_inputs
|
||||
|
||||
self.assertEqual(
|
||||
tokenized_task_inputs.shape,
|
||||
(self.processing_tester.batch_size, expected_sequence_length),
|
||||
)
|
||||
|
||||
def test_call_pytorch(self):
|
||||
# Initialize processor
|
||||
processor = self.processing_class(**self.processor_dict)
|
||||
# create random PyTorch tensors
|
||||
image_inputs = prepare_image_inputs(self.processing_tester, equal_resolution=False, torchify=True)
|
||||
for image in image_inputs:
|
||||
self.assertIsInstance(image, torch.Tensor)
|
||||
|
||||
# Test not batched input
|
||||
encoded_images = processor(image_inputs[0], ["semantic"], return_tensors="pt").pixel_values
|
||||
|
||||
expected_height, expected_width, expected_sequence_length = self.processing_tester.get_expected_values(
|
||||
image_inputs
|
||||
)
|
||||
|
||||
self.assertEqual(
|
||||
encoded_images.shape,
|
||||
(1, self.processing_tester.num_channels, expected_height, expected_width),
|
||||
)
|
||||
|
||||
tokenized_task_inputs = processor(image_inputs[0], ["semantic"], return_tensors="pt").task_inputs
|
||||
|
||||
self.assertEqual(
|
||||
tokenized_task_inputs.shape,
|
||||
(1, expected_sequence_length),
|
||||
)
|
||||
|
||||
# Test batched
|
||||
expected_height, expected_width, expected_sequence_length = self.processing_tester.get_expected_values(
|
||||
image_inputs, batched=True
|
||||
)
|
||||
|
||||
encoded_images = processor(image_inputs, ["semantic"] * len(image_inputs), return_tensors="pt").pixel_values
|
||||
self.assertEqual(
|
||||
encoded_images.shape,
|
||||
(
|
||||
self.processing_tester.batch_size,
|
||||
self.processing_tester.num_channels,
|
||||
expected_height,
|
||||
expected_width,
|
||||
),
|
||||
)
|
||||
|
||||
tokenized_task_inputs = processor(
|
||||
image_inputs, ["semantic"] * len(image_inputs), return_tensors="pt"
|
||||
).task_inputs
|
||||
|
||||
self.assertEqual(
|
||||
tokenized_task_inputs.shape,
|
||||
(self.processing_tester.batch_size, expected_sequence_length),
|
||||
)
|
||||
|
||||
def test_equivalence_pad_and_create_pixel_mask(self):
|
||||
# Initialize processors
|
||||
processor_1 = self.processing_class(**self.processor_dict)
|
||||
|
||||
image_processor = OneFormerImageProcessor(
|
||||
do_resize=False,
|
||||
do_normalize=False,
|
||||
do_rescale=False,
|
||||
num_labels=self.processing_tester.num_classes,
|
||||
class_info_file="ade20k_panoptic.json",
|
||||
num_text=self.processing_tester.num_text,
|
||||
)
|
||||
tokenizer = CLIPTokenizer.from_pretrained("shi-labs/oneformer_ade20k_swin_tiny")
|
||||
processor_2 = self.processing_class(
|
||||
image_processor=image_processor, tokenizer=tokenizer, max_seq_length=77, task_seq_length=77
|
||||
)
|
||||
|
||||
# create random PyTorch tensors
|
||||
image_inputs = prepare_image_inputs(self.processing_tester, equal_resolution=False, torchify=True)
|
||||
for image in image_inputs:
|
||||
self.assertIsInstance(image, torch.Tensor)
|
||||
|
||||
# Test whether the method "pad_and_return_pixel_mask" and calling the image processor return the same tensors
|
||||
encoded_images_with_method = processor_1.encode_inputs(
|
||||
image_inputs, ["semantic"] * len(image_inputs), return_tensors="pt"
|
||||
)
|
||||
encoded_images = processor_2(image_inputs, ["semantic"] * len(image_inputs), return_tensors="pt")
|
||||
|
||||
self.assertTrue(
|
||||
torch.allclose(encoded_images_with_method["pixel_values"], encoded_images["pixel_values"], atol=1e-4)
|
||||
)
|
||||
self.assertTrue(
|
||||
torch.allclose(encoded_images_with_method["pixel_mask"], encoded_images["pixel_mask"], atol=1e-4)
|
||||
)
|
||||
|
||||
def comm_get_processor_inputs(self, with_segmentation_maps=False, is_instance_map=False, segmentation_type="np"):
|
||||
processor = self.processing_class(**self.processor_dict)
|
||||
# prepare image and target
|
||||
num_labels = self.processing_tester.num_labels
|
||||
annotations = None
|
||||
instance_id_to_semantic_id = None
|
||||
image_inputs = prepare_image_inputs(self.processing_tester, equal_resolution=False)
|
||||
if with_segmentation_maps:
|
||||
high = num_labels
|
||||
if is_instance_map:
|
||||
labels_expanded = list(range(num_labels)) * 2
|
||||
instance_id_to_semantic_id = {
|
||||
instance_id: label_id for instance_id, label_id in enumerate(labels_expanded)
|
||||
}
|
||||
annotations = [
|
||||
np.random.randint(0, high * 2, (img.size[1], img.size[0])).astype(np.uint8) for img in image_inputs
|
||||
]
|
||||
if segmentation_type == "pil":
|
||||
annotations = [Image.fromarray(annotation) for annotation in annotations]
|
||||
|
||||
inputs = processor(
|
||||
image_inputs,
|
||||
["semantic"] * len(image_inputs),
|
||||
annotations,
|
||||
return_tensors="pt",
|
||||
instance_id_to_semantic_id=instance_id_to_semantic_id,
|
||||
pad_and_return_pixel_mask=True,
|
||||
)
|
||||
|
||||
return inputs
|
||||
|
||||
def test_init_without_params(self):
|
||||
pass
|
||||
|
||||
def test_feat_extract_from_and_save_pretrained(self):
|
||||
feat_extract_first = self.feature_extraction_class(**self.processor_dict)
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdirname:
|
||||
feat_extract_first.save_pretrained(tmpdirname)
|
||||
check_json_file_has_correct_format(os.path.join(tmpdirname, "preprocessor_config.json"))
|
||||
feat_extract_second = self.feature_extraction_class.from_pretrained(tmpdirname)
|
||||
|
||||
self.assertEqual(feat_extract_second.image_processor.to_dict(), feat_extract_first.image_processor.to_dict())
|
||||
self.assertIsInstance(feat_extract_first.image_processor, OneFormerImageProcessor)
|
||||
self.assertIsInstance(feat_extract_first.tokenizer, CLIPTokenizer)
|
||||
|
||||
def test_call_with_segmentation_maps(self):
|
||||
def common(is_instance_map=False, segmentation_type=None):
|
||||
inputs = self.comm_get_processor_inputs(
|
||||
with_segmentation_maps=True, is_instance_map=is_instance_map, segmentation_type=segmentation_type
|
||||
)
|
||||
|
||||
mask_labels = inputs["mask_labels"]
|
||||
class_labels = inputs["class_labels"]
|
||||
pixel_values = inputs["pixel_values"]
|
||||
text_inputs = inputs["text_inputs"]
|
||||
|
||||
# check the batch_size
|
||||
for mask_label, class_label, text_input in zip(mask_labels, class_labels, text_inputs):
|
||||
self.assertEqual(mask_label.shape[0], class_label.shape[0])
|
||||
# this ensure padding has happened
|
||||
self.assertEqual(mask_label.shape[1:], pixel_values.shape[2:])
|
||||
self.assertEqual(text_input.shape[0], self.processing_tester.num_text)
|
||||
|
||||
common()
|
||||
common(is_instance_map=True)
|
||||
common(is_instance_map=False, segmentation_type="pil")
|
||||
common(is_instance_map=True, segmentation_type="pil")
|
||||
|
||||
def test_integration_semantic_segmentation(self):
|
||||
# load 2 images and corresponding panoptic annotations from the hub
|
||||
dataset = load_dataset("nielsr/ade20k-panoptic-demo")
|
||||
image1 = dataset["train"][0]["image"]
|
||||
image2 = dataset["train"][1]["image"]
|
||||
segments_info1 = dataset["train"][0]["segments_info"]
|
||||
segments_info2 = dataset["train"][1]["segments_info"]
|
||||
annotation1 = dataset["train"][0]["label"]
|
||||
annotation2 = dataset["train"][1]["label"]
|
||||
|
||||
def rgb_to_id(color):
|
||||
if isinstance(color, np.ndarray) and len(color.shape) == 3:
|
||||
if color.dtype == np.uint8:
|
||||
color = color.astype(np.int32)
|
||||
return color[:, :, 0] + 256 * color[:, :, 1] + 256 * 256 * color[:, :, 2]
|
||||
return int(color[0] + 256 * color[1] + 256 * 256 * color[2])
|
||||
|
||||
def create_panoptic_map(annotation, segments_info):
|
||||
annotation = np.array(annotation)
|
||||
# convert RGB to segment IDs per pixel
|
||||
# 0 is the "ignore" label, for which we don't need to make binary masks
|
||||
panoptic_map = rgb_to_id(annotation)
|
||||
|
||||
# create mapping between segment IDs and semantic classes
|
||||
inst2class = {segment["id"]: segment["category_id"] for segment in segments_info}
|
||||
|
||||
return panoptic_map, inst2class
|
||||
|
||||
panoptic_map1, inst2class1 = create_panoptic_map(annotation1, segments_info1)
|
||||
panoptic_map2, inst2class2 = create_panoptic_map(annotation2, segments_info2)
|
||||
|
||||
image_processor = OneFormerImageProcessor(
|
||||
reduce_labels=True,
|
||||
ignore_index=0,
|
||||
size=(512, 512),
|
||||
class_info_file="ade20k_panoptic.json",
|
||||
num_text=self.processing_tester.num_text,
|
||||
)
|
||||
|
||||
tokenizer = CLIPTokenizer.from_pretrained("shi-labs/oneformer_ade20k_swin_tiny")
|
||||
|
||||
processor = OneFormerProcessor(
|
||||
image_processor=image_processor,
|
||||
tokenizer=tokenizer,
|
||||
max_seq_length=77,
|
||||
task_seq_length=77,
|
||||
)
|
||||
|
||||
# prepare the images and annotations
|
||||
pixel_values_list = [np.moveaxis(np.array(image1), -1, 0), np.moveaxis(np.array(image2), -1, 0)]
|
||||
inputs = processor.encode_inputs(
|
||||
pixel_values_list,
|
||||
["semantic", "semantic"],
|
||||
[panoptic_map1, panoptic_map2],
|
||||
instance_id_to_semantic_id=[inst2class1, inst2class2],
|
||||
return_tensors="pt",
|
||||
)
|
||||
|
||||
# verify the pixel values, task inputs, text inputs and pixel mask
|
||||
self.assertEqual(inputs["pixel_values"].shape, (2, 3, 512, 711))
|
||||
self.assertEqual(inputs["pixel_mask"].shape, (2, 512, 711))
|
||||
self.assertEqual(inputs["task_inputs"].shape, (2, 77))
|
||||
self.assertEqual(inputs["text_inputs"].shape, (2, self.processing_tester.num_text, 77))
|
||||
|
||||
# verify the class labels
|
||||
self.assertEqual(len(inputs["class_labels"]), 2)
|
||||
# fmt: off
|
||||
expected_class_labels = torch.tensor([4, 17, 32, 42, 12, 3, 5, 0, 43, 96, 104, 31, 125, 138, 87, 149]) # noqa: E231
|
||||
# fmt: on
|
||||
self.assertTrue(torch.allclose(inputs["class_labels"][0], expected_class_labels))
|
||||
# fmt: off
|
||||
expected_class_labels = torch.tensor([19, 67, 82, 17, 12, 42, 3, 14, 5, 0, 115, 43, 8, 138, 125, 143]) # noqa: E231
|
||||
# fmt: on
|
||||
self.assertTrue(torch.allclose(inputs["class_labels"][1], expected_class_labels))
|
||||
|
||||
# verify the task inputs
|
||||
self.assertEqual(len(inputs["task_inputs"]), 2)
|
||||
self.assertEqual(inputs["task_inputs"][0].sum().item(), 141082)
|
||||
self.assertEqual(inputs["task_inputs"][0].sum().item(), inputs["task_inputs"][1].sum().item())
|
||||
|
||||
# verify the text inputs
|
||||
self.assertEqual(len(inputs["text_inputs"]), 2)
|
||||
self.assertEqual(inputs["text_inputs"][0].sum().item(), 1095752)
|
||||
self.assertEqual(inputs["text_inputs"][1].sum().item(), 1062468)
|
||||
|
||||
# verify the mask labels
|
||||
self.assertEqual(len(inputs["mask_labels"]), 2)
|
||||
self.assertEqual(inputs["mask_labels"][0].shape, (16, 512, 711))
|
||||
self.assertEqual(inputs["mask_labels"][1].shape, (16, 512, 711))
|
||||
self.assertEqual(inputs["mask_labels"][0].sum().item(), 315193.0)
|
||||
self.assertEqual(inputs["mask_labels"][1].sum().item(), 350747.0)
|
||||
|
||||
def test_integration_instance_segmentation(self):
|
||||
# load 2 images and corresponding panoptic annotations from the hub
|
||||
dataset = load_dataset("nielsr/ade20k-panoptic-demo")
|
||||
image1 = dataset["train"][0]["image"]
|
||||
image2 = dataset["train"][1]["image"]
|
||||
segments_info1 = dataset["train"][0]["segments_info"]
|
||||
segments_info2 = dataset["train"][1]["segments_info"]
|
||||
annotation1 = dataset["train"][0]["label"]
|
||||
annotation2 = dataset["train"][1]["label"]
|
||||
|
||||
def rgb_to_id(color):
|
||||
if isinstance(color, np.ndarray) and len(color.shape) == 3:
|
||||
if color.dtype == np.uint8:
|
||||
color = color.astype(np.int32)
|
||||
return color[:, :, 0] + 256 * color[:, :, 1] + 256 * 256 * color[:, :, 2]
|
||||
return int(color[0] + 256 * color[1] + 256 * 256 * color[2])
|
||||
|
||||
def create_panoptic_map(annotation, segments_info):
|
||||
annotation = np.array(annotation)
|
||||
# convert RGB to segment IDs per pixel
|
||||
# 0 is the "ignore" label, for which we don't need to make binary masks
|
||||
panoptic_map = rgb_to_id(annotation)
|
||||
|
||||
# create mapping between segment IDs and semantic classes
|
||||
inst2class = {segment["id"]: segment["category_id"] for segment in segments_info}
|
||||
|
||||
return panoptic_map, inst2class
|
||||
|
||||
panoptic_map1, inst2class1 = create_panoptic_map(annotation1, segments_info1)
|
||||
panoptic_map2, inst2class2 = create_panoptic_map(annotation2, segments_info2)
|
||||
|
||||
image_processor = OneFormerImageProcessor(
|
||||
reduce_labels=True,
|
||||
ignore_index=0,
|
||||
size=(512, 512),
|
||||
class_info_file="ade20k_panoptic.json",
|
||||
num_text=self.processing_tester.num_text,
|
||||
)
|
||||
|
||||
tokenizer = CLIPTokenizer.from_pretrained("shi-labs/oneformer_ade20k_swin_tiny")
|
||||
|
||||
processor = OneFormerProcessor(
|
||||
image_processor=image_processor,
|
||||
tokenizer=tokenizer,
|
||||
max_seq_length=77,
|
||||
task_seq_length=77,
|
||||
)
|
||||
|
||||
# prepare the images and annotations
|
||||
pixel_values_list = [np.moveaxis(np.array(image1), -1, 0), np.moveaxis(np.array(image2), -1, 0)]
|
||||
inputs = processor.encode_inputs(
|
||||
pixel_values_list,
|
||||
["instance", "instance"],
|
||||
[panoptic_map1, panoptic_map2],
|
||||
instance_id_to_semantic_id=[inst2class1, inst2class2],
|
||||
return_tensors="pt",
|
||||
)
|
||||
|
||||
# verify the pixel values, task inputs, text inputs and pixel mask
|
||||
self.assertEqual(inputs["pixel_values"].shape, (2, 3, 512, 711))
|
||||
self.assertEqual(inputs["pixel_mask"].shape, (2, 512, 711))
|
||||
self.assertEqual(inputs["task_inputs"].shape, (2, 77))
|
||||
self.assertEqual(inputs["text_inputs"].shape, (2, self.processing_tester.num_text, 77))
|
||||
|
||||
# verify the class labels
|
||||
self.assertEqual(len(inputs["class_labels"]), 2)
|
||||
# fmt: off
|
||||
expected_class_labels = torch.tensor([32, 42, 42, 42, 42, 42, 42, 42, 32, 12, 12, 12, 12, 12, 42, 42, 12, 12, 12, 42, 12, 12, 12, 12, 12, 12, 12, 12, 12, 42, 42, 42, 12, 42, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 43, 43, 43, 43, 104, 43, 31, 125, 31, 125, 138, 87, 125, 149, 138, 125, 87, 87]) # noqa: E231
|
||||
# fmt: on
|
||||
self.assertTrue(torch.allclose(inputs["class_labels"][0], expected_class_labels))
|
||||
# fmt: off
|
||||
expected_class_labels = torch.tensor([19, 19, 19, 19, 19, 19, 19, 19, 19, 19, 19, 67, 82, 19, 19, 19, 19, 19, 19, 19, 19, 19, 19, 19, 12, 12, 42, 12, 12, 12, 12, 14, 12, 12, 12, 12, 12, 12, 12, 12, 14, 12, 12, 115, 43, 43, 115, 43, 43, 43, 8, 8, 8, 138, 138, 125, 143]) # noqa: E231
|
||||
# fmt: on
|
||||
self.assertTrue(torch.allclose(inputs["class_labels"][1], expected_class_labels))
|
||||
|
||||
# verify the task inputs
|
||||
self.assertEqual(len(inputs["task_inputs"]), 2)
|
||||
self.assertEqual(inputs["task_inputs"][0].sum().item(), 144985)
|
||||
self.assertEqual(inputs["task_inputs"][0].sum().item(), inputs["task_inputs"][1].sum().item())
|
||||
|
||||
# verify the text inputs
|
||||
self.assertEqual(len(inputs["text_inputs"]), 2)
|
||||
self.assertEqual(inputs["text_inputs"][0].sum().item(), 1037040)
|
||||
self.assertEqual(inputs["text_inputs"][1].sum().item(), 1044078)
|
||||
|
||||
# verify the mask labels
|
||||
self.assertEqual(len(inputs["mask_labels"]), 2)
|
||||
self.assertEqual(inputs["mask_labels"][0].shape, (73, 512, 711))
|
||||
self.assertEqual(inputs["mask_labels"][1].shape, (57, 512, 711))
|
||||
self.assertEqual(inputs["mask_labels"][0].sum().item(), 35040.0)
|
||||
self.assertEqual(inputs["mask_labels"][1].sum().item(), 98228.0)
|
||||
|
||||
def test_integration_panoptic_segmentation(self):
|
||||
# load 2 images and corresponding panoptic annotations from the hub
|
||||
dataset = load_dataset("nielsr/ade20k-panoptic-demo")
|
||||
image1 = dataset["train"][0]["image"]
|
||||
image2 = dataset["train"][1]["image"]
|
||||
segments_info1 = dataset["train"][0]["segments_info"]
|
||||
segments_info2 = dataset["train"][1]["segments_info"]
|
||||
annotation1 = dataset["train"][0]["label"]
|
||||
annotation2 = dataset["train"][1]["label"]
|
||||
|
||||
def rgb_to_id(color):
|
||||
if isinstance(color, np.ndarray) and len(color.shape) == 3:
|
||||
if color.dtype == np.uint8:
|
||||
color = color.astype(np.int32)
|
||||
return color[:, :, 0] + 256 * color[:, :, 1] + 256 * 256 * color[:, :, 2]
|
||||
return int(color[0] + 256 * color[1] + 256 * 256 * color[2])
|
||||
|
||||
def create_panoptic_map(annotation, segments_info):
|
||||
annotation = np.array(annotation)
|
||||
# convert RGB to segment IDs per pixel
|
||||
# 0 is the "ignore" label, for which we don't need to make binary masks
|
||||
panoptic_map = rgb_to_id(annotation)
|
||||
|
||||
# create mapping between segment IDs and semantic classes
|
||||
inst2class = {segment["id"]: segment["category_id"] for segment in segments_info}
|
||||
|
||||
return panoptic_map, inst2class
|
||||
|
||||
panoptic_map1, inst2class1 = create_panoptic_map(annotation1, segments_info1)
|
||||
panoptic_map2, inst2class2 = create_panoptic_map(annotation2, segments_info2)
|
||||
|
||||
image_processor = OneFormerImageProcessor(
|
||||
reduce_labels=True,
|
||||
ignore_index=0,
|
||||
size=(512, 512),
|
||||
class_info_file="ade20k_panoptic.json",
|
||||
num_text=self.processing_tester.num_text,
|
||||
)
|
||||
|
||||
tokenizer = CLIPTokenizer.from_pretrained("shi-labs/oneformer_ade20k_swin_tiny")
|
||||
|
||||
processor = OneFormerProcessor(
|
||||
image_processor=image_processor,
|
||||
tokenizer=tokenizer,
|
||||
max_seq_length=77,
|
||||
task_seq_length=77,
|
||||
)
|
||||
|
||||
# prepare the images and annotations
|
||||
pixel_values_list = [np.moveaxis(np.array(image1), -1, 0), np.moveaxis(np.array(image2), -1, 0)]
|
||||
inputs = processor.encode_inputs(
|
||||
pixel_values_list,
|
||||
["panoptic", "panoptic"],
|
||||
[panoptic_map1, panoptic_map2],
|
||||
instance_id_to_semantic_id=[inst2class1, inst2class2],
|
||||
return_tensors="pt",
|
||||
)
|
||||
|
||||
# verify the pixel values, task inputs, text inputs and pixel mask
|
||||
self.assertEqual(inputs["pixel_values"].shape, (2, 3, 512, 711))
|
||||
self.assertEqual(inputs["pixel_mask"].shape, (2, 512, 711))
|
||||
self.assertEqual(inputs["task_inputs"].shape, (2, 77))
|
||||
self.assertEqual(inputs["text_inputs"].shape, (2, self.processing_tester.num_text, 77))
|
||||
|
||||
# verify the class labels
|
||||
self.assertEqual(len(inputs["class_labels"]), 2)
|
||||
# fmt: off
|
||||
expected_class_labels = torch.tensor([4, 17, 32, 42, 42, 42, 42, 42, 42, 42, 32, 12, 12, 12, 12, 12, 42, 42, 12, 12, 12, 42, 12, 12, 12, 12, 12, 3, 12, 12, 12, 12, 42, 42, 42, 12, 42, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 5, 12, 12, 12, 12, 12, 12, 12, 0, 43, 43, 43, 96, 43, 104, 43, 31, 125, 31, 125, 138, 87, 125, 149, 138, 125, 87, 87]) # noqa: E231
|
||||
# fmt: on
|
||||
self.assertTrue(torch.allclose(inputs["class_labels"][0], expected_class_labels))
|
||||
# fmt: off
|
||||
expected_class_labels = torch.tensor([19, 19, 19, 19, 19, 19, 19, 19, 19, 19, 19, 67, 82, 19, 19, 17, 19, 19, 19, 19, 19, 19, 19, 19, 19, 12, 12, 42, 12, 12, 12, 12, 3, 14, 12, 12, 12, 12, 12, 12, 12, 12, 14, 5, 12, 12, 0, 115, 43, 43, 115, 43, 43, 43, 8, 8, 8, 138, 138, 125, 143]) # noqa: E231
|
||||
# fmt: on
|
||||
self.assertTrue(torch.allclose(inputs["class_labels"][1], expected_class_labels))
|
||||
|
||||
# verify the task inputs
|
||||
self.assertEqual(len(inputs["task_inputs"]), 2)
|
||||
self.assertEqual(inputs["task_inputs"][0].sum().item(), 136240)
|
||||
self.assertEqual(inputs["task_inputs"][0].sum().item(), inputs["task_inputs"][1].sum().item())
|
||||
|
||||
# verify the text inputs
|
||||
self.assertEqual(len(inputs["text_inputs"]), 2)
|
||||
self.assertEqual(inputs["text_inputs"][0].sum().item(), 1048653)
|
||||
self.assertEqual(inputs["text_inputs"][1].sum().item(), 1067160)
|
||||
|
||||
# verify the mask labels
|
||||
self.assertEqual(len(inputs["mask_labels"]), 2)
|
||||
self.assertEqual(inputs["mask_labels"][0].shape, (79, 512, 711))
|
||||
self.assertEqual(inputs["mask_labels"][1].shape, (61, 512, 711))
|
||||
self.assertEqual(inputs["mask_labels"][0].sum().item(), 315193.0)
|
||||
self.assertEqual(inputs["mask_labels"][1].sum().item(), 350747.0)
|
||||
|
||||
def test_binary_mask_to_rle(self):
|
||||
fake_binary_mask = np.zeros((20, 50))
|
||||
fake_binary_mask[0, 20:] = 1
|
||||
fake_binary_mask[1, :15] = 1
|
||||
fake_binary_mask[5, :10] = 1
|
||||
|
||||
rle = binary_mask_to_rle(fake_binary_mask)
|
||||
self.assertEqual(len(rle), 4)
|
||||
self.assertEqual(rle[0], 21)
|
||||
self.assertEqual(rle[1], 45)
|
||||
|
||||
def test_post_process_semantic_segmentation(self):
|
||||
image_processor = OneFormerImageProcessor(
|
||||
reduce_labels=True,
|
||||
ignore_index=0,
|
||||
size=(512, 512),
|
||||
class_info_file="ade20k_panoptic.json",
|
||||
num_text=self.processing_tester.num_text,
|
||||
)
|
||||
tokenizer = CLIPTokenizer.from_pretrained("shi-labs/oneformer_ade20k_swin_tiny")
|
||||
processor = OneFormerProcessor(
|
||||
image_processor=image_processor,
|
||||
tokenizer=tokenizer,
|
||||
max_seq_length=77,
|
||||
task_seq_length=77,
|
||||
)
|
||||
|
||||
outputs = self.processing_tester.get_fake_oneformer_outputs()
|
||||
|
||||
segmentation = processor.post_process_semantic_segmentation(outputs)
|
||||
|
||||
self.assertEqual(len(segmentation), self.processing_tester.batch_size)
|
||||
self.assertEqual(
|
||||
segmentation[0].shape,
|
||||
(
|
||||
self.processing_tester.height,
|
||||
self.processing_tester.width,
|
||||
),
|
||||
)
|
||||
|
||||
target_sizes = [(1, 4) for i in range(self.processing_tester.batch_size)]
|
||||
segmentation = processor.post_process_semantic_segmentation(outputs, target_sizes=target_sizes)
|
||||
|
||||
self.assertEqual(segmentation[0].shape, target_sizes[0])
|
||||
|
||||
def test_post_process_instance_segmentation(self):
|
||||
image_processor = OneFormerImageProcessor(
|
||||
reduce_labels=True,
|
||||
ignore_index=0,
|
||||
size=(512, 512),
|
||||
class_info_file="ade20k_panoptic.json",
|
||||
num_text=self.processing_tester.num_text,
|
||||
)
|
||||
tokenizer = CLIPTokenizer.from_pretrained("shi-labs/oneformer_ade20k_swin_tiny")
|
||||
processor = OneFormerProcessor(
|
||||
image_processor=image_processor,
|
||||
tokenizer=tokenizer,
|
||||
max_seq_length=77,
|
||||
task_seq_length=77,
|
||||
)
|
||||
|
||||
outputs = self.processing_tester.get_fake_oneformer_outputs()
|
||||
segmentation = processor.post_process_instance_segmentation(outputs, threshold=0)
|
||||
|
||||
self.assertTrue(len(segmentation) == self.processing_tester.batch_size)
|
||||
for el in segmentation:
|
||||
self.assertTrue("segmentation" in el)
|
||||
self.assertTrue("segments_info" in el)
|
||||
self.assertEqual(type(el["segments_info"]), list)
|
||||
self.assertEqual(el["segmentation"].shape, (self.processing_tester.height, self.processing_tester.width))
|
||||
|
||||
def test_post_process_panoptic_segmentation(self):
|
||||
image_processor = OneFormerImageProcessor(
|
||||
reduce_labels=True,
|
||||
ignore_index=0,
|
||||
size=(512, 512),
|
||||
class_info_file="ade20k_panoptic.json",
|
||||
num_text=self.processing_tester.num_text,
|
||||
)
|
||||
tokenizer = CLIPTokenizer.from_pretrained("shi-labs/oneformer_ade20k_swin_tiny")
|
||||
processor = OneFormerProcessor(
|
||||
image_processor=image_processor,
|
||||
tokenizer=tokenizer,
|
||||
max_seq_length=77,
|
||||
task_seq_length=77,
|
||||
)
|
||||
|
||||
outputs = self.processing_tester.get_fake_oneformer_outputs()
|
||||
segmentation = processor.post_process_panoptic_segmentation(outputs, threshold=0)
|
||||
|
||||
self.assertTrue(len(segmentation) == self.processing_tester.batch_size)
|
||||
for el in segmentation:
|
||||
self.assertTrue("segmentation" in el)
|
||||
self.assertTrue("segments_info" in el)
|
||||
self.assertEqual(type(el["segments_info"]), list)
|
||||
self.assertEqual(el["segmentation"].shape, (self.processing_tester.height, self.processing_tester.width))
|
||||
Reference in New Issue
Block a user