



'''simple docstring'''




import json
import pathlib
import unittest

import numpy as np

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

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
							import torch

if is_vision_available():
							from PIL import Image

							from transformers import DetaImageProcessor

class 			UpperCamelCase_ (							unittest.TestCase			):





	def __init__(  self		,		A		,		A=7		,		A=3		,		A=30		,		A=400		,		A=True		,		A=None		,		A=True		,		A=[0.5, 0.5, 0.5]		,		A=[0.5, 0.5, 0.5]		,		A=True		,		A=1 / 255		,		A=True		,		)			->			Optional[Any]:
								# by setting size["longest_edge"] > max_resolution we're effectively not testing this :p
								UpperCAmelCase     :		Optional[int]							     =		size if size is not None else {"""shortest_edge""": 18, """longest_edge""": 1333}
								UpperCAmelCase     :		Optional[Any]							     =		parent
								UpperCAmelCase     :		Optional[Any]							     =		batch_size
								UpperCAmelCase     :		Any							     =		num_channels
								UpperCAmelCase     :		int							     =		min_resolution
								UpperCAmelCase     :		List[Any]							     =		max_resolution
								UpperCAmelCase     :		int							     =		do_resize
								UpperCAmelCase     :		Union[str, Any]							     =		size
								UpperCAmelCase     :		Any							     =		do_normalize
								UpperCAmelCase     :		str							     =		image_mean
								UpperCAmelCase     :		Any							     =		image_std
								UpperCAmelCase     :		Union[str, Any]							     =		do_rescale
								UpperCAmelCase     :		List[str]							     =		rescale_factor
								UpperCAmelCase     :		List[Any]							     =		do_pad





	def  _lowercase(  self    )			->			List[str]:
								return {
								    "do_resize": self.do_resize,
								    "size": self.size,
								    "do_normalize": self.do_normalize,
								    "image_mean": self.image_mean,
								    "image_std": self.image_std,
								    "do_rescale": self.do_rescale,
								    "rescale_factor": self.rescale_factor,
								    "do_pad": self.do_pad,
								}






	def  _lowercase(  self		,		A		,		A=False    )			->			Dict:
								if not batched:
															UpperCAmelCase     :		str							     =		image_inputs[0]
															if isinstance(A		,		Image.Image    ):
																						UpperCAmelCase     ,     UpperCAmelCase     :		Tuple							     =		image.size
															else:
																						UpperCAmelCase     ,     UpperCAmelCase     :		str							     =		image.shape[1], image.shape[2]
															if w < h:
																						UpperCAmelCase     :		Tuple							     =		int(self.size["""shortest_edge"""] * h / w    )
																						UpperCAmelCase     :		Any							     =		self.size["""shortest_edge"""]
															elif w > h:
																						UpperCAmelCase     :		Tuple							     =		self.size["""shortest_edge"""]
																						UpperCAmelCase     :		int							     =		int(self.size["""shortest_edge"""] * w / h    )
															else:
																						UpperCAmelCase     :		Any							     =		self.size["""shortest_edge"""]
																						UpperCAmelCase     :		Optional[int]							     =		self.size["""shortest_edge"""]

								else:
															UpperCAmelCase     :		int							     =		[]
															for image in image_inputs:
																						UpperCAmelCase     ,     UpperCAmelCase     :		Union[str, Any]							     =		self.get_expected_values([image]    )
																						expected_values.append((expected_height, expected_width)    )
															UpperCAmelCase     :		List[Any]							     =		max(A		,		key=lambda A    : item[0]    )[0]
															UpperCAmelCase     :		Dict							     =		max(A		,		key=lambda A    : item[1]    )[1]

								return expected_height, expected_width




