


import time
from contextlib import contextmanager
from pathlib import Path

import pytest
import requests
from huggingface_hub.hf_api import HfApi, HfFolder


UpperCAmelCase      =   """__DUMMY_TRANSFORMERS_USER__"""
UpperCAmelCase      =   """Dummy User"""
UpperCAmelCase      =   """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"""

UpperCAmelCase      =   """https://hub-ci.huggingface.co"""
UpperCAmelCase      =   CI_HUB_ENDPOINT + """/datasets/{repo_id}/resolve/{revision}/{path}"""
UpperCAmelCase      =   CI_HUB_ENDPOINT + """/{repo_id}/resolve/{revision}/{filename}"""
UpperCAmelCase      =   Path("""~/.huggingface/hub_ci_token""").expanduser()






@pytest.fixture
def 	__lowerCAmelCase   (SCREAMING_SNAKE_CASE )->  List[Any]:




						"""simple docstring"""

						monkeypatch.setattr(
						    '''huggingface_hub.file_download.HUGGINGFACE_CO_URL_TEMPLATE''' , SCREAMING_SNAKE_CASE )






@pytest.fixture
def 	__lowerCAmelCase   (SCREAMING_SNAKE_CASE )->  Any:




						"""simple docstring"""

						monkeypatch.setattr('''datasets.config.HF_ENDPOINT''' , SCREAMING_SNAKE_CASE )
						monkeypatch.setattr('''datasets.config.HUB_DATASETS_URL''' , SCREAMING_SNAKE_CASE )






@pytest.fixture
def 	__lowerCAmelCase   (SCREAMING_SNAKE_CASE )->  Optional[Any]:




						"""simple docstring"""

						monkeypatch.setattr('''huggingface_hub.hf_api.HfFolder.path_token''' , SCREAMING_SNAKE_CASE )






@pytest.fixture
def 	__lowerCAmelCase   (SCREAMING_SNAKE_CASE , SCREAMING_SNAKE_CASE )->  List[Any]:




						"""simple docstring"""

						HfFolder.save_token(SCREAMING_SNAKE_CASE )
						yield
						HfFolder.delete_token()






@pytest.fixture(scope='''session''' )
def 	__lowerCAmelCase   ()->  Any:




						"""simple docstring"""

						return HfApi(endpoint=SCREAMING_SNAKE_CASE )






@pytest.fixture(scope='''session''' )
def 	__lowerCAmelCase   (SCREAMING_SNAKE_CASE )->  List[str]:




						"""simple docstring"""

						snake_case_	=				HfFolder.get_token()
						HfFolder.save_token(SCREAMING_SNAKE_CASE )
						yield CI_HUB_USER_TOKEN
						if previous_token is not None:
												HfFolder.save_token(SCREAMING_SNAKE_CASE )






@pytest.fixture
def 	__lowerCAmelCase   (SCREAMING_SNAKE_CASE )->  Tuple:




						"""simple docstring"""

						def _cleanup_repo(SCREAMING_SNAKE_CASE ):
												hf_api.delete_repo(SCREAMING_SNAKE_CASE , token=SCREAMING_SNAKE_CASE , repo_type='''dataset''' )

						return _cleanup_repo






@pytest.fixture
def 	__lowerCAmelCase   (SCREAMING_SNAKE_CASE )->  int:




						"""simple docstring"""

						@contextmanager
						def _temporary_repo(SCREAMING_SNAKE_CASE ):
												try:
																		yield repo_id
												finally:
																		cleanup_repo(SCREAMING_SNAKE_CASE )

						return _temporary_repo






@pytest.fixture(scope='''session''' )
def 	__lowerCAmelCase   (SCREAMING_SNAKE_CASE , SCREAMING_SNAKE_CASE , SCREAMING_SNAKE_CASE )->  Any:




						"""simple docstring"""

						snake_case_	=				f'''repo_txt_data-{int(time.time() * 10E3 )}'''
						snake_case_	=				f'''{CI_HUB_USER}/{repo_name}'''
						hf_api.create_repo(SCREAMING_SNAKE_CASE , token=SCREAMING_SNAKE_CASE , repo_type='''dataset''' , private=SCREAMING_SNAKE_CASE )
						hf_api.upload_file(
						    token=SCREAMING_SNAKE_CASE , path_or_fileobj=str(SCREAMING_SNAKE_CASE ) , path_in_repo='''data/text_data.txt''' , repo_id=SCREAMING_SNAKE_CASE , repo_type='''dataset''' , )
						yield repo_id
						try:
												hf_api.delete_repo(SCREAMING_SNAKE_CASE , token=SCREAMING_SNAKE_CASE , repo_type='''dataset''' )
						except (requests.exceptions.HTTPError, ValueError):  # catch http error and token invalid error
												pass






@pytest.fixture()
def 	__lowerCAmelCase   (SCREAMING_SNAKE_CASE , SCREAMING_SNAKE_CASE , SCREAMING_SNAKE_CASE )->  List[Any]:




						"""simple docstring"""

						return hf_private_dataset_repo_txt_data_






