

'''simple docstring'''

import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    UNetaDConditionModel,
)
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 PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()







class lowerCamelCase   (    __lowerCAmelCase				,				__lowerCAmelCase				,				unittest.TestCase				):
				snake_case_							     = StableDiffusionParadigmsPipeline
				snake_case_							     = TEXT_TO_IMAGE_PARAMS
				snake_case_							     = TEXT_TO_IMAGE_BATCH_PARAMS
				snake_case_							     = TEXT_TO_IMAGE_IMAGE_PARAMS
				snake_case_							     = TEXT_TO_IMAGE_IMAGE_PARAMS

				def 					_lowerCamelCase    (       self )    ->							int:
					torch.manual_seed(0 )
					snake_case							=					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,						attention_head_dim=(2, 4),						use_linear_projection=lowercase_,						)
					snake_case							=					DDIMParallelScheduler(
					    beta_start=0.00_085,						beta_end=0.012,						beta_schedule='scaled_linear',						clip_sample=lowercase_,						set_alpha_to_one=lowercase_,						)
					torch.manual_seed(0 )
					snake_case							=					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,						sample_size=128,						)
					torch.manual_seed(0 )
					snake_case							=					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=1000,						hidden_act='gelu',						projection_dim=512,						)
					snake_case							=					CLIPTextModel(lowercase_ )
					snake_case							=					CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip' )

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

				def 					_lowerCamelCase    (       self,						lowercase_,						lowercase_=0 )    ->							List[str]:
					if str(lowercase_ ).startswith('mps' ):
						snake_case							=					torch.manual_seed(lowercase_ )
					else:
						snake_case							=					torch.Generator(device=lowercase_ ).manual_seed(lowercase_ )
					snake_case							=					{
					    'prompt': 'a photograph of an astronaut riding a horse',
					    'generator': generator,
					    'num_inference_steps': 10,
					    'guidance_scale': 6.0,
					    'output_type': 'numpy',
					    'parallel': 3,
					    'debug': True,
					}
					return inputs

				def 					_lowerCamelCase    (       self )    ->							int:
					snake_case							=					'cpu'  # ensure determinism for the device-dependent torch.Generator
					snake_case							=					self.get_dummy_components()
					snake_case							=					StableDiffusionParadigmsPipeline(**lowercase_ )
					snake_case							=					sd_pipe.to(lowercase_ )
					sd_pipe.set_progress_bar_config(disable=lowercase_ )

					snake_case							=					self.get_dummy_inputs(lowercase_ )
					snake_case							=					sd_pipe(**lowercase_ ).images
					snake_case							=					image[0, -3:, -3:, -1]
					assert image.shape == (1, 64, 64, 3)

					snake_case							=					np.array([0.4_773, 0.5_417, 0.4_723, 0.4_925, 0.5_631, 0.4_752, 0.5_240, 0.4_935, 0.5_023] )

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

				def 					_lowerCamelCase    (       self )    ->							Optional[Any]:
					snake_case							=					'cpu'  # ensure determinism for the device-dependent torch.Generator
					snake_case							=					self.get_dummy_components()
					torch.manual_seed(0 )
					snake_case							=					DDPMParallelScheduler()
					torch.manual_seed(0 )
					snake_case							=					StableDiffusionParadigmsPipeline(**lowercase_ )
					snake_case							=					sd_pipe.to(lowercase_ )
					sd_pipe.set_progress_bar_config(disable=lowercase_ )

					snake_case							=					self.get_dummy_inputs(lowercase_ )
					snake_case							=					sd_pipe(**lowercase_ ).images
					snake_case							=					image[0, -3:, -3:, -1]
					assert image.shape == (1, 64, 64, 3)

					snake_case							=					np.array([0.3_573, 0.4_420, 0.4_960, 0.4_799, 0.3_796, 0.3_879, 0.4_819, 0.4_365, 0.4_468] )

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

				def 					_lowerCamelCase    (       self )    ->							List[Any]:
					super().test_inference_batch_consistent(batch_sizes=[1, 2] )

				def 					_lowerCamelCase    (       self )    ->							Optional[Any]:
					super().test_inference_batch_single_identical(batch_size=2,						expected_max_diff=3E-3 )






				def 					_lowerCamelCase    (       self )    ->							str:
					snake_case							=					'cpu'  # ensure determinism for the device-dependent torch.Generator
					snake_case							=					self.get_dummy_components()
					snake_case							=					StableDiffusionParadigmsPipeline(**lowercase_ )
					snake_case							=					sd_pipe.to(lowercase_ )
					sd_pipe.set_progress_bar_config(disable=lowercase_ )

					snake_case							=					self.get_dummy_inputs(lowercase_ )
					snake_case							=					'french fries'
					snake_case							=					sd_pipe(**lowercase_,						negative_prompt=lowercase_ )
					snake_case							=					output.images
					snake_case							=					image[0, -3:, -3:, -1]

					assert image.shape == (1, 64, 64, 3)

					snake_case							=					np.array([0.4_771, 0.5_420, 0.4_683, 0.4_918, 0.5_636, 0.4_725, 0.5_230, 0.4_923, 0.5_015] )

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







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

				def 					_lowerCamelCase    (       self )    ->							Tuple:
					super().tearDown()
					gc.collect()
					torch.cuda.empty_cache()

				def 					_lowerCamelCase    (       self,						lowercase_=0 )    ->							Optional[Any]:
					snake_case							=					torch.Generator(device=lowercase_ ).manual_seed(lowercase_ )
					snake_case							=					{
					    'prompt': 'a photograph of an astronaut riding a horse',
					    'generator': generator,
					    'num_inference_steps': 10,
					    'guidance_scale': 7.5,
					    'output_type': 'numpy',
					    'parallel': 3,
					    'debug': True,
					}
					return inputs






				def 					_lowerCamelCase    (       self )    ->							Optional[int]:
					snake_case							=					'stabilityai/stable-diffusion-2-base'
					snake_case							=					DDIMParallelScheduler.from_pretrained(lowercase_,						subfolder='scheduler' )
					snake_case							=					StableDiffusionParadigmsPipeline.from_pretrained(lowercase_,						scheduler=lowercase_,						safety_checker=lowercase_ )
					pipe.to(lowercase_ )
					pipe.set_progress_bar_config(disable=lowercase_ )
					pipe.enable_attention_slicing()

					snake_case							=					self.get_inputs()
					snake_case							=					pipe(**lowercase_ ).images
					snake_case							=					image[0, -3:, -3:, -1].flatten()

					assert image.shape == (1, 512, 512, 3)

					snake_case							=					np.array([0.9_622, 0.9_602, 0.9_748, 0.9_591, 0.9_630, 0.9_691, 0.9_661, 0.9_631, 0.9_741] )

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








