

from typing import Any, Dict, List, Union

from ..utils import add_end_docstrings, is_torch_available, is_vision_available, logging, requires_backends
from .base import PIPELINE_INIT_ARGS, ChunkPipeline


if is_vision_available():
		from PIL import Image

		from ..image_utils import load_image

if is_torch_available():
		import torch

		from transformers.modeling_outputs import BaseModelOutput

		from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_OBJECT_DETECTION_MAPPING

lowercase			:				Optional[Any]   =     logging.get_logger(__name__)







@add_end_docstrings(_SCREAMING_SNAKE_CASE				)
class 					__lowercase  (						_SCREAMING_SNAKE_CASE				):





							"""simple docstring"""

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

														if self.framework == "tf":
																					raise ValueError(f'The {self.__class__} is only available in PyTorch.'						)

														requires_backends(self    ,						'''vision'''						)
														self.check_model_type(__UpperCAmelCase						)




							def __call__(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase = None    ,						**__UpperCAmelCase    ,						)		->       Dict:
														if "text_queries" in kwargs:
																					A      :					Optional[int]      		=					kwargs.pop('''text_queries'''						)

														if isinstance(__UpperCAmelCase    ,						(str, Image.Image)						):
																					A      :					Optional[Any]      		=					{'''image''': image, '''candidate_labels''': candidate_labels}
														else:
																					A      :					Optional[Any]      		=					image
														A      :					int      		=					super().__call__(__UpperCAmelCase    ,						**__UpperCAmelCase						)
														return results




							def       snake_case			(					self    ,						**__UpperCAmelCase						)		->       Tuple:
														A      :					Optional[int]      		=					{}
														if "threshold" in kwargs:
																					A      :					Optional[int]      		=					kwargs['''threshold''']
														if "top_k" in kwargs:
																					A      :					int      		=					kwargs['''top_k''']
														return {}, {}, postprocess_params




							def       snake_case			(					self    ,						__UpperCAmelCase						)		->       List[Any]:
														A      :					str      		=					load_image(inputs['''image''']						)
														A      :					Optional[int]      		=					inputs['''candidate_labels''']
														if isinstance(__UpperCAmelCase    ,						__UpperCAmelCase						):
																					A      :					Optional[Any]      		=					candidate_labels.split(''','''						)

														A      :					Optional[Any]      		=					torch.tensor([[image.height, image.width]]    ,						dtype=torch.intaa						)
														for i, candidate_label in enumerate(__UpperCAmelCase						):
																					A      :					str      		=					self.tokenizer(__UpperCAmelCase    ,						return_tensors=self.framework						)
																					A      :					Optional[Any]      		=					self.image_processor(__UpperCAmelCase    ,						return_tensors=self.framework						)
																					yield {
																					    "is_last": i == len(__UpperCAmelCase						) - 1,
																					    "target_size": target_size,
																					    "candidate_label": candidate_label,
																					    **text_inputs,
																					    **image_features,
																					}




							def       snake_case			(					self    ,						__UpperCAmelCase						)		->       Tuple:
														A      :					Optional[int]      		=					model_inputs.pop('''target_size'''						)
														A      :					Any      		=					model_inputs.pop('''candidate_label'''						)
														A      :					List[Any]      		=					model_inputs.pop('''is_last'''						)

														A      :					Dict      		=					self.model(**__UpperCAmelCase						)

														A      :					Tuple      		=					{'''target_size''': target_size, '''candidate_label''': candidate_label, '''is_last''': is_last, **outputs}
														return model_outputs




							def       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase=0.1    ,						__UpperCAmelCase=None						)		->       str:
														A      :					List[Any]      		=					[]
														for model_output in model_outputs:
																					A      :					Tuple      		=					model_output['''candidate_label''']
																					A      :					Any      		=					BaseModelOutput(__UpperCAmelCase						)
																					A      :					List[Any]      		=					self.image_processor.post_process_object_detection(
																					    outputs=__UpperCAmelCase    ,						threshold=__UpperCAmelCase    ,						target_sizes=model_output['''target_size''']						)[0]

																					for index in outputs["scores"].nonzero():
																												A      :					int      		=					outputs['''scores'''][index].item()
																												A      :					int      		=					self._get_bounding_box(outputs['''boxes'''][index][0]						)

																												A      :					int      		=					{'''score''': score, '''label''': label, '''box''': box}
																												results.append(__UpperCAmelCase						)

														A      :					List[str]      		=					sorted(__UpperCAmelCase    ,						key=lambda __UpperCAmelCase						: x["score"]    ,						reverse=__UpperCAmelCase						)
														if top_k:
																					A      :					Any      		=					results[:top_k]

														return results






							def       snake_case			(					self    ,						__UpperCAmelCase						)		->       Dict[str, int]:
														if self.framework != "pt":
																					raise ValueError('''The ZeroShotObjectDetectionPipeline is only available in PyTorch.'''						)
														A		,		A		,		A		,		A      :					List[str]      		=					box.int().tolist()
														A      :					Any      		=					{
														    '''xmin''': xmin,
														    '''ymin''': ymin,
														    '''xmax''': xmax,
														    '''ymax''': ymax,
														}
														return bbox









