





import gc
import unittest

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

from diffusers import PriorTransformer, UnCLIPPipeline, UnCLIPScheduler, UNetaDConditionModel, UNetaDModel
from diffusers.pipelines.unclip.text_proj import UnCLIPTextProjModel
from diffusers.utils import load_numpy, nightly, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()

class 			__A(    __lowercase    ,      unittest.TestCase			):
     __A							=							UnCLIPPipeline
     __A							=							TEXT_TO_IMAGE_PARAMS - {
         "negative_prompt",
         "height",
         "width",
         "negative_prompt_embeds",
         "guidance_scale",
         "prompt_embeds",
         "cross_attention_kwargs",
     }
     __A							=							TEXT_TO_IMAGE_BATCH_PARAMS
     __A							=							[
         "generator",
         "return_dict",
         "prior_num_inference_steps",
         "decoder_num_inference_steps",
         "super_res_num_inference_steps",
     ]
     __A							=							False

     @property
     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            return 32

     @property
     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            return 32

     @property
     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            return self.time_input_dim

     @property
     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            return self.time_input_dim * 4

     @property
     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            return 100

     @property
     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            _UpperCamelCase														=    CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''   )
            return tokenizer

     @property
     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            torch.manual_seed(0   )
            _UpperCamelCase														=    CLIPTextConfig(
                bos_token_id=0,							eos_token_id=2,							hidden_size=self.text_embedder_hidden_size,							projection_dim=self.text_embedder_hidden_size,							intermediate_size=37,							layer_norm_eps=1E-0_5,							num_attention_heads=4,							num_hidden_layers=5,							pad_token_id=1,							vocab_size=1000,							)
            return CLIPTextModelWithProjection(A__   )

     @property
     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            torch.manual_seed(0   )

            _UpperCamelCase														=    {
                '''num_attention_heads''': 2,
                '''attention_head_dim''': 12,
                '''embedding_dim''': self.text_embedder_hidden_size,
                '''num_layers''': 1,
            }

            _UpperCamelCase														=    PriorTransformer(**A__   )
            return model

     @property
     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            torch.manual_seed(0   )

            _UpperCamelCase														=    {
                '''clip_embeddings_dim''': self.text_embedder_hidden_size,
                '''time_embed_dim''': self.time_embed_dim,
                '''cross_attention_dim''': self.cross_attention_dim,
            }

            _UpperCamelCase														=    UnCLIPTextProjModel(**A__   )
            return model

     @property
     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            torch.manual_seed(0   )

            _UpperCamelCase														=    {
                '''sample_size''': 32,
                # RGB in channels
                '''in_channels''': 3,
                # Out channels is double in channels because predicts mean and variance
                '''out_channels''': 6,
                '''down_block_types''': ('''ResnetDownsampleBlock2D''', '''SimpleCrossAttnDownBlock2D'''),
                '''up_block_types''': ('''SimpleCrossAttnUpBlock2D''', '''ResnetUpsampleBlock2D'''),
                '''mid_block_type''': '''UNetMidBlock2DSimpleCrossAttn''',
                '''block_out_channels''': (self.block_out_channels_a, self.block_out_channels_a * 2),
                '''layers_per_block''': 1,
                '''cross_attention_dim''': self.cross_attention_dim,
                '''attention_head_dim''': 4,
                '''resnet_time_scale_shift''': '''scale_shift''',
                '''class_embed_type''': '''identity''',
            }

            _UpperCamelCase														=    UNetaDConditionModel(**A__   )
            return model

     @property
     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            return {
                "sample_size": 64,
                "layers_per_block": 1,
                "down_block_types": ("ResnetDownsampleBlock2D", "ResnetDownsampleBlock2D"),
                "up_block_types": ("ResnetUpsampleBlock2D", "ResnetUpsampleBlock2D"),
                "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
                "in_channels": 6,
                "out_channels": 3,
            }

     @property
     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            torch.manual_seed(0   )

            _UpperCamelCase														=    UNetaDModel(**self.dummy_super_res_kwargs   )
            return model

     @property
     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            torch.manual_seed(1   )

            _UpperCamelCase														=    UNetaDModel(**self.dummy_super_res_kwargs   )
            return model

     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            _UpperCamelCase														=    self.dummy_prior
            _UpperCamelCase														=    self.dummy_decoder
            _UpperCamelCase														=    self.dummy_text_proj
            _UpperCamelCase														=    self.dummy_text_encoder
            _UpperCamelCase														=    self.dummy_tokenizer
            _UpperCamelCase														=    self.dummy_super_res_first
            _UpperCamelCase														=    self.dummy_super_res_last

            _UpperCamelCase														=    UnCLIPScheduler(
                variance_type='''fixed_small_log''',							prediction_type='''sample''',							num_train_timesteps=1000,							clip_sample_range=5.0,							)

            _UpperCamelCase														=    UnCLIPScheduler(
                variance_type='''learned_range''',							prediction_type='''epsilon''',							num_train_timesteps=1000,							)

            _UpperCamelCase														=    UnCLIPScheduler(
                variance_type='''fixed_small_log''',							prediction_type='''epsilon''',							num_train_timesteps=1000,							)

            _UpperCamelCase														=    {
                '''prior''': prior,
                '''decoder''': decoder,
                '''text_proj''': text_proj,
                '''text_encoder''': text_encoder,
                '''tokenizer''': tokenizer,
                '''super_res_first''': super_res_first,
                '''super_res_last''': super_res_last,
                '''prior_scheduler''': prior_scheduler,
                '''decoder_scheduler''': decoder_scheduler,
                '''super_res_scheduler''': super_res_scheduler,
            }

            return components

     def 						_UpperCamelCase  (   self,							A,							A=0   ):





            """simple docstring"""

            if str(A__   ).startswith('''mps'''   ):
                   _UpperCamelCase														=    torch.manual_seed(A__   )
            else:
                   _UpperCamelCase														=    torch.Generator(device=A__   ).manual_seed(A__   )
            _UpperCamelCase														=    {
                '''prompt''': '''horse''',
                '''generator''': generator,
                '''prior_num_inference_steps''': 2,
                '''decoder_num_inference_steps''': 2,
                '''super_res_num_inference_steps''': 2,
                '''output_type''': '''numpy''',
            }
            return inputs

     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            _UpperCamelCase														=    '''cpu'''

            _UpperCamelCase														=    self.get_dummy_components()

            _UpperCamelCase														=    self.pipeline_class(**A__   )
            _UpperCamelCase														=    pipe.to(A__   )

            pipe.set_progress_bar_config(disable=A__   )

            _UpperCamelCase														=    pipe(**self.get_dummy_inputs(A__   )   )
            _UpperCamelCase														=    output.images

            _UpperCamelCase														=    pipe(
                **self.get_dummy_inputs(A__   ),							return_dict=A__,							)[0]

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

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

            _UpperCamelCase														=    np.array(
                [
                    0.9_997,
                    0.9_988,
                    0.0_028,
                    0.9_997,
                    0.9_984,
                    0.9_965,
                    0.0_029,
                    0.9_986,
                    0.0_025,
                ]   )

            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   ):





            """simple docstring"""

            _UpperCamelCase														=    torch.device('''cpu'''   )

            class 			__A:
                 __A							=							1






            _UpperCamelCase														=    self.get_dummy_components()

            _UpperCamelCase														=    self.pipeline_class(**A__   )
            _UpperCamelCase														=    pipe.to(A__   )

            _UpperCamelCase														=    components['''prior''']
            _UpperCamelCase														=    components['''decoder''']
            _UpperCamelCase														=    components['''super_res_first''']
            _UpperCamelCase														=    components['''tokenizer''']
            _UpperCamelCase														=    components['''text_encoder''']

            _UpperCamelCase														=    torch.Generator(device=A__   ).manual_seed(0   )
            _UpperCamelCase														=    prior.dtype
            _UpperCamelCase														=    1

            _UpperCamelCase														=    (batch_size, prior.config.embedding_dim)
            _UpperCamelCase														=    pipe.prepare_latents(
                A__,							dtype=A__,							device=A__,							generator=A__,							latents=A__,							scheduler=DummyScheduler()   )
            _UpperCamelCase														=    (batch_size, decoder.config.in_channels, decoder.config.sample_size, decoder.config.sample_size)
            _UpperCamelCase														=    pipe.prepare_latents(
                A__,							dtype=A__,							device=A__,							generator=A__,							latents=A__,							scheduler=DummyScheduler()   )

            _UpperCamelCase														=    (
                batch_size,
                super_res_first.config.in_channels // 2,
                super_res_first.config.sample_size,
                super_res_first.config.sample_size,
            )
            _UpperCamelCase														=    pipe.prepare_latents(
                A__,							dtype=A__,							device=A__,							generator=A__,							latents=A__,							scheduler=DummyScheduler()   )

            pipe.set_progress_bar_config(disable=A__   )

            _UpperCamelCase														=    '''this is a prompt example'''

            _UpperCamelCase														=    torch.Generator(device=A__   ).manual_seed(0   )
            _UpperCamelCase														=    pipe(
                [prompt],							generator=A__,							prior_num_inference_steps=2,							decoder_num_inference_steps=2,							super_res_num_inference_steps=2,							prior_latents=A__,							decoder_latents=A__,							super_res_latents=A__,							output_type='''np''',							)
            _UpperCamelCase														=    output.images

            _UpperCamelCase														=    tokenizer(
                A__,							padding='''max_length''',							max_length=tokenizer.model_max_length,							return_tensors='''pt''',							)
            _UpperCamelCase														=    text_encoder(text_inputs.input_ids   )
            _UpperCamelCase														=    text_inputs.attention_mask

            _UpperCamelCase														=    torch.Generator(device=A__   ).manual_seed(0   )
            _UpperCamelCase														=    pipe(
                generator=A__,							prior_num_inference_steps=2,							decoder_num_inference_steps=2,							super_res_num_inference_steps=2,							prior_latents=A__,							decoder_latents=A__,							super_res_latents=A__,							text_model_output=A__,							text_attention_mask=A__,							output_type='''np''',							)[0]

            # make sure passing text embeddings manually is identical
            assert np.abs(image - image_from_text   ).max() < 1E-4

     @skip_mps
     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            _UpperCamelCase														=    torch_device == '''cpu'''

            self._test_attention_slicing_forward_pass(test_max_difference=A__,							expected_max_diff=0.01   )

     @skip_mps
     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            _UpperCamelCase														=    torch_device == '''cpu'''
            _UpperCamelCase														=    True
            _UpperCamelCase														=    [
                '''prior_num_inference_steps''',
                '''decoder_num_inference_steps''',
                '''super_res_num_inference_steps''',
            ]

            self._test_inference_batch_single_identical(
                test_max_difference=A__,							relax_max_difference=A__,							additional_params_copy_to_batched_inputs=A__,							)

     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            _UpperCamelCase														=    [
                '''prior_num_inference_steps''',
                '''decoder_num_inference_steps''',
                '''super_res_num_inference_steps''',
            ]

            if torch_device == "mps":
                   # TODO: MPS errors with larger batch sizes
                   _UpperCamelCase														=    [2, 3]
                   self._test_inference_batch_consistent(
                       batch_sizes=A__,							additional_params_copy_to_batched_inputs=A__,							)
            else:
                   self._test_inference_batch_consistent(
                       additional_params_copy_to_batched_inputs=A__   )

     @skip_mps
     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            return super().test_dict_tuple_outputs_equivalent()

     @skip_mps
     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            return super().test_save_load_local()



     @skip_mps
     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            return super().test_save_load_optional_components()






