

from typing import List, Union

import numpy as np

from ..tokenization_utils import TruncationStrategy
from ..utils import add_end_docstrings, logging
from .base import PIPELINE_INIT_ARGS, ArgumentHandler, ChunkPipeline


_lowerCamelCase    :       List[Any]        =							logging.get_logger(__name__)





class  __UpperCAmelCase					(	lowerCamelCase__						):



			def 	__magic_name__   (							self							:					str,		__A							:					Optional[int]						):
					if isinstance(__A,		__A						):
							UpperCAmelCase	:							str						=			[label.strip() for label in labels.split(''','''						) if label.strip()]
					return labels





			def __call__(							self							:					Any,		__A							:					Any,		__A							:					Tuple,		__A							:					Dict						):
					if len(__A						) == 0 or len(__A						) == 0:
							raise ValueError('''You must include at least one label and at least one sequence.'''						)
					if hypothesis_template.format(labels[0]						) == hypothesis_template:
							raise ValueError(
							    (
							        '''The provided hypothesis_template "{}" was not able to be formatted with the target labels. '''
							        '''Make sure the passed template includes formatting syntax such as {{}} where the label should go.'''
							    ).format(__A						)						)

					if isinstance(__A,		__A						):
							UpperCAmelCase	:							int						=			[sequences]

					UpperCAmelCase	:							Dict						=			[]
					for sequence in sequences:
							sequence_pairs.extend([[sequence, hypothesis_template.format(__A						)] for label in labels]						)

					return sequence_pairs, sequences






