



"""simple docstring"""
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    :str     =					logging.get_logger(__name__)
class 							_UpperCAmelCase							( a      ):





     '''simple docstring'''



     def  __lowerCAmelCase	(					self       ,	A		)							->					str:
         if isinstance(A       ,	A		):
             _UpperCAmelCase							:  List[str]											=    [label.strip() for label in labels.split(''','''		) if label.strip()]
         return labels





     def __call__(					self       ,	A       ,	A       ,	A		)							->					str:
         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							:  Union[str, Any]											=    [sequences]

         _UpperCAmelCase							:  List[Any]											=    []
         for sequence in sequences:
             sequence_pairs.extend([[sequence, hypothesis_template.format(A		)] for label in labels]		)

         return sequence_pairs, sequences



@add_end_docstrings(a      )
class 							_UpperCAmelCase							( a      ):





     '''simple docstring'''



     def __init__(					self       ,	A=ZeroShotClassificationArgumentHandler()       ,	*A       ,	**A		)							->					int:
         _UpperCAmelCase							:  List[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  __lowerCAmelCase	(					self		)							->					Union[str, Any]:
         for label, ind in self.model.config.labelaid.items():
             if label.lower().startswith('''entail'''		):
                 return ind
         return -1





     def  __lowerCAmelCase	(					self       ,	A       ,	A=True       ,	A=True       ,	A=TruncationStrategy.ONLY_FIRST       ,	**A		)							->					List[str]:
         _UpperCAmelCase							:  str											=    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							:  List[Any]											=    self.tokenizer(
                     A       ,	add_special_tokens=A       ,	return_tensors=A       ,	padding=A       ,	truncation=TruncationStrategy.DO_NOT_TRUNCATE       ,	)
             else:
                 raise e

         return inputs





     def  __lowerCAmelCase	(					self       ,	**A		)							->					Dict:
         if kwargs.get('''multi_class'''       ,	A		) is not None:
             _UpperCAmelCase							:  List[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							:  Any											=    {}
         if "candidate_labels" in kwargs:
             _UpperCAmelCase							:  str											=    self._args_parser._parse_labels(kwargs['''candidate_labels''']		)
         if "hypothesis_template" in kwargs:
             _UpperCAmelCase							:  Tuple											=    kwargs['''hypothesis_template''']

         _UpperCAmelCase							:  Tuple											=    {}
         if "multi_label" in kwargs:
             _UpperCAmelCase							:  Optional[int]											=    kwargs['''multi_label''']
         return preprocess_params, {}, postprocess_params





     def __call__(					self       ,	A       ,	*A       ,	**A       ,	)							->					Dict:
         if len(A		) == 0:
             pass
         elif len(A		) == 1 and "candidate_labels" not in kwargs:
             _UpperCAmelCase							:  List[str]											=    args[0]
         else:
             raise ValueError(f'Unable to understand extra arguments {args}'		)

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





     def  __lowerCAmelCase	(					self       ,	A       ,	A=None       ,	A="This example is {}."		)							->					Dict:
         _UpperCAmelCase							,				_UpperCAmelCase							:  str											=    self._args_parser(A       ,	A       ,	A		)

         for i, (candidate_label, sequence_pair) in enumerate(zip(A       ,	A		)		):
             _UpperCAmelCase							:  List[Any]											=    self._parse_and_tokenize([sequence_pair]		)

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





     def  __lowerCAmelCase	(					self       ,	A		)							->					List[str]:
         _UpperCAmelCase							:  int											=    inputs['''candidate_label''']
         _UpperCAmelCase							:  Dict											=    inputs['''sequence''']
         _UpperCAmelCase							:  Optional[Any]											=    {k: inputs[k] for k in self.tokenizer.model_input_names}
         _UpperCAmelCase							:  Optional[Any]											=    self.model(**A		)

         _UpperCAmelCase							:  Optional[int]											=    {
             '''candidate_label''': candidate_label,
             '''sequence''': sequence,
             '''is_last''': inputs['''is_last'''],
             **outputs,
         }
         return model_outputs





     def  __lowerCAmelCase	(					self       ,	A       ,	A=False		)							->					Tuple:
         _UpperCAmelCase							:  List[str]											=    [outputs['''candidate_label'''] for outputs in model_outputs]
         _UpperCAmelCase							:  Optional[int]											=    [outputs['''sequence'''] for outputs in model_outputs]
         _UpperCAmelCase							:  int											=    np.concatenate([output['''logits'''].numpy() for output in model_outputs]		)
         _UpperCAmelCase							:  Optional[int]											=    logits.shape[0]
         _UpperCAmelCase							:  int											=    len(A		)
         _UpperCAmelCase							:  int											=    N // n
         _UpperCAmelCase							:  Any											=    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							:  Optional[int]											=    self.entailment_id
             _UpperCAmelCase							:  Any											=    -1 if entailment_id == 0 else 0
             _UpperCAmelCase							:  Optional[Any]											=    reshaped_outputs[..., [contradiction_id, entailment_id]]
             _UpperCAmelCase							:  Union[str, Any]											=    np.exp(A		) / np.exp(A		).sum(-1       ,	keepdims=A		)
             _UpperCAmelCase							:  int											=    scores[..., 1]
         else:
             # softmax the "entailment" logits over all candidate labels
             _UpperCAmelCase							:  int											=    reshaped_outputs[..., self.entailment_id]
             _UpperCAmelCase							:  List[Any]											=    np.exp(A		) / np.exp(A		).sum(-1       ,	keepdims=A		)

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







"""simple docstring"""
import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionAttendAndExcitePipeline,
    UNetaDConditionModel,
)
from diffusers.utils import load_numpy, skip_mps, slow
from diffusers.utils.testing_utils import require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


_lowerCAmelCase    :Any     =					False
@skip_mps
class 							_UpperCAmelCase							( a		,a		,a		,unittest.TestCase      ):





     '''simple docstring'''



     a__									=StableDiffusionAttendAndExcitePipeline
     a__									=False
     a__									=TEXT_TO_IMAGE_PARAMS
     a__									=TEXT_TO_IMAGE_BATCH_PARAMS.union({'''token_indices'''}      )
     a__									=TEXT_TO_IMAGE_IMAGE_PARAMS
     a__									=TEXT_TO_IMAGE_IMAGE_PARAMS





     @classmethod
     def  __lowerCAmelCase	(					cls		)							->					List[str]:
         super().setUpClass()
         torch.use_deterministic_algorithms(A		)





     @classmethod
     def  __lowerCAmelCase	(					cls		)							->					Union[str, Any]:
         super().tearDownClass()
         torch.use_deterministic_algorithms(A		)





     def  __lowerCAmelCase	(					self		)							->					Tuple:
         torch.manual_seed(0		)
         _UpperCAmelCase							:  Optional[int]											=    UNetaDConditionModel(
             block_out_channels=(3_2, 6_4)       ,	layers_per_block=1       ,	sample_size=3_2       ,	in_channels=4       ,	out_channels=4       ,	down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')       ,	up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')       ,	cross_attention_dim=3_2       ,	attention_head_dim=(2, 4)       ,	use_linear_projection=A       ,	)
         _UpperCAmelCase							:  List[Any]											=    DDIMScheduler(
             beta_start=0.00_085       ,	beta_end=0.012       ,	beta_schedule='''scaled_linear'''       ,	clip_sample=A       ,	set_alpha_to_one=A       ,	)
         torch.manual_seed(0		)
         _UpperCAmelCase							:  int											=    AutoencoderKL(
             block_out_channels=[3_2, 6_4]       ,	in_channels=3       ,	out_channels=3       ,	down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']       ,	up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']       ,	latent_channels=4       ,	sample_size=1_2_8       ,	)
         torch.manual_seed(0		)
         _UpperCAmelCase							:  int											=    CLIPTextConfig(
             bos_token_id=0       ,	eos_token_id=2       ,	hidden_size=3_2       ,	intermediate_size=3_7       ,	layer_norm_eps=1E-05       ,	num_attention_heads=4       ,	num_hidden_layers=5       ,	pad_token_id=1       ,	vocab_size=1_0_0_0       ,	hidden_act='''gelu'''       ,	projection_dim=5_1_2       ,	)
         _UpperCAmelCase							:  List[str]											=    CLIPTextModel(A		)
         _UpperCAmelCase							:  str											=    CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''		)

         _UpperCAmelCase							:  Union[str, Any]											=    {
             '''unet''': unet,
             '''scheduler''': scheduler,
             '''vae''': vae,
             '''text_encoder''': text_encoder,
             '''tokenizer''': tokenizer,
             '''safety_checker''': None,
             '''feature_extractor''': None,
         }

         return components





     def  __lowerCAmelCase	(					self       ,	A       ,	A=0		)							->					List[Any]:
         if str(A		).startswith('''mps'''		):
             _UpperCAmelCase							:  Optional[int]											=    torch.manual_seed(A		)
         else:
             _UpperCAmelCase							:  Union[str, Any]											=    torch.Generator(device=A		).manual_seed(A		)
         _UpperCAmelCase							:  List[str]											=    {
             '''prompt''': '''a cat and a frog''',
             '''token_indices''': [2, 5],
             '''generator''': generator,
             '''num_inference_steps''': 1,
             '''guidance_scale''': 6.0,
             '''output_type''': '''numpy''',
             '''max_iter_to_alter''': 2,
             '''thresholds''': {0: 0.7},
         }
         return inputs





     def  __lowerCAmelCase	(					self		)							->					int:
         _UpperCAmelCase							:  List[str]											=    '''cpu'''

         _UpperCAmelCase							:  Tuple											=    self.get_dummy_components()
         _UpperCAmelCase							:  int											=    self.pipeline_class(**A		)
         pipe.to(A		)
         pipe.set_progress_bar_config(disable=A		)

         _UpperCAmelCase							:  Dict											=    self.get_dummy_inputs(A		)
         _UpperCAmelCase							:  Union[str, Any]											=    pipe(**A		).images
         _UpperCAmelCase							:  Tuple											=    image[0, -3:, -3:, -1]

         self.assertEqual(image.shape       ,	(1, 6_4, 6_4, 3)		)
         _UpperCAmelCase							:  int											=    np.array(
             [0.63_905_364, 0.62_897_307, 0.48_599_017, 0.5_133_624, 0.5_550_048, 0.45_769_516, 0.50_326_973, 0.5_023_139, 0.45_384_496]		)
         _UpperCAmelCase							:  Tuple											=    np.abs(image_slice.flatten() - expected_slice		).max()
         self.assertLessEqual(A       ,	1E-3		)





     def  __lowerCAmelCase	(					self		)							->					Dict:
         super().test_cpu_offload_forward_pass(expected_max_diff=5E-4		)





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





     def  __lowerCAmelCase	(					self		)							->					Union[str, Any]:
         self._test_inference_batch_single_identical(batch_size=2       ,	expected_max_diff=7E-4		)





     def  __lowerCAmelCase	(					self		)							->					List[str]:
         super().test_dict_tuple_outputs_equivalent(expected_max_difference=3E-3		)





     def  __lowerCAmelCase	(					self		)							->					List[str]:
         super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5E-4		)





     def  __lowerCAmelCase	(					self		)							->					str:
         super().test_save_load_local(expected_max_difference=5E-4		)





     def  __lowerCAmelCase	(					self		)							->					Optional[int]:
         super().test_save_load_optional_components(expected_max_difference=4E-4		)



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





     '''simple docstring'''



     @classmethod
     def  __lowerCAmelCase	(					cls		)							->					Union[str, Any]:
         super().setUpClass()
         torch.use_deterministic_algorithms(A		)





     @classmethod
     def  __lowerCAmelCase	(					cls		)							->					Optional[int]:
         super().tearDownClass()
         torch.use_deterministic_algorithms(A		)





     def  __lowerCAmelCase	(					self		)							->					List[str]:
         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()





     def  __lowerCAmelCase	(					self		)							->					str:
         _UpperCAmelCase							:  Any											=    torch.manual_seed(5_1		)

         _UpperCAmelCase							:  Optional[Any]											=    StableDiffusionAttendAndExcitePipeline.from_pretrained(
             '''CompVis/stable-diffusion-v1-4'''       ,	safety_checker=A       ,	torch_dtype=torch.floataa		)
         pipe.to('''cuda'''		)

         _UpperCAmelCase							:  Optional[int]											=    '''a painting of an elephant with glasses'''
         _UpperCAmelCase							:  int											=    [5, 7]

         _UpperCAmelCase							:  Dict											=    pipe(
             prompt=A       ,	token_indices=A       ,	guidance_scale=7.5       ,	generator=A       ,	num_inference_steps=5       ,	max_iter_to_alter=5       ,	output_type='''numpy'''       ,	).images[0]

         _UpperCAmelCase							:  List[Any]											=    load_numpy(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/attend-and-excite/elephant_glasses.npy'''		)
         assert np.abs((expected_image - image).max()		) < 5E-1


