






'''simple docstring'''





import unittest

from transformers import AutoTokenizer, FalconConfig, is_torch_available
from transformers.testing_utils import require_torch, slow, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
						import torch

						from transformers import (
						    FalconForCausalLM,
						    FalconForQuestionAnswering,
						    FalconForSequenceClassification,
						    FalconForTokenClassification,
						    FalconModel,
						)


class  snake_case__      :




					def __init__(  self  :		str				, _A  :		str				, _A  :		List[Any]=3				, _A  :		Tuple=7				, _A  :		List[Any]=True				, _A  :		Optional[int]=True				, _A  :		List[Any]=False				, _A  :		Any=True				, _A  :		List[str]=99				, _A  :		Dict=32				, _A  :		Tuple=5				, _A  :		List[Any]=4				, _A  :		List[Any]=37				, _A  :		int="gelu"				, _A  :		Dict=0.1				, _A  :		Dict=0.1				, _A  :		Any=5_12				, _A  :		Optional[int]=16				, _A  :		Any=2				, _A  :		Optional[Any]=0.02				, _A  :		List[str]=3				, _A  :		int=4				, _A  :		Tuple=None				, )    ->							str:
								UpperCAmelCase_      :       Union[str, Any]          =       parent
								UpperCAmelCase_      :       int          =       batch_size
								UpperCAmelCase_      :       Any          =       seq_length
								UpperCAmelCase_      :       Optional[Any]          =       is_training
								UpperCAmelCase_      :       Optional[Any]          =       use_input_mask
								UpperCAmelCase_      :       Any          =       use_token_type_ids
								UpperCAmelCase_      :       Dict          =       use_labels
								UpperCAmelCase_      :       Union[str, Any]          =       vocab_size
								UpperCAmelCase_      :       Any          =       hidden_size
								UpperCAmelCase_      :       Optional[Any]          =       num_hidden_layers
								UpperCAmelCase_      :       Optional[int]          =       num_attention_heads
								UpperCAmelCase_      :       Dict          =       intermediate_size
								UpperCAmelCase_      :       List[Any]          =       hidden_act
								UpperCAmelCase_      :       str          =       hidden_dropout_prob
								UpperCAmelCase_      :       Union[str, Any]          =       attention_probs_dropout_prob
								UpperCAmelCase_      :       Optional[Any]          =       max_position_embeddings
								UpperCAmelCase_      :       str          =       type_vocab_size
								UpperCAmelCase_      :       Union[str, Any]          =       type_sequence_label_size
								UpperCAmelCase_      :       Tuple          =       initializer_range
								UpperCAmelCase_      :       List[Any]          =       num_labels
								UpperCAmelCase_      :       str          =       num_choices
								UpperCAmelCase_      :       Union[str, Any]          =       scope




					def 		A      (  self  :		Union[str, Any]      )    ->							Optional[Any]:
								UpperCAmelCase_      :       Any          =       ids_tensor([self.batch_size, self.seq_length]				, self.vocab_size      )

								UpperCAmelCase_      :       Optional[int]          =       None
								if self.use_input_mask:
											UpperCAmelCase_      :       Any          =       random_attention_mask([self.batch_size, self.seq_length]      )

								UpperCAmelCase_      :       Dict          =       None

								UpperCAmelCase_      :       List[Any]          =       None
								UpperCAmelCase_      :       int          =       None
								UpperCAmelCase_      :       Optional[Any]          =       None
								if self.use_labels:
											UpperCAmelCase_      :       int          =       ids_tensor([self.batch_size]				, self.type_sequence_label_size      )
											UpperCAmelCase_      :       str          =       ids_tensor([self.batch_size, self.seq_length]				, self.num_labels      )
											UpperCAmelCase_      :       Union[str, Any]          =       ids_tensor([self.batch_size]				, self.num_choices      )

								UpperCAmelCase_      :       List[str]          =       self.get_config()

								return config, input_ids, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels




					def 		A      (  self  :		List[str]      )    ->							Optional[int]:
								return FalconConfig(
								    vocab_size=self.vocab_size				, hidden_size=self.hidden_size				, num_hidden_layers=self.num_hidden_layers				, num_attention_heads=self.num_attention_heads				, intermediate_size=self.intermediate_size				, hidden_act=self.hidden_act				, hidden_dropout_prob=self.hidden_dropout_prob				, attention_probs_dropout_prob=self.attention_probs_dropout_prob				, max_position_embeddings=self.max_position_embeddings				, type_vocab_size=self.type_vocab_size				, is_decoder=_A				, initializer_range=self.initializer_range				, pad_token_id=1				, new_decoder_architecture=_A				, )




					def 		A      (  self  :		Optional[int]				, _A  :		Union[str, Any]				, _A  :		Optional[Any]				, _A  :		Any				, _A  :		int				, _A  :		Optional[int]				, _A  :		Optional[Any]				, _A  :		Optional[int]      )    ->							Any:
								UpperCAmelCase_      :       Tuple          =       FalconModel(config=_A      )
								model.to(_A      )
								model.eval()
								UpperCAmelCase_      :       Union[str, Any]          =       model(_A				, attention_mask=_A      )
								UpperCAmelCase_      :       Optional[Any]          =       model(_A      )
								self.parent.assertEqual(result.last_hidden_state.shape				, (self.batch_size, self.seq_length, self.hidden_size)      )




					def 		A      (  self  :		Any				, _A  :		int				, _A  :		Optional[Any]				, _A  :		Tuple				, _A  :		Any				, _A  :		Optional[int]				, _A  :		Union[str, Any]				, _A  :		Tuple				, _A  :		str				, _A  :		Union[str, Any]				, )    ->							Dict:
								UpperCAmelCase_      :       Union[str, Any]          =       True
								UpperCAmelCase_      :       List[Any]          =       FalconModel(_A      )
								model.to(_A      )
								model.eval()
								UpperCAmelCase_      :       Optional[Any]          =       model(
								    _A				, attention_mask=_A				, encoder_hidden_states=_A				, encoder_attention_mask=_A				, )
								UpperCAmelCase_      :       Any          =       model(
								    _A				, attention_mask=_A				, encoder_hidden_states=_A				, )
								UpperCAmelCase_      :       str          =       model(_A				, attention_mask=_A      )
								self.parent.assertEqual(result.last_hidden_state.shape				, (self.batch_size, self.seq_length, self.hidden_size)      )




					def 		A      (  self  :		List[str]				, _A  :		Optional[Any]				, _A  :		Dict				, _A  :		int				, _A  :		List[Any]				, _A  :		Any				, _A  :		List[Any]				, _A  :		Tuple				, _A  :		Union[str, Any]				, _A  :		Tuple				, )    ->							Optional[Any]:
								UpperCAmelCase_      :       Tuple          =       FalconForCausalLM(config=_A      )
								model.to(_A      )
								model.eval()
								UpperCAmelCase_      :       str          =       model(_A				, attention_mask=_A				, labels=_A      )
								self.parent.assertEqual(result.logits.shape				, (self.batch_size, self.seq_length, self.vocab_size)      )




					def 		A      (  self  :		List[str]				, _A  :		List[str]				, _A  :		Tuple				, _A  :		Tuple				, _A  :		List[Any]				, _A  :		Optional[int]				, _A  :		List[Any]				, _A  :		Tuple				, _A  :		Dict				, _A  :		int				, )    ->							int:
								UpperCAmelCase_      :       Dict          =       True
								UpperCAmelCase_      :       Optional[int]          =       True
								UpperCAmelCase_      :       Union[str, Any]          =       FalconForCausalLM(config=_A      )
								model.to(_A      )
								model.eval()

								# first forward pass
								UpperCAmelCase_      :       Tuple          =       model(
								    _A				, attention_mask=_A				, encoder_hidden_states=_A				, encoder_attention_mask=_A				, use_cache=_A				, )
								UpperCAmelCase_      :       List[Any]          =       outputs.past_key_values

								# create hypothetical multiple next token and extent to next_input_ids
								UpperCAmelCase_      :       int          =       ids_tensor((self.batch_size, 3)				, config.vocab_size      )
								UpperCAmelCase_      :       int          =       ids_tensor((self.batch_size, 3)				, vocab_size=2      )

								# append to next input_ids and
								UpperCAmelCase_      :       List[Any]          =       torch.cat([input_ids, next_tokens]				, dim=-1      )
								UpperCAmelCase_      :       List[str]          =       torch.cat([input_mask, next_mask]				, dim=-1      )

								UpperCAmelCase_      :       Optional[int]          =       model(
								    _A				, attention_mask=_A				, encoder_hidden_states=_A				, encoder_attention_mask=_A				, output_hidden_states=_A				, )['''hidden_states'''][0]
								UpperCAmelCase_      :       Any          =       model(
								    _A				, attention_mask=_A				, encoder_hidden_states=_A				, encoder_attention_mask=_A				, past_key_values=_A				, output_hidden_states=_A				, )['''hidden_states'''][0]

								# select random slice
								UpperCAmelCase_      :       List[str]          =       ids_tensor((1,)				, output_from_past.shape[-1]      ).item()
								UpperCAmelCase_      :       int          =       output_from_no_past[:, -3:, random_slice_idx].detach()
								UpperCAmelCase_      :       Optional[Any]          =       output_from_past[:, :, random_slice_idx].detach()

								self.parent.assertTrue(output_from_past_slice.shape[1] == next_tokens.shape[1]      )

								# test that outputs are equal for slice
								self.parent.assertTrue(torch.allclose(_A				, _A				, atol=1e-3      )      )





					def 		A      (  self  :		int      )    ->							Optional[Any]:
								UpperCAmelCase_      :       Dict          =       self.prepare_config_and_inputs()
								(
								    (
								    UpperCAmelCase_
								)    ,      (
								    UpperCAmelCase_
								)    ,      (
								    UpperCAmelCase_
								)    ,      (
								    UpperCAmelCase_
								)    ,      (
								    UpperCAmelCase_
								)    ,      (
								    UpperCAmelCase_
								)    ,      (
								    UpperCAmelCase_
								)    ,      
								)      :       Any          =       config_and_inputs
								UpperCAmelCase_      :       Optional[int]          =       {'''input_ids''': input_ids, '''attention_mask''': input_mask}
								return config, inputs_dict






