import gc
import random
import unittest

import numpy as np
import torch

from diffusers import (
    DDIMScheduler,
    KandinskyVaaControlnetPipeline,
    KandinskyVaaPriorPipeline,
    UNetaDConditionModel,
    VQModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()






class 						A   (   __snake_case       ,	unittest.TestCase   ):
							a_        =      KandinskyVaaControlnetPipeline
							a_        =      ['image_embeds', 'negative_image_embeds', 'hint']
							a_        =      ['image_embeds', 'negative_image_embeds', 'hint']
							a_        =      [
							    'generator',
							    'height',
							    'width',
							    'latents',
							    'guidance_scale',
							    'num_inference_steps',
							    'return_dict',
							    'guidance_scale',
							    'num_images_per_prompt',
							    'output_type',
							    'return_dict',
							]
							a_        =      False





							@property
							def       snake_case__						(      self				:							List[str]	)							->	Dict:
								return 3_2





							@property
							def       snake_case__						(      self				:							Any	)							->	Optional[Any]:
								return 3_2





							@property
							def       snake_case__						(      self				:							Tuple	)							->	Any:
								return self.time_input_dim





							@property
							def       snake_case__						(      self				:							int	)							->	Union[str, Any]:
								return self.time_input_dim * 4





							@property
							def       snake_case__						(      self				:							str	)							->	Any:
								return 1_0_0





							@property
							def       snake_case__						(      self				:							List[Any]	)							->	str:
								torch.manual_seed(0	)

								__UpperCAmelCase							=			{
								    "in_channels": 8,
								    # Out channels is double in channels because predicts mean and variance
								    "out_channels": 8,
								    "addition_embed_type": "image_hint",
								    "down_block_types": ("ResnetDownsampleBlock2D", "SimpleCrossAttnDownBlock2D"),
								    "up_block_types": ("SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"),
								    "mid_block_type": "UNetMidBlock2DSimpleCrossAttn",
								    "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
								    "layers_per_block": 1,
								    "encoder_hid_dim": self.text_embedder_hidden_size,
								    "encoder_hid_dim_type": "image_proj",
								    "cross_attention_dim": self.cross_attention_dim,
								    "attention_head_dim": 4,
								    "resnet_time_scale_shift": "scale_shift",
								    "class_embed_type": None,
								}

								__UpperCAmelCase							=			UNetaDConditionModel(**A_	)
								return model





							@property
							def       snake_case__						(      self				:							int	)							->	List[str]:
								return {
								    "block_out_channels": [3_2, 3_2, 6_4, 6_4],
								    "down_block_types": [
								        "DownEncoderBlock2D",
								        "DownEncoderBlock2D",
								        "DownEncoderBlock2D",
								        "AttnDownEncoderBlock2D",
								    ],
								    "in_channels": 3,
								    "latent_channels": 4,
								    "layers_per_block": 1,
								    "norm_num_groups": 8,
								    "norm_type": "spatial",
								    "num_vq_embeddings": 1_2,
								    "out_channels": 3,
								    "up_block_types": ["AttnUpDecoderBlock2D", "UpDecoderBlock2D", "UpDecoderBlock2D", "UpDecoderBlock2D"],
								    "vq_embed_dim": 4,
								}





							@property
							def       snake_case__						(      self				:							Optional[int]	)							->	str:
								torch.manual_seed(0	)
								__UpperCAmelCase							=			VQModel(**self.dummy_movq_kwargs	)
								return model





							def       snake_case__						(      self				:							Dict	)							->	List[str]:
								__UpperCAmelCase							=			self.dummy_unet
								__UpperCAmelCase							=			self.dummy_movq

								__UpperCAmelCase							=			DDIMScheduler(
								    num_train_timesteps=1_0_0_0     ,		beta_schedule='''linear'''     ,		beta_start=0.0_0_0_8_5     ,		beta_end=0.0_1_2     ,		clip_sample=A_     ,		set_alpha_to_one=A_     ,		steps_offset=1     ,		prediction_type='''epsilon'''     ,		thresholding=A_     ,		)

								__UpperCAmelCase							=			{
								    "unet": unet,
								    "scheduler": scheduler,
								    "movq": movq,
								}
								return components





							def       snake_case__						(      self				:							int     ,		__a				:							Optional[Any]     ,		__a				:							Union[str, Any]=0	)							->	List[str]:
								__UpperCAmelCase							=			floats_tensor((1, self.text_embedder_hidden_size)     ,		rng=random.Random(A_	)	).to(A_	)
								__UpperCAmelCase							=			floats_tensor((1, self.text_embedder_hidden_size)     ,		rng=random.Random(seed + 1	)	).to(
								    A_	)

								# create hint
								__UpperCAmelCase							=			floats_tensor((1, 3, 6_4, 6_4)     ,		rng=random.Random(A_	)	).to(A_	)

								if str(A_	).startswith('''mps'''	):
									__UpperCAmelCase							=			torch.manual_seed(A_	)
								else:
									__UpperCAmelCase							=			torch.Generator(device=A_	).manual_seed(A_	)
								__UpperCAmelCase							=			{
								    "image_embeds": image_embeds,
								    "negative_image_embeds": negative_image_embeds,
								    "hint": hint,
								    "generator": generator,
								    "height": 6_4,
								    "width": 6_4,
								    "guidance_scale": 4.0,
								    "num_inference_steps": 2,
								    "output_type": "np",
								}
								return inputs






							def       snake_case__						(      self				:							Dict	)							->	List[Any]:
								__UpperCAmelCase							=			"cpu"

								__UpperCAmelCase							=			self.get_dummy_components()

								__UpperCAmelCase							=			self.pipeline_class(**A_	)
								__UpperCAmelCase							=			pipe.to(A_	)

								pipe.set_progress_bar_config(disable=A_	)

								__UpperCAmelCase							=			pipe(**self.get_dummy_inputs(A_	)	)
								__UpperCAmelCase							=			output.images

								__UpperCAmelCase							=			pipe(
								    **self.get_dummy_inputs(A_	)     ,		return_dict=A_     ,		)[0]

								__UpperCAmelCase							=			image[0, -3:, -3:, -1]
								__UpperCAmelCase							=			image_from_tuple[0, -3:, -3:, -1]

								assert image.shape == (1, 6_4, 6_4, 3)

								__UpperCAmelCase							=			np.array(
								    [0.6_9_5_9_8_2_6, 0.8_6_8_2_7_9, 0.7_5_5_8_0_9_2, 0.6_8_7_6_9_4_6_7, 0.8_5_8_0_5_8_0_4, 0.6_5_9_7_7_4_9_6, 0.4_4_8_8_5_3_0_2, 0.5_9_5_9_1_1_1, 0.4_2_5_1_5_9_5]	)

								assert (
								    np.abs(image_slice.flatten() - expected_slice	).max() < 1e-2
								), f""" expected_slice {expected_slice}, but got {image_slice.flatten()}"""

								assert (
								    np.abs(image_from_tuple_slice.flatten() - expected_slice	).max() < 1e-2
								), f""" expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}"""






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





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






							def       snake_case__						(      self				:							Optional[int]	)							->	Union[str, Any]:
								__UpperCAmelCase							=			load_numpy(
								    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
								    '''/kandinskyv22/kandinskyv22_controlnet_robotcat_fp16.npy'''	)

								__UpperCAmelCase							=			load_image(
								    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
								    '''/kandinskyv22/hint_image_cat.png'''	)
								__UpperCAmelCase							=			torch.from_numpy(np.array(A_	)	).float() / 2_5_5.0
								__UpperCAmelCase							=			hint.permute(2     ,		0     ,		1	).unsqueeze(0	)

								__UpperCAmelCase							=			KandinskyVaaPriorPipeline.from_pretrained(
								    '''kandinsky-community/kandinsky-2-2-prior'''     ,		torch_dtype=torch.floataa	)
								pipe_prior.to(A_	)

								__UpperCAmelCase							=			KandinskyVaaControlnetPipeline.from_pretrained(
								    '''kandinsky-community/kandinsky-2-2-controlnet-depth'''     ,		torch_dtype=torch.floataa	)
								__UpperCAmelCase							=			pipeline.to(A_	)
								pipeline.set_progress_bar_config(disable=A_	)

								__UpperCAmelCase							=			"A robot, 4k photo"

								__UpperCAmelCase							=			torch.Generator(device='''cuda'''	).manual_seed(0	)
								__UpperCAmelCase							=			pipe_prior(
								    A_     ,		generator=A_     ,		num_inference_steps=5     ,		negative_prompt=''''''     ,		).to_tuple()

								__UpperCAmelCase							=			torch.Generator(device='''cuda'''	).manual_seed(0	)
								__UpperCAmelCase							=			pipeline(
								    image_embeds=A_     ,		negative_image_embeds=A_     ,		hint=A_     ,		generator=A_     ,		num_inference_steps=1_0_0     ,		output_type='''np'''     ,		)

								__UpperCAmelCase							=			output.images[0]

								assert image.shape == (5_1_2, 5_1_2, 3)

								assert_mean_pixel_difference(A_     ,		A_	)

'''simple docstring'''


from __future__ import annotations

import unittest

from transformers import DistilBertConfig, is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
					import tensorflow as tf

					from transformers.models.distilbert.modeling_tf_distilbert import (
					    TF_DISTILBERT_PRETRAINED_MODEL_ARCHIVE_LIST,
					    TFDistilBertForMaskedLM,
					    TFDistilBertForMultipleChoice,
					    TFDistilBertForQuestionAnswering,
					    TFDistilBertForSequenceClassification,
					    TFDistilBertForTokenClassification,
					    TFDistilBertModel,
					)






class 						A   :





							def __init__(      self				:							List[Any]     ,		__a				:							Any     ,		)							->	Dict:
								__UpperCAmelCase							=			parent
								__UpperCAmelCase							=			1_3
								__UpperCAmelCase							=			7
								__UpperCAmelCase							=			True
								__UpperCAmelCase							=			True
								__UpperCAmelCase							=			False
								__UpperCAmelCase							=			True
								__UpperCAmelCase							=			9_9
								__UpperCAmelCase							=			3_2
								__UpperCAmelCase							=			2
								__UpperCAmelCase							=			4
								__UpperCAmelCase							=			3_7
								__UpperCAmelCase							=			'''gelu'''
								__UpperCAmelCase							=			0.1
								__UpperCAmelCase							=			0.1
								__UpperCAmelCase							=			5_1_2
								__UpperCAmelCase							=			1_6
								__UpperCAmelCase							=			2
								__UpperCAmelCase							=			0.0_2
								__UpperCAmelCase							=			3
								__UpperCAmelCase							=			4
								__UpperCAmelCase							=			None





							def       snake_case__						(      self				:							Optional[int]	)							->	Dict:
								__UpperCAmelCase							=			ids_tensor([self.batch_size, self.seq_length]     ,		self.vocab_size	)

								__UpperCAmelCase							=			None
								if self.use_input_mask:
									__UpperCAmelCase							=			random_attention_mask([self.batch_size, self.seq_length]	)

								__UpperCAmelCase							=			None
								__UpperCAmelCase							=			None
								__UpperCAmelCase							=			None
								if self.use_labels:
									__UpperCAmelCase							=			ids_tensor([self.batch_size]     ,		self.type_sequence_label_size	)
									__UpperCAmelCase							=			ids_tensor([self.batch_size, self.seq_length]     ,		self.num_labels	)
									__UpperCAmelCase							=			ids_tensor([self.batch_size]     ,		self.num_choices	)

								__UpperCAmelCase							=			DistilBertConfig(
								    vocab_size=self.vocab_size     ,		dim=self.hidden_size     ,		n_layers=self.num_hidden_layers     ,		n_heads=self.num_attention_heads     ,		hidden_dim=self.intermediate_size     ,		hidden_act=self.hidden_act     ,		dropout=self.hidden_dropout_prob     ,		attention_dropout=self.attention_probs_dropout_prob     ,		max_position_embeddings=self.max_position_embeddings     ,		initializer_range=self.initializer_range     ,		)

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





							def       snake_case__						(      self				:							Union[str, Any]     ,		__a				:							List[str]     ,		__a				:							int     ,		__a				:							Union[str, Any]     ,		__a				:							Union[str, Any]     ,		__a				:							List[Any]     ,		__a				:							int	)							->	Any:
								__UpperCAmelCase							=			TFDistilBertModel(config=__a	)
								__UpperCAmelCase							=			{'''input_ids''': input_ids, '''attention_mask''': input_mask}

								__UpperCAmelCase							=			model(__a	)

								__UpperCAmelCase							=			[input_ids, input_mask]

								__UpperCAmelCase							=			model(__a	)

								self.parent.assertEqual(result.last_hidden_state.shape     ,		(self.batch_size, self.seq_length, self.hidden_size)	)





							def       snake_case__						(      self				:							Tuple     ,		__a				:							List[Any]     ,		__a				:							int     ,		__a				:							Tuple     ,		__a				:							List[Any]     ,		__a				:							Union[str, Any]     ,		__a				:							List[Any]	)							->	int:
								__UpperCAmelCase							=			TFDistilBertForMaskedLM(config=__a	)
								__UpperCAmelCase							=			{'''input_ids''': input_ids, '''attention_mask''': input_mask}
								__UpperCAmelCase							=			model(__a	)
								self.parent.assertEqual(result.logits.shape     ,		(self.batch_size, self.seq_length, self.vocab_size)	)





							def       snake_case__						(      self				:							Optional[int]     ,		__a				:							Any     ,		__a				:							Union[str, Any]     ,		__a				:							Optional[int]     ,		__a				:							int     ,		__a				:							Optional[Any]     ,		__a				:							Optional[int]	)							->	Dict:
								__UpperCAmelCase							=			TFDistilBertForQuestionAnswering(config=__a	)
								__UpperCAmelCase							=			{
								    '''input_ids''': input_ids,
								    '''attention_mask''': input_mask,
								}
								__UpperCAmelCase							=			model(__a	)
								self.parent.assertEqual(result.start_logits.shape     ,		(self.batch_size, self.seq_length)	)
								self.parent.assertEqual(result.end_logits.shape     ,		(self.batch_size, self.seq_length)	)





							def       snake_case__						(      self				:							Any     ,		__a				:							Optional[Any]     ,		__a				:							List[str]     ,		__a				:							Dict     ,		__a				:							Dict     ,		__a				:							int     ,		__a				:							List[Any]	)							->	Dict:
								__UpperCAmelCase							=			self.num_labels
								__UpperCAmelCase							=			TFDistilBertForSequenceClassification(__a	)
								__UpperCAmelCase							=			{'''input_ids''': input_ids, '''attention_mask''': input_mask}
								__UpperCAmelCase							=			model(__a	)
								self.parent.assertEqual(result.logits.shape     ,		(self.batch_size, self.num_labels)	)





							def       snake_case__						(      self				:							Union[str, Any]     ,		__a				:							int     ,		__a				:							str     ,		__a				:							Union[str, Any]     ,		__a				:							Optional[int]     ,		__a				:							List[str]     ,		__a				:							Dict	)							->	str:
								__UpperCAmelCase							=			self.num_choices
								__UpperCAmelCase							=			TFDistilBertForMultipleChoice(__a	)
								__UpperCAmelCase							=			tf.tile(tf.expand_dims(__a     ,		1	)     ,		(1, self.num_choices, 1)	)
								__UpperCAmelCase							=			tf.tile(tf.expand_dims(__a     ,		1	)     ,		(1, self.num_choices, 1)	)
								__UpperCAmelCase							=			{
								    '''input_ids''': multiple_choice_inputs_ids,
								    '''attention_mask''': multiple_choice_input_mask,
								}
								__UpperCAmelCase							=			model(__a	)
								self.parent.assertEqual(result.logits.shape     ,		(self.batch_size, self.num_choices)	)





							def       snake_case__						(      self				:							int     ,		__a				:							Optional[Any]     ,		__a				:							int     ,		__a				:							Tuple     ,		__a				:							int     ,		__a				:							Optional[int]     ,		__a				:							Optional[int]	)							->	int:
								__UpperCAmelCase							=			self.num_labels
								__UpperCAmelCase							=			TFDistilBertForTokenClassification(__a	)
								__UpperCAmelCase							=			{'''input_ids''': input_ids, '''attention_mask''': input_mask}
								__UpperCAmelCase							=			model(__a	)
								self.parent.assertEqual(result.logits.shape     ,		(self.batch_size, self.seq_length, self.num_labels)	)






							def       snake_case__						(      self				:							str	)							->	Any:
								__UpperCAmelCase							=			self.prepare_config_and_inputs()
								((__UpperCAmelCase)				,			(__UpperCAmelCase)				,			(__UpperCAmelCase)				,			(__UpperCAmelCase)				,			(__UpperCAmelCase)				,			(__UpperCAmelCase))							=			config_and_inputs
								__UpperCAmelCase							=			{'''input_ids''': input_ids, '''attention_mask''': input_mask}
								return config, inputs_dict






@require_tf
class 						A   (   UpperCAmelCase       ,	UpperCAmelCase       ,	unittest.TestCase   ):
							a_        =      (
							    (
							        TFDistilBertModel,
							        TFDistilBertForMaskedLM,
							        TFDistilBertForQuestionAnswering,
							        TFDistilBertForSequenceClassification,
							        TFDistilBertForTokenClassification,
							        TFDistilBertForMultipleChoice,
							    )
							    if is_tf_available()
							    else None
							)
							a_        =      (
							    {
							        '''feature-extraction''': TFDistilBertModel,
							        '''fill-mask''': TFDistilBertForMaskedLM,
							        '''question-answering''': TFDistilBertForQuestionAnswering,
							        '''text-classification''': TFDistilBertForSequenceClassification,
							        '''token-classification''': TFDistilBertForTokenClassification,
							        '''zero-shot''': TFDistilBertForSequenceClassification,
							    }
							    if is_tf_available()
							    else {}
							)
							a_        =      False
							a_        =      False





							def       snake_case__						(      self				:							Any	)							->	Any:
								__UpperCAmelCase							=			TFDistilBertModelTester(self	)
								__UpperCAmelCase							=			ConfigTester(self     ,		config_class=__a     ,		dim=3_7	)





							def       snake_case__						(      self				:							List[Any]	)							->	Optional[int]:
								self.config_tester.run_common_tests()





							def       snake_case__						(      self				:							Any	)							->	str:
								__UpperCAmelCase							=			self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_distilbert_model(*__a	)





							def       snake_case__						(      self				:							Tuple	)							->	Dict:
								__UpperCAmelCase							=			self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_distilbert_for_masked_lm(*__a	)





							def       snake_case__						(      self				:							Union[str, Any]	)							->	Any:
								__UpperCAmelCase							=			self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_distilbert_for_question_answering(*__a	)





							def       snake_case__						(      self				:							Optional[Any]	)							->	Dict:
								__UpperCAmelCase							=			self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_distilbert_for_sequence_classification(*__a	)





							def       snake_case__						(      self				:							Any	)							->	int:
								__UpperCAmelCase							=			self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_distilbert_for_multiple_choice(*__a	)





							def       snake_case__						(      self				:							List[str]	)							->	List[Any]:
								__UpperCAmelCase							=			self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_distilbert_for_token_classification(*__a	)






							@slow
							def       snake_case__						(      self				:							Dict	)							->	Tuple:
								for model_name in list(TF_DISTILBERT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]	):
									__UpperCAmelCase							=			TFDistilBertModel.from_pretrained(__a	)
									self.assertIsNotNone(__a	)






@require_tf
class 						A   (   unittest.TestCase   ):





							@slow
							def       snake_case__						(      self				:							int	)							->	Dict:
								__UpperCAmelCase							=			TFDistilBertModel.from_pretrained('''distilbert-base-uncased'''	)
								__UpperCAmelCase							=			tf.constant([[0, 1, 2, 3, 4, 5]]	)
								__UpperCAmelCase							=			model(__a	)[0]

								__UpperCAmelCase							=			[1, 6, 7_6_8]
								self.assertEqual(output.shape     ,		__a	)

								__UpperCAmelCase							=			tf.constant(
								    [
								        [
								            [0.1_9_2_6_1_8_8_5, -0.1_3_7_3_2_9_5_5, 0.4_1_1_9_7_9_9],
								            [0.2_2_1_5_0_1_5_6, -0.0_7_4_2_2_6_6_1, 0.3_9_0_3_7_2_0_4],
								            [0.2_2_7_5_6_0_1_8, -0.0_8_9_6_4_1_4, 0.3_7_0_1_4_6_7],
								        ]
								    ]	)
								tf.debugging.assert_near(output[:, :3, :3]     ,		__a     ,		atol=1e-4	)