@require_torch
@require_vision
class 			UpperCamelCase_ (							__magic_name__      ,     unittest.TestCase			):
	lowercase          = DetaImageProcessor if is_vision_available() else None





	def  _lowercase(  self    )			->			Any:
								UpperCAmelCase     :		str							     =		DetaImageProcessingTester(self    )





	@property
	def  _lowercase(  self    )			->			Optional[int]:
								return self.image_processor_tester.prepare_image_processor_dict()





	def  _lowercase(  self    )			->			List[str]:
								UpperCAmelCase     :		int							     =		self.image_processing_class(**self.image_processor_dict    )
								self.assertTrue(hasattr(A		,		"""image_mean"""    )    )
								self.assertTrue(hasattr(A		,		"""image_std"""    )    )
								self.assertTrue(hasattr(A		,		"""do_normalize"""    )    )
								self.assertTrue(hasattr(A		,		"""do_resize"""    )    )
								self.assertTrue(hasattr(A		,		"""do_rescale"""    )    )
								self.assertTrue(hasattr(A		,		"""do_pad"""    )    )
								self.assertTrue(hasattr(A		,		"""size"""    )    )





	def  _lowercase(  self    )			->			Tuple:
								UpperCAmelCase     :		str							     =		self.image_processing_class.from_dict(self.image_processor_dict    )
								self.assertEqual(image_processor.size		,		{"""shortest_edge""": 18, """longest_edge""": 1333}    )
								self.assertEqual(image_processor.do_pad		,		A    )





	def  _lowercase(  self    )			->			int:
								pass





	def  _lowercase(  self    )			->			Dict:
								# Initialize image_processing
								UpperCAmelCase     :		Any							     =		self.image_processing_class(**self.image_processor_dict    )
								# create random PIL images
								UpperCAmelCase     :		Union[str, Any]							     =		prepare_image_inputs(self.image_processor_tester		,		equal_resolution=A    )
								for image in image_inputs:
															self.assertIsInstance(A		,		Image.Image    )

								# Test not batched input
								UpperCAmelCase     :		Optional[int]							     =		image_processing(image_inputs[0]		,		return_tensors="""pt"""    ).pixel_values

								UpperCAmelCase     ,     UpperCAmelCase     :		Any							     =		self.image_processor_tester.get_expected_values(A    )

								self.assertEqual(
								    encoded_images.shape		,		(1, self.image_processor_tester.num_channels, expected_height, expected_width)		,		)

								# Test batched
								UpperCAmelCase     ,     UpperCAmelCase     :		Dict							     =		self.image_processor_tester.get_expected_values(A		,		batched=A    )

								UpperCAmelCase     :		Union[str, Any]							     =		image_processing(A		,		return_tensors="""pt"""    ).pixel_values
								self.assertEqual(
								    encoded_images.shape		,		(
								        self.image_processor_tester.batch_size,
								        self.image_processor_tester.num_channels,
								        expected_height,
								        expected_width,
								    )		,		)





	def  _lowercase(  self    )			->			List[Any]:
								# Initialize image_processing
								UpperCAmelCase     :		Optional[Any]							     =		self.image_processing_class(**self.image_processor_dict    )
								# create random numpy tensors
								UpperCAmelCase     :		Union[str, Any]							     =		prepare_image_inputs(self.image_processor_tester		,		equal_resolution=A		,		numpify=A    )
								for image in image_inputs:
															self.assertIsInstance(A		,		np.ndarray    )

								# Test not batched input
								UpperCAmelCase     :		str							     =		image_processing(image_inputs[0]		,		return_tensors="""pt"""    ).pixel_values

								UpperCAmelCase     ,     UpperCAmelCase     :		List[Any]							     =		self.image_processor_tester.get_expected_values(A    )

								self.assertEqual(
								    encoded_images.shape		,		(1, self.image_processor_tester.num_channels, expected_height, expected_width)		,		)

								# Test batched
								UpperCAmelCase     :		List[str]							     =		image_processing(A		,		return_tensors="""pt"""    ).pixel_values

								UpperCAmelCase     ,     UpperCAmelCase     :		List[Any]							     =		self.image_processor_tester.get_expected_values(A		,		batched=A    )

								self.assertEqual(
								    encoded_images.shape		,		(
								        self.image_processor_tester.batch_size,
								        self.image_processor_tester.num_channels,
								        expected_height,
								        expected_width,
								    )		,		)





	def  _lowercase(  self    )			->			str:
								# Initialize image_processing
								UpperCAmelCase     :		str							     =		self.image_processing_class(**self.image_processor_dict    )
								# create random PyTorch tensors
								UpperCAmelCase     :		str							     =		prepare_image_inputs(self.image_processor_tester		,		equal_resolution=A		,		torchify=A    )
								for image in image_inputs:
															self.assertIsInstance(A		,		torch.Tensor    )

								# Test not batched input
								UpperCAmelCase     :		List[Any]							     =		image_processing(image_inputs[0]		,		return_tensors="""pt"""    ).pixel_values

								UpperCAmelCase     ,     UpperCAmelCase     :		int							     =		self.image_processor_tester.get_expected_values(A    )

								self.assertEqual(
								    encoded_images.shape		,		(1, self.image_processor_tester.num_channels, expected_height, expected_width)		,		)

								# Test batched
								UpperCAmelCase     :		Dict							     =		image_processing(A		,		return_tensors="""pt"""    ).pixel_values

								UpperCAmelCase     ,     UpperCAmelCase     :		Dict							     =		self.image_processor_tester.get_expected_values(A		,		batched=A    )

								self.assertEqual(
								    encoded_images.shape		,		(
								        self.image_processor_tester.batch_size,
								        self.image_processor_tester.num_channels,
								        expected_height,
								        expected_width,
								    )		,		)





	@slow
	def  _lowercase(  self    )			->			List[str]:
								# prepare image and target
								UpperCAmelCase     :		Union[str, Any]							     =		Image.open("""./tests/fixtures/tests_samples/COCO/000000039769.png"""    )
								with open("""./tests/fixtures/tests_samples/COCO/coco_annotations.txt"""		,		"""r"""    ) as f:
															UpperCAmelCase     :		str							     =		json.loads(f.read()    )

								UpperCAmelCase     :		List[str]							     =		{"""image_id""": 39769, """annotations""": target}

								# encode them
								UpperCAmelCase     :		Union[str, Any]							     =		DetaImageProcessor()
								UpperCAmelCase     :		Tuple							     =		image_processing(images=A		,		annotations=A		,		return_tensors="""pt"""    )

								# verify pixel values
								UpperCAmelCase     :		List[str]							     =		torch.Size([1, 3, 800, 1066]    )
								self.assertEqual(encoding["""pixel_values"""].shape		,		A    )

								UpperCAmelCase     :		Union[str, Any]							     =		torch.tensor([0.2_7_9_6, 0.3_1_3_8, 0.3_4_8_1]    )
								self.assertTrue(torch.allclose(encoding["""pixel_values"""][0, 0, 0, :3]		,		A		,		atol=1e-4    )    )

								# verify area
								UpperCAmelCase     :		Optional[int]							     =		torch.tensor([5_8_8_7.9_6_0_0, 1_1_2_5_0.2_0_6_1, 4_8_9_3_5_3.8_4_3_8, 8_3_7_1_2_2.7_5_0_0, 1_4_7_9_6_7.5_1_5_6, 1_6_5_7_3_2.3_4_3_8]    )
								self.assertTrue(torch.allclose(encoding["""labels"""][0]["""area"""]		,		A    )    )
								# verify boxes
								UpperCAmelCase     :		Optional[int]							     =		torch.Size([6, 4]    )
								self.assertEqual(encoding["""labels"""][0]["""boxes"""].shape		,		A    )
								UpperCAmelCase     :		str							     =		torch.tensor([0.5_5_0_3, 0.2_7_6_5, 0.0_6_0_4, 0.2_2_1_5]    )
								self.assertTrue(torch.allclose(encoding["""labels"""][0]["""boxes"""][0]		,		A		,		atol=1e-3    )    )
								# verify image_id
								UpperCAmelCase     :		Tuple							     =		torch.tensor([39769]    )
								self.assertTrue(torch.allclose(encoding["""labels"""][0]["""image_id"""]		,		A    )    )
								# verify is_crowd
								UpperCAmelCase     :		List[str]							     =		torch.tensor([0, 0, 0, 0, 0, 0]    )
								self.assertTrue(torch.allclose(encoding["""labels"""][0]["""iscrowd"""]		,		A    )    )
								# verify class_labels
								UpperCAmelCase     :		Dict							     =		torch.tensor([75, 75, 63, 65, 17, 17]    )
								self.assertTrue(torch.allclose(encoding["""labels"""][0]["""class_labels"""]		,		A    )    )
								# verify orig_size
								UpperCAmelCase     :		int							     =		torch.tensor([480, 640]    )
								self.assertTrue(torch.allclose(encoding["""labels"""][0]["""orig_size"""]		,		A    )    )
								# verify size
								UpperCAmelCase     :		Dict							     =		torch.tensor([800, 1066]    )
								self.assertTrue(torch.allclose(encoding["""labels"""][0]["""size"""]		,		A    )    )






	@slow
	def  _lowercase(  self    )			->			int:
								# prepare image, target and masks_path
								UpperCAmelCase     :		List[str]							     =		Image.open("""./tests/fixtures/tests_samples/COCO/000000039769.png"""    )
								with open("""./tests/fixtures/tests_samples/COCO/coco_panoptic_annotations.txt"""		,		"""r"""    ) as f:
															UpperCAmelCase     :		Any							     =		json.loads(f.read()    )

								UpperCAmelCase     :		int							     =		{"""file_name""": """000000039769.png""", """image_id""": 39769, """segments_info""": target}

								UpperCAmelCase     :		Union[str, Any]							     =		pathlib.Path("""./tests/fixtures/tests_samples/COCO/coco_panoptic"""    )

								# encode them
								UpperCAmelCase     :		Optional[Any]							     =		DetaImageProcessor(format="""coco_panoptic"""    )
								UpperCAmelCase     :		Union[str, Any]							     =		image_processing(images=A		,		annotations=A		,		masks_path=A		,		return_tensors="""pt"""    )

								# verify pixel values
								UpperCAmelCase     :		Union[str, Any]							     =		torch.Size([1, 3, 800, 1066]    )
								self.assertEqual(encoding["""pixel_values"""].shape		,		A    )

								UpperCAmelCase     :		Union[str, Any]							     =		torch.tensor([0.2_7_9_6, 0.3_1_3_8, 0.3_4_8_1]    )
								self.assertTrue(torch.allclose(encoding["""pixel_values"""][0, 0, 0, :3]		,		A		,		atol=1e-4    )    )

								# verify area
								UpperCAmelCase     :		int							     =		torch.tensor([1_4_7_9_7_9.6_8_7_5, 1_6_5_5_2_7.0_4_6_9, 4_8_4_6_3_8.5_9_3_8, 1_1_2_9_2.9_3_7_5, 5_8_7_9.6_5_6_2, 7_6_3_4.1_1_4_7]    )
								self.assertTrue(torch.allclose(encoding["""labels"""][0]["""area"""]		,		A    )    )
								# verify boxes
								UpperCAmelCase     :		List[Any]							     =		torch.Size([6, 4]    )
								self.assertEqual(encoding["""labels"""][0]["""boxes"""].shape		,		A    )
								UpperCAmelCase     :		Tuple							     =		torch.tensor([0.2_6_2_5, 0.5_4_3_7, 0.4_6_8_8, 0.8_6_2_5]    )
								self.assertTrue(torch.allclose(encoding["""labels"""][0]["""boxes"""][0]		,		A		,		atol=1e-3    )    )
								# verify image_id
								UpperCAmelCase     :		Dict							     =		torch.tensor([39769]    )
								self.assertTrue(torch.allclose(encoding["""labels"""][0]["""image_id"""]		,		A    )    )
								# verify is_crowd
								UpperCAmelCase     :		Optional[int]							     =		torch.tensor([0, 0, 0, 0, 0, 0]    )
								self.assertTrue(torch.allclose(encoding["""labels"""][0]["""iscrowd"""]		,		A    )    )
								# verify class_labels
								UpperCAmelCase     :		List[Any]							     =		torch.tensor([17, 17, 63, 75, 75, 93]    )
								self.assertTrue(torch.allclose(encoding["""labels"""][0]["""class_labels"""]		,		A    )    )
								# verify masks
								UpperCAmelCase     :		Any							     =		822873
								self.assertEqual(encoding["""labels"""][0]["""masks"""].sum().item()		,		A    )
								# verify orig_size
								UpperCAmelCase     :		Any							     =		torch.tensor([480, 640]    )
								self.assertTrue(torch.allclose(encoding["""labels"""][0]["""orig_size"""]		,		A    )    )
								# verify size
								UpperCAmelCase     :		Any							     =		torch.tensor([800, 1066]    )
								self.assertTrue(torch.allclose(encoding["""labels"""][0]["""size"""]		,		A    )    )






