





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

_lowerCAmelCase 				=     collections.namedtuple("""_Datasets""", ["""train""", """validation""", """test"""])

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



def     _lowerCAmelCase		(  _lowerCAmelCase     ):
   '''simple docstring'''


   A_     :				Tuple			      =	numpy.dtype(numpy.uintaa     ).newbyteorder(""">"""     )
   return numpy.frombuffer(bytestream.read(4     )  ,dtype=_lowerCAmelCase     )[0]



@deprecated(_lowerCAmelCase  ,"""Please use tf.data to implement this functionality."""     )
def     _lowerCAmelCase		(  _lowerCAmelCase     ):
   '''simple docstring'''


   print("""Extracting"""  ,f.name     )
   with gzip.GzipFile(fileobj=_lowerCAmelCase     ) as bytestream:
      A_     :				Union[str, Any]			      =	_readaa(_lowerCAmelCase     )
      if magic != 2_0_5_1:
         raise ValueError(
             """Invalid magic number %d in MNIST image file: %s""" % (magic, f.name)     )
      A_     :				Any			      =	_readaa(_lowerCAmelCase     )
      A_     :				int			      =	_readaa(_lowerCAmelCase     )
      A_     :				List[str]			      =	_readaa(_lowerCAmelCase     )
      A_     :				List[str]			      =	bytestream.read(rows * cols * num_images     )
      A_     :				Any			      =	numpy.frombuffer(_lowerCAmelCase  ,dtype=numpy.uinta     )
      A_     :				int			      =	data.reshape(_lowerCAmelCase  ,_lowerCAmelCase  ,_lowerCAmelCase  ,1     )
      return data



@deprecated(_lowerCAmelCase  ,"""Please use tf.one_hot on tensors."""     )
def     _lowerCAmelCase		(  _lowerCAmelCase  ,_lowerCAmelCase     ):
   '''simple docstring'''


   A_     :				int			      =	labels_dense.shape[0]
   A_     :				Dict			      =	numpy.arange(_lowerCAmelCase     ) * num_classes
   A_     :				Optional[Any]			      =	numpy.zeros((num_labels, num_classes)     )
   A_     :				int			      =	1
   return labels_one_hot



@deprecated(_lowerCAmelCase  ,"""Please use tf.data to implement this functionality."""     )
def     _lowerCAmelCase		(  _lowerCAmelCase  ,_lowerCAmelCase=False  ,_lowerCAmelCase=1_0     ):
   '''simple docstring'''


   print("""Extracting"""  ,f.name     )
   with gzip.GzipFile(fileobj=_lowerCAmelCase     ) as bytestream:
      A_     :				Dict			      =	_readaa(_lowerCAmelCase     )
      if magic != 2_0_4_9:
         raise ValueError(
             """Invalid magic number %d in MNIST label file: %s""" % (magic, f.name)     )
      A_     :				Optional[Any]			      =	_readaa(_lowerCAmelCase     )
      A_     :				Optional[Any]			      =	bytestream.read(_lowerCAmelCase     )
      A_     :				Optional[int]			      =	numpy.frombuffer(_lowerCAmelCase  ,dtype=numpy.uinta     )
      if one_hot:
         return _dense_to_one_hot(_lowerCAmelCase  ,_lowerCAmelCase     )
      return labels