@require_torch
class  snake_case__      (					UpperCamelCase     ,				UpperCamelCase     ,				UpperCamelCase     ,				unittest.TestCase):
					a_        =					(
					    (
					        FalconModel,
					        FalconForCausalLM,
					        FalconForSequenceClassification,
					        FalconForTokenClassification,
					        FalconForQuestionAnswering,
					    )
					    if is_torch_available()
					    else ()
					)
					a_        =					(FalconForCausalLM,) if is_torch_available() else ()
					a_        =					(
					    {
					        "feature-extraction": FalconModel,
					        "text-classification": FalconForSequenceClassification,
					        "text-generation": FalconForCausalLM,
					        "question-answering": FalconForQuestionAnswering,
					        "token-classification": FalconForTokenClassification,
					        "zero-shot": FalconForSequenceClassification,
					    }
					    if is_torch_available()
					    else {}
					)
					a_        =					False
					a_        =					False




					def 		A      (  self  :		Union[str, Any]      )    ->							int:
								UpperCAmelCase_      :       List[Any]          =       FalconModelTester(self      )
								UpperCAmelCase_      :       List[Any]          =       ConfigTester(self				, config_class=_A				, hidden_size=37      )




					def 		A      (  self  :		Optional[Any]      )    ->							Dict:
								self.config_tester.run_common_tests()




					def 		A      (  self  :		List[Any]      )    ->							Tuple:
								UpperCAmelCase_      :       Union[str, Any]          =       self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_model(*_A      )




					def 		A      (  self  :		List[Any]      )    ->							Union[str, Any]:
								UpperCAmelCase_    ,      *UpperCAmelCase_      :       str          =       self.model_tester.prepare_config_and_inputs()
								for alibi in [True, False]:
											UpperCAmelCase_      :       Optional[Any]          =       alibi
											self.model_tester.create_and_check_model(_A				, *_A      )




					def 		A      (  self  :		Optional[int]      )    ->							List[str]:
								UpperCAmelCase_    ,      UpperCAmelCase_      :       Dict          =       self.model_tester.prepare_config_and_inputs_for_common()
								UpperCAmelCase_      :       Any          =       3
								UpperCAmelCase_      :       List[str]          =       input_dict['''input_ids''']
								UpperCAmelCase_      :       Tuple          =       input_ids.ne(1      ).to(_A      )
								UpperCAmelCase_      :       int          =       ids_tensor([self.model_tester.batch_size]				, self.model_tester.type_sequence_label_size      )
								UpperCAmelCase_      :       Tuple          =       FalconForSequenceClassification(_A      )
								model.to(_A      )
								model.eval()
								UpperCAmelCase_      :       str          =       model(_A				, attention_mask=_A				, labels=_A      )
								self.assertEqual(result.logits.shape				, (self.model_tester.batch_size, self.model_tester.num_labels)      )




					def 		A      (  self  :		str      )    ->							List[Any]:
								UpperCAmelCase_    ,      UpperCAmelCase_      :       Any          =       self.model_tester.prepare_config_and_inputs_for_common()
								UpperCAmelCase_      :       Tuple          =       3
								UpperCAmelCase_      :       str          =       '''single_label_classification'''
								UpperCAmelCase_      :       Optional[Any]          =       input_dict['''input_ids''']
								UpperCAmelCase_      :       str          =       input_ids.ne(1      ).to(_A      )
								UpperCAmelCase_      :       str          =       ids_tensor([self.model_tester.batch_size]				, self.model_tester.type_sequence_label_size      )
								UpperCAmelCase_      :       int          =       FalconForSequenceClassification(_A      )
								model.to(_A      )
								model.eval()
								UpperCAmelCase_      :       int          =       model(_A				, attention_mask=_A				, labels=_A      )
								self.assertEqual(result.logits.shape				, (self.model_tester.batch_size, self.model_tester.num_labels)      )




					def 		A      (  self  :		Union[str, Any]      )    ->							List[Any]:
								UpperCAmelCase_    ,      UpperCAmelCase_      :       int          =       self.model_tester.prepare_config_and_inputs_for_common()
								UpperCAmelCase_      :       Optional[Any]          =       input_dict['''input_ids''']
								UpperCAmelCase_      :       Dict          =       FalconForCausalLM(_A      )
								model.to(_A      )
								model.eval()
								UpperCAmelCase_      :       List[str]          =       model(_A				, use_cache=_A      )
								UpperCAmelCase_      :       int          =       input_ids.shape[0]
								UpperCAmelCase_      :       Optional[int]          =       model._convert_to_rw_cache(result.past_key_values      )
								UpperCAmelCase_      :       Any          =       model._convert_cache_to_standard_format(_A				, _A      )
								for layer in range(len(_A      )      ):
											for tensor_idx in range(2      ):
														self.assertTrue(rw_cache[layer][tensor_idx].ndim == 3      )
														self.assertTrue(result.past_key_values[layer][tensor_idx].ndim == 4      )
														self.assertTrue(
														    torch.all(result.past_key_values[layer][tensor_idx] == standard_cache[layer][tensor_idx]      )      )




					def 		A      (  self  :		Optional[Any]      )    ->							str:
								UpperCAmelCase_    ,      UpperCAmelCase_      :       List[str]          =       self.model_tester.prepare_config_and_inputs_for_common()
								UpperCAmelCase_      :       Optional[int]          =       3
								UpperCAmelCase_      :       List[str]          =       '''multi_label_classification'''
								UpperCAmelCase_      :       Optional[int]          =       input_dict['''input_ids''']
								UpperCAmelCase_      :       List[Any]          =       input_ids.ne(1      ).to(_A      )
								UpperCAmelCase_      :       List[str]          =       ids_tensor(
								    [self.model_tester.batch_size, config.num_labels]				, self.model_tester.type_sequence_label_size      ).to(torch.float      )
								UpperCAmelCase_      :       Dict          =       FalconForSequenceClassification(_A      )
								model.to(_A      )
								model.eval()
								UpperCAmelCase_      :       Union[str, Any]          =       model(_A				, attention_mask=_A				, labels=_A      )
								self.assertEqual(result.logits.shape				, (self.model_tester.batch_size, self.model_tester.num_labels)      )





					def 		A      (  self  :		List[Any]      )    ->							Dict:
								# Falcon can have different numbers of KV-heads than the number of query heads, so we need
								# to override this test to use the right head counts.
								for model_class in self.all_generative_model_classes:
											UpperCAmelCase_    ,      UpperCAmelCase_      :       Optional[int]          =       self.model_tester.prepare_config_and_inputs_for_common()

											# If it doesn't support cache, pass the test
											if not hasattr(_A				, '''use_cache'''      ):
														return

											UpperCAmelCase_      :       Union[str, Any]          =       model_class(_A      ).to(_A      )
											if "use_cache" not in inputs:
														UpperCAmelCase_      :       int          =       True
											UpperCAmelCase_      :       List[str]          =       model(**_A      )

											# If "past_key_values" is not returned, pass the test (e.g. RWKV uses a different cache name and format)
											if "past_key_values" not in outputs:
														return

											UpperCAmelCase_      :       Any          =       (
											    getattr(_A				, '''decoder_layers'''				, _A      )
											    or getattr(_A				, '''num_decoder_layers'''				, _A      )
											    or config.num_hidden_layers
											)
											UpperCAmelCase_      :       Union[str, Any]          =       getattr(_A				, '''num_kv_heads'''				, config.num_attention_heads      )
											UpperCAmelCase_      :       str          =       getattr(_A				, '''d_model'''				, config.hidden_size      )
											UpperCAmelCase_      :       List[str]          =       embed_dim // num_attention_heads

											UpperCAmelCase_      :       str          =       outputs['''past_key_values''']
											self.assertEqual(len(_A      )				, _A      )

											UpperCAmelCase_    ,      UpperCAmelCase_      :       str          =       inputs['''input_ids'''].shape
											for i in range(_A      ):
														if config.new_decoder_architecture:
																	UpperCAmelCase_      :       Dict          =       config.num_attention_heads
														elif config.multi_query:
																	UpperCAmelCase_      :       str          =       1
														self.assertEqual(len(past_kv[0]      )				, 2      )  # K V for the decoder = 2
														self.assertEqual(
														    past_kv[i][0].shape				, (batch_size, num_attention_heads, seq_length, per_head_embed_dim)      )
														self.assertEqual(
														    past_kv[i][1].shape				, (batch_size, num_attention_heads, seq_length, per_head_embed_dim)      )