'''simple docstring'''




from collections import UserDict
from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
							from PIL import Image

							from ..image_utils import load_image

if is_torch_available():
							from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING

if is_tf_available():
							from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
							from ..tf_utils import stable_softmax

a   :			Union[str, Any]					  =    logging.get_logger(__name__)

@add_end_docstrings(__magic_name__			)
class 			UpperCamelCase_ (							__magic_name__			):





	def __init__(  self		,		**A    )			->			List[str]:
								super().__init__(**A    )

								requires_backends(self		,		"""vision"""    )
								self.check_model_type(
								    TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
								    if self.framework == """tf"""
								    else MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING    )





	def __call__(  self		,		A		,		**A    )			->			Optional[Any]:
								return super().__call__(A		,		**A    )





	def  _lowercase(  self		,		**A    )			->			Optional[Any]:
								UpperCAmelCase     :		List[Any]							     =		{}
								if "candidate_labels" in kwargs:
															UpperCAmelCase     :		Dict							     =		kwargs["""candidate_labels"""]
								if "hypothesis_template" in kwargs:
															UpperCAmelCase     :		Optional[Any]							     =		kwargs["""hypothesis_template"""]

								return preprocess_params, {}, {}





	def  _lowercase(  self		,		A		,		A=None		,		A="This is a photo of {}."    )			->			Optional[Any]:
								UpperCAmelCase     :		int							     =		load_image(A    )
								UpperCAmelCase     :		List[str]							     =		self.image_processor(images=[image]		,		return_tensors=self.framework    )
								UpperCAmelCase     :		List[str]							     =		candidate_labels
								UpperCAmelCase     :		Tuple							     =		[hypothesis_template.format(A    ) for x in candidate_labels]
								UpperCAmelCase     :		Union[str, Any]							     =		self.tokenizer(A		,		return_tensors=self.framework		,		padding=A    )
								UpperCAmelCase     :		Union[str, Any]							     =		[text_inputs]
								return inputs





	def  _lowercase(  self		,		A    )			->			Optional[int]:
								UpperCAmelCase     :		List[Any]							     =		model_inputs.pop("""candidate_labels"""    )
								UpperCAmelCase     :		Optional[Any]							     =		model_inputs.pop("""text_inputs"""    )
								if isinstance(text_inputs[0]		,		A    ):
															UpperCAmelCase     :		Optional[Any]							     =		text_inputs[0]
								else:
															# Batching case.
															UpperCAmelCase     :		Any							     =		text_inputs[0][0]

								UpperCAmelCase     :		Dict							     =		self.model(**A		,		**A    )

								UpperCAmelCase     :		List[Any]							     =		{
								    """candidate_labels""": candidate_labels,
								    """logits""": outputs.logits_per_image,
								}
								return model_outputs






	def  _lowercase(  self		,		A    )			->			Union[str, Any]:
								UpperCAmelCase     :		int							     =		model_outputs.pop("""candidate_labels"""    )
								UpperCAmelCase     :		int							     =		model_outputs["""logits"""][0]
								if self.framework == "pt":
															UpperCAmelCase     :		Optional[int]							     =		logits.softmax(dim=-1    ).squeeze(-1    )
															UpperCAmelCase     :		Any							     =		probs.tolist()
															if not isinstance(A		,		A    ):
																						UpperCAmelCase     :		Any							     =		[scores]
								elif self.framework == "tf":
															UpperCAmelCase     :		List[str]							     =		stable_softmax(A		,		axis=-1    )
															UpperCAmelCase     :		Union[str, Any]							     =		probs.numpy().tolist()
								else:
															raise ValueError(f'''Unsupported framework: {self.framework}'''    )

								UpperCAmelCase     :		Any							     =		[
								    {"""score""": score, """label""": candidate_label}
								    for score, candidate_label in sorted(zip(A		,		A    )		,		key=lambda A    : -x[0]    )
								]
								return result