@pytest.fixture(scope='''session''' )
def 	__lowerCAmelCase   (SCREAMING_SNAKE_CASE , SCREAMING_SNAKE_CASE , SCREAMING_SNAKE_CASE )->  Optional[int]:




						"""simple docstring"""

						snake_case_	=				f'''repo_zipped_txt_data-{int(time.time() * 10E3 )}'''
						snake_case_	=				f'''{CI_HUB_USER}/{repo_name}'''
						hf_api.create_repo(SCREAMING_SNAKE_CASE , token=SCREAMING_SNAKE_CASE , repo_type='''dataset''' , private=SCREAMING_SNAKE_CASE )
						hf_api.upload_file(
						    token=SCREAMING_SNAKE_CASE , path_or_fileobj=str(SCREAMING_SNAKE_CASE ) , path_in_repo='''data.zip''' , repo_id=SCREAMING_SNAKE_CASE , repo_type='''dataset''' , )
						yield repo_id
						try:
												hf_api.delete_repo(SCREAMING_SNAKE_CASE , token=SCREAMING_SNAKE_CASE , repo_type='''dataset''' )
						except (requests.exceptions.HTTPError, ValueError):  # catch http error and token invalid error
												pass






@pytest.fixture()
def 	__lowerCAmelCase   (SCREAMING_SNAKE_CASE , SCREAMING_SNAKE_CASE , SCREAMING_SNAKE_CASE )->  Optional[Any]:




						"""simple docstring"""

						return hf_private_dataset_repo_zipped_txt_data_






@pytest.fixture(scope='''session''' )
def 	__lowerCAmelCase   (SCREAMING_SNAKE_CASE , SCREAMING_SNAKE_CASE , SCREAMING_SNAKE_CASE )->  Tuple:




						"""simple docstring"""

						snake_case_	=				f'''repo_zipped_img_data-{int(time.time() * 10E3 )}'''
						snake_case_	=				f'''{CI_HUB_USER}/{repo_name}'''
						hf_api.create_repo(SCREAMING_SNAKE_CASE , token=SCREAMING_SNAKE_CASE , repo_type='''dataset''' , private=SCREAMING_SNAKE_CASE )
						hf_api.upload_file(
						    token=SCREAMING_SNAKE_CASE , path_or_fileobj=str(SCREAMING_SNAKE_CASE ) , path_in_repo='''data.zip''' , repo_id=SCREAMING_SNAKE_CASE , repo_type='''dataset''' , )
						yield repo_id
						try:
												hf_api.delete_repo(SCREAMING_SNAKE_CASE , token=SCREAMING_SNAKE_CASE , repo_type='''dataset''' )
						except (requests.exceptions.HTTPError, ValueError):  # catch http error and token invalid error
												pass






@pytest.fixture()
def 	__lowerCAmelCase   (SCREAMING_SNAKE_CASE , SCREAMING_SNAKE_CASE , SCREAMING_SNAKE_CASE )->  List[str]:




						"""simple docstring"""

						return hf_private_dataset_repo_zipped_img_data_



import gc
import unittest

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

from diffusers import TransformeraDModel, VQDiffusionPipeline, VQDiffusionScheduler, VQModel
from diffusers.pipelines.vq_diffusion.pipeline_vq_diffusion import LearnedClassifierFreeSamplingEmbeddings
from diffusers.utils import load_numpy, slow, torch_device
from diffusers.utils.testing_utils import require_torch_gpu


UpperCAmelCase      =   False