@require_torch
class  snake_case__      (					unittest.TestCase):




					@slow
					def 		A      (  self  :		Dict      )    ->							List[Any]:
								UpperCAmelCase_      :       Union[str, Any]          =       AutoTokenizer.from_pretrained('''Rocketknight1/falcon-rw-1b'''      )
								UpperCAmelCase_      :       str          =       FalconForCausalLM.from_pretrained('''Rocketknight1/falcon-rw-1b'''      )
								model.eval()
								model.to(_A      )
								UpperCAmelCase_      :       Tuple          =       tokenizer('''My favorite food is'''				, return_tensors='''pt'''      ).to(_A      )

								UpperCAmelCase_      :       Any          =       (
								    '''My favorite food is pizza. I love it so much that I have a pizza party every year for my birthday.'''
								)

								UpperCAmelCase_      :       Optional[int]          =       model.generate(**_A				, do_sample=_A				, max_new_tokens=19      )
								UpperCAmelCase_      :       Optional[Any]          =       tokenizer.batch_decode(_A      )[0]

								self.assertEqual(_A				, _A      )




					@slow
					def 		A      (  self  :		str      )    ->							Dict:
								# The big models are way too big for the CI, so we use tiny random models that resemble their
								# architectures but with much smaller and fewer layers
								for repo in ["Rocketknight1/tiny-random-falcon-7b", "Rocketknight1/tiny-random-falcon-40b"]:
											UpperCAmelCase_      :       Dict          =       AutoTokenizer.from_pretrained(_A      )
											UpperCAmelCase_      :       Union[str, Any]          =       FalconForCausalLM.from_pretrained(_A      )
											model.eval()
											model.to(_A      )
											UpperCAmelCase_      :       Optional[Any]          =       tokenizer('''My favorite food is'''				, return_tensors='''pt'''      ).to(_A      )

											# We just test that these run without errors - the models are randomly initialized
											# and so the actual text outputs will be garbage
											model.generate(**_A				, do_sample=_A				, max_new_tokens=4      )
											model.generate(**_A				, do_sample=_A				, max_new_tokens=4      )
											model.generate(**_A				, num_beams=2				, max_new_tokens=4      )





					@slow
					def 		A      (  self  :		Union[str, Any]      )    ->							int:
								# The big models are way too big for the CI, so we use tiny random models that resemble their
								# architectures but with much smaller and fewer layers
								with torch.no_grad():
											for repo in [
											    "Rocketknight1/falcon-rw-1b",
											    "Rocketknight1/tiny-random-falcon-7b",
											    "Rocketknight1/tiny-random-falcon-40b",
											]:
														UpperCAmelCase_      :       Any          =       AutoTokenizer.from_pretrained(_A      )
														UpperCAmelCase_      :       Optional[int]          =       FalconForCausalLM.from_pretrained(_A      )
														model.eval()
														model.to(device=_A      )
														UpperCAmelCase_      :       Optional[int]          =       tokenizer('''My favorite food is'''				, return_tensors='''pt'''      ).to(_A      )

														# Test results are the same with and without cache
														UpperCAmelCase_      :       List[Any]          =       model.generate(**_A				, do_sample=_A				, max_new_tokens=20				, use_cache=_A      )
														UpperCAmelCase_      :       Optional[int]          =       model.generate(**_A				, do_sample=_A				, max_new_tokens=20				, use_cache=_A      )
														self.assertTrue((outputs_cache - outputs_no_cache).sum().item() == 0      )










