




"""simple docstring"""






import collections
import gzip
import os
import urllib

import numpy
from tensorflow.python.framework import dtypes, random_seed
from tensorflow.python.platform import gfile
from tensorflow.python.util.deprecation import deprecated

UpperCamelCase__    =      collections.namedtuple('''_Datasets''', ['''train''', '''validation''', '''test'''])

# CVDF mirror of http://yann.lecun.com/exdb/mnist/
UpperCamelCase__    =      '''https://storage.googleapis.com/cvdf-datasets/mnist/'''





def 				UpperCAmelCase		(		snake_case      :       str     ):
					_lowerCAmelCase:Any						=	numpy.dtype(numpy.uintaa     ).newbyteorder('''>'''     )
					return numpy.frombuffer(bytestream.read(4     )   ,							dtype=snake_case     )[0]





@deprecated(snake_case   ,							'''Please use tf.data to implement this functionality.'''     )
def 				UpperCAmelCase		(		snake_case      :       List[Any]     ):
					print('''Extracting'''   ,							f.name     )
					with gzip.GzipFile(fileobj=snake_case     ) as bytestream:
										_lowerCAmelCase:Optional[Any]						=	_readaa(snake_case     )
										if magic != 2051:
															raise ValueError(
															    '''Invalid magic number %d in MNIST image file: %s''' % (magic, f.name)     )
										_lowerCAmelCase:Optional[Any]						=	_readaa(snake_case     )
										_lowerCAmelCase:int						=	_readaa(snake_case     )
										_lowerCAmelCase:Optional[Any]						=	_readaa(snake_case     )
										_lowerCAmelCase:Optional[int]						=	bytestream.read(rows * cols * num_images     )
										_lowerCAmelCase:Optional[int]						=	numpy.frombuffer(snake_case   ,							dtype=numpy.uinta     )
										_lowerCAmelCase:Any						=	data.reshape(snake_case   ,							snake_case   ,							snake_case   ,							1     )
										return data





@deprecated(snake_case   ,							'''Please use tf.one_hot on tensors.'''     )
def 				UpperCAmelCase		(		snake_case      :       List[str]   ,							snake_case      :       Tuple     ):
					_lowerCAmelCase:Tuple						=	labels_dense.shape[0]
					_lowerCAmelCase:Dict						=	numpy.arange(snake_case     ) * num_classes
					_lowerCAmelCase:List[Any]						=	numpy.zeros((num_labels, num_classes)     )
					_lowerCAmelCase:Any						=	1
					return labels_one_hot





@deprecated(snake_case   ,							'''Please use tf.data to implement this functionality.'''     )
def 				UpperCAmelCase		(		snake_case      :       List[str]   ,							snake_case      :       Tuple=False   ,							snake_case      :       Optional[int]=10     ):
					print('''Extracting'''   ,							f.name     )
					with gzip.GzipFile(fileobj=snake_case     ) as bytestream:
										_lowerCAmelCase:Dict						=	_readaa(snake_case     )
										if magic != 2049:
															raise ValueError(
															    '''Invalid magic number %d in MNIST label file: %s''' % (magic, f.name)     )
										_lowerCAmelCase:int						=	_readaa(snake_case     )
										_lowerCAmelCase:Any						=	bytestream.read(snake_case     )
										_lowerCAmelCase:int						=	numpy.frombuffer(snake_case   ,							dtype=numpy.uinta     )
										if one_hot:
															return _dense_to_one_hot(snake_case   ,							snake_case     )
										return labels




