


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


lowercase_    		=     logging.get_logger(__name__)




class        _UpperCamelCase			(   a__						):







							'''simple docstring'''







							def 		_UpperCAmelCase						(		self	:					str		,					SCREAMING_SNAKE_CASE_	:					List[str]		):
													if isinstance(lowerCamelCase_		,					lowerCamelCase_		):
																			_a     			=      [label.strip() for label in labels.split(','		) if label.strip()]
													return labels







							def __call__(		self	:					List[Any]		,					SCREAMING_SNAKE_CASE_	:					Any		,					SCREAMING_SNAKE_CASE_	:					Optional[int]		,					SCREAMING_SNAKE_CASE_	:					Tuple		):
													if len(lowerCamelCase_		) == 0 or len(lowerCamelCase_		) == 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(lowerCamelCase_		)		)

													if isinstance(lowerCamelCase_		,					lowerCamelCase_		):
																			_a     			=      [sequences]

													_a     			=      []
													for sequence in sequences:
																			sequence_pairs.extend([[sequence, hypothesis_template.format(lowerCamelCase_		)] for label in labels]		)

													return sequence_pairs, sequences




@add_end_docstrings(a__						)
class        _UpperCamelCase			(   a__						):







							'''simple docstring'''







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







							def 		_UpperCAmelCase						(		self	:					int		,					SCREAMING_SNAKE_CASE_	:					Union[str, Any]		,					SCREAMING_SNAKE_CASE_	:					Tuple=True		,					SCREAMING_SNAKE_CASE_	:					str=True		,					SCREAMING_SNAKE_CASE_	:					str=TruncationStrategy.ONLY_FIRST		,					**SCREAMING_SNAKE_CASE_	:					int		):
													_a     			=      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`'		)
																			_a     			=      self.tokenizer.eos_token
													try:
																			_a     			=      self.tokenizer(
																			    lowerCamelCase_		,					add_special_tokens=lowerCamelCase_		,					return_tensors=lowerCamelCase_		,					padding=lowerCamelCase_		,					truncation=lowerCamelCase_		,					)
													except Exception as e:
																			if "too short" in str(lowerCamelCase_		):
																									# 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.

																									_a     			=      self.tokenizer(
																									    lowerCamelCase_		,					add_special_tokens=lowerCamelCase_		,					return_tensors=lowerCamelCase_		,					padding=lowerCamelCase_		,					truncation=TruncationStrategy.DO_NOT_TRUNCATE		,					)
																			else:
																									raise e

													return inputs







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

													_a     			=      {}
													if "multi_label" in kwargs:
																			_a     			=      kwargs['multi_label']
													return preprocess_params, {}, postprocess_params







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

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







							def 		_UpperCAmelCase						(		self	:					str		,					SCREAMING_SNAKE_CASE_	:					Tuple		,					SCREAMING_SNAKE_CASE_	:					Dict=None		,					SCREAMING_SNAKE_CASE_	:					Tuple="This example is {}."		):
													_a			,						_a     			=      self._args_parser(lowerCamelCase_		,					lowerCamelCase_		,					lowerCamelCase_		)

													for i, (candidate_label, sequence_pair) in enumerate(zip(lowerCamelCase_		,					lowerCamelCase_		)		):
																			_a     			=      self._parse_and_tokenize([sequence_pair]		)

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







							def 		_UpperCAmelCase						(		self	:					str		,					SCREAMING_SNAKE_CASE_	:					Optional[int]		):
													_a     			=      inputs['candidate_label']
													_a     			=      inputs['sequence']
													_a     			=      {k: inputs[k] for k in self.tokenizer.model_input_names}
													_a     			=      self.model(**lowerCamelCase_		)

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







							def 		_UpperCAmelCase						(		self	:					Optional[int]		,					SCREAMING_SNAKE_CASE_	:					Any		,					SCREAMING_SNAKE_CASE_	:					Tuple=False		):
													_a     			=      [outputs['candidate_label'] for outputs in model_outputs]
													_a     			=      [outputs['sequence'] for outputs in model_outputs]
													_a     			=      np.concatenate([output['logits'].numpy() for output in model_outputs]		)
													_a     			=      logits.shape[0]
													_a     			=      len(lowerCamelCase_		)
													_a     			=      N // n
													_a     			=      logits.reshape((num_sequences, n, -1)		)

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

													_a     			=      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, XLMRobertaTokenizer

from diffusers import AltDiffusionPipeline, AutoencoderKL, DDIMScheduler, PNDMScheduler, UNetaDConditionModel
from diffusers.pipelines.alt_diffusion.modeling_roberta_series import (
    RobertaSeriesConfig,
    RobertaSeriesModelWithTransformation,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, 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


enable_full_determinism()




class 						a__							(				a__							,							a__							,							a__							,							unittest.TestCase    ):





							'''simple docstring'''






							lowercase__     :  Optional[Any]      		= AltDiffusionPipeline
							lowercase__     :  Dict      		= TEXT_TO_IMAGE_PARAMS
							lowercase__     :  str      		= TEXT_TO_IMAGE_BATCH_PARAMS
							lowercase__     :  Any      		= TEXT_TO_IMAGE_IMAGE_PARAMS
							lowercase__     :  List[str]      		= TEXT_TO_IMAGE_IMAGE_PARAMS







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				str:
														torch.manual_seed(0  )
														lowerCAmelCase__			  =      UNetaDConditionModel(
														    block_out_channels=(32, 64)   ,       layers_per_block=2   ,       sample_size=32   ,       in_channels=4   ,       out_channels=4   ,       down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')   ,       up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')   ,       cross_attention_dim=32   ,       )
														lowerCAmelCase__			  =      DDIMScheduler(
														    beta_start=0.00_085   ,       beta_end=0.012   ,       beta_schedule='''scaled_linear'''   ,       clip_sample=lowerCamelCase_   ,       set_alpha_to_one=lowerCamelCase_   ,       )
														torch.manual_seed(0  )
														lowerCAmelCase__			  =      AutoencoderKL(
														    block_out_channels=[32, 64]   ,       in_channels=3   ,       out_channels=3   ,       down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']   ,       up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']   ,       latent_channels=4   ,       )

														# TODO: address the non-deterministic text encoder (fails for save-load tests)
														# torch.manual_seed(0)
														# text_encoder_config = RobertaSeriesConfig(
														#     hidden_size=32,
														#     project_dim=32,
														#     intermediate_size=37,
														#     layer_norm_eps=1e-05,
														#     num_attention_heads=4,
														#     num_hidden_layers=5,
														#     vocab_size=5002,
														# )
														# text_encoder = RobertaSeriesModelWithTransformation(text_encoder_config)

														torch.manual_seed(0  )
														lowerCAmelCase__			  =      CLIPTextConfig(
														    bos_token_id=0   ,       eos_token_id=2   ,       hidden_size=32   ,       projection_dim=32   ,       intermediate_size=37   ,       layer_norm_eps=1e-05   ,       num_attention_heads=4   ,       num_hidden_layers=5   ,       pad_token_id=1   ,       vocab_size=50_02   ,       )
														lowerCAmelCase__			  =      CLIPTextModel(lowerCamelCase_  )

														lowerCAmelCase__			  =      XLMRobertaTokenizer.from_pretrained('''hf-internal-testing/tiny-xlm-roberta'''  )
														lowerCAmelCase__			  =      77

														lowerCAmelCase__			  =      {
														    '''unet''': unet,
														    '''scheduler''': scheduler,
														    '''vae''': vae,
														    '''text_encoder''': text_encoder,
														    '''tokenizer''': tokenizer,
														    '''safety_checker''': None,
														    '''feature_extractor''': None,
														}
														return components







							def     __SCREAMING_SNAKE_CASE				(						self   ,       lowerCamelCase_   ,       lowerCamelCase_=0  )			->				List[str]:
														if str(lowerCamelCase_  ).startswith('''mps'''  ):
																					lowerCAmelCase__			  =      torch.manual_seed(lowerCamelCase_  )
														else:
																					lowerCAmelCase__			  =      torch.Generator(device=lowerCamelCase_  ).manual_seed(lowerCamelCase_  )
														lowerCAmelCase__			  =      {
														    '''prompt''': '''A painting of a squirrel eating a burger''',
														    '''generator''': generator,
														    '''num_inference_steps''': 2,
														    '''guidance_scale''': 6.0,
														    '''output_type''': '''numpy''',
														}
														return inputs







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Union[str, Any]:
														super().test_attention_slicing_forward_pass(expected_max_diff=3e-3  )







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Optional[int]:
														super().test_inference_batch_single_identical(expected_max_diff=3e-3  )







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				List[str]:
														lowerCAmelCase__			  =      '''cpu'''  # ensure determinism for the device-dependent torch.Generator

														lowerCAmelCase__			  =      self.get_dummy_components()
														torch.manual_seed(0  )
														lowerCAmelCase__			  =      RobertaSeriesConfig(
														    hidden_size=32   ,       project_dim=32   ,       intermediate_size=37   ,       layer_norm_eps=1e-05   ,       num_attention_heads=4   ,       num_hidden_layers=5   ,       vocab_size=50_02   ,       )
														# TODO: remove after fixing the non-deterministic text encoder
														lowerCAmelCase__			  =      RobertaSeriesModelWithTransformation(lowerCamelCase_  )
														lowerCAmelCase__			  =      text_encoder

														lowerCAmelCase__			  =      AltDiffusionPipeline(**lowerCamelCase_  )
														lowerCAmelCase__			  =      alt_pipe.to(lowerCamelCase_  )
														alt_pipe.set_progress_bar_config(disable=lowerCamelCase_  )

														lowerCAmelCase__			  =      self.get_dummy_inputs(lowerCamelCase_  )
														lowerCAmelCase__			  =      '''A photo of an astronaut'''
														lowerCAmelCase__			  =      alt_pipe(**lowerCamelCase_  )
														lowerCAmelCase__			  =      output.images
														lowerCAmelCase__			  =      image[0, -3:, -3:, -1]

														assert image.shape == (1, 64, 64, 3)
														lowerCAmelCase__			  =      np.array(
														    [0.5_748_162, 0.60_447_145, 0.48_821_217, 0.50_100_636, 0.5_431_185, 0.45_763_683, 0.49_657_696, 0.48_132_733, 0.47_573_093]  )

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







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				List[Any]:
														lowerCAmelCase__			  =      '''cpu'''  # ensure determinism for the device-dependent torch.Generator

														lowerCAmelCase__			  =      self.get_dummy_components()
														lowerCAmelCase__			  =      PNDMScheduler(skip_prk_steps=lowerCamelCase_  )
														torch.manual_seed(0  )
														lowerCAmelCase__			  =      RobertaSeriesConfig(
														    hidden_size=32   ,       project_dim=32   ,       intermediate_size=37   ,       layer_norm_eps=1e-05   ,       num_attention_heads=4   ,       num_hidden_layers=5   ,       vocab_size=50_02   ,       )
														# TODO: remove after fixing the non-deterministic text encoder
														lowerCAmelCase__			  =      RobertaSeriesModelWithTransformation(lowerCamelCase_  )
														lowerCAmelCase__			  =      text_encoder
														lowerCAmelCase__			  =      AltDiffusionPipeline(**lowerCamelCase_  )
														lowerCAmelCase__			  =      alt_pipe.to(lowerCamelCase_  )
														alt_pipe.set_progress_bar_config(disable=lowerCamelCase_  )

														lowerCAmelCase__			  =      self.get_dummy_inputs(lowerCamelCase_  )
														lowerCAmelCase__			  =      alt_pipe(**lowerCamelCase_  )
														lowerCAmelCase__			  =      output.images
														lowerCAmelCase__			  =      image[0, -3:, -3:, -1]

														assert image.shape == (1, 64, 64, 3)
														lowerCAmelCase__			  =      np.array(
														    [0.51_605_093, 0.5_707_241, 0.47_365_507, 0.50_578_886, 0.5_633_877, 0.4_642_503, 0.5_182_081, 0.48_763_484, 0.49_084_237]  )

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




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





							'''simple docstring'''






							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Tuple:
														# clean up the VRAM after each test
														super().tearDown()
														gc.collect()
														torch.cuda.empty_cache()







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				List[Any]:
														# make sure here that pndm scheduler skips prk
														lowerCAmelCase__			  =      AltDiffusionPipeline.from_pretrained('''BAAI/AltDiffusion'''   ,       safety_checker=lowerCamelCase_  )
														lowerCAmelCase__			  =      alt_pipe.to(lowerCamelCase_  )
														alt_pipe.set_progress_bar_config(disable=lowerCamelCase_  )

														lowerCAmelCase__			  =      '''A painting of a squirrel eating a burger'''
														lowerCAmelCase__			  =      torch.manual_seed(0  )
														lowerCAmelCase__			  =      alt_pipe([prompt]   ,       generator=lowerCamelCase_   ,       guidance_scale=6.0   ,       num_inference_steps=20   ,       output_type='''np'''  )

														lowerCAmelCase__			  =      output.images

														lowerCAmelCase__			  =      image[0, -3:, -3:, -1]

														assert image.shape == (1, 5_12, 5_12, 3)
														lowerCAmelCase__			  =      np.array([0.1_010, 0.0_800, 0.0_794, 0.0_885, 0.0_843, 0.0_762, 0.0_769, 0.0_729, 0.0_586]  )

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







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Any:
														lowerCAmelCase__			  =      DDIMScheduler.from_pretrained('''BAAI/AltDiffusion'''   ,       subfolder='''scheduler'''  )

														lowerCAmelCase__			  =      AltDiffusionPipeline.from_pretrained('''BAAI/AltDiffusion'''   ,       scheduler=lowerCamelCase_   ,       safety_checker=lowerCamelCase_  )
														lowerCAmelCase__			  =      alt_pipe.to(lowerCamelCase_  )
														alt_pipe.set_progress_bar_config(disable=lowerCamelCase_  )

														lowerCAmelCase__			  =      '''A painting of a squirrel eating a burger'''
														lowerCAmelCase__			  =      torch.manual_seed(0  )

														lowerCAmelCase__			  =      alt_pipe([prompt]   ,       generator=lowerCamelCase_   ,       num_inference_steps=2   ,       output_type='''numpy'''  )
														lowerCAmelCase__			  =      output.images

														lowerCAmelCase__			  =      image[0, -3:, -3:, -1]

														assert image.shape == (1, 5_12, 5_12, 3)
														lowerCAmelCase__			  =      np.array([0.4_019, 0.4_052, 0.3_810, 0.4_119, 0.3_916, 0.3_982, 0.4_651, 0.4_195, 0.5_323]  )

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