


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__						(    _a    ,							unittest.TestCase     ):
       lowerCamelCase__:  Any				     =  UnCLIPPipeline
       lowerCamelCase__:  Optional[Any]				     =  TEXT_TO_IMAGE_PARAMS - {
           "negative_prompt",
           "height",
           "width",
           "negative_prompt_embeds",
           "guidance_scale",
           "prompt_embeds",
           "cross_attention_kwargs",
       }
       lowerCamelCase__:  str				     =  TEXT_TO_IMAGE_BATCH_PARAMS
       lowerCamelCase__:  Optional[int]				     =  [
           "generator",
           "return_dict",
           "prior_num_inference_steps",
           "decoder_num_inference_steps",
           "super_res_num_inference_steps",
       ]
       lowerCamelCase__:  Union[str, Any]				     =  False

       @property
       def     UpperCAmelCase( self   :  Tuple     ):
            return 3_2

       @property
       def     UpperCAmelCase( self   :  Tuple     ):
            return 3_2

       @property
       def     UpperCAmelCase( self   :  Any     ):
            return self.time_input_dim

       @property
       def     UpperCAmelCase( self   :  Optional[int]     ):
            return self.time_input_dim * 4

       @property
       def     UpperCAmelCase( self   :  Union[str, Any]     ):
            return 1_0_0

       @property
       def     UpperCAmelCase( self   :  List[str]     ):
            a_     :  Any				      =     CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""     )
            return tokenizer

       @property
       def     UpperCAmelCase( self   :  Union[str, Any]     ):
            torch.manual_seed(0     )
            a_     :  Dict				      =     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=3_7   ,				layer_norm_eps=1E-05   ,				num_attention_heads=4   ,				num_hidden_layers=5   ,				pad_token_id=1   ,				vocab_size=1_0_0_0   ,				)
            return CLIPTextModelWithProjection(__SCREAMING_SNAKE_CASE     )

       @property
       def     UpperCAmelCase( self   :  Dict     ):
            torch.manual_seed(0     )

            a_     :  Union[str, Any]				      =     {
                '''num_attention_heads''': 2,
                '''attention_head_dim''': 1_2,
                '''embedding_dim''': self.text_embedder_hidden_size,
                '''num_layers''': 1,
            }

            a_     :  Tuple				      =     PriorTransformer(**__SCREAMING_SNAKE_CASE     )
            return model

       @property
       def     UpperCAmelCase( self   :  Any     ):
            torch.manual_seed(0     )

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

            a_     :  Optional[Any]				      =     UnCLIPTextProjModel(**__SCREAMING_SNAKE_CASE     )
            return model

       @property
       def     UpperCAmelCase( self   :  Dict     ):
            torch.manual_seed(0     )

            a_     :  Union[str, Any]				      =     {
                '''sample_size''': 3_2,
                # 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''',
            }

            a_     :  Tuple				      =     UNetaDConditionModel(**__SCREAMING_SNAKE_CASE     )
            return model

       @property
       def     UpperCAmelCase( self   :  Optional[Any]     ):
            return {
                "sample_size": 6_4,
                "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   :  Optional[int]     ):
            torch.manual_seed(0     )

            a_     :  Tuple				      =     UNetaDModel(**self.dummy_super_res_kwargs     )
            return model

       @property
       def     UpperCAmelCase( self   :  Optional[int]     ):
            torch.manual_seed(1     )

            a_     :  Optional[int]				      =     UNetaDModel(**self.dummy_super_res_kwargs     )
            return model

       def     UpperCAmelCase( self   :  List[Any]     ):
            a_     :  Any				      =     self.dummy_prior
            a_     :  List[str]				      =     self.dummy_decoder
            a_     :  Any				      =     self.dummy_text_proj
            a_     :  Dict				      =     self.dummy_text_encoder
            a_     :  Union[str, Any]				      =     self.dummy_tokenizer
            a_     :  Tuple				      =     self.dummy_super_res_first
            a_     :  Optional[Any]				      =     self.dummy_super_res_last

            a_     :  Optional[int]				      =     UnCLIPScheduler(
                variance_type="""fixed_small_log"""   ,				prediction_type="""sample"""   ,				num_train_timesteps=1_0_0_0   ,				clip_sample_range=5.0   ,				)

            a_     :  Dict				      =     UnCLIPScheduler(
                variance_type="""learned_range"""   ,				prediction_type="""epsilon"""   ,				num_train_timesteps=1_0_0_0   ,				)

            a_     :  Union[str, Any]				      =     UnCLIPScheduler(
                variance_type="""fixed_small_log"""   ,				prediction_type="""epsilon"""   ,				num_train_timesteps=1_0_0_0   ,				)

            a_     :  List[Any]				      =     {
                '''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   :  Dict   ,				lowerCamelCase_   :  List[str]   ,				lowerCamelCase_   :  Dict=0     ):




            if str(__SCREAMING_SNAKE_CASE     ).startswith("""mps"""     ):
                 a_     :  Optional[Any]				      =     torch.manual_seed(__SCREAMING_SNAKE_CASE     )
            else:
                 a_     :  Optional[int]				      =     torch.Generator(device=__SCREAMING_SNAKE_CASE     ).manual_seed(__SCREAMING_SNAKE_CASE     )
            a_     :  List[str]				      =     {
                '''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   :  Optional[Any]     ):
            a_     :  Optional[Any]				      =     '''cpu'''

            a_     :  Tuple				      =     self.get_dummy_components()

            a_     :  Any				      =     self.pipeline_class(**__SCREAMING_SNAKE_CASE     )
            a_     :  Optional[Any]				      =     pipe.to(__SCREAMING_SNAKE_CASE     )

            pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE     )

            a_     :  Optional[Any]				      =     pipe(**self.get_dummy_inputs(__SCREAMING_SNAKE_CASE     )     )
            a_     :  Dict				      =     output.images

            a_     :  str				      =     pipe(
                **self.get_dummy_inputs(__SCREAMING_SNAKE_CASE     )   ,				return_dict=__SCREAMING_SNAKE_CASE   ,				)[0]

            a_     :  List[str]				      =     image[0, -3:, -3:, -1]
            a_     :  Optional[Any]				      =     image_from_tuple[0, -3:, -3:, -1]

            assert image.shape == (1, 6_4, 6_4, 3)

            a_     :  str				      =     np.array(
                [
                    0.9_9_9_7,
                    0.9_9_8_8,
                    0.0_0_2_8,
                    0.9_9_9_7,
                    0.9_9_8_4,
                    0.9_9_6_5,
                    0.0_0_2_9,
                    0.9_9_8_6,
                    0.0_0_2_5,
                ]     )

            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   :  str     ):
            a_     :  List[str]				      =     torch.device("""cpu"""     )


            class 	a__						:
                   lowerCamelCase__:  List[Any]				     =  1

            a_     :  Optional[int]				      =     self.get_dummy_components()

            a_     :  Any				      =     self.pipeline_class(**__SCREAMING_SNAKE_CASE     )
            a_     :  Any				      =     pipe.to(__SCREAMING_SNAKE_CASE     )

            a_     :  List[str]				      =     components['''prior''']
            a_     :  List[Any]				      =     components['''decoder''']
            a_     :  Optional[Any]				      =     components['''super_res_first''']
            a_     :  int				      =     components['''tokenizer''']
            a_     :  Optional[Any]				      =     components['''text_encoder''']

            a_     :  int				      =     torch.Generator(device=__SCREAMING_SNAKE_CASE     ).manual_seed(0     )
            a_     :  int				      =     prior.dtype
            a_     :  Optional[int]				      =     1

            a_     :  List[str]				      =     (batch_size, prior.config.embedding_dim)
            a_     :  int				      =     pipe.prepare_latents(
                __SCREAMING_SNAKE_CASE   ,				dtype=__SCREAMING_SNAKE_CASE   ,				device=__SCREAMING_SNAKE_CASE   ,				generator=__SCREAMING_SNAKE_CASE   ,				latents=__SCREAMING_SNAKE_CASE   ,				scheduler=DummyScheduler()     )
            a_     :  Any				      =     (batch_size, decoder.config.in_channels, decoder.config.sample_size, decoder.config.sample_size)
            a_     :  Optional[int]				      =     pipe.prepare_latents(
                __SCREAMING_SNAKE_CASE   ,				dtype=__SCREAMING_SNAKE_CASE   ,				device=__SCREAMING_SNAKE_CASE   ,				generator=__SCREAMING_SNAKE_CASE   ,				latents=__SCREAMING_SNAKE_CASE   ,				scheduler=DummyScheduler()     )

            a_     :  Optional[Any]				      =     (
                batch_size,
                super_res_first.config.in_channels // 2,
                super_res_first.config.sample_size,
                super_res_first.config.sample_size,
            )
            a_     :  List[str]				      =     pipe.prepare_latents(
                __SCREAMING_SNAKE_CASE   ,				dtype=__SCREAMING_SNAKE_CASE   ,				device=__SCREAMING_SNAKE_CASE   ,				generator=__SCREAMING_SNAKE_CASE   ,				latents=__SCREAMING_SNAKE_CASE   ,				scheduler=DummyScheduler()     )

            pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE     )

            a_     :  str				      =     '''this is a prompt example'''

            a_     :  List[Any]				      =     torch.Generator(device=__SCREAMING_SNAKE_CASE     ).manual_seed(0     )
            a_     :  Union[str, Any]				      =     pipe(
                [prompt]   ,				generator=__SCREAMING_SNAKE_CASE   ,				prior_num_inference_steps=2   ,				decoder_num_inference_steps=2   ,				super_res_num_inference_steps=2   ,				prior_latents=__SCREAMING_SNAKE_CASE   ,				decoder_latents=__SCREAMING_SNAKE_CASE   ,				super_res_latents=__SCREAMING_SNAKE_CASE   ,				output_type="""np"""   ,				)
            a_     :  List[Any]				      =     output.images

            a_     :  Dict				      =     tokenizer(
                __SCREAMING_SNAKE_CASE   ,				padding="""max_length"""   ,				max_length=tokenizer.model_max_length   ,				return_tensors="""pt"""   ,				)
            a_     :  Optional[Any]				      =     text_encoder(text_inputs.input_ids     )
            a_     :  Dict				      =     text_inputs.attention_mask

            a_     :  List[Any]				      =     torch.Generator(device=__SCREAMING_SNAKE_CASE     ).manual_seed(0     )
            a_     :  Union[str, Any]				      =     pipe(
                generator=__SCREAMING_SNAKE_CASE   ,				prior_num_inference_steps=2   ,				decoder_num_inference_steps=2   ,				super_res_num_inference_steps=2   ,				prior_latents=__SCREAMING_SNAKE_CASE   ,				decoder_latents=__SCREAMING_SNAKE_CASE   ,				super_res_latents=__SCREAMING_SNAKE_CASE   ,				text_model_output=__SCREAMING_SNAKE_CASE   ,				text_attention_mask=__SCREAMING_SNAKE_CASE   ,				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   :  int     ):
            a_     :  Any				      =     torch_device == '''cpu'''

            self._test_attention_slicing_forward_pass(test_max_difference=__SCREAMING_SNAKE_CASE   ,				expected_max_diff=0.0_1     )

       @skip_mps
       def     UpperCAmelCase( self   :  List[str]     ):
            a_     :  List[Any]				      =     torch_device == '''cpu'''
            a_     :  Optional[Any]				      =     True
            a_     :  Dict				      =     [
                '''prior_num_inference_steps''',
                '''decoder_num_inference_steps''',
                '''super_res_num_inference_steps''',
            ]

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

       def     UpperCAmelCase( self   :  int     ):
            a_     :  Any				      =     [
                '''prior_num_inference_steps''',
                '''decoder_num_inference_steps''',
                '''super_res_num_inference_steps''',
            ]

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

       @skip_mps
       def     UpperCAmelCase( self   :  str     ):
            return super().test_dict_tuple_outputs_equivalent()

       @skip_mps
       def     UpperCAmelCase( self   :  List[str]     ):
            return super().test_save_load_local()

       @skip_mps
       def     UpperCAmelCase( self   :  List[Any]     ):
            return super().test_save_load_optional_components()