class    _UpperCAmelCase							:
      @deprecated(
          a__      ,       """Please use alternatives such as official/mnist/_DataSet.py"""
          """ from tensorflow/models."""      ,       )
      def __init__(		self      ,       a__      ,       a__      ,       a__=False      ,       a__=False      ,       a__=dtypes.floataa      ,       a__=True      ,       a__=None      ,       ):
         A_      ,				A_     :				Union[str, Any]			      =	random_seed.get_seed(a__       )
         # If op level seed is not set, use whatever graph level seed is returned
         numpy.random.seed(seeda if seed is None else seeda       )
         A_     :				int			      =	dtypes.as_dtype(a__       ).base_dtype
         if dtype not in (dtypes.uinta, dtypes.floataa):
            raise TypeError("""Invalid image dtype %r, expected uint8 or float32""" % dtype       )
         if fake_data:
            A_     :				Any			      =	10000
            A_     :				Dict			      =	one_hot
         else:
            assert (
                images.shape[0] == labels.shape[0]
            ), F"""images.shape: {images.shape} labels.shape: {labels.shape}"""
            A_     :				int			      =	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
               A_     :				Any			      =	images.reshape(
                   images.shape[0]      ,       images.shape[1] * images.shape[2]       )
            if dtype == dtypes.floataa:
               # Convert from [0, 255] -> [0.0, 1.0].
               A_     :				List[Any]			      =	images.astype(numpy.floataa       )
               A_     :				Optional[Any]			      =	numpy.multiply(a__      ,       1.0 / 255.0       )
         A_     :				int			      =	images
         A_     :				Any			      =	labels
         A_     :				Union[str, Any]			      =	0
         A_     :				str			      =	0
      @property
      def 		_lowerCamelCase      (		self       ):
         return self._images
      @property
      def 		_lowerCamelCase      (		self       ):
         return self._labels
      @property
      def 		_lowerCamelCase      (		self       ):
         return self._num_examples
      @property
      def 		_lowerCamelCase      (		self       ):
         return self._epochs_completed






      def 		_lowerCamelCase      (		self      ,       a__      ,       a__=False      ,       a__=True       ):
         if fake_data:
            A_     :				Any			      =	[1] * 784
            A_     :				Tuple			      =	[1] + [0] * 9 if self.one_hot else 0
            return (
                [fake_image for _ in range(a__       )],
                [fake_label for _ in range(a__       )],
            )
         A_     :				Tuple			      =	self._index_in_epoch
         # Shuffle for the first epoch
         if self._epochs_completed == 0 and start == 0 and shuffle:
            A_     :				Any			      =	numpy.arange(self._num_examples       )
            numpy.random.shuffle(a__       )
            A_     :				int			      =	self.images[perma]
            A_     :				str			      =	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
            A_     :				str			      =	self._num_examples - start
            A_     :				List[Any]			      =	self._images[start : self._num_examples]
            A_     :				str			      =	self._labels[start : self._num_examples]
            # Shuffle the data
            if shuffle:
               A_     :				Optional[Any]			      =	numpy.arange(self._num_examples       )
               numpy.random.shuffle(a__       )
               A_     :				int			      =	self.images[perm]
               A_     :				Any			      =	self.labels[perm]
            # Start next epoch
            A_     :				Dict			      =	0
            A_     :				Tuple			      =	batch_size - rest_num_examples
            A_     :				List[str]			      =	self._index_in_epoch
            A_     :				str			      =	self._images[start:end]
            A_     :				List[Any]			      =	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
            A_     :				str			      =	self._index_in_epoch
            return self._images[start:end], self._labels[start:end]






@deprecated(_lowerCAmelCase  ,"""Please write your own downloading logic."""     )
def     _lowerCAmelCase		(  _lowerCAmelCase  ,_lowerCAmelCase  ,_lowerCAmelCase     ):
   '''simple docstring'''


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







@deprecated(
    _lowerCAmelCase  ,"""Please use alternatives such as:""" """ tensorflow_datasets.load('mnist')"""     )