@nightly
class 			__A(    unittest.TestCase			):

     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

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



     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            _UpperCamelCase														=    load_numpy(
                '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
                '''/unclip/karlo_v1_alpha_horse_cpu.npy'''   )

            _UpperCamelCase														=    UnCLIPPipeline.from_pretrained('''kakaobrain/karlo-v1-alpha'''   )
            pipeline.set_progress_bar_config(disable=A__   )

            _UpperCamelCase														=    torch.manual_seed(0   )
            _UpperCamelCase														=    pipeline(
                '''horse''',							num_images_per_prompt=1,							generator=A__,							output_type='''np''',							)

            _UpperCamelCase														=    output.images[0]

            assert image.shape == (256, 256, 3)
            assert np.abs(expected_image - image   ).max() < 1E-1






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

     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

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

     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            _UpperCamelCase														=    load_numpy(
                '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
                '''/unclip/karlo_v1_alpha_horse_fp16.npy'''   )

            _UpperCamelCase														=    UnCLIPPipeline.from_pretrained('''kakaobrain/karlo-v1-alpha''',							torch_dtype=torch.floataa   )
            _UpperCamelCase														=    pipeline.to(A__   )
            pipeline.set_progress_bar_config(disable=A__   )

            _UpperCamelCase														=    torch.Generator(device='''cpu'''   ).manual_seed(0   )
            _UpperCamelCase														=    pipeline(
                '''horse''',							generator=A__,							output_type='''np''',							)

            _UpperCamelCase														=    output.images[0]

            assert image.shape == (256, 256, 3)

            assert_mean_pixel_difference(A__,							A__   )



     def 						_UpperCamelCase  (   self   ):





            """simple docstring"""

            torch.cuda.empty_cache()
            torch.cuda.reset_max_memory_allocated()
            torch.cuda.reset_peak_memory_stats()

            _UpperCamelCase														=    UnCLIPPipeline.from_pretrained('''kakaobrain/karlo-v1-alpha''',							torch_dtype=torch.floataa   )
            _UpperCamelCase														=    pipe.to(A__   )
            pipe.set_progress_bar_config(disable=A__   )
            pipe.enable_attention_slicing()
            pipe.enable_sequential_cpu_offload()

            _UpperCamelCase														=    pipe(
                '''horse''',							num_images_per_prompt=1,							prior_num_inference_steps=2,							decoder_num_inference_steps=2,							super_res_num_inference_steps=2,							output_type='''np''',							)

            _UpperCamelCase														=    torch.cuda.max_memory_allocated()
            # make sure that less than 7 GB is allocated
            assert mem_bytes < 7 * 10**9