@nightly
class 	a__						(    unittest.TestCase     ):

       def     UpperCAmelCase( self   :  str     ):
            super().tearDown()
            gc.collect()
            torch.cuda.empty_cache()

       def     UpperCAmelCase( self   :  List[Any]     ):
            a_     :  Optional[int]				      =     load_numpy(
                """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
                """/unclip/karlo_v1_alpha_horse_cpu.npy"""     )

            a_     :  Tuple				      =     UnCLIPPipeline.from_pretrained("""kakaobrain/karlo-v1-alpha"""     )
            pipeline.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE     )

            a_     :  Any				      =     torch.manual_seed(0     )
            a_     :  Any				      =     pipeline(
                """horse"""   ,				num_images_per_prompt=1   ,				generator=__SCREAMING_SNAKE_CASE   ,				output_type="""np"""   ,				)

            a_     :  Any				      =     output.images[0]

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


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

       def     UpperCAmelCase( self   :  Optional[Any]     ):
            super().tearDown()
            gc.collect()
            torch.cuda.empty_cache()

       def     UpperCAmelCase( self   :  int     ):
            a_     :  str				      =     load_numpy(
                """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
                """/unclip/karlo_v1_alpha_horse_fp16.npy"""     )

            a_     :  Any				      =     UnCLIPPipeline.from_pretrained("""kakaobrain/karlo-v1-alpha"""   ,				torch_dtype=torch.floataa     )
            a_     :  Dict				      =     pipeline.to(__SCREAMING_SNAKE_CASE     )
            pipeline.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE     )

            a_     :  Tuple				      =     torch.Generator(device="""cpu"""     ).manual_seed(0     )
            a_     :  List[Any]				      =     pipeline(
                """horse"""   ,				generator=__SCREAMING_SNAKE_CASE   ,				output_type="""np"""   ,				)

            a_     :  Union[str, Any]				      =     output.images[0]

            assert image.shape == (2_5_6, 2_5_6, 3)

            assert_mean_pixel_difference(__SCREAMING_SNAKE_CASE   ,				__SCREAMING_SNAKE_CASE     )

       def     UpperCAmelCase( self   :  List[Any]     ):
            torch.cuda.empty_cache()
            torch.cuda.reset_max_memory_allocated()
            torch.cuda.reset_peak_memory_stats()

            a_     :  Optional[Any]				      =     UnCLIPPipeline.from_pretrained("""kakaobrain/karlo-v1-alpha"""   ,				torch_dtype=torch.floataa     )
            a_     :  Union[str, Any]				      =     pipe.to(__SCREAMING_SNAKE_CASE     )
            pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE     )
            pipe.enable_attention_slicing()
            pipe.enable_sequential_cpu_offload()

            a_     :  str				      =     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"""   ,				)

            a_     :  Optional[Any]				      =     torch.cuda.max_memory_allocated()
            # make sure that less than 7 GB is allocated
            assert mem_bytes < 7 * 1_0**9







import os

from apache_beam.io.filesystems import FileSystems
from apache_beam.pipeline import Pipeline

from .logging import get_logger


__UpperCAmelCase			:			Optional[Any]			      =					2 << 20  # 2mb
__UpperCAmelCase			:			List[Any]			      =					get_logger(__name__)



class  UpperCAmelCase_						(				_a):
       '''simple docstring'''






       def 			_lowercase (  self    ):




              """simple docstring"""




              UpperCamelCase     :						Optional[Any] 						=		self._options.get_all_options().get('''runner'''    )
              return runner in [None, "DirectRunner", "PortableRunner"]




def   a						(				SCREAMING_SNAKE_CASE_   :		str				,      SCREAMING_SNAKE_CASE_   :		int				,      SCREAMING_SNAKE_CASE_   :		Optional[Any]=False							):






       """simple docstring"""





       UpperCamelCase     :						int 						=		FileSystems
       if fs.exists(SCREAMING_SNAKE_CASE_							):
              if force_upload:
                     logger.info(F"""Remote path already exist: {remote_file_path}. Overwriting it as force_upload=True."""							)
              else:
                     logger.info(F"""Remote path already exist: {remote_file_path}. Skipping it as force_upload=False."""							)
                     return
       with fs.create(SCREAMING_SNAKE_CASE_							) as remote_file:
              with open(SCREAMING_SNAKE_CASE_				,      '''rb'''							) as local_file:
                     UpperCamelCase     :						List[str] 						=		local_file.read(SCREAMING_SNAKE_CASE_							)
                     while chunk:
                            remote_file.write(SCREAMING_SNAKE_CASE_							)
                            UpperCamelCase     :						Any 						=		local_file.read(SCREAMING_SNAKE_CASE_							)


def   a						(				SCREAMING_SNAKE_CASE_   :		str				,      SCREAMING_SNAKE_CASE_   :		Optional[Any]				,      SCREAMING_SNAKE_CASE_   :		List[str]=False							):






       """simple docstring"""





       UpperCamelCase     :						List[Any] 						=		FileSystems
       if os.path.exists(SCREAMING_SNAKE_CASE_							):
              if force_download:
                     logger.info(F"""Local path already exist: {remote_file_path}. Overwriting it as force_upload=True."""							)
              else:
                     logger.info(F"""Local path already exist: {remote_file_path}. Skipping it as force_upload=False."""							)
                     return
       with fs.open(SCREAMING_SNAKE_CASE_							) as remote_file:
              with open(SCREAMING_SNAKE_CASE_				,      '''wb'''							) as local_file:
                     UpperCamelCase     :						List[str] 						=		remote_file.read(SCREAMING_SNAKE_CASE_							)
                     while chunk:
                            local_file.write(SCREAMING_SNAKE_CASE_							)
                            UpperCamelCase     :						str 						=		remote_file.read(SCREAMING_SNAKE_CASE_							)