'''simple docstring'''





import gc
import random
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    EulerAncestralDiscreteScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionInstructPixaPixPipeline,
    UNetaDConditionModel,
)
from diffusers.image_processor import VaeImageProcessor
from diffusers.utils import floats_tensor, load_image, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()


class  snake_case__      (					UpperCamelCase     ,				UpperCamelCase     ,				UpperCamelCase     ,				unittest.TestCase):
					a_        =					StableDiffusionInstructPixaPixPipeline
					a_        =					TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"height", "width", "cross_attention_kwargs"}
					a_        =					TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
					a_        =					IMAGE_TO_IMAGE_IMAGE_PARAMS
					a_        =					IMAGE_TO_IMAGE_IMAGE_PARAMS




					def 		A      (  self  :		str      )    ->							Optional[Any]:
								torch.manual_seed(0      )
								UpperCAmelCase_      :       Tuple          =       UNetaDConditionModel(
								    block_out_channels=(32, 64)				, layers_per_block=2				, sample_size=32				, in_channels=8				, out_channels=4				, down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')				, up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')				, cross_attention_dim=32				, )
								UpperCAmelCase_      :       Any          =       PNDMScheduler(skip_prk_steps=_A      )
								torch.manual_seed(0      )
								UpperCAmelCase_      :       int          =       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				, )
								torch.manual_seed(0      )
								UpperCAmelCase_      :       Tuple          =       CLIPTextConfig(
								    bos_token_id=0				, eos_token_id=2				, hidden_size=32				, intermediate_size=37				, layer_norm_eps=1e-05				, num_attention_heads=4				, num_hidden_layers=5				, pad_token_id=1				, vocab_size=10_00				, )
								UpperCAmelCase_      :       List[str]          =       CLIPTextModel(_A      )
								UpperCAmelCase_      :       List[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 		A      (  self  :		List[str]				, _A  :		Tuple				, _A  :		str=0      )    ->							int:
								UpperCAmelCase_      :       Tuple          =       floats_tensor((1, 3, 32, 32)				, rng=random.Random(_A      )      ).to(_A      )
								UpperCAmelCase_      :       Dict          =       image.cpu().permute(0				, 2				, 3				, 1      )[0]
								UpperCAmelCase_      :       Optional[Any]          =       Image.fromarray(np.uinta(_A      )      ).convert('''RGB'''      )
								if str(_A      ).startswith('''mps'''      ):
											UpperCAmelCase_      :       str          =       torch.manual_seed(_A      )
								else:
											UpperCAmelCase_      :       Optional[Any]          =       torch.Generator(device=_A      ).manual_seed(_A      )
								UpperCAmelCase_      :       List[str]          =       {
								    '''prompt''': '''A painting of a squirrel eating a burger''',
								    '''image''': image,
								    '''generator''': generator,
								    '''num_inference_steps''': 2,
								    '''guidance_scale''': 6.0,
								    '''image_guidance_scale''': 1,
								    '''output_type''': '''numpy''',
								}
								return inputs




					def 		A      (  self  :		Tuple      )    ->							List[Any]:
								UpperCAmelCase_      :       List[str]          =       '''cpu'''  # ensure determinism for the device-dependent torch.Generator
								UpperCAmelCase_      :       Any          =       self.get_dummy_components()
								UpperCAmelCase_      :       int          =       StableDiffusionInstructPixaPixPipeline(**_A      )
								UpperCAmelCase_      :       Tuple          =       sd_pipe.to(_A      )
								sd_pipe.set_progress_bar_config(disable=_A      )

								UpperCAmelCase_      :       str          =       self.get_dummy_inputs(_A      )
								UpperCAmelCase_      :       Union[str, Any]          =       sd_pipe(**_A      ).images
								UpperCAmelCase_      :       int          =       image[0, -3:, -3:, -1]
								assert image.shape == (1, 32, 32, 3)
								UpperCAmelCase_      :       Union[str, Any]          =       np.array([0.7_526, 0.3_750, 0.4_547, 0.6_117, 0.5_866, 0.5_016, 0.4_327, 0.5_642, 0.4_815]      )

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




					def 		A      (  self  :		int      )    ->							str:
								UpperCAmelCase_      :       List[Any]          =       '''cpu'''  # ensure determinism for the device-dependent torch.Generator
								UpperCAmelCase_      :       List[str]          =       self.get_dummy_components()
								UpperCAmelCase_      :       Dict          =       StableDiffusionInstructPixaPixPipeline(**_A      )
								UpperCAmelCase_      :       str          =       sd_pipe.to(_A      )
								sd_pipe.set_progress_bar_config(disable=_A      )

								UpperCAmelCase_      :       Any          =       self.get_dummy_inputs(_A      )
								UpperCAmelCase_      :       List[Any]          =       '''french fries'''
								UpperCAmelCase_      :       Optional[Any]          =       sd_pipe(**_A				, negative_prompt=_A      )
								UpperCAmelCase_      :       Tuple          =       output.images
								UpperCAmelCase_      :       Union[str, Any]          =       image[0, -3:, -3:, -1]

								assert image.shape == (1, 32, 32, 3)
								UpperCAmelCase_      :       Dict          =       np.array([0.7_511, 0.3_642, 0.4_553, 0.6_236, 0.5_797, 0.5_013, 0.4_343, 0.5_611, 0.4_831]      )

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




					def 		A      (  self  :		Dict      )    ->							str:
								UpperCAmelCase_      :       List[str]          =       '''cpu'''  # ensure determinism for the device-dependent torch.Generator
								UpperCAmelCase_      :       List[Any]          =       self.get_dummy_components()
								UpperCAmelCase_      :       Any          =       StableDiffusionInstructPixaPixPipeline(**_A      )
								UpperCAmelCase_      :       int          =       sd_pipe.to(_A      )
								sd_pipe.set_progress_bar_config(disable=_A      )

								UpperCAmelCase_      :       Any          =       self.get_dummy_inputs(_A      )
								UpperCAmelCase_      :       Any          =       [inputs['''prompt''']] * 2

								UpperCAmelCase_      :       int          =       np.array(inputs['''image''']      ).astype(np.floataa      ) / 255.0
								UpperCAmelCase_      :       int          =       torch.from_numpy(_A      ).unsqueeze(0      ).to(_A      )
								UpperCAmelCase_      :       List[str]          =       image / 2 + 0.5
								UpperCAmelCase_      :       str          =       image.permute(0				, 3				, 1				, 2      )
								UpperCAmelCase_      :       Any          =       image.repeat(2				, 1				, 1				, 1      )

								UpperCAmelCase_      :       List[Any]          =       sd_pipe(**_A      ).images
								UpperCAmelCase_      :       str          =       image[-1, -3:, -3:, -1]

								assert image.shape == (2, 32, 32, 3)
								UpperCAmelCase_      :       Optional[Any]          =       np.array([0.5_812, 0.5_748, 0.5_222, 0.5_908, 0.5_695, 0.7_174, 0.6_804, 0.5_523, 0.5_579]      )

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




					def 		A      (  self  :		List[str]      )    ->							List[Any]:
								UpperCAmelCase_      :       Union[str, Any]          =       '''cpu'''  # ensure determinism for the device-dependent torch.Generator
								UpperCAmelCase_      :       Tuple          =       self.get_dummy_components()
								UpperCAmelCase_      :       List[str]          =       EulerAncestralDiscreteScheduler(
								    beta_start=0.00_085				, beta_end=0.012				, beta_schedule='''scaled_linear'''      )
								UpperCAmelCase_      :       List[Any]          =       StableDiffusionInstructPixaPixPipeline(**_A      )
								UpperCAmelCase_      :       Dict          =       sd_pipe.to(_A      )
								sd_pipe.set_progress_bar_config(disable=_A      )

								UpperCAmelCase_      :       Tuple          =       self.get_dummy_inputs(_A      )
								UpperCAmelCase_      :       str          =       sd_pipe(**_A      ).images
								UpperCAmelCase_      :       Dict          =       image[0, -3:, -3:, -1]

								UpperCAmelCase_      :       Dict          =       [round(_A				, 4      ) for x in image_slice.flatten().tolist()]
								print(''','''.join([str(_A      ) for x in slice]      )      )

								assert image.shape == (1, 32, 32, 3)
								UpperCAmelCase_      :       Any          =       np.array([0.7_417, 0.3_842, 0.4_732, 0.5_776, 0.5_891, 0.5_139, 0.4_052, 0.5_673, 0.4_986]      )

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




					def 		A      (  self  :		str      )    ->							List[str]:
								super().test_inference_batch_single_identical(expected_max_diff=3e-3      )





					def 		A      (  self  :		List[Any]      )    ->							Optional[Any]:
								UpperCAmelCase_      :       Tuple          =       self.get_dummy_components()
								UpperCAmelCase_      :       Tuple          =       StableDiffusionInstructPixaPixPipeline(**_A      )
								UpperCAmelCase_      :       Any          =       VaeImageProcessor(do_resize=_A				, do_normalize=_A      )
								UpperCAmelCase_      :       Union[str, Any]          =       pipe.to(_A      )
								pipe.set_progress_bar_config(disable=_A      )

								UpperCAmelCase_      :       int          =       pipe(**self.get_dummy_inputs_by_type(_A				, input_image_type='''pt'''      )      )[0]

								UpperCAmelCase_      :       Union[str, Any]          =       components['''vae''']
								UpperCAmelCase_      :       Any          =       self.get_dummy_inputs_by_type(_A				, input_image_type='''pt'''      )

								for image_param in self.image_latents_params:
											if image_param in inputs.keys():
														UpperCAmelCase_      :       int          =       vae.encode(inputs[image_param]      ).latent_dist.mode()

								UpperCAmelCase_      :       List[Any]          =       pipe(**_A      )[0]

								UpperCAmelCase_      :       int          =       np.abs(out - out_latents_inputs      ).max()
								self.assertLess(_A				, 1e-4				, '''passing latents as image input generate different result from passing image'''      )






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




					def 		A      (  self  :		Any      )    ->							Optional[Any]:
								super().tearDown()
								gc.collect()
								torch.cuda.empty_cache()




					def 		A      (  self  :		List[str]				, _A  :		Optional[Any]=0      )    ->							List[Any]:
								UpperCAmelCase_      :       Dict          =       torch.manual_seed(_A      )
								UpperCAmelCase_      :       Union[str, Any]          =       load_image(
								    '''https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/stable_diffusion_pix2pix/example.jpg'''      )
								UpperCAmelCase_      :       Optional[int]          =       {
								    '''prompt''': '''turn him into a cyborg''',
								    '''image''': image,
								    '''generator''': generator,
								    '''num_inference_steps''': 3,
								    '''guidance_scale''': 7.5,
								    '''image_guidance_scale''': 1.0,
								    '''output_type''': '''numpy''',
								}
								return inputs




					def 		A      (  self  :		Optional[int]      )    ->							Optional[Any]:
								UpperCAmelCase_      :       List[Any]          =       StableDiffusionInstructPixaPixPipeline.from_pretrained(
								    '''timbrooks/instruct-pix2pix'''				, safety_checker=_A      )
								pipe.to(_A      )
								pipe.set_progress_bar_config(disable=_A      )
								pipe.enable_attention_slicing()

								UpperCAmelCase_      :       List[str]          =       self.get_inputs()
								UpperCAmelCase_      :       int          =       pipe(**_A      ).images
								UpperCAmelCase_      :       Dict          =       image[0, -3:, -3:, -1].flatten()

								assert image.shape == (1, 5_12, 5_12, 3)
								UpperCAmelCase_      :       Any          =       np.array([0.5_902, 0.6_015, 0.6_027, 0.5_983, 0.6_092, 0.6_061, 0.5_765, 0.5_785, 0.5_555]      )

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




					def 		A      (  self  :		Optional[Any]      )    ->							Optional[Any]:
								UpperCAmelCase_      :       Optional[Any]          =       StableDiffusionInstructPixaPixPipeline.from_pretrained(
								    '''timbrooks/instruct-pix2pix'''				, safety_checker=_A      )
								UpperCAmelCase_      :       Dict          =       LMSDiscreteScheduler.from_config(pipe.scheduler.config      )
								pipe.to(_A      )
								pipe.set_progress_bar_config(disable=_A      )
								pipe.enable_attention_slicing()

								UpperCAmelCase_      :       List[Any]          =       self.get_inputs()
								UpperCAmelCase_      :       str          =       pipe(**_A      ).images
								UpperCAmelCase_      :       Optional[int]          =       image[0, -3:, -3:, -1].flatten()

								assert image.shape == (1, 5_12, 5_12, 3)
								UpperCAmelCase_      :       List[str]          =       np.array([0.6_578, 0.6_817, 0.6_972, 0.6_761, 0.6_856, 0.6_916, 0.6_428, 0.6_516, 0.6_301]      )

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




					def 		A      (  self  :		List[str]      )    ->							Union[str, Any]:
								UpperCAmelCase_      :       str          =       StableDiffusionInstructPixaPixPipeline.from_pretrained(
								    '''timbrooks/instruct-pix2pix'''				, safety_checker=_A      )
								UpperCAmelCase_      :       str          =       DDIMScheduler.from_config(pipe.scheduler.config      )
								pipe.to(_A      )
								pipe.set_progress_bar_config(disable=_A      )
								pipe.enable_attention_slicing()

								UpperCAmelCase_      :       Any          =       self.get_inputs()
								UpperCAmelCase_      :       Any          =       pipe(**_A      ).images
								UpperCAmelCase_      :       Optional[Any]          =       image[0, -3:, -3:, -1].flatten()

								assert image.shape == (1, 5_12, 5_12, 3)
								UpperCAmelCase_      :       Dict          =       np.array([0.3_828, 0.3_834, 0.3_818, 0.3_792, 0.3_865, 0.3_752, 0.3_792, 0.3_847, 0.3_753]      )

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




					def 		A      (  self  :		Tuple      )    ->							List[Any]:
								UpperCAmelCase_      :       List[str]          =       0

								def callback_fn(_A  :		int				, _A  :		int				, _A  :		torch.FloatTensor      ) -> None:
											UpperCAmelCase_      :       Tuple          =       True
											nonlocal number_of_steps
											number_of_steps += 1
											if step == 1:
														UpperCAmelCase_      :       List[str]          =       latents.detach().cpu().numpy()
														assert latents.shape == (1, 4, 64, 64)
														UpperCAmelCase_      :       Optional[Any]          =       latents[0, -3:, -3:, -1]
														UpperCAmelCase_      :       List[Any]          =       np.array([-0.2_463, -0.4_644, -0.9_756, 1.5_176, 1.4_414, 0.7_866, 0.9_897, 0.8_521, 0.7_983]      )

														assert np.abs(latents_slice.flatten() - expected_slice      ).max() < 5e-2
											elif step == 2:
														UpperCAmelCase_      :       Union[str, Any]          =       latents.detach().cpu().numpy()
														assert latents.shape == (1, 4, 64, 64)
														UpperCAmelCase_      :       Dict          =       latents[0, -3:, -3:, -1]
														UpperCAmelCase_      :       Tuple          =       np.array([-0.2_644, -0.4_626, -0.9_653, 1.5_176, 1.4_551, 0.7_686, 0.9_805, 0.8_452, 0.8_115]      )

														assert np.abs(latents_slice.flatten() - expected_slice      ).max() < 5e-2

								UpperCAmelCase_      :       Dict          =       False

								UpperCAmelCase_      :       Dict          =       StableDiffusionInstructPixaPixPipeline.from_pretrained(
								    '''timbrooks/instruct-pix2pix'''				, safety_checker=_A				, torch_dtype=torch.floataa      )
								UpperCAmelCase_      :       Any          =       pipe.to(_A      )
								pipe.set_progress_bar_config(disable=_A      )
								pipe.enable_attention_slicing()

								UpperCAmelCase_      :       Dict          =       self.get_inputs()
								pipe(**_A				, callback=_A				, callback_steps=1      )
								assert callback_fn.has_been_called
								assert number_of_steps == 3




					def 		A      (  self  :		int      )    ->							int:
								torch.cuda.empty_cache()
								torch.cuda.reset_max_memory_allocated()
								torch.cuda.reset_peak_memory_stats()

								UpperCAmelCase_      :       Union[str, Any]          =       StableDiffusionInstructPixaPixPipeline.from_pretrained(
								    '''timbrooks/instruct-pix2pix'''				, safety_checker=_A				, torch_dtype=torch.floataa      )
								UpperCAmelCase_      :       Tuple          =       pipe.to(_A      )
								pipe.set_progress_bar_config(disable=_A      )
								pipe.enable_attention_slicing(1      )
								pipe.enable_sequential_cpu_offload()

								UpperCAmelCase_      :       Dict          =       self.get_inputs()
								UpperCAmelCase_      :       List[str]          =       pipe(**_A      )

								UpperCAmelCase_      :       Dict          =       torch.cuda.max_memory_allocated()
								# make sure that less than 2.2 GB is allocated
								assert mem_bytes < 2.2 * 10**9





					def 		A      (  self  :		str      )    ->							int:
								UpperCAmelCase_      :       List[Any]          =       self.get_inputs()
								# resize to resolution that is divisible by 8 but not 16 or 32
								UpperCAmelCase_      :       List[str]          =       inputs['''image'''].resize((5_04, 5_04)      )

								UpperCAmelCase_      :       int          =       '''timbrooks/instruct-pix2pix'''
								UpperCAmelCase_      :       Tuple          =       StableDiffusionInstructPixaPixPipeline.from_pretrained(
								    _A				, safety_checker=_A				, )
								pipe.to(_A      )
								pipe.set_progress_bar_config(disable=_A      )
								pipe.enable_attention_slicing()

								UpperCAmelCase_      :       Optional[Any]          =       pipe(**_A      )
								UpperCAmelCase_      :       Union[str, Any]          =       output.images[0]

								UpperCAmelCase_      :       List[str]          =       image[2_55:2_58, 3_83:3_86, -1]

								assert image.shape == (5_04, 5_04, 3)
								UpperCAmelCase_      :       List[str]          =       np.array([0.2_726, 0.2_529, 0.2_664, 0.2_655, 0.2_641, 0.2_642, 0.2_591, 0.2_649, 0.2_590]      )

								assert np.abs(image_slice.flatten() - expected_slice      ).max() < 5e-3