from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import center_crop, normalize, rescale, resize, to_channel_dimension_format
from ...image_utils import (
    IMAGENET_STANDARD_MEAN,
    IMAGENET_STANDARD_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


if is_vision_available():
		import PIL


lowercase			:				Tuple   =     logging.get_logger(__name__)







class 					__lowercase  (						_SCREAMING_SNAKE_CASE				):





							"""simple docstring"""

							UpperCAmelCase_						:		Optional[Any]								=	['''pixel_values''']




							def __init__(					self    ,						__UpperCAmelCase = True    ,						__UpperCAmelCase = None    ,						__UpperCAmelCase = PIL.Image.BICUBIC    ,						__UpperCAmelCase = True    ,						__UpperCAmelCase = None    ,						__UpperCAmelCase = 1 / 2_55    ,						__UpperCAmelCase = True    ,						__UpperCAmelCase = True    ,						__UpperCAmelCase = None    ,						__UpperCAmelCase = None    ,						**__UpperCAmelCase    ,						)		->       None:
														super().__init__(**__UpperCAmelCase						)
														A      :					Any      		=					size if size is not None else {'''height''': 2_56, '''width''': 2_56}
														A      :					Any      		=					get_size_dict(__UpperCAmelCase						)
														A      :					List[Any]      		=					crop_size if crop_size is not None else {'''height''': 2_24, '''width''': 2_24}
														A      :					List[Any]      		=					get_size_dict(__UpperCAmelCase    ,						param_name='''crop_size'''						)

														A      :					Dict      		=					do_resize
														A      :					Tuple      		=					size
														A      :					Union[str, Any]      		=					resample
														A      :					Dict      		=					do_center_crop
														A      :					int      		=					crop_size
														A      :					Union[str, Any]      		=					do_rescale
														A      :					str      		=					rescale_factor
														A      :					Optional[Any]      		=					do_normalize
														A      :					Dict      		=					image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
														A      :					Dict      		=					image_std if image_std is not None else IMAGENET_STANDARD_STD




							def       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase = PIL.Image.BICUBIC    ,						__UpperCAmelCase = None    ,						**__UpperCAmelCase    ,						)		->       np.ndarray:
														A      :					int      		=					get_size_dict(__UpperCAmelCase						)
														if "height" not in size or "width" not in size:
																					raise ValueError(f'The size dictionary must have keys \'height\' and \'width\'. Got {size.keys()}'						)
														return resize(
														    __UpperCAmelCase    ,						size=(size['''height'''], size['''width'''])    ,						resample=__UpperCAmelCase    ,						data_format=__UpperCAmelCase    ,						**__UpperCAmelCase						)




							def       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase = None    ,						**__UpperCAmelCase    ,						)		->       np.ndarray:
														A      :					Optional[Any]      		=					get_size_dict(__UpperCAmelCase						)
														if "height" not in size or "width" not in size:
																					raise ValueError(f'The size dictionary must have keys \'height\' and \'width\'. Got {size.keys()}'						)
														return center_crop(__UpperCAmelCase    ,						size=(size['''height'''], size['''width'''])    ,						data_format=__UpperCAmelCase    ,						**__UpperCAmelCase						)




							def       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase = None    ,						**__UpperCAmelCase    ,						)		->       List[str]:
														return rescale(__UpperCAmelCase    ,						scale=__UpperCAmelCase    ,						data_format=__UpperCAmelCase    ,						**__UpperCAmelCase						)




							def       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase = None    ,						**__UpperCAmelCase    ,						)		->       np.ndarray:
														return normalize(__UpperCAmelCase    ,						mean=__UpperCAmelCase    ,						std=__UpperCAmelCase    ,						data_format=__UpperCAmelCase    ,						**__UpperCAmelCase						)






							def       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase = None    ,						__UpperCAmelCase = None    ,						__UpperCAmelCase=None    ,						__UpperCAmelCase = None    ,						__UpperCAmelCase = None    ,						__UpperCAmelCase = None    ,						__UpperCAmelCase = None    ,						__UpperCAmelCase = None    ,						__UpperCAmelCase = None    ,						__UpperCAmelCase = None    ,						__UpperCAmelCase = None    ,						__UpperCAmelCase = ChannelDimension.FIRST    ,						**__UpperCAmelCase    ,						)		->       PIL.Image.Image:
														A      :					Optional[int]      		=					do_resize if do_resize is not None else self.do_resize
														A      :					Optional[Any]      		=					resample if resample is not None else self.resample
														A      :					Tuple      		=					do_center_crop if do_center_crop is not None else self.do_center_crop
														A      :					Optional[Any]      		=					do_rescale if do_rescale is not None else self.do_rescale
														A      :					str      		=					rescale_factor if rescale_factor is not None else self.rescale_factor
														A      :					Tuple      		=					do_normalize if do_normalize is not None else self.do_normalize
														A      :					Optional[int]      		=					image_mean if image_mean is not None else self.image_mean
														A      :					Dict      		=					image_std if image_std is not None else self.image_std

														A      :					List[Any]      		=					size if size is not None else self.size
														A      :					List[Any]      		=					get_size_dict(__UpperCAmelCase						)
														A      :					Dict      		=					crop_size if crop_size is not None else self.crop_size
														A      :					Tuple      		=					get_size_dict(__UpperCAmelCase    ,						param_name='''crop_size'''						)

														A      :					Any      		=					make_list_of_images(__UpperCAmelCase						)

														if not valid_images(__UpperCAmelCase						):
																					raise ValueError(
																					    '''Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, '''
																					    '''torch.Tensor, tf.Tensor or jax.ndarray.'''						)

														if do_resize and size is None or resample is None:
																					raise ValueError('''Size and resample must be specified if do_resize is True.'''						)

														if do_center_crop and crop_size is None:
																					raise ValueError('''Crop size must be specified if do_center_crop is True.'''						)

														if do_rescale and rescale_factor is None:
																					raise ValueError('''Rescale factor must be specified if do_rescale is True.'''						)

														if do_normalize and (image_mean is None or image_std is None):
																					raise ValueError('''Image mean and std must be specified if do_normalize is True.'''						)

														# All transformations expect numpy arrays.
														A      :					List[Any]      		=					[to_numpy_array(__UpperCAmelCase						) for image in images]

														if do_resize:
																					A      :					List[str]      		=					[self.resize(image=__UpperCAmelCase    ,						size=__UpperCAmelCase    ,						resample=__UpperCAmelCase						) for image in images]

														if do_center_crop:
																					A      :					Dict      		=					[self.center_crop(image=__UpperCAmelCase    ,						size=__UpperCAmelCase						) for image in images]

														if do_rescale:
																					A      :					Union[str, Any]      		=					[self.rescale(image=__UpperCAmelCase    ,						scale=__UpperCAmelCase						) for image in images]

														if do_normalize:
																					A      :					Tuple      		=					[self.normalize(image=__UpperCAmelCase    ,						mean=__UpperCAmelCase    ,						std=__UpperCAmelCase						) for image in images]

														A      :					Optional[Any]      		=					[to_channel_dimension_format(__UpperCAmelCase    ,						__UpperCAmelCase						) for image in images]

														A      :					List[Any]      		=					{'''pixel_values''': images}
														return BatchFeature(data=__UpperCAmelCase    ,						tensor_type=__UpperCAmelCase						)