class lowerCAmelCase_						(			unittest.TestCase	):

 '''simple docstring'''

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



 @property
 def 						UpperCamelCase__					(     self ):
       return 12



 @property
 def 						UpperCamelCase__					(     self ):
       return 12



 @property
 def 						UpperCamelCase__					(     self ):
       return 32



 @property
 def 						UpperCamelCase__					(     self ):
       torch.manual_seed(0 )
       snake_case_	=				VQModel(
           block_out_channels=[32, 64]		,				in_channels=3		,				out_channels=3		,				down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']		,				up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']		,				latent_channels=3		,				num_vq_embeddings=self.num_embed		,				vq_embed_dim=3		,				)
       return model



 @property
 def 						UpperCamelCase__					(     self ):
       snake_case_	=				CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip''' )
       return tokenizer



 @property
 def 						UpperCamelCase__					(     self ):
       torch.manual_seed(0 )
       snake_case_	=				CLIPTextConfig(
           bos_token_id=0		,				eos_token_id=2		,				hidden_size=self.text_embedder_hidden_size		,				intermediate_size=37		,				layer_norm_eps=1E-05		,				num_attention_heads=4		,				num_hidden_layers=5		,				pad_token_id=1		,				vocab_size=10_00		,				)
       return CLIPTextModel(_UpperCAmelCase )



 @property
 def 						UpperCamelCase__					(     self ):
       torch.manual_seed(0 )

       snake_case_	=				12
       snake_case_	=				12

       snake_case_	=				{
           '''attention_bias''': True,
           '''cross_attention_dim''': 32,
           '''attention_head_dim''': height * width,
           '''num_attention_heads''': 1,
           '''num_vector_embeds''': self.num_embed,
           '''num_embeds_ada_norm''': self.num_embeds_ada_norm,
           '''norm_num_groups''': 32,
           '''sample_size''': width,
           '''activation_fn''': '''geglu-approximate''',
       }

       snake_case_	=				TransformeraDModel(**_UpperCAmelCase )
       return model



 def 						UpperCamelCase__					(     self ):
       snake_case_	=				'''cpu'''

       snake_case_	=				self.dummy_vqvae
       snake_case_	=				self.dummy_text_encoder
       snake_case_	=				self.dummy_tokenizer
       snake_case_	=				self.dummy_transformer
       snake_case_	=				VQDiffusionScheduler(self.num_embed )
       snake_case_	=				LearnedClassifierFreeSamplingEmbeddings(learnable=_UpperCAmelCase )

       snake_case_	=				VQDiffusionPipeline(
           vqvae=_UpperCAmelCase		,				text_encoder=_UpperCAmelCase		,				tokenizer=_UpperCAmelCase		,				transformer=_UpperCAmelCase		,				scheduler=_UpperCAmelCase		,				learned_classifier_free_sampling_embeddings=_UpperCAmelCase		,				)
       snake_case_	=				pipe.to(_UpperCAmelCase )
       pipe.set_progress_bar_config(disable=_UpperCAmelCase )

       snake_case_	=				'''teddy bear playing in the pool'''

       snake_case_	=				torch.Generator(device=_UpperCAmelCase ).manual_seed(0 )
       snake_case_	=				pipe([prompt]		,				generator=_UpperCAmelCase		,				num_inference_steps=2		,				output_type='''np''' )
       snake_case_	=				output.images

       snake_case_	=				torch.Generator(device=_UpperCAmelCase ).manual_seed(0 )
       snake_case_	=				pipe(
           [prompt]		,				generator=_UpperCAmelCase		,				output_type='''np'''		,				return_dict=_UpperCAmelCase		,				num_inference_steps=2 )[0]

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

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

       snake_case_	=				np.array([0.6_551, 0.6_168, 0.5_008, 0.5_676, 0.5_659, 0.4_295, 0.6_073, 0.5_599, 0.4_992] )

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





 def 						UpperCamelCase__					(     self ):
       snake_case_	=				'''cpu'''

       snake_case_	=				self.dummy_vqvae
       snake_case_	=				self.dummy_text_encoder
       snake_case_	=				self.dummy_tokenizer
       snake_case_	=				self.dummy_transformer
       snake_case_	=				VQDiffusionScheduler(self.num_embed )
       snake_case_	=				LearnedClassifierFreeSamplingEmbeddings(
           learnable=_UpperCAmelCase		,				hidden_size=self.text_embedder_hidden_size		,				length=tokenizer.model_max_length )

       snake_case_	=				VQDiffusionPipeline(
           vqvae=_UpperCAmelCase		,				text_encoder=_UpperCAmelCase		,				tokenizer=_UpperCAmelCase		,				transformer=_UpperCAmelCase		,				scheduler=_UpperCAmelCase		,				learned_classifier_free_sampling_embeddings=_UpperCAmelCase		,				)
       snake_case_	=				pipe.to(_UpperCAmelCase )
       pipe.set_progress_bar_config(disable=_UpperCAmelCase )

       snake_case_	=				'''teddy bear playing in the pool'''

       snake_case_	=				torch.Generator(device=_UpperCAmelCase ).manual_seed(0 )
       snake_case_	=				pipe([prompt]		,				generator=_UpperCAmelCase		,				num_inference_steps=2		,				output_type='''np''' )
       snake_case_	=				output.images

       snake_case_	=				torch.Generator(device=_UpperCAmelCase ).manual_seed(0 )
       snake_case_	=				pipe(
           [prompt]		,				generator=_UpperCAmelCase		,				output_type='''np'''		,				return_dict=_UpperCAmelCase		,				num_inference_steps=2 )[0]

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

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

       snake_case_	=				np.array([0.6_693, 0.6_075, 0.4_959, 0.5_701, 0.5_583, 0.4_333, 0.6_171, 0.5_684, 0.4_988] )

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







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

 '''simple docstring'''

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





 def 						UpperCamelCase__					(     self ):
       snake_case_	=				load_numpy(
           '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
           '''/vq_diffusion/teddy_bear_pool_classifier_free_sampling.npy''' )

       snake_case_	=				VQDiffusionPipeline.from_pretrained('''microsoft/vq-diffusion-ithq''' )
       snake_case_	=				pipeline.to(_UpperCAmelCase )
       pipeline.set_progress_bar_config(disable=_UpperCAmelCase )

       # requires GPU generator for gumbel softmax
       # don't use GPU generator in tests though
       snake_case_	=				torch.Generator(device=_UpperCAmelCase ).manual_seed(0 )
       snake_case_	=				pipeline(
           '''teddy bear playing in the pool'''		,				num_images_per_prompt=1		,				generator=_UpperCAmelCase		,				output_type='''np'''		,				)

       snake_case_	=				output.images[0]

       assert image.shape == (2_56, 2_56, 3)
       assert np.abs(expected_image - image ).max() < 2.0