@add_end_docstrings(lowerCamelCase__						)
class  __UpperCAmelCase					(	lowerCamelCase__						):



			def __init__(							self							:					Union[str, Any],		__A							:					List[str]=ZeroShotClassificationArgumentHandler(),		*__A							:					Optional[int],		**__A							:					Optional[Any]						):
					UpperCAmelCase	:							Optional[Any]						=			args_parser
					super().__init__(*__A,		**__A						)
					if self.entailment_id == -1:
							logger.warning(
							    '''Failed to determine \'entailment\' label id from the label2id mapping in the model config. Setting to '''
							    '''-1. Define a descriptive label2id mapping in the model config to ensure correct outputs.'''						)



			@property
			def 	__magic_name__   (							self							:					Union[str, Any]						):
					for label, ind in self.model.config.labelaid.items():
							if label.lower().startswith('''entail'''						):
									return ind
					return -1



			def 	__magic_name__   (							self							:					str,		__A							:					int,		__A							:					List[Any]=True,		__A							:					Any=True,		__A							:					List[str]=TruncationStrategy.ONLY_FIRST,		**__A							:					List[Any]						):
					UpperCAmelCase	:							Dict						=			self.framework
					if self.tokenizer.pad_token is None:
							# Override for tokenizers not supporting padding
							logger.error(
							    '''Tokenizer was not supporting padding necessary for zero-shot, attempting to use '''
							    ''' `pad_token=eos_token`'''						)
							UpperCAmelCase	:							Optional[int]						=			self.tokenizer.eos_token
					try:
							UpperCAmelCase	:							int						=			self.tokenizer(
							    __A,		add_special_tokens=__A,		return_tensors=__A,		padding=__A,		truncation=__A,		)
					except Exception as e:
							if "too short" in str(__A						):
									# tokenizers might yell that we want to truncate
									# to a value that is not even reached by the input.
									# In that case we don't want to truncate.
									# It seems there's not a really better way to catch that
									# exception.

									UpperCAmelCase	:							Dict						=			self.tokenizer(
									    __A,		add_special_tokens=__A,		return_tensors=__A,		padding=__A,		truncation=TruncationStrategy.DO_NOT_TRUNCATE,		)
							else:
									raise e

					return inputs



			def 	__magic_name__   (							self							:					str,		**__A							:					int						):
					if kwargs.get('''multi_class''',		__A						) is not None:
							UpperCAmelCase	:							Any						=			kwargs['''multi_class''']
							logger.warning(
							    '''The `multi_class` argument has been deprecated and renamed to `multi_label`. '''
							    '''`multi_class` will be removed in a future version of Transformers.'''						)
					UpperCAmelCase	:							Tuple						=			{}
					if "candidate_labels" in kwargs:
							UpperCAmelCase	:							List[str]						=			self._args_parser._parse_labels(kwargs['''candidate_labels''']						)
					if "hypothesis_template" in kwargs:
							UpperCAmelCase	:							int						=			kwargs['''hypothesis_template''']

					UpperCAmelCase	:							int						=			{}
					if "multi_label" in kwargs:
							UpperCAmelCase	:							List[str]						=			kwargs['''multi_label''']
					return preprocess_params, {}, postprocess_params



			def __call__(							self							:					List[Any],		__A							:					Union[str, List[str]],		*__A							:					Tuple,		**__A							:					Union[str, Any],		):
					if len(__A						) == 0:
							pass
					elif len(__A						) == 1 and "candidate_labels" not in kwargs:
							UpperCAmelCase	:							int						=			args[0]
					else:
							raise ValueError(F'''Unable to understand extra arguments {args}'''						)

					return super().__call__(__A,		**__A						)



			def 	__magic_name__   (							self							:					List[Any],		__A							:					Tuple,		__A							:					Optional[Any]=None,		__A							:					List[str]="This example is {}."						):
					UpperCAmelCase	:							List[Any]						=			self._args_parser(__A,		__A,		__A						)

					for i, (candidate_label, sequence_pair) in enumerate(zip(__A,		__A						)						):
							UpperCAmelCase	:							Dict						=			self._parse_and_tokenize([sequence_pair]						)

							yield {
							    "candidate_label": candidate_label,
							    "sequence": sequences[0],
							    "is_last": i == len(__A						) - 1,
							    **model_input,
							}



			def 	__magic_name__   (							self							:					Dict,		__A							:					Dict						):
					UpperCAmelCase	:							str						=			inputs['''candidate_label''']
					UpperCAmelCase	:							List[Any]						=			inputs['''sequence''']
					UpperCAmelCase	:							int						=			{k: inputs[k] for k in self.tokenizer.model_input_names}
					UpperCAmelCase	:							List[Any]						=			self.model(**__A						)

					UpperCAmelCase	:							List[str]						=			{
					    '''candidate_label''': candidate_label,
					    '''sequence''': sequence,
					    '''is_last''': inputs['''is_last'''],
					    **outputs,
					}
					return model_outputs





			def 	__magic_name__   (							self							:					List[str],		__A							:					Optional[Any],		__A							:					int=False						):
					UpperCAmelCase	:							Union[str, Any]						=			[outputs['''candidate_label'''] for outputs in model_outputs]
					UpperCAmelCase	:							Union[str, Any]						=			[outputs['''sequence'''] for outputs in model_outputs]
					UpperCAmelCase	:							Any						=			np.concatenate([output['''logits'''].numpy() for output in model_outputs]						)
					UpperCAmelCase	:							List[Any]						=			logits.shape[0]
					UpperCAmelCase	:							Union[str, Any]						=			len(__A						)
					UpperCAmelCase	:							str						=			N // n
					UpperCAmelCase	:							Dict						=			logits.reshape((num_sequences, n, -1)						)

					if multi_label or len(__A						) == 1:
							# softmax over the entailment vs. contradiction dim for each label independently
							UpperCAmelCase	:							Union[str, Any]						=			self.entailment_id
							UpperCAmelCase	:							List[str]						=			-1 if entailment_id == 0 else 0
							UpperCAmelCase	:							Tuple						=			reshaped_outputs[..., [contradiction_id, entailment_id]]
							UpperCAmelCase	:							Dict						=			np.exp(__A						) / np.exp(__A						).sum(-1,		keepdims=__A						)
							UpperCAmelCase	:							Dict						=			scores[..., 1]
					else:
							# softmax the "entailment" logits over all candidate labels
							UpperCAmelCase	:							List[str]						=			reshaped_outputs[..., self.entailment_id]
							UpperCAmelCase	:							str						=			np.exp(__A						) / np.exp(__A						).sum(-1,		keepdims=__A						)

					UpperCAmelCase	:							Any						=			list(reversed(scores[0].argsort()						)						)
					return {
					    "sequence": sequences[0],
					    "labels": [candidate_labels[i] for i in top_inds],
					    "scores": scores[0, top_inds].tolist(),
					}








import gc
import random
import unittest

import numpy as np
import torch
from transformers import CLIPImageProcessor, CLIPVisionConfig, CLIPVisionModel

from diffusers import HeunDiscreteScheduler, PriorTransformer, ShapEImgaImgPipeline
from diffusers.pipelines.shap_e import ShapERenderer
from diffusers.utils import floats_tensor, load_image, load_numpy, slow
from diffusers.utils.testing_utils import require_torch_gpu, torch_device

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference





class  __UpperCAmelCase					(	lowerCamelCase__      ,  unittest.TestCase						):
   UpperCamelCase       					=					ShapEImgaImgPipeline
   UpperCamelCase       					=					["""image"""]
   UpperCamelCase       					=					["""image"""]
   UpperCamelCase       					=					[
       """num_images_per_prompt""",
       """num_inference_steps""",
       """generator""",
       """latents""",
       """guidance_scale""",
       """frame_size""",
       """output_type""",
       """return_dict""",
   ]
   UpperCamelCase       					=					False



   @property
   def 	__magic_name__   (							self							:					Optional[Any]						):
     return 3_2



   @property
   def 	__magic_name__   (							self							:					Optional[int]						):
     return 3_2



   @property
   def 	__magic_name__   (							self							:					Union[str, Any]						):
     return self.time_input_dim * 4



   @property
   def 	__magic_name__   (							self							:					Any						):
     return 8



   @property
   def 	__magic_name__   (							self							:					List[str]						):
     torch.manual_seed(0						)
     UpperCAmelCase	:							str						=			CLIPVisionConfig(
         hidden_size=self.text_embedder_hidden_size,		image_size=6_4,		projection_dim=self.text_embedder_hidden_size,		intermediate_size=3_7,		num_attention_heads=4,		num_channels=3,		num_hidden_layers=5,		patch_size=1,		)

     UpperCAmelCase	:							Any						=			CLIPVisionModel(__A						)
     return model



   @property
   def 	__magic_name__   (							self							:					int						):
     UpperCAmelCase	:							Optional[int]						=			CLIPImageProcessor(
         crop_size=2_2_4,		do_center_crop=__A,		do_normalize=__A,		do_resize=__A,		image_mean=[0.4_8_1_4_5_4_6_6, 0.4_5_7_8_2_7_5, 0.4_0_8_2_1_0_7_3],		image_std=[0.2_6_8_6_2_9_5_4, 0.2_6_1_3_0_2_5_8, 0.2_7_5_7_7_7_1_1],		resample=3,		size=2_2_4,		)

     return image_processor



   @property
   def 	__magic_name__   (							self							:					Dict						):
     torch.manual_seed(0						)

     UpperCAmelCase	:							Any						=			{
         '''num_attention_heads''': 2,
         '''attention_head_dim''': 1_6,
         '''embedding_dim''': self.time_input_dim,
         '''num_embeddings''': 3_2,
         '''embedding_proj_dim''': self.text_embedder_hidden_size,
         '''time_embed_dim''': self.time_embed_dim,
         '''num_layers''': 1,
         '''clip_embed_dim''': self.time_input_dim * 2,
         '''additional_embeddings''': 0,
         '''time_embed_act_fn''': '''gelu''',
         '''norm_in_type''': '''layer''',
         '''embedding_proj_norm_type''': '''layer''',
         '''encoder_hid_proj_type''': None,
         '''added_emb_type''': None,
     }

     UpperCAmelCase	:							List[Any]						=			PriorTransformer(**__A						)
     return model



   @property
   def 	__magic_name__   (							self							:					Union[str, Any]						):
     torch.manual_seed(0						)

     UpperCAmelCase	:							List[Any]						=			{
         '''param_shapes''': (
             (self.renderer_dim, 9_3),
             (self.renderer_dim, 8),
             (self.renderer_dim, 8),
             (self.renderer_dim, 8),
         ),
         '''d_latent''': self.time_input_dim,
         '''d_hidden''': self.renderer_dim,
         '''n_output''': 1_2,
         '''background''': (
             0.1,
             0.1,
             0.1,
         ),
     }
     UpperCAmelCase	:							List[str]						=			ShapERenderer(**__A						)
     return model



   def 	__magic_name__   (							self							:					List[Any]						):
     UpperCAmelCase	:							str						=			self.dummy_prior
     UpperCAmelCase	:							List[str]						=			self.dummy_image_encoder
     UpperCAmelCase	:							List[Any]						=			self.dummy_image_processor
     UpperCAmelCase	:							Dict						=			self.dummy_renderer

     UpperCAmelCase	:							int						=			HeunDiscreteScheduler(
         beta_schedule='''exp''',		num_train_timesteps=1_0_2_4,		prediction_type='''sample''',		use_karras_sigmas=__A,		clip_sample=__A,		clip_sample_range=1.0,		)
     UpperCAmelCase	:							List[str]						=			{
         '''prior''': prior,
         '''image_encoder''': image_encoder,
         '''image_processor''': image_processor,
         '''renderer''': renderer,
         '''scheduler''': scheduler,
     }

     return components



   def 	__magic_name__   (							self							:					List[Any],		__A							:					Dict,		__A							:					List[Any]=0						):
     UpperCAmelCase	:							int						=			floats_tensor((1, 3, 6_4, 6_4),		rng=random.Random(__A						)						).to(__A						)

     if str(__A						).startswith('''mps'''						):
       UpperCAmelCase	:							List[str]						=			torch.manual_seed(__A						)
     else:
       UpperCAmelCase	:							Optional[Any]						=			torch.Generator(device=__A						).manual_seed(__A						)
     UpperCAmelCase	:							Dict						=			{
         '''image''': input_image,
         '''generator''': generator,
         '''num_inference_steps''': 1,
         '''frame_size''': 3_2,
         '''output_type''': '''np''',
     }
     return inputs



   def 	__magic_name__   (							self							:					str						):
     UpperCAmelCase	:							Dict						=			'''cpu'''

     UpperCAmelCase	:							Any						=			self.get_dummy_components()

     UpperCAmelCase	:							Tuple						=			self.pipeline_class(**__A						)
     UpperCAmelCase	:							Dict						=			pipe.to(__A						)

     pipe.set_progress_bar_config(disable=__A						)

     UpperCAmelCase	:							List[Any]						=			pipe(**self.get_dummy_inputs(__A						)						)
     UpperCAmelCase	:							Any						=			output.images[0]
     UpperCAmelCase	:							List[str]						=			image[0, -3:, -3:, -1]

     assert image.shape == (2_0, 3_2, 3_2, 3)

     UpperCAmelCase	:							Optional[int]						=			np.array(
         [
             0.0_0_0_3_9_2_1_6,
             0.0_0_0_3_9_2_1_6,
             0.0_0_0_3_9_2_1_6,
             0.0_0_0_3_9_2_1_6,
             0.0_0_0_3_9_2_1_6,
             0.0_0_0_3_9_2_1_6,
             0.0_0_0_3_9_2_1_6,
             0.0_0_0_3_9_2_1_6,
             0.0_0_0_3_9_2_1_6,
         ]						)

     assert np.abs(image_slice.flatten() - expected_slice						).max() < 1E-2



   def 	__magic_name__   (							self							:					Dict						):
     # NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
     self._test_inference_batch_consistent(batch_sizes=[1, 2]						)



   def 	__magic_name__   (							self							:					str						):
     UpperCAmelCase	:							Tuple						=			torch_device == '''cpu'''
     UpperCAmelCase	:							List[Any]						=			True
     self._test_inference_batch_single_identical(
         batch_size=2,		test_max_difference=__A,		relax_max_difference=__A,		)





   def 	__magic_name__   (							self							:					Dict						):
     UpperCAmelCase	:							Dict						=			self.get_dummy_components()
     UpperCAmelCase	:							Union[str, Any]						=			self.pipeline_class(**__A						)
     UpperCAmelCase	:							List[str]						=			pipe.to(__A						)
     pipe.set_progress_bar_config(disable=__A						)

     UpperCAmelCase	:							Any						=			1
     UpperCAmelCase	:							Any						=			2

     UpperCAmelCase	:							Optional[int]						=			self.get_dummy_inputs(__A						)

     for key in inputs.keys():
       if key in self.batch_params:
         UpperCAmelCase	:							Any						=			batch_size * [inputs[key]]

     UpperCAmelCase	:							int						=			pipe(**__A,		num_images_per_prompt=__A						)[0]

     assert images.shape[0] == batch_size * num_images_per_prompt






