
"""simple docstring"""



import gc
import random
import unittest

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

import diffusers
from diffusers import (
    AutoencoderKL,
    EulerDiscreteScheduler,
    StableDiffusionLatentUpscalePipeline,
    StableDiffusionPipeline,
    UNetaDConditionModel,
)
from diffusers.schedulers import KarrasDiffusionSchedulers
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 ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()






def 						__snake_case	(     SCREAMING_SNAKE_CASE:       List[str]    ):






							"""simple docstring"""



							_lowerCAmelCase							=      [tensor.shape for tensor in tensor_list]
							return all(shape == shapes[0] for shape in shapes[1:]    )
class 							_SCREAMING_SNAKE_CASE	(	UpperCamelCase__   ,					UpperCamelCase__   ,					UpperCamelCase__   ,					unittest.TestCase				):






				'''simple docstring'''



				SCREAMING_SNAKE_CASE_:				Optional[int]				=       StableDiffusionLatentUpscalePipeline
				SCREAMING_SNAKE_CASE_:				Any				=       TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
				    "height",
				    "width",
				    "cross_attention_kwargs",
				    "negative_prompt_embeds",
				    "prompt_embeds",
				}
				SCREAMING_SNAKE_CASE_:				Dict				=       PipelineTesterMixin.required_optional_params - {"num_images_per_prompt"}
				SCREAMING_SNAKE_CASE_:				Optional[Any]				=       TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
				SCREAMING_SNAKE_CASE_:				Any				=       frozenset(
				    []				)  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
				SCREAMING_SNAKE_CASE_:				Tuple				=       frozenset([]				)

				SCREAMING_SNAKE_CASE_:				List[str]				=       True


				@property
				def 					__lowerCamelCase						(       self						:      Union[str, Any]	)		->	Optional[Any]:







											"""simple docstring"""




											_lowerCAmelCase							=      1
											_lowerCAmelCase							=      4
											_lowerCAmelCase							=      (16, 16)

											_lowerCAmelCase							=      floats_tensor((batch_size, num_channels) + sizes					,  rng=random.Random(0	)	).to(__lowerCamelCase	)
											return image


				def 					__lowerCamelCase						(       self						:      List[str]	)		->	int:







											"""simple docstring"""




											torch.manual_seed(0	)
											_lowerCAmelCase							=      UNetaDConditionModel(
											    act_fn='gelu'					,  attention_head_dim=8					,  norm_num_groups=__lowerCamelCase					,  block_out_channels=[32, 32, 64, 64]					,  time_cond_proj_dim=160					,  conv_in_kernel=1					,  conv_out_kernel=1					,  cross_attention_dim=32					,  down_block_types=(
											        'KDownBlock2D',
											        'KCrossAttnDownBlock2D',
											        'KCrossAttnDownBlock2D',
											        'KCrossAttnDownBlock2D',
											    )					,  in_channels=8					,  mid_block_type=__lowerCamelCase					,  only_cross_attention=__lowerCamelCase					,  out_channels=5					,  resnet_time_scale_shift='scale_shift'					,  time_embedding_type='fourier'					,  timestep_post_act='gelu'					,  up_block_types=('KCrossAttnUpBlock2D', 'KCrossAttnUpBlock2D', 'KCrossAttnUpBlock2D', 'KUpBlock2D')					,  )
											_lowerCAmelCase							=      AutoencoderKL(
											    block_out_channels=[32, 32, 64, 64]					,  in_channels=3					,  out_channels=3					,  down_block_types=[
											        'DownEncoderBlock2D',
											        'DownEncoderBlock2D',
											        'DownEncoderBlock2D',
											        'DownEncoderBlock2D',
											    ]					,  up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D', 'UpDecoderBlock2D', 'UpDecoderBlock2D']					,  latent_channels=4					,  )
											_lowerCAmelCase							=      EulerDiscreteScheduler(prediction_type='sample'	)
											_lowerCAmelCase							=      CLIPTextConfig(
											    bos_token_id=0					,  eos_token_id=2					,  hidden_size=32					,  intermediate_size=37					,  layer_norm_eps=1E-0_5					,  num_attention_heads=4					,  num_hidden_layers=5					,  pad_token_id=1					,  vocab_size=1_000					,  hidden_act='quick_gelu'					,  projection_dim=512					,  )
											_lowerCAmelCase							=      CLIPTextModel(__lowerCamelCase	)
											_lowerCAmelCase							=      CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'	)

											_lowerCAmelCase							=      {
											    'unet': model.eval(),
											    'vae': vae.eval(),
											    'scheduler': scheduler,
											    'text_encoder': text_encoder,
											    'tokenizer': tokenizer,
											}

											return components


				def 					__lowerCamelCase						(       self						:      Dict					,  UpperCAmelCase_						:      List[Any]					,  UpperCAmelCase_						:      Any=0	)		->	int:







											"""simple docstring"""




											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',
											    'image': self.dummy_image.cpu(),
											    'generator': generator,
											    'num_inference_steps': 2,
											    'output_type': 'numpy',
											}
											return inputs


				def 					__lowerCamelCase						(       self						:      List[Any]	)		->	List[str]:







											"""simple docstring"""




											_lowerCAmelCase							=      'cpu'

											_lowerCAmelCase							=      self.get_dummy_components()
											_lowerCAmelCase							=      self.pipeline_class(**__lowerCamelCase	)
											pipe.to(__lowerCamelCase	)
											pipe.set_progress_bar_config(disable=__lowerCamelCase	)

											_lowerCAmelCase							=      self.get_dummy_inputs(__lowerCamelCase	)
											_lowerCAmelCase							=      pipe(**__lowerCamelCase	).images
											_lowerCAmelCase							=      image[0, -3:, -3:, -1]

											self.assertEqual(image.shape					,  (1, 256, 256, 3)	)
											_lowerCAmelCase							=      np.array(
											    [0.47222412, 0.41921633, 0.44717434, 0.46874192, 0.42588258, 0.46150726, 0.4677534, 0.45583832, 0.48579055]	)
											_lowerCAmelCase							=      np.abs(image_slice.flatten() - expected_slice	).max()
											self.assertLessEqual(__lowerCamelCase					,  1E-3	)


				def 					__lowerCamelCase						(       self						:      Union[str, Any]	)		->	Dict:







											"""simple docstring"""




											super().test_attention_slicing_forward_pass(expected_max_diff=7E-3	)


				def 					__lowerCamelCase						(       self						:      Optional[Any]	)		->	List[str]:







											"""simple docstring"""




											super().test_cpu_offload_forward_pass(expected_max_diff=3E-3	)


				def 					__lowerCamelCase						(       self						:      Optional[Any]	)		->	List[str]:







											"""simple docstring"""




											super().test_dict_tuple_outputs_equivalent(expected_max_difference=3E-3	)


				def 					__lowerCamelCase						(       self						:      Any	)		->	List[Any]:







											"""simple docstring"""




											super().test_inference_batch_single_identical(expected_max_diff=7E-3	)


				def 					__lowerCamelCase						(       self						:      str	)		->	Dict:







											"""simple docstring"""




											super().test_pt_np_pil_outputs_equivalent(expected_max_diff=3E-3	)


				def 					__lowerCamelCase						(       self						:      str	)		->	Any:







											"""simple docstring"""




											super().test_save_load_local(expected_max_difference=3E-3	)


				def 					__lowerCamelCase						(       self						:      List[Any]	)		->	Optional[Any]:







											"""simple docstring"""




											super().test_save_load_optional_components(expected_max_difference=3E-3	)







				def 					__lowerCamelCase						(       self						:      Optional[Any]	)		->	int:







											"""simple docstring"""




											_lowerCAmelCase							=      [
											    'DDIMScheduler',
											    'DDPMScheduler',
											    'PNDMScheduler',
											    'HeunDiscreteScheduler',
											    'EulerAncestralDiscreteScheduler',
											    'KDPM2DiscreteScheduler',
											    'KDPM2AncestralDiscreteScheduler',
											    'DPMSolverSDEScheduler',
											]
											_lowerCAmelCase							=      self.get_dummy_components()
											_lowerCAmelCase							=      self.pipeline_class(**__lowerCamelCase	)

											# make sure that PNDM does not need warm-up
											pipe.scheduler.register_to_config(skip_prk_steps=__lowerCamelCase	)

											pipe.to(__lowerCamelCase	)
											pipe.set_progress_bar_config(disable=__lowerCamelCase	)
											_lowerCAmelCase							=      self.get_dummy_inputs(__lowerCamelCase	)
											_lowerCAmelCase							=      2

											_lowerCAmelCase							=      []
											for scheduler_enum in KarrasDiffusionSchedulers:
																		if scheduler_enum.name in skip_schedulers:
																									# no sigma schedulers are not supported
																									# no schedulers
																									continue

																		_lowerCAmelCase							=      getattr(__lowerCamelCase					,  scheduler_enum.name	)
																		_lowerCAmelCase							=      scheduler_cls.from_config(pipe.scheduler.config	)
																		_lowerCAmelCase							=      pipe(**__lowerCamelCase	)[0]
																		outputs.append(__lowerCamelCase	)

											assert check_same_shape(__lowerCamelCase	)






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






				'''simple docstring'''



				def 					__lowerCamelCase						(       self						:      Optional[int]	)		->	Tuple:







											"""simple docstring"""




											super().tearDown()
											gc.collect()
											torch.cuda.empty_cache()


				def 					__lowerCamelCase						(       self						:      List[str]	)		->	int:







											"""simple docstring"""




											_lowerCAmelCase							=      torch.manual_seed(33	)

											_lowerCAmelCase							=      StableDiffusionPipeline.from_pretrained('CompVis/stable-diffusion-v1-4'					,  torch_dtype=torch.floataa	)
											pipe.to('cuda'	)

											_lowerCAmelCase							=      StableDiffusionLatentUpscalePipeline.from_pretrained(
											    'stabilityai/sd-x2-latent-upscaler'					,  torch_dtype=torch.floataa	)
											upscaler.to('cuda'	)

											_lowerCAmelCase							=      'a photo of an astronaut high resolution, unreal engine, ultra realistic'

											_lowerCAmelCase							=      pipe(__lowerCamelCase					,  generator=__lowerCamelCase					,  output_type='latent'	).images

											_lowerCAmelCase							=      upscaler(
											    prompt=__lowerCamelCase					,  image=__lowerCamelCase					,  num_inference_steps=20					,  guidance_scale=0					,  generator=__lowerCamelCase					,  output_type='np'					,  ).images[0]

											_lowerCAmelCase							=      load_numpy(
											    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/astronaut_1024.npy'	)
											assert np.abs((expected_image - image).mean()	) < 5E-2







				def 					__lowerCamelCase						(       self						:      str	)		->	Union[str, Any]:







											"""simple docstring"""




											_lowerCAmelCase							=      torch.manual_seed(33	)

											_lowerCAmelCase							=      StableDiffusionLatentUpscalePipeline.from_pretrained(
											    'stabilityai/sd-x2-latent-upscaler'					,  torch_dtype=torch.floataa	)
											upscaler.to('cuda'	)

											_lowerCAmelCase							=      'the temple of fire by Ross Tran and Gerardo Dottori, oil on canvas'

											_lowerCAmelCase							=      load_image(
											    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/fire_temple_512.png'	)

											_lowerCAmelCase							=      upscaler(
											    prompt=__lowerCamelCase					,  image=__lowerCamelCase					,  num_inference_steps=20					,  guidance_scale=0					,  generator=__lowerCamelCase					,  output_type='np'					,  ).images[0]

											_lowerCAmelCase							=      load_numpy(
											    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/fire_temple_1024.npy'	)
											assert np.abs((expected_image - image).max()	) < 5E-2













import json
import os
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from requests.exceptions import HTTPError

from transformers.utils import (
    CONFIG_NAME,
    FLAX_WEIGHTS_NAME,
    TF2_WEIGHTS_NAME,
    TRANSFORMERS_CACHE,
    WEIGHTS_NAME,
    cached_file,
    get_file_from_repo,
    has_file,
)


a__: Union[str, Any]							   =						'hf-internal-testing/tiny-random-bert'
a__: Optional[int]							   =						os.path.join(TRANSFORMERS_CACHE, 'models--hf-internal-testing--tiny-random-bert')
a__: List[str]							   =						'9b8c223d42b2188cb49d29af482996f9d0f3e5a6'




class        SCREAMING_SNAKE_CASE__					(		unittest.TestCase ):



 def 							UpperCamelCase     (				self					):
        A__		   =		cached_file(__lowerCamelCase,__lowerCamelCase					)
        # Should have downloaded the file in here
        self.assertTrue(os.path.isdir(__lowerCamelCase					)					)
        # Cache should contain at least those three subfolders:
        for subfolder in ["blobs", "refs", "snapshots"]:
               self.assertTrue(os.path.isdir(os.path.join(__lowerCamelCase,__lowerCamelCase					)					)					)
        with open(os.path.join(__lowerCamelCase,'''refs''','''main'''					)					) as f:
               A__		   =		f.read()
        self.assertEqual(__lowerCamelCase,os.path.join(__lowerCamelCase,'''snapshots''',__lowerCamelCase,__lowerCamelCase					)					)
        self.assertTrue(os.path.isfile(__lowerCamelCase					)					)

        # File is cached at the same place the second time.
        A__		   =		cached_file(__lowerCamelCase,__lowerCamelCase					)
        self.assertEqual(__lowerCamelCase,__lowerCamelCase					)

        # Using a specific revision to test the full commit hash.
        A__		   =		cached_file(__lowerCamelCase,__lowerCamelCase,revision='''9b8c223'''					)
        self.assertEqual(__lowerCamelCase,os.path.join(__lowerCamelCase,'''snapshots''',__lowerCamelCase,__lowerCamelCase					)					)



 def 							UpperCamelCase     (				self					):
        with self.assertRaisesRegex(__lowerCamelCase,'''is not a valid model identifier'''					):
               A__		   =		cached_file('''tiny-random-bert''',__lowerCamelCase					)

        with self.assertRaisesRegex(__lowerCamelCase,'''is not a valid git identifier'''					):
               A__		   =		cached_file(__lowerCamelCase,__lowerCamelCase,revision='''aaaa'''					)

        with self.assertRaisesRegex(__lowerCamelCase,'''does not appear to have a file named'''					):
               A__		   =		cached_file(__lowerCamelCase,'''conf'''					)



 def 							UpperCamelCase     (				self					):
        with self.assertRaisesRegex(__lowerCamelCase,'''does not appear to have a file named'''					):
               A__		   =		cached_file(__lowerCamelCase,'''conf'''					)

        with open(os.path.join(__lowerCamelCase,'''refs''','''main'''					)					) as f:
               A__		   =		f.read()
        self.assertTrue(os.path.isfile(os.path.join(__lowerCamelCase,'''.no_exist''',__lowerCamelCase,'''conf'''					)					)					)

        A__		   =		cached_file(__lowerCamelCase,'''conf''',_raise_exceptions_for_missing_entries=__lowerCamelCase					)
        self.assertIsNone(__lowerCamelCase					)

        A__		   =		cached_file(__lowerCamelCase,'''conf''',local_files_only=__lowerCamelCase,_raise_exceptions_for_missing_entries=__lowerCamelCase					)
        self.assertIsNone(__lowerCamelCase					)

        A__		   =		mock.Mock()
        A__		   =		500
        A__		   =		{}
        A__		   =		HTTPError
        A__		   =		{}

        # Under the mock environment we get a 500 error when trying to reach the tokenizer.
        with mock.patch('''requests.Session.request''',return_value=__lowerCamelCase					) as mock_head:
               A__		   =		cached_file(__lowerCamelCase,'''conf''',_raise_exceptions_for_connection_errors=__lowerCamelCase					)
               self.assertIsNone(__lowerCamelCase					)
               # This check we did call the fake head request
               mock_head.assert_called()



 def 							UpperCamelCase     (				self					):
        self.assertTrue(has_file('''hf-internal-testing/tiny-bert-pt-only''',__lowerCamelCase					)					)
        self.assertFalse(has_file('''hf-internal-testing/tiny-bert-pt-only''',__lowerCamelCase					)					)
        self.assertFalse(has_file('''hf-internal-testing/tiny-bert-pt-only''',__lowerCamelCase					)					)



 def 							UpperCamelCase     (				self					):
        # `get_file_from_repo` returns None if the file does not exist
        self.assertIsNone(get_file_from_repo('''bert-base-cased''','''ahah.txt'''					)					)

        # The function raises if the repository does not exist.
        with self.assertRaisesRegex(__lowerCamelCase,'''is not a valid model identifier'''					):
               get_file_from_repo('''bert-base-case''',__lowerCamelCase					)

        # The function raises if the revision does not exist.
        with self.assertRaisesRegex(__lowerCamelCase,'''is not a valid git identifier'''					):
               get_file_from_repo('''bert-base-cased''',__lowerCamelCase,revision='''ahaha'''					)

        A__		   =		get_file_from_repo('''bert-base-cased''',__lowerCamelCase					)
        # The name is the cached name which is not very easy to test, so instead we load the content.
        A__		   =		json.loads(open(__lowerCamelCase,'''r'''					).read()					)
        self.assertEqual(config['''hidden_size'''],768					)




 def 							UpperCamelCase     (				self					):
        with tempfile.TemporaryDirectory() as tmp_dir:
               A__		   =		Path(__lowerCamelCase					) / '''a.txt'''
               filename.touch()
               self.assertEqual(get_file_from_repo(__lowerCamelCase,'''a.txt'''					),str(__lowerCamelCase					)					)

               self.assertIsNone(get_file_from_repo(__lowerCamelCase,'''b.txt'''					)					)