def     _lowerCAmelCase		(  _lowerCAmelCase  ,_lowerCAmelCase=False  ,_lowerCAmelCase=False  ,_lowerCAmelCase=dtypes.floataa  ,_lowerCAmelCase=True  ,_lowerCAmelCase=5_0_0_0  ,_lowerCAmelCase=None  ,_lowerCAmelCase=DEFAULT_SOURCE_URL  ,):
   '''simple docstring'''


   if fake_data:

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

      A_     :				Union[str, Any]			      =	fake()
      A_     :				Union[str, Any]			      =	fake()
      A_     :				Any			      =	fake()
      return _Datasets(train=_lowerCAmelCase  ,validation=_lowerCAmelCase  ,test=_lowerCAmelCase     )

   if not source_url:  # empty string check
      A_     :				int			      =	DEFAULT_SOURCE_URL

   A_     :				Optional[int]			      =	"""train-images-idx3-ubyte.gz"""
   A_     :				str			      =	"""train-labels-idx1-ubyte.gz"""
   A_     :				str			      =	"""t10k-images-idx3-ubyte.gz"""
   A_     :				Optional[Any]			      =	"""t10k-labels-idx1-ubyte.gz"""

   A_     :				List[str]			      =	_maybe_download(
       _lowerCAmelCase  ,_lowerCAmelCase  ,source_url + train_images_file     )
   with gfile.Open(_lowerCAmelCase  ,"""rb"""     ) as f:
      A_     :				Any			      =	_extract_images(_lowerCAmelCase     )

   A_     :				Dict			      =	_maybe_download(
       _lowerCAmelCase  ,_lowerCAmelCase  ,source_url + train_labels_file     )
   with gfile.Open(_lowerCAmelCase  ,"""rb"""     ) as f:
      A_     :				List[Any]			      =	_extract_labels(_lowerCAmelCase  ,one_hot=_lowerCAmelCase     )

   A_     :				Tuple			      =	_maybe_download(
       _lowerCAmelCase  ,_lowerCAmelCase  ,source_url + test_images_file     )
   with gfile.Open(_lowerCAmelCase  ,"""rb"""     ) as f:
      A_     :				Dict			      =	_extract_images(_lowerCAmelCase     )

   A_     :				List[Any]			      =	_maybe_download(
       _lowerCAmelCase  ,_lowerCAmelCase  ,source_url + test_labels_file     )
   with gfile.Open(_lowerCAmelCase  ,"""rb"""     ) as f:
      A_     :				List[Any]			      =	_extract_labels(_lowerCAmelCase  ,one_hot=_lowerCAmelCase     )

   if not 0 <= validation_size <= len(_lowerCAmelCase     ):
      A_     :				Optional[int]			      =	(
          """Validation size should be between 0 and """
          f"""{len(_lowerCAmelCase     )}. Received: {validation_size}."""
      )
      raise ValueError(_lowerCAmelCase     )

   A_     :				Union[str, Any]			      =	train_images[:validation_size]
   A_     :				str			      =	train_labels[:validation_size]
   A_     :				Tuple			      =	train_images[validation_size:]
   A_     :				int			      =	train_labels[validation_size:]

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

   A_     :				Tuple			      =	_DataSet(_lowerCAmelCase  ,_lowerCAmelCase  ,**_lowerCAmelCase     )
   A_     :				List[Any]			      =	_DataSet(_lowerCAmelCase  ,_lowerCAmelCase  ,**_lowerCAmelCase     )
   A_     :				Optional[int]			      =	_DataSet(_lowerCAmelCase  ,_lowerCAmelCase  ,**_lowerCAmelCase     )

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













import gc
import unittest

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

from diffusers import AltDiffusionPipeline, AutoencoderKL, DDIMScheduler, PNDMScheduler, UNetaDConditionModel
from diffusers.pipelines.alt_diffusion.modeling_roberta_series import (
    RobertaSeriesConfig,
    RobertaSeriesModelWithTransformation,
)
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 PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()

class    _UpperCAmelCase							(		_lowerCamelCase			,       _lowerCamelCase			,       _lowerCamelCase			,       unittest.TestCase     ):
      a						       =							AltDiffusionPipeline
      a						       =							TEXT_TO_IMAGE_PARAMS
      a						       =							TEXT_TO_IMAGE_BATCH_PARAMS
      a						       =							TEXT_TO_IMAGE_IMAGE_PARAMS
      a						       =							TEXT_TO_IMAGE_IMAGE_PARAMS
      def 		_lowerCamelCase      (		self       ):
         torch.manual_seed(0       )
         A_     :				Optional[int]			      =	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      ,       )
         A_     :				List[Any]			      =	DDIMScheduler(
             beta_start=0.0_0085      ,       beta_end=0.012      ,       beta_schedule="""scaled_linear"""      ,       clip_sample=a__      ,       set_alpha_to_one=a__      ,       )
         torch.manual_seed(0       )
         A_     :				int			      =	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      ,       )

         # TODO: address the non-deterministic text encoder (fails for save-load tests)
         # torch.manual_seed(0)
         # text_encoder_config = RobertaSeriesConfig(
         #     hidden_size=32,
         #     project_dim=32,
         #     intermediate_size=37,
         #     layer_norm_eps=1e-05,
         #     num_attention_heads=4,
         #     num_hidden_layers=5,
         #     vocab_size=5002,
         # )
         # text_encoder = RobertaSeriesModelWithTransformation(text_encoder_config)

         torch.manual_seed(0       )
         A_     :				Optional[int]			      =	CLIPTextConfig(
             bos_token_id=0      ,       eos_token_id=2      ,       hidden_size=32      ,       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=5002      ,       )
         A_     :				Optional[Any]			      =	CLIPTextModel(a__       )

         A_     :				Union[str, Any]			      =	XLMRobertaTokenizer.from_pretrained("""hf-internal-testing/tiny-xlm-roberta"""       )
         A_     :				str			      =	77

         A_     :				Optional[Any]			      =	{
             """unet""": unet,
             """scheduler""": scheduler,
             """vae""": vae,
             """text_encoder""": text_encoder,
             """tokenizer""": tokenizer,
             """safety_checker""": None,
             """feature_extractor""": None,
         }
         return components
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__=0       ):
         if str(a__       ).startswith("""mps"""       ):
            A_     :				Union[str, Any]			      =	torch.manual_seed(a__       )
         else:
            A_     :				Dict			      =	torch.Generator(device=a__       ).manual_seed(a__       )
         A_     :				Dict			      =	{
             """prompt""": """A painting of a squirrel eating a burger""",
             """generator""": generator,
             """num_inference_steps""": 2,
             """guidance_scale""": 6.0,
             """output_type""": """numpy""",
         }
         return inputs
      def 		_lowerCamelCase      (		self       ):
         super().test_attention_slicing_forward_pass(expected_max_diff=3E-3       )
      def 		_lowerCamelCase      (		self       ):
         super().test_inference_batch_single_identical(expected_max_diff=3E-3       )
      def 		_lowerCamelCase      (		self       ):
         A_     :				Optional[Any]			      =	"""cpu"""  # ensure determinism for the device-dependent torch.Generator

         A_     :				Union[str, Any]			      =	self.get_dummy_components()
         torch.manual_seed(0       )
         A_     :				List[str]			      =	RobertaSeriesConfig(
             hidden_size=32      ,       project_dim=32      ,       intermediate_size=37      ,       layer_norm_eps=1E-05      ,       num_attention_heads=4      ,       num_hidden_layers=5      ,       vocab_size=5002      ,       )
         # TODO: remove after fixing the non-deterministic text encoder
         A_     :				Dict			      =	RobertaSeriesModelWithTransformation(a__       )
         A_     :				Any			      =	text_encoder

         A_     :				Tuple			      =	AltDiffusionPipeline(**a__       )
         A_     :				int			      =	alt_pipe.to(a__       )
         alt_pipe.set_progress_bar_config(disable=a__       )

         A_     :				Union[str, Any]			      =	self.get_dummy_inputs(a__       )
         A_     :				Tuple			      =	"""A photo of an astronaut"""
         A_     :				Dict			      =	alt_pipe(**a__       )
         A_     :				List[str]			      =	output.images
         A_     :				List[Any]			      =	image[0, -3:, -3:, -1]

         assert image.shape == (1, 64, 64, 3)
         A_     :				Tuple			      =	np.array(
             [0.574_8162, 0.6044_7145, 0.4882_1217, 0.5010_0636, 0.543_1185, 0.4576_3683, 0.4965_7696, 0.4813_2733, 0.4757_3093]       )

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






      def 		_lowerCamelCase      (		self       ):
         A_     :				Tuple			      =	"""cpu"""  # ensure determinism for the device-dependent torch.Generator

         A_     :				Union[str, Any]			      =	self.get_dummy_components()
         A_     :				str			      =	PNDMScheduler(skip_prk_steps=a__       )
         torch.manual_seed(0       )
         A_     :				Any			      =	RobertaSeriesConfig(
             hidden_size=32      ,       project_dim=32      ,       intermediate_size=37      ,       layer_norm_eps=1E-05      ,       num_attention_heads=4      ,       num_hidden_layers=5      ,       vocab_size=5002      ,       )
         # TODO: remove after fixing the non-deterministic text encoder
         A_     :				int			      =	RobertaSeriesModelWithTransformation(a__       )
         A_     :				Any			      =	text_encoder
         A_     :				Any			      =	AltDiffusionPipeline(**a__       )
         A_     :				int			      =	alt_pipe.to(a__       )
         alt_pipe.set_progress_bar_config(disable=a__       )

         A_     :				Union[str, Any]			      =	self.get_dummy_inputs(a__       )
         A_     :				Union[str, Any]			      =	alt_pipe(**a__       )
         A_     :				List[str]			      =	output.images
         A_     :				List[Any]			      =	image[0, -3:, -3:, -1]

         assert image.shape == (1, 64, 64, 3)
         A_     :				List[Any]			      =	np.array(
             [0.5160_5093, 0.570_7241, 0.4736_5507, 0.5057_8886, 0.563_3877, 0.464_2503, 0.518_2081, 0.4876_3484, 0.4908_4237]       )

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






@slow
@require_torch_gpu
class    _UpperCAmelCase							(		unittest.TestCase     ):
      def 		_lowerCamelCase      (		self       ):
         # clean up the VRAM after each test
         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()
      def 		_lowerCamelCase      (		self       ):
         # make sure here that pndm scheduler skips prk
         A_     :				Any			      =	AltDiffusionPipeline.from_pretrained("""BAAI/AltDiffusion"""      ,       safety_checker=a__       )
         A_     :				Optional[int]			      =	alt_pipe.to(a__       )
         alt_pipe.set_progress_bar_config(disable=a__       )

         A_     :				Tuple			      =	"""A painting of a squirrel eating a burger"""
         A_     :				Tuple			      =	torch.manual_seed(0       )
         A_     :				int			      =	alt_pipe([prompt]      ,       generator=a__      ,       guidance_scale=6.0      ,       num_inference_steps=20      ,       output_type="""np"""       )

         A_     :				Optional[Any]			      =	output.images

         A_     :				int			      =	image[0, -3:, -3:, -1]

         assert image.shape == (1, 512, 512, 3)
         A_     :				str			      =	np.array([0.1010, 0.0800, 0.0794, 0.0885, 0.0843, 0.0762, 0.0769, 0.0729, 0.0586]       )

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






      def 		_lowerCamelCase      (		self       ):
         A_     :				Dict			      =	DDIMScheduler.from_pretrained("""BAAI/AltDiffusion"""      ,       subfolder="""scheduler"""       )

         A_     :				int			      =	AltDiffusionPipeline.from_pretrained("""BAAI/AltDiffusion"""      ,       scheduler=a__      ,       safety_checker=a__       )
         A_     :				Dict			      =	alt_pipe.to(a__       )
         alt_pipe.set_progress_bar_config(disable=a__       )

         A_     :				Optional[int]			      =	"""A painting of a squirrel eating a burger"""
         A_     :				Any			      =	torch.manual_seed(0       )

         A_     :				List[str]			      =	alt_pipe([prompt]      ,       generator=a__      ,       num_inference_steps=2      ,       output_type="""numpy"""       )
         A_     :				List[Any]			      =	output.images

         A_     :				Tuple			      =	image[0, -3:, -3:, -1]

         assert image.shape == (1, 512, 512, 3)
         A_     :				Any			      =	np.array([0.4019, 0.4052, 0.3810, 0.4119, 0.3916, 0.3982, 0.4651, 0.4195, 0.5323]       )

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