@slow
@require_torch_gpu
class  __UpperCAmelCase					(	unittest.TestCase						):



   def 	__magic_name__   (							self							:					List[Any]						):
     # clean up the VRAM after each test
     super().tearDown()
     gc.collect()
     torch.cuda.empty_cache()





   def 	__magic_name__   (							self							:					Tuple						):
     UpperCAmelCase	:							List[str]						=			load_image(
         '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main''' '''/shap_e/corgi.png'''						)
     UpperCAmelCase	:							Dict						=			load_numpy(
         '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
         '''/shap_e/test_shap_e_img2img_out.npy'''						)
     UpperCAmelCase	:							Union[str, Any]						=			ShapEImgaImgPipeline.from_pretrained('''openai/shap-e-img2img'''						)
     UpperCAmelCase	:							int						=			pipe.to(__A						)
     pipe.set_progress_bar_config(disable=__A						)

     UpperCAmelCase	:							Dict						=			torch.Generator(device=__A						).manual_seed(0						)

     UpperCAmelCase	:							List[str]						=			pipe(
         __A,		generator=__A,		guidance_scale=3.0,		num_inference_steps=6_4,		frame_size=6_4,		output_type='''np''',		).images[0]

     assert images.shape == (2_0, 6_4, 6_4, 3)

     assert_mean_pixel_difference(__A,		__A						)