'''simple docstring'''

from __future__ import annotations

import inspect
import unittest
from typing import List, Tuple

from transformers import ConvNextConfig
from transformers.testing_utils import require_tf, require_vision, slow
from transformers.utils import cached_property, is_tf_available, is_vision_available

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


if is_tf_available():
						import tensorflow as tf

						from transformers import TFConvNextForImageClassification, TFConvNextModel


if is_vision_available():
						from PIL import Image

						from transformers import ConvNextImageProcessor







class lowerCamelCase   :

				def __init__(       self,						lowercase_,						lowercase_=13,						lowercase_=32,						lowercase_=3,						lowercase_=4,						lowercase_=[10, 20, 30, 40],						lowercase_=[2, 2, 3, 2],						lowercase_=True,						lowercase_=True,						lowercase_=37,						lowercase_="gelu",						lowercase_=10,						lowercase_=0.02,						lowercase_=3,						lowercase_=None,						)    ->							Dict:
					snake_case							=					parent
					snake_case							=					batch_size
					snake_case							=					image_size
					snake_case							=					num_channels
					snake_case							=					num_stages
					snake_case							=					hidden_sizes
					snake_case							=					depths
					snake_case							=					is_training
					snake_case							=					use_labels
					snake_case							=					intermediate_size
					snake_case							=					hidden_act
					snake_case							=					type_sequence_label_size
					snake_case							=					initializer_range
					snake_case							=					scope

				def 					_lowerCamelCase    (       self )    ->							List[Any]:
					snake_case							=					floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size] )

					snake_case							=					None
					if self.use_labels:
						snake_case							=					ids_tensor([self.batch_size],						self.type_sequence_label_size )

					snake_case							=					self.get_config()

					return config, pixel_values, labels

				def 					_lowerCamelCase    (       self )    ->							Any:
					return ConvNextConfig(
					    num_channels=self.num_channels,						hidden_sizes=self.hidden_sizes,						depths=self.depths,						num_stages=self.num_stages,						hidden_act=self.hidden_act,						is_decoder=lowercase_,						initializer_range=self.initializer_range,						)

				def 					_lowerCamelCase    (       self,						lowercase_,						lowercase_,						lowercase_ )    ->							List[Any]:
					snake_case							=					TFConvNextModel(config=lowercase_ )
					snake_case							=					model(lowercase_,						training=lowercase_ )
					# expected last hidden states: B, C, H // 32, W // 32
					self.parent.assertEqual(
					    result.last_hidden_state.shape,						(self.batch_size, self.hidden_sizes[-1], self.image_size // 32, self.image_size // 32),						)

				def 					_lowerCamelCase    (       self,						lowercase_,						lowercase_,						lowercase_ )    ->							Union[str, Any]:
					snake_case							=					self.type_sequence_label_size
					snake_case							=					TFConvNextForImageClassification(lowercase_ )
					snake_case							=					model(lowercase_,						labels=lowercase_,						training=lowercase_ )
					self.parent.assertEqual(result.logits.shape,						(self.batch_size, self.type_sequence_label_size) )






				def 					_lowerCamelCase    (       self )    ->							Optional[int]:
					snake_case							=					self.prepare_config_and_inputs()
					snake_case			,			snake_case			,			snake_case							=					config_and_inputs
					snake_case							=					{'pixel_values': pixel_values}
					return config, inputs_dict







@require_tf
class lowerCamelCase   (    __lowerCAmelCase				,				__lowerCAmelCase				,				unittest.TestCase				):
				snake_case_							     = (TFConvNextModel, TFConvNextForImageClassification) if is_tf_available() else ()
				snake_case_							     = (
				    {'''feature-extraction''': TFConvNextModel, '''image-classification''': TFConvNextForImageClassification}
				    if is_tf_available()
				    else {}
				)

				snake_case_							     = False
				snake_case_							     = False
				snake_case_							     = False
				snake_case_							     = False
				snake_case_							     = False

				def 					_lowerCamelCase    (       self )    ->							str:
					snake_case							=					TFConvNextModelTester(self )
					snake_case							=					ConfigTester(
					    self,						config_class=lowercase_,						has_text_modality=lowercase_,						hidden_size=37,						)

				@unittest.skip(reason='ConvNext does not use inputs_embeds' )
				def 					_lowerCamelCase    (       self )    ->							Tuple:
					pass

				@unittest.skipIf(
				    not is_tf_available() or len(tf.config.list_physical_devices('GPU' ) ) == 0,						reason='TF does not support backprop for grouped convolutions on CPU.',						)
				@slow
				def 					_lowerCamelCase    (       self )    ->							Optional[Any]:
					super().test_keras_fit()

				@unittest.skip(reason='ConvNext does not support input and output embeddings' )
				def 					_lowerCamelCase    (       self )    ->							Dict:
					pass

				def 					_lowerCamelCase    (       self )    ->							List[Any]:
					snake_case			,			snake_case							=					self.model_tester.prepare_config_and_inputs_for_common()

					for model_class in self.all_model_classes:
						snake_case							=					model_class(lowercase_ )
						snake_case							=					inspect.signature(model.call )
						# signature.parameters is an OrderedDict => so arg_names order is deterministic
						snake_case							=					[*signature.parameters.keys()]

						snake_case							=					['pixel_values']
						self.assertListEqual(arg_names[:1],						lowercase_ )

				def 					_lowerCamelCase    (       self )    ->							str:
					snake_case							=					self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_model(*lowercase_ )

				@unittest.skipIf(
				    not is_tf_available() or len(tf.config.list_physical_devices('GPU' ) ) == 0,						reason='TF does not support backprop for grouped convolutions on CPU.',						)
				def 					_lowerCamelCase    (       self )    ->							Optional[int]:
					super().test_dataset_conversion()

				def 					_lowerCamelCase    (       self )    ->							Union[str, Any]:
					def check_hidden_states_output(lowercase_,						lowercase_,						lowercase_ ):
						snake_case							=					model_class(lowercase_ )

						snake_case							=					model(**self._prepare_for_class(lowercase_,						lowercase_ ) )
						snake_case							=					outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states

						snake_case							=					self.model_tester.num_stages
						self.assertEqual(len(lowercase_ ),						expected_num_stages + 1 )

						# ConvNext's feature maps are of shape (batch_size, num_channels, height, width)
						self.assertListEqual(
						    list(hidden_states[0].shape[-2:] ),						[self.model_tester.image_size // 4, self.model_tester.image_size // 4],						)

					snake_case			,			snake_case							=					self.model_tester.prepare_config_and_inputs_for_common()

					for model_class in self.all_model_classes:
						snake_case							=					True
						check_hidden_states_output(lowercase_,						lowercase_,						lowercase_ )

						# check that output_hidden_states also work using config
						del inputs_dict["output_hidden_states"]
						snake_case							=					True

						check_hidden_states_output(lowercase_,						lowercase_,						lowercase_ )

				def 					_lowerCamelCase    (       self )    ->							Dict:
					snake_case			,			snake_case							=					self.model_tester.prepare_config_and_inputs_for_common()

					def check_equivalence(lowercase_,						lowercase_,						lowercase_,						lowercase_={} ):
						snake_case							=					model(lowercase_,						return_dict=lowercase_,						**lowercase_ )
						snake_case							=					model(lowercase_,						return_dict=lowercase_,						**lowercase_ ).to_tuple()

						def recursive_check(lowercase_,						lowercase_ ):
							if isinstance(lowercase_,						(List, Tuple) ):
								for tuple_iterable_value, dict_iterable_value in zip(lowercase_,						lowercase_ ):
									recursive_check(lowercase_,						lowercase_ )
							elif tuple_object is None:
								return
							else:
								self.assertTrue(
								    all(tf.equal(lowercase_,						lowercase_ ) ),						msg=(
								        'Tuple and dict output are not equal. Difference:'
								        F''' {tf.math.reduce_max(tf.abs(tuple_object - dict_object ) )}'''
								    ),						)

							recursive_check(lowercase_,						lowercase_ )

					for model_class in self.all_model_classes:
						snake_case							=					model_class(lowercase_ )

						snake_case							=					self._prepare_for_class(lowercase_,						lowercase_ )
						snake_case							=					self._prepare_for_class(lowercase_,						lowercase_ )
						check_equivalence(lowercase_,						lowercase_,						lowercase_ )

						snake_case							=					self._prepare_for_class(lowercase_,						lowercase_,						return_labels=lowercase_ )
						snake_case							=					self._prepare_for_class(lowercase_,						lowercase_,						return_labels=lowercase_ )
						check_equivalence(lowercase_,						lowercase_,						lowercase_ )

						snake_case							=					self._prepare_for_class(lowercase_,						lowercase_ )
						snake_case							=					self._prepare_for_class(lowercase_,						lowercase_ )
						check_equivalence(lowercase_,						lowercase_,						lowercase_,						{'output_hidden_states': True} )

						snake_case							=					self._prepare_for_class(lowercase_,						lowercase_,						return_labels=lowercase_ )
						snake_case							=					self._prepare_for_class(lowercase_,						lowercase_,						return_labels=lowercase_ )
						check_equivalence(lowercase_,						lowercase_,						lowercase_,						{'output_hidden_states': True} )

				def 					_lowerCamelCase    (       self )    ->							List[str]:
					snake_case							=					self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_for_image_classification(*lowercase_ )






				@slow
				def 					_lowerCamelCase    (       self )    ->							str:
					snake_case							=					TFConvNextModel.from_pretrained('facebook/convnext-tiny-224' )
					self.assertIsNotNone(lowercase_ )

def      __magic_name__							(			)      ->		Optional[int]:
	snake_case							=					Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png'				)
	return image







@require_tf
@require_vision
class lowerCamelCase   (    unittest.TestCase				):

				@cached_property
				def 					_lowerCamelCase    (       self )    ->							List[str]:
					return ConvNextImageProcessor.from_pretrained('facebook/convnext-tiny-224' ) if is_vision_available() else None






				@slow
				def 					_lowerCamelCase    (       self )    ->							Dict:
					snake_case							=					TFConvNextForImageClassification.from_pretrained('facebook/convnext-tiny-224' )

					snake_case							=					self.default_image_processor
					snake_case							=					prepare_img()
					snake_case							=					image_processor(images=lowercase_,						return_tensors='tf' )

					# forward pass
					snake_case							=					model(**lowercase_ )

					# verify the logits
					snake_case							=					tf.TensorShape((1, 1000) )
					self.assertEqual(outputs.logits.shape,						lowercase_ )

					snake_case							=					tf.constant([-0.0_260, -0.4_739, 0.1_911] )

					tf.debugging.assert_near(outputs.logits[0, :3],						lowercase_,						atol=1E-4 )