class 	a__      :





						@deprecated(
						    snake_case__  ,'''Please use alternatives such as official/mnist/_DataSet.py'''
						    ''' from tensorflow/models.'''  ,)
						def __init__(				self   :						Tuple  ,a__   :						List[str]  ,a__   :						str  ,a__   :						int=False  ,a__   :						str=False  ,a__   :						Tuple=dtypes.floataa  ,a__   :						int=True  ,a__   :						List[Any]=None  ,)    ->							Any:




											"""simple docstring"""







											_lowerCAmelCase:Optional[int]						=	random_seed.get_seed(snake_case__)
											# If op level seed is not set, use whatever graph level seed is returned
											numpy.random.seed(seeda if seed is None else seeda)
											_lowerCAmelCase:Optional[Any]						=	dtypes.as_dtype(snake_case__).base_dtype
											if dtype not in (dtypes.uinta, dtypes.floataa):
																raise TypeError('''Invalid image dtype %r, expected uint8 or float32''' % dtype)
											if fake_data:
																_lowerCAmelCase:Tuple						=	1_0000
																_lowerCAmelCase:Any						=	one_hot
											else:
																assert (
																    images.shape[0] == labels.shape[0]
																), F'images.shape: {images.shape} labels.shape: {labels.shape}'
																_lowerCAmelCase:Tuple						=	images.shape[0]

																# Convert shape from [num examples, rows, columns, depth]
																# to [num examples, rows*columns] (assuming depth == 1)
																if reshape:
																					assert images.shape[3] == 1
																					_lowerCAmelCase:Dict						=	images.reshape(
																					    images.shape[0]  ,images.shape[1] * images.shape[2])
																if dtype == dtypes.floataa:
																					# Convert from [0, 255] -> [0.0, 1.0].
																					_lowerCAmelCase:Optional[Any]						=	images.astype(numpy.floataa)
																					_lowerCAmelCase:Dict						=	numpy.multiply(snake_case__  ,1.0 / 255.0)
											_lowerCAmelCase:Union[str, Any]						=	images
											_lowerCAmelCase:List[Any]						=	labels
											_lowerCAmelCase:Union[str, Any]						=	0
											_lowerCAmelCase:int						=	0





						@property
						def    __UpperCamelCase  (				self   :						List[str])    ->							Optional[Any]:




											"""simple docstring"""







											return self._images





						@property
						def    __UpperCamelCase  (				self   :						Optional[int])    ->							Dict:




											"""simple docstring"""







											return self._labels





						@property
						def    __UpperCamelCase  (				self   :						str)    ->							Optional[Any]:




											"""simple docstring"""







											return self._num_examples





						@property
						def    __UpperCamelCase  (				self   :						List[str])    ->							int:




											"""simple docstring"""







											return self._epochs_completed







						def    __UpperCamelCase  (				self   :						int  ,a__   :						str  ,a__   :						Dict=False  ,a__   :						Any=True)    ->							int:




											"""simple docstring"""







											if fake_data:
																_lowerCAmelCase:Optional[int]						=	[1] * 784
																_lowerCAmelCase:Optional[Any]						=	[1] + [0] * 9 if self.one_hot else 0
																return (
																    [fake_image for _ in range(snake_case__)],
																    [fake_label for _ in range(snake_case__)],
																)
											_lowerCAmelCase:Optional[int]						=	self._index_in_epoch
											# Shuffle for the first epoch
											if self._epochs_completed == 0 and start == 0 and shuffle:
																_lowerCAmelCase:Optional[Any]						=	numpy.arange(self._num_examples)
																numpy.random.shuffle(snake_case__)
																_lowerCAmelCase:Any						=	self.images[perma]
																_lowerCAmelCase:List[Any]						=	self.labels[perma]
											# Go to the next epoch
											if start + batch_size > self._num_examples:
																# Finished epoch
																self._epochs_completed += 1
																# Get the rest examples in this epoch
																_lowerCAmelCase:Tuple						=	self._num_examples - start
																_lowerCAmelCase:List[Any]						=	self._images[start : self._num_examples]
																_lowerCAmelCase:Tuple						=	self._labels[start : self._num_examples]
																# Shuffle the data
																if shuffle:
																					_lowerCAmelCase:Optional[int]						=	numpy.arange(self._num_examples)
																					numpy.random.shuffle(snake_case__)
																					_lowerCAmelCase:Any						=	self.images[perm]
																					_lowerCAmelCase:List[str]						=	self.labels[perm]
																# Start next epoch
																_lowerCAmelCase:Any						=	0
																_lowerCAmelCase:List[str]						=	batch_size - rest_num_examples
																_lowerCAmelCase:List[str]						=	self._index_in_epoch
																_lowerCAmelCase:str						=	self._images[start:end]
																_lowerCAmelCase:Tuple						=	self._labels[start:end]
																return (
																    numpy.concatenate((images_rest_part, images_new_part)  ,axis=0),
																    numpy.concatenate((labels_rest_part, labels_new_part)  ,axis=0),
																)
											else:
																self._index_in_epoch += batch_size
																_lowerCAmelCase:Union[str, Any]						=	self._index_in_epoch
																return self._images[start:end], self._labels[start:end]





