

import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionSAGPipeline,
    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   _a					(			snake_case_     ,				snake_case_     ,				unittest.TestCase				):




      """simple docstring"""

      _lowerCamelCase       :			str      		=       StableDiffusionSAGPipeline
      _lowerCamelCase       :			Any      		=       TEXT_TO_IMAGE_PARAMS
      _lowerCamelCase       :			List[Any]      		=       TEXT_TO_IMAGE_BATCH_PARAMS
      _lowerCamelCase       :			Optional[int]      		=       TEXT_TO_IMAGE_IMAGE_PARAMS
      _lowerCamelCase       :			Optional[int]      		=       TEXT_TO_IMAGE_IMAGE_PARAMS
      _lowerCamelCase       :			Tuple      		=       False





      def 		__A	(   self							:				Tuple	):
            torch.manual_seed(0	)
            A_						  =				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_						  =				DDIMScheduler(
                beta_start=0.00_085						,			beta_end=0.012						,			beta_schedule="scaled_linear"						,			clip_sample=UpperCAmelCase						,			set_alpha_to_one=UpperCAmelCase						,			)
            torch.manual_seed(0	)
            A_						  =				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						,			)
            torch.manual_seed(0	)
            A_						  =				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						,			)
            A_						  =				CLIPTextModel(UpperCAmelCase	)
            A_						  =				CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"	)

            A_						  =				{
                "unet": unet,
                "scheduler": scheduler,
                "vae": vae,
                "text_encoder": text_encoder,
                "tokenizer": tokenizer,
                "safety_checker": None,
                "feature_extractor": None,
            }
            return components





      def 		__A	(   self							:				Union[str, Any]						,			UpperCAmelCase							:				Dict						,			UpperCAmelCase							:				List[str]=0	):
            if str(UpperCAmelCase	).startswith("mps"	):
                  A_						  =				torch.manual_seed(UpperCAmelCase	)
            else:
                  A_						  =				torch.Generator(device=UpperCAmelCase	).manual_seed(UpperCAmelCase	)
            A_						  =				{
                "prompt": ".",
                "generator": generator,
                "num_inference_steps": 2,
                "guidance_scale": 1.0,
                "sag_scale": 1.0,
                "output_type": "numpy",
            }
            return inputs







      def 		__A	(   self							:				Tuple	):
            super().test_inference_batch_single_identical(expected_max_diff=3E-3	)




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




      """simple docstring"""

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





      def 		__A	(   self							:				int	):
            A_						  =				StableDiffusionSAGPipeline.from_pretrained("CompVis/stable-diffusion-v1-4"	)
            A_						  =				sag_pipe.to(UpperCAmelCase	)
            sag_pipe.set_progress_bar_config(disable=UpperCAmelCase	)

            A_						  =				"."
            A_						  =				torch.manual_seed(0	)
            A_						  =				sag_pipe(
                [prompt]						,			generator=UpperCAmelCase						,			guidance_scale=7.5						,			sag_scale=1.0						,			num_inference_steps=20						,			output_type="np"	)

            A_						  =				output.images

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

            assert image.shape == (1, 512, 512, 3)
            A_						  =				np.array([0.1_568, 0.1_738, 0.1_695, 0.1_693, 0.1_507, 0.1_705, 0.1_547, 0.1_751, 0.1_949]	)

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





      def 		__A	(   self							:				Optional[Any]	):
            A_						  =				StableDiffusionSAGPipeline.from_pretrained("stabilityai/stable-diffusion-2-1-base"	)
            A_						  =				sag_pipe.to(UpperCAmelCase	)
            sag_pipe.set_progress_bar_config(disable=UpperCAmelCase	)

            A_						  =				"."
            A_						  =				torch.manual_seed(0	)
            A_						  =				sag_pipe(
                [prompt]						,			generator=UpperCAmelCase						,			guidance_scale=7.5						,			sag_scale=1.0						,			num_inference_steps=20						,			output_type="np"	)

            A_						  =				output.images

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

            assert image.shape == (1, 512, 512, 3)
            A_						  =				np.array([0.3_459, 0.2_876, 0.2_537, 0.3_002, 0.2_671, 0.2_160, 0.3_026, 0.2_262, 0.2_371]	)

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







      def 		__A	(   self							:				Tuple	):
            A_						  =				StableDiffusionSAGPipeline.from_pretrained("stabilityai/stable-diffusion-2-1-base"	)
            A_						  =				sag_pipe.to(UpperCAmelCase	)
            sag_pipe.set_progress_bar_config(disable=UpperCAmelCase	)

            A_						  =				"."
            A_						  =				torch.manual_seed(0	)
            A_						  =				sag_pipe(
                [prompt]						,			width=768						,			height=512						,			generator=UpperCAmelCase						,			guidance_scale=7.5						,			sag_scale=1.0						,			num_inference_steps=20						,			output_type="np"						,			)

            A_						  =				output.images

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