from __future__ import annotations



def  snake_case_(_UpperCamelCase					,			_UpperCamelCase					,			_UpperCamelCase	)				->      int:



  """simple docstring"""






  assert n > 1 and a > 0 and greatest_common_divisor(_UpperCamelCase					,			_UpperCamelCase	) == 1
  ((_snake_case),	(_snake_case),	(_snake_case))		=				extended_gcd(_UpperCamelCase					,			_UpperCamelCase	)  # Implemented below
  _snake_case		=				(b * s) % n
  return x



def  snake_case_(_UpperCamelCase					,			_UpperCamelCase	)				->      int:



  """simple docstring"""






  ((_snake_case),	(_snake_case))		=				extended_euclid(_UpperCamelCase					,			_UpperCamelCase	)  # Implemented below
  if b < 0:
    _snake_case		=				(b % n + n) % n
  return b



def  snake_case_(_UpperCamelCase					,			_UpperCamelCase					,			_UpperCamelCase	)				->      int:



  """simple docstring"""






  _snake_case		=				invert_modulo(_UpperCamelCase					,			_UpperCamelCase	)
  _snake_case		=				(b * s) % n
  return x



def  snake_case_(_UpperCamelCase					,			_UpperCamelCase	)				->      tuple[int, int, int]:



  """simple docstring"""






  assert a >= 0 and b >= 0

  if b == 0:
    _snake_case,	_snake_case,	_snake_case		=				a, 1, 0
  else:
    ((_snake_case),	(_snake_case),	(_snake_case))		=				extended_gcd(_UpperCamelCase					,			a % b	)
    _snake_case		=				q
    _snake_case		=				p - q * (a // b)

  assert a % d == 0 and b % d == 0
  assert d == a * x + b * y

  return (d, x, y)



def  snake_case_(_UpperCamelCase					,			_UpperCamelCase	)				->      tuple[int, int]:



  """simple docstring"""






  if b == 0:
    return (1, 0)
  ((_snake_case),	(_snake_case))		=				extended_euclid(_UpperCamelCase					,			a % b	)
  _snake_case		=				a // b
  return (y, x - k * y)



def  snake_case_(_UpperCamelCase					,			_UpperCamelCase	)				->      int:



  """simple docstring"""






  if a < b:
    _snake_case,	_snake_case		=				b, a

  while a % b != 0:
    _snake_case,	_snake_case		=				b, a % b

  return b


if __name__ == "__main__":
 from doctest import testmod

 testmod(name='''modular_division''', verbose=True)
 testmod(name='''modular_division2''', verbose=True)
 testmod(name='''invert_modulo''', verbose=True)
 testmod(name='''extended_gcd''', verbose=True)
 testmod(name='''extended_euclid''', verbose=True)
 testmod(name='''greatest_common_divisor''', verbose=True)