@deprecated(snake_case   ,							'''Please write your own downloading logic.'''     )
def 				UpperCAmelCase		(		snake_case      :       int   ,							snake_case      :       int   ,							snake_case      :       Optional[Any]     ):



					if not gfile.Exists(snake_case     ):
										gfile.MakeDirs(snake_case     )
					_lowerCAmelCase:Tuple						=	os.path.join(snake_case   ,							snake_case     )
					if not gfile.Exists(snake_case     ):
										urllib.request.urlretrieve(snake_case   ,							snake_case     )  # noqa: S310
										with gfile.GFile(snake_case     ) as f:
															_lowerCAmelCase:Optional[int]						=	f.size()
										print('''Successfully downloaded'''   ,							snake_case   ,							snake_case   ,							'''bytes.'''     )
					return filepath





@deprecated(
    snake_case   ,							'''Please use alternatives such as:''' ''' tensorflow_datasets.load(\'mnist\')'''     )
def 				UpperCAmelCase		(		snake_case      :       str   ,							snake_case      :       List[Any]=False   ,							snake_case      :       List[str]=False   ,							snake_case      :       Optional[Any]=dtypes.floataa   ,							snake_case      :       int=True   ,							snake_case      :       Tuple=5000   ,							snake_case      :       Optional[int]=None   ,							snake_case      :       Tuple=DEFAULT_SOURCE_URL   ,							):



					if fake_data:

										def fake():
															return _DataSet(
															    []   ,							[]   ,							fake_data=snake_case   ,							one_hot=snake_case   ,							dtype=snake_case   ,							seed=snake_case     )

										_lowerCAmelCase:Dict						=	fake()
										_lowerCAmelCase:List[Any]						=	fake()
										_lowerCAmelCase:List[str]						=	fake()
										return _Datasets(train=snake_case   ,							validation=snake_case   ,							test=snake_case     )

					if not source_url:  # empty string check
										_lowerCAmelCase:Union[str, Any]						=	DEFAULT_SOURCE_URL

					_lowerCAmelCase:List[str]						=	"""train-images-idx3-ubyte.gz"""
					_lowerCAmelCase:str						=	"""train-labels-idx1-ubyte.gz"""
					_lowerCAmelCase:Union[str, Any]						=	"""t10k-images-idx3-ubyte.gz"""
					_lowerCAmelCase:Optional[int]						=	"""t10k-labels-idx1-ubyte.gz"""

					_lowerCAmelCase:Optional[Any]						=	_maybe_download(
					    snake_case   ,							snake_case   ,							source_url + train_images_file     )
					with gfile.Open(snake_case   ,							'''rb'''     ) as f:
										_lowerCAmelCase:List[Any]						=	_extract_images(snake_case     )

					_lowerCAmelCase:int						=	_maybe_download(
					    snake_case   ,							snake_case   ,							source_url + train_labels_file     )
					with gfile.Open(snake_case   ,							'''rb'''     ) as f:
										_lowerCAmelCase:int						=	_extract_labels(snake_case   ,							one_hot=snake_case     )

					_lowerCAmelCase:Dict						=	_maybe_download(
					    snake_case   ,							snake_case   ,							source_url + test_images_file     )
					with gfile.Open(snake_case   ,							'''rb'''     ) as f:
										_lowerCAmelCase:int						=	_extract_images(snake_case     )

					_lowerCAmelCase:List[Any]						=	_maybe_download(
					    snake_case   ,							snake_case   ,							source_url + test_labels_file     )
					with gfile.Open(snake_case   ,							'''rb'''     ) as f:
										_lowerCAmelCase:Optional[int]						=	_extract_labels(snake_case   ,							one_hot=snake_case     )

					if not 0 <= validation_size <= len(snake_case     ):
										_lowerCAmelCase:Union[str, Any]						=	(
										    """Validation size should be between 0 and """
										    F'{len(snake_case     )}. Received: {validation_size}.'
										)
										raise ValueError(snake_case     )

					_lowerCAmelCase:List[str]						=	train_images[:validation_size]
					_lowerCAmelCase:Any						=	train_labels[:validation_size]
					_lowerCAmelCase:Optional[Any]						=	train_images[validation_size:]
					_lowerCAmelCase:Union[str, Any]						=	train_labels[validation_size:]

					_lowerCAmelCase:Optional[Any]						=	{"""dtype""": dtype, """reshape""": reshape, """seed""": seed}

					_lowerCAmelCase:Any						=	_DataSet(snake_case   ,							snake_case   ,							**snake_case     )
					_lowerCAmelCase:List[str]						=	_DataSet(snake_case   ,							snake_case   ,							**snake_case     )
					_lowerCAmelCase:str						=	_DataSet(snake_case   ,							snake_case   ,							**snake_case     )

					return _Datasets(train=snake_case   ,							validation=snake_case   ,							test=snake_case     )