'''simple docstring'''



__UpperCamelCase   :							List[Any]         =   """ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"""



def  __UpperCAmelCase						(					SCREAMING_SNAKE_CASE__:						bytes     )	->      bytes:

   """simple docstring"""
   # Make sure the supplied data is a bytes-like object
   if not isinstance(SCREAMING_SNAKE_CASE__,       SCREAMING_SNAKE_CASE__     ):
      __a					  =     f"""a bytes-like object is required, not '{data.__class__.__name__}'"""
      raise TypeError(SCREAMING_SNAKE_CASE__     )

   __a					  =     ''.join(bin(SCREAMING_SNAKE_CASE__     )[2:].zfill(8     ) for byte in data     )

   __a					  =     len(SCREAMING_SNAKE_CASE__     ) % 6 != 0

   if padding_needed:
      # The padding that will be added later
      __a					  =     b'=' * ((6 - len(SCREAMING_SNAKE_CASE__     ) % 6) // 2)

      # Append binary_stream with arbitrary binary digits (0's by default) to make its
      # length a multiple of 6.
      binary_stream += "0" * (6 - len(SCREAMING_SNAKE_CASE__     ) % 6)
   else:
      __a					  =     b''

   # Encode every 6 binary digits to their corresponding Base64 character
   return (
       "".join(
           B64_CHARSET[int(binary_stream[index : index + 6],       2     )]
           for index in range(0,       len(SCREAMING_SNAKE_CASE__     ),       6     )     ).encode()
       + padding
   )



def  __UpperCAmelCase						(					SCREAMING_SNAKE_CASE__:						str     )	->      bytes:

   """simple docstring"""
   # Make sure encoded_data is either a string or a bytes-like object
   if not isinstance(SCREAMING_SNAKE_CASE__,       SCREAMING_SNAKE_CASE__     ) and not isinstance(SCREAMING_SNAKE_CASE__,       SCREAMING_SNAKE_CASE__     ):
      __a					  =     (
          'argument should be a bytes-like object or ASCII string, '
          f"""not '{encoded_data.__class__.__name__}'"""
      )
      raise TypeError(SCREAMING_SNAKE_CASE__     )

   # In case encoded_data is a bytes-like object, make sure it contains only
   # ASCII characters so we convert it to a string object
   if isinstance(SCREAMING_SNAKE_CASE__,       SCREAMING_SNAKE_CASE__     ):
      try:
         __a					  =     encoded_data.decode('utf-8'     )
      except UnicodeDecodeError:
         raise ValueError('base64 encoded data should only contain ASCII characters'     )

   __a					  =     encoded_data.count('='     )

   # Check if the encoded string contains non base64 characters
   if padding:
      assert all(
          char in B64_CHARSET for char in encoded_data[:-padding]     ), "Invalid base64 character(s) found."
   else:
      assert all(
          char in B64_CHARSET for char in encoded_data     ), "Invalid base64 character(s) found."

   # Check the padding
   assert len(SCREAMING_SNAKE_CASE__     ) % 4 == 0 and padding < 3, "Incorrect padding"

   if padding:
      # Remove padding if there is one
      __a					  =     encoded_data[:-padding]

      __a					  =     ''.join(
          bin(B64_CHARSET.index(SCREAMING_SNAKE_CASE__     )     )[2:].zfill(6     ) for char in encoded_data     )[: -padding * 2]
   else:
      __a					  =     ''.join(
          bin(B64_CHARSET.index(SCREAMING_SNAKE_CASE__     )     )[2:].zfill(6     ) for char in encoded_data     )

   __a					  =     [
       int(binary_stream[index : index + 8],       2     )
       for index in range(0,       len(SCREAMING_SNAKE_CASE__     ),       8     )
   ]

   return bytes(SCREAMING_SNAKE_CASE__     )


if __name__ == "__main__":
 import doctest

 doctest.testmod()