import gc
import unittest

import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    PriorTransformer,
    StableUnCLIPPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.testing_utils import enable_full_determinism, load_numpy, require_torch_gpu, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()
class     UpperCamelCase__					(     lowerCamelCase__				, lowerCamelCase__				, lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       Optional[int]       =	StableUnCLIPPipeline
		__a       :       int       =	TEXT_TO_IMAGE_PARAMS
		__a       :       Optional[Any]       =	TEXT_TO_IMAGE_BATCH_PARAMS
		__a       :       Tuple       =	TEXT_TO_IMAGE_IMAGE_PARAMS
		__a       :       Optional[int]       =	TEXT_TO_IMAGE_IMAGE_PARAMS

		# TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
		__a       :       Tuple       =	False






		def 						snake_case__				(							self  )       -> Optional[Any]:






									"""simple docstring"""
									lowercase_	:   int  				=     32
									lowercase_	:   Tuple  				=     embedder_hidden_size

									# prior components

									torch.manual_seed(0  )
									lowercase_	:   Optional[Any]  				=     CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""  )

									torch.manual_seed(0  )
									lowercase_	:   Dict  				=     CLIPTextModelWithProjection(
									    CLIPTextConfig(
									        bos_token_id=0,   eos_token_id=2,   hidden_size=snake_case__,   projection_dim=snake_case__,   intermediate_size=37,   layer_norm_eps=1E-05,   num_attention_heads=4,   num_hidden_layers=5,   pad_token_id=1,   vocab_size=10_00,   )  )

									torch.manual_seed(0  )
									lowercase_	:   Tuple  				=     PriorTransformer(
									    num_attention_heads=2,   attention_head_dim=12,   embedding_dim=snake_case__,   num_layers=1,   )

									torch.manual_seed(0  )
									lowercase_	:   Optional[Any]  				=     DDPMScheduler(
									    variance_type="""fixed_small_log""",   prediction_type="""sample""",   num_train_timesteps=10_00,   clip_sample=snake_case__,   clip_sample_range=5.0,   beta_schedule="""squaredcos_cap_v2""",   )

									# regular denoising components

									torch.manual_seed(0  )
									lowercase_	:   Union[str, Any]  				=     StableUnCLIPImageNormalizer(embedding_dim=snake_case__  )
									lowercase_	:   str  				=     DDPMScheduler(beta_schedule="""squaredcos_cap_v2"""  )

									torch.manual_seed(0  )
									lowercase_	:   str  				=     CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""  )

									torch.manual_seed(0  )
									lowercase_	:   List[Any]  				=     CLIPTextModel(
									    CLIPTextConfig(
									        bos_token_id=0,   eos_token_id=2,   hidden_size=snake_case__,   projection_dim=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,   )  )

									torch.manual_seed(0  )
									lowercase_	:   Optional[int]  				=     UNetaDConditionModel(
									    sample_size=32,   in_channels=4,   out_channels=4,   down_block_types=("""CrossAttnDownBlock2D""", """DownBlock2D"""),   up_block_types=("""UpBlock2D""", """CrossAttnUpBlock2D"""),   block_out_channels=(32, 64),   attention_head_dim=(2, 4),   class_embed_type="""projection""",   projection_class_embeddings_input_dim=embedder_projection_dim * 2,   cross_attention_dim=snake_case__,   layers_per_block=1,   upcast_attention=snake_case__,   use_linear_projection=snake_case__,   )

									torch.manual_seed(0  )
									lowercase_	:   Dict  				=     DDIMScheduler(
									    beta_schedule="""scaled_linear""",   beta_start=0.00085,   beta_end=0.012,   prediction_type="""v_prediction""",   set_alpha_to_one=snake_case__,   steps_offset=1,   )

									torch.manual_seed(0  )
									lowercase_	:   str  				=     AutoencoderKL()

									lowercase_	:   str  				=     {
									    # prior components
									    """prior_tokenizer""": prior_tokenizer,
									    """prior_text_encoder""": prior_text_encoder,
									    """prior""": prior,
									    """prior_scheduler""": prior_scheduler,
									    # image noising components
									    """image_normalizer""": image_normalizer,
									    """image_noising_scheduler""": image_noising_scheduler,
									    # regular denoising components
									    """tokenizer""": tokenizer,
									    """text_encoder""": text_encoder,
									    """unet""": unet,
									    """scheduler""": scheduler,
									    """vae""": vae,
									}

									return components






		def 						snake_case__				(							self,   snake_case__,   snake_case__=0  )       -> str:






									"""simple docstring"""
									if str(snake_case__  ).startswith("""mps"""  ):
																lowercase_	:   Tuple  				=     torch.manual_seed(snake_case__  )
									else:
																lowercase_	:   int  				=     torch.Generator(device=snake_case__  ).manual_seed(snake_case__  )
									lowercase_	:   Dict  				=     {
									    """prompt""": """A painting of a squirrel eating a burger""",
									    """generator""": generator,
									    """num_inference_steps""": 2,
									    """prior_num_inference_steps""": 2,
									    """output_type""": """numpy""",
									}
									return inputs






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






									"""simple docstring"""
									lowercase_	:   int  				=     torch_device == """cpu"""

									self._test_attention_slicing_forward_pass(test_max_difference=snake_case__  )






		def 						snake_case__				(							self  )       -> Any:






									"""simple docstring"""
									lowercase_	:   List[str]  				=     torch_device in ["""cpu""", """mps"""]

									self._test_inference_batch_single_identical(test_max_difference=snake_case__  )




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






		'''simple docstring'''






		def 						snake_case__				(							self  )       -> Any:






									"""simple docstring"""
									# clean up the VRAM after each test
									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()






		def 						snake_case__				(							self  )       -> List[str]:






									"""simple docstring"""
									lowercase_	:   Union[str, Any]  				=     load_numpy(
									    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy"""  )

									lowercase_	:   List[Any]  				=     StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l""",   torch_dtype=torch.floataa  )
									pipe.to(snake_case__  )
									pipe.set_progress_bar_config(disable=snake_case__  )
									# stable unclip will oom when integration tests are run on a V100,
									# so turn on memory savings
									pipe.enable_attention_slicing()
									pipe.enable_sequential_cpu_offload()

									lowercase_	:   Any  				=     torch.Generator(device="""cpu"""  ).manual_seed(0  )
									lowercase_	:   Dict  				=     pipe("""anime turle""",   generator=snake_case__,   output_type="""np"""  )

									lowercase_	:   str  				=     output.images[0]

									assert image.shape == (7_68, 7_68, 3)

									assert_mean_pixel_difference(snake_case__,   snake_case__  )






		def 						snake_case__				(							self  )       -> Dict:






									"""simple docstring"""
									torch.cuda.empty_cache()
									torch.cuda.reset_max_memory_allocated()
									torch.cuda.reset_peak_memory_stats()

									lowercase_	:   str  				=     StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l""",   torch_dtype=torch.floataa  )
									lowercase_	:   Optional[int]  				=     pipe.to(snake_case__  )
									pipe.set_progress_bar_config(disable=snake_case__  )
									pipe.enable_attention_slicing()
									pipe.enable_sequential_cpu_offload()

									lowercase_	:   List[Any]  				=     pipe(
									    """anime turtle""",   prior_num_inference_steps=2,   num_inference_steps=2,   output_type="""np""",   )

									lowercase_	:   Optional[int]  				=     torch.cuda.max_memory_allocated()
									# make sure that less than 7 GB is allocated
									assert mem_bytes < 7 * 10**9