



'''simple docstring'''





import gc
import random
import unittest

import numpy as np
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModelWithProjection,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import (
    DiffusionPipeline,
    UnCLIPImageVariationPipeline,
    UnCLIPScheduler,
    UNetaDConditionModel,
    UNetaDModel,
)
from diffusers.pipelines.unclip.text_proj import UnCLIPTextProjModel
from diffusers.utils import floats_tensor, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, load_image, require_torch_gpu, skip_mps

from ..pipeline_params import IMAGE_VARIATION_BATCH_PARAMS, IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()







class        UpperCAmelCase_			(			_a						,						unittest.TestCase      ):
    """simple docstring"""







    lowercase   =						UnCLIPImageVariationPipeline
    lowercase   =						IMAGE_VARIATION_PARAMS - {"height", "width", "guidance_scale"}
    lowercase   =						IMAGE_VARIATION_BATCH_PARAMS

    lowercase   =						[
        "generator",
        "return_dict",
        "decoder_num_inference_steps",
        "super_res_num_inference_steps",
    ]
    lowercase   =						False




    @property
    def lowerCamelCase			(       self   :	Tuple		):
          return 32




    @property
    def lowerCamelCase			(       self   :	Tuple		):
          return 32




    @property
    def lowerCamelCase			(       self   :	List[Any]		):
          return self.time_input_dim




    @property
    def lowerCamelCase			(       self   :	Any		):
          return self.time_input_dim * 4




    @property
    def lowerCamelCase			(       self   :	List[Any]		):
          return 100




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




    @property
    def lowerCamelCase			(       self   :	List[str]		):
          torch.manual_seed(0		)
          snake_case__    :       List[str]       		= 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=1_000					,	)
          return CLIPTextModelWithProjection(snake_case_		)




    @property
    def lowerCamelCase			(       self   :	Optional[Any]		):
          torch.manual_seed(0		)
          snake_case__    :       Optional[Any]       		= CLIPVisionConfig(
              hidden_size=self.text_embedder_hidden_size					,	projection_dim=self.text_embedder_hidden_size					,	num_hidden_layers=5					,	num_attention_heads=4					,	image_size=32					,	intermediate_size=37					,	patch_size=1					,	)
          return CLIPVisionModelWithProjection(snake_case_		)




    @property
    def lowerCamelCase			(       self   :	Tuple		):
          torch.manual_seed(0		)

          snake_case__    :       Optional[int]       		= {
              """clip_embeddings_dim""": self.text_embedder_hidden_size,
              """time_embed_dim""": self.time_embed_dim,
              """cross_attention_dim""": self.cross_attention_dim,
          }

          snake_case__    :       Optional[Any]       		= UnCLIPTextProjModel(**snake_case_		)
          return model




    @property
    def lowerCamelCase			(       self   :	List[str]		):
          torch.manual_seed(0		)

          snake_case__    :       List[Any]       		= {
              """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""",
          }

          snake_case__    :       Union[str, Any]       		= UNetaDConditionModel(**snake_case_		)
          return model




    @property
    def lowerCamelCase			(       self   :	Optional[Any]		):
          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 lowerCamelCase			(       self   :	Tuple		):
          torch.manual_seed(0		)

          snake_case__    :       Optional[Any]       		= UNetaDModel(**self.dummy_super_res_kwargs		)
          return model




    @property
    def lowerCamelCase			(       self   :	List[Any]		):
          # seeded differently to get different unet than `self.dummy_super_res_first`
          torch.manual_seed(1		)

          snake_case__    :       List[str]       		= UNetaDModel(**self.dummy_super_res_kwargs		)
          return model




    def lowerCamelCase			(       self   :	Tuple		):
          snake_case__    :       Union[str, Any]       		= self.dummy_decoder
          snake_case__    :       Optional[int]       		= self.dummy_text_proj
          snake_case__    :       Any       		= self.dummy_text_encoder
          snake_case__    :       List[Any]       		= self.dummy_tokenizer
          snake_case__    :       Dict       		= self.dummy_super_res_first
          snake_case__    :       List[Any]       		= self.dummy_super_res_last

          snake_case__    :       Tuple       		= UnCLIPScheduler(
              variance_type="""learned_range"""					,	prediction_type="""epsilon"""					,	num_train_timesteps=1_000					,	)

          snake_case__    :       List[Any]       		= UnCLIPScheduler(
              variance_type="""fixed_small_log"""					,	prediction_type="""epsilon"""					,	num_train_timesteps=1_000					,	)

          snake_case__    :       str       		= CLIPImageProcessor(crop_size=32					,	size=32		)

          snake_case__    :       int       		= self.dummy_image_encoder

          return {
              "decoder": decoder,
              "text_encoder": text_encoder,
              "tokenizer": tokenizer,
              "text_proj": text_proj,
              "feature_extractor": feature_extractor,
              "image_encoder": image_encoder,
              "super_res_first": super_res_first,
              "super_res_last": super_res_last,
              "decoder_scheduler": decoder_scheduler,
              "super_res_scheduler": super_res_scheduler,
          }




    def lowerCamelCase			(       self   :	str					,	snake_case_   :	Dict					,	snake_case_   :	Union[str, Any]=0					,	snake_case_   :	List[str]=True		):
          snake_case__    :       Optional[Any]       		= floats_tensor((1, 3, 32, 32)					,	rng=random.Random(snake_case_		)		).to(snake_case_		)
          if str(snake_case_		).startswith("""mps"""		):
                snake_case__    :       Any       		= torch.manual_seed(snake_case_		)
          else:
                snake_case__    :       Dict       		= torch.Generator(device=snake_case_		).manual_seed(snake_case_		)

          if pil_image:
                snake_case__    :       Optional[Any]       		= input_image * 0.5 + 0.5
                snake_case__    :       List[Any]       		= input_image.clamp(0					,	1		)
                snake_case__    :       Optional[int]       		= input_image.cpu().permute(0					,	2					,	3					,	1		).float().numpy()
                snake_case__    :       Dict       		= DiffusionPipeline.numpy_to_pil(snake_case_		)[0]

          return {
              "image": input_image,
              "generator": generator,
              "decoder_num_inference_steps": 2,
              "super_res_num_inference_steps": 2,
              "output_type": "np",
          }




    def lowerCamelCase			(       self   :	int		):
          snake_case__    :       List[Any]       		= """cpu"""

          snake_case__    :       List[Any]       		= self.get_dummy_components()

          snake_case__    :       str       		= self.pipeline_class(**snake_case_		)
          snake_case__    :       List[str]       		= pipe.to(snake_case_		)

          pipe.set_progress_bar_config(disable=snake_case_		)

          snake_case__    :       Optional[Any]       		= self.get_dummy_inputs(snake_case_					,	pil_image=snake_case_		)

          snake_case__    :       Dict       		= pipe(**snake_case_		)
          snake_case__    :       List[str]       		= output.images

          snake_case__    :       str       		= self.get_dummy_inputs(snake_case_					,	pil_image=snake_case_		)

          snake_case__    :       Optional[Any]       		= pipe(
              **snake_case_					,	return_dict=snake_case_					,	)[0]

          snake_case__    :       List[Any]       		= image[0, -3:, -3:, -1]
          snake_case__    :       List[Any]       		= image_from_tuple[0, -3:, -3:, -1]

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

          snake_case__    :       List[str]       		= np.array(
              [
                  0.9997,
                  0.0002,
                  0.9997,
                  0.9997,
                  0.9969,
                  0.0023,
                  0.9997,
                  0.9969,
                  0.9970,
              ]		)

          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 lowerCamelCase			(       self   :	int		):
          snake_case__    :       Optional[Any]       		= """cpu"""

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

          snake_case__    :       Tuple       		= self.pipeline_class(**snake_case_		)
          snake_case__    :       Any       		= pipe.to(snake_case_		)

          pipe.set_progress_bar_config(disable=snake_case_		)

          snake_case__    :       List[str]       		= self.get_dummy_inputs(snake_case_					,	pil_image=snake_case_		)

          snake_case__    :       Optional[Any]       		= pipe(**snake_case_		)
          snake_case__    :       List[Any]       		= output.images

          snake_case__    :       Optional[int]       		= self.get_dummy_inputs(snake_case_					,	pil_image=snake_case_		)

          snake_case__    :       Optional[Any]       		= pipe(
              **snake_case_					,	return_dict=snake_case_					,	)[0]

          snake_case__    :       List[str]       		= image[0, -3:, -3:, -1]
          snake_case__    :       Optional[int]       		= image_from_tuple[0, -3:, -3:, -1]

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

          snake_case__    :       Optional[Any]       		= np.array([0.9997, 0.0003, 0.9997, 0.9997, 0.9970, 0.0024, 0.9997, 0.9971, 0.9971]		)

          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 lowerCamelCase			(       self   :	Tuple		):
          snake_case__    :       Tuple       		= """cpu"""

          snake_case__    :       List[str]       		= self.get_dummy_components()

          snake_case__    :       Dict       		= self.pipeline_class(**snake_case_		)
          snake_case__    :       int       		= pipe.to(snake_case_		)

          pipe.set_progress_bar_config(disable=snake_case_		)

          snake_case__    :       Any       		= self.get_dummy_inputs(snake_case_					,	pil_image=snake_case_		)
          snake_case__    :       str       		= [
              pipeline_inputs["""image"""],
              pipeline_inputs["""image"""],
          ]

          snake_case__    :       str       		= pipe(**snake_case_		)
          snake_case__    :       List[str]       		= output.images

          snake_case__    :       List[Any]       		= self.get_dummy_inputs(snake_case_					,	pil_image=snake_case_		)
          snake_case__    :       Any       		= [
              tuple_pipeline_inputs["""image"""],
              tuple_pipeline_inputs["""image"""],
          ]

          snake_case__    :       List[str]       		= pipe(
              **snake_case_					,	return_dict=snake_case_					,	)[0]

          snake_case__    :       Tuple       		= image[0, -3:, -3:, -1]
          snake_case__    :       Union[str, Any]       		= image_from_tuple[0, -3:, -3:, -1]

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

          snake_case__    :       Optional[int]       		= np.array(
              [
                  0.9997,
                  0.9989,
                  0.0008,
                  0.0021,
                  0.9960,
                  0.0018,
                  0.0014,
                  0.0002,
                  0.9933,
              ]		)

          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 lowerCamelCase			(       self   :	Union[str, Any]		):
          snake_case__    :       Dict       		= torch.device("""cpu"""		)







          class        UpperCAmelCase_			:
              """simple docstring"""







              lowercase   =						1
          snake_case__    :       Union[str, Any]       		= self.get_dummy_components()

          snake_case__    :       Union[str, Any]       		= self.pipeline_class(**snake_case_		)
          snake_case__    :       Tuple       		= pipe.to(snake_case_		)

          pipe.set_progress_bar_config(disable=snake_case_		)

          snake_case__    :       Optional[int]       		= torch.Generator(device=snake_case_		).manual_seed(0		)
          snake_case__    :       Union[str, Any]       		= pipe.decoder.dtype
          snake_case__    :       List[str]       		= 1

          snake_case__    :       str       		= (
              batch_size,
              pipe.decoder.config.in_channels,
              pipe.decoder.config.sample_size,
              pipe.decoder.config.sample_size,
          )
          snake_case__    :       List[Any]       		= pipe.prepare_latents(
              snake_case_					,	dtype=snake_case_					,	device=snake_case_					,	generator=snake_case_					,	latents=snake_case_					,	scheduler=DummyScheduler()		)

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

          snake_case__    :       Dict       		= self.get_dummy_inputs(snake_case_					,	pil_image=snake_case_		)

          snake_case__    :       Dict       		= pipe(
              **snake_case_					,	decoder_latents=snake_case_					,	super_res_latents=snake_case_		).images

          snake_case__    :       int       		= self.get_dummy_inputs(snake_case_					,	pil_image=snake_case_		)
          # Don't pass image, instead pass embedding
          snake_case__    :       Union[str, Any]       		= pipeline_inputs.pop("""image"""		)
          snake_case__    :       List[Any]       		= pipe.image_encoder(snake_case_		).image_embeds

          snake_case__    :       str       		= pipe(
              **snake_case_					,	decoder_latents=snake_case_					,	super_res_latents=snake_case_					,	image_embeddings=snake_case_					,	).images

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




    @skip_mps
    def lowerCamelCase			(       self   :	Optional[int]		):
          snake_case__    :       Optional[Any]       		= torch_device == """cpu"""

          # Check is relaxed because there is not a torch 2.0 sliced attention added kv processor
          snake_case__    :       Any       		= 1E-2

          self._test_attention_slicing_forward_pass(
              test_max_difference=snake_case_					,	expected_max_diff=snake_case_		)




    @skip_mps
    def lowerCamelCase			(       self   :	List[str]		):
          snake_case__    :       List[str]       		= torch_device == """cpu"""
          snake_case__    :       Any       		= True
          snake_case__    :       List[str]       		= [
              """decoder_num_inference_steps""",
              """super_res_num_inference_steps""",
          ]

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




    def lowerCamelCase			(       self   :	List[str]		):
          snake_case__    :       Dict       		= [
              """decoder_num_inference_steps""",
              """super_res_num_inference_steps""",
          ]

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




    @skip_mps
    def lowerCamelCase			(       self   :	Tuple		):
          return super().test_dict_tuple_outputs_equivalent()




    @skip_mps
    def lowerCamelCase			(       self   :	str		):
          return super().test_save_load_local()




    @skip_mps
    def lowerCamelCase			(       self   :	List[str]		):
          return super().test_save_load_optional_components()







@slow
@require_torch_gpu
class        UpperCAmelCase_			(			unittest.TestCase      ):
    """simple docstring"""







    def lowerCamelCase			(       self   :	Union[str, Any]		):
          # clean up the VRAM after each test
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()




    def lowerCamelCase			(       self   :	Optional[int]		):
          snake_case__    :       Optional[Any]       		= load_image(
              """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/unclip/cat.png"""		)
          snake_case__    :       Dict       		= load_numpy(
              """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
              """/unclip/karlo_v1_alpha_cat_variation_fp16.npy"""		)

          snake_case__    :       Optional[Any]       		= UnCLIPImageVariationPipeline.from_pretrained(
              """kakaobrain/karlo-v1-alpha-image-variations"""					,	torch_dtype=torch.floataa		)
          snake_case__    :       Any       		= pipeline.to(snake_case_		)
          pipeline.set_progress_bar_config(disable=snake_case_		)

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

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

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

          assert_mean_pixel_difference(snake_case_					,	snake_case_					,	15		)





'''simple docstring'''





import os
import unittest

from huggingface_hub.utils import are_progress_bars_disabled

import transformers.models.bart.tokenization_bart
from transformers import logging
from transformers.testing_utils import CaptureLogger, mockenv, mockenv_context
from transformers.utils.logging import disable_progress_bar, enable_progress_bar







class        UpperCAmelCase_			(			unittest.TestCase      ):
    """simple docstring"""







    def lowerCamelCase			(       self   :	Optional[int]		):
          snake_case__    :       int       		= logging.get_logger()

          # the current default level is logging.WARNING
          snake_case__    :       str       		= logging.get_verbosity()

          logging.set_verbosity_error()
          self.assertEqual(logger.getEffectiveLevel()					,	logging.get_verbosity()		)

          logging.set_verbosity_warning()
          self.assertEqual(logger.getEffectiveLevel()					,	logging.get_verbosity()		)

          logging.set_verbosity_info()
          self.assertEqual(logger.getEffectiveLevel()					,	logging.get_verbosity()		)

          logging.set_verbosity_debug()
          self.assertEqual(logger.getEffectiveLevel()					,	logging.get_verbosity()		)

          # restore to the original level
          logging.set_verbosity(snake_case_		)




    def lowerCamelCase			(       self   :	Union[str, Any]		):
          snake_case__    :       List[Any]       		= logging.get_verbosity()

          snake_case__    :       Any       		= logging.get_logger("""transformers.models.bart.tokenization_bart"""		)
          snake_case__    :       int       		= """Testing 1, 2, 3"""

          # should be able to log warnings (if default settings weren't overridden by `pytest --log-level-all`)
          if level_origin <= logging.WARNING:
                with CaptureLogger(snake_case_		) as cl:
                      logger.warning(snake_case_		)
                self.assertEqual(cl.out					,	msg + """\n"""		)

          # this is setting the level for all of `transformers.*` loggers
          logging.set_verbosity_error()

          # should not be able to log warnings
          with CaptureLogger(snake_case_		) as cl:
                logger.warning(snake_case_		)
          self.assertEqual(cl.out					,	""""""		)

          # should be able to log warnings again
          logging.set_verbosity_warning()
          with CaptureLogger(snake_case_		) as cl:
                logger.warning(snake_case_		)
          self.assertEqual(cl.out					,	msg + """\n"""		)

          # restore to the original level
          logging.set_verbosity(snake_case_		)




    @mockenv(TRANSFORMERS_VERBOSITY="""error"""		)
    def lowerCamelCase			(       self   :	Any		):
          # reset for the env var to take effect, next time some logger call is made
          transformers.utils.logging._reset_library_root_logger()
          # this action activates the env var
          snake_case__    :       Union[str, Any]       		= logging.get_logger("""transformers.models.bart.tokenization_bart"""		)

          snake_case__    :       int       		= os.getenv("""TRANSFORMERS_VERBOSITY"""					,	snake_case_		)
          snake_case__    :       List[Any]       		= logging.log_levels[env_level_str]

          snake_case__    :       str       		= logging.get_verbosity()
          self.assertEqual(
              snake_case_					,	snake_case_					,	f"TRANSFORMERS_VERBOSITY={env_level_str}/{env_level}, but internal verbosity is {current_level}"					,	)

          # restore to the original level
          snake_case__    :       List[Any]       		= """"""
          transformers.utils.logging._reset_library_root_logger()




    @mockenv(TRANSFORMERS_VERBOSITY="""super-error"""		)
    def lowerCamelCase			(       self   :	int		):
          # reset for the env var to take effect, next time some logger call is made
          transformers.utils.logging._reset_library_root_logger()
          snake_case__    :       List[Any]       		= logging.logging.getLogger()
          with CaptureLogger(snake_case_		) as cl:
                # this action activates the env var
                logging.get_logger("""transformers.models.bart.tokenization_bart"""		)
          self.assertIn("""Unknown option TRANSFORMERS_VERBOSITY=super-error"""					,	cl.out		)

          # no need to restore as nothing was changed




    def lowerCamelCase			(       self   :	int		):
          # testing `logger.warning_advice()`
          transformers.utils.logging._reset_library_root_logger()

          snake_case__    :       Optional[int]       		= logging.get_logger("""transformers.models.bart.tokenization_bart"""		)
          snake_case__    :       Tuple       		= """Testing 1, 2, 3"""

          with mockenv_context(TRANSFORMERS_NO_ADVISORY_WARNINGS="""1"""		):
                # nothing should be logged as env var disables this method
                with CaptureLogger(snake_case_		) as cl:
                      logger.warning_advice(snake_case_		)
                self.assertEqual(cl.out					,	""""""		)

          with mockenv_context(TRANSFORMERS_NO_ADVISORY_WARNINGS=""""""		):
                # should log normally as TRANSFORMERS_NO_ADVISORY_WARNINGS is unset
                with CaptureLogger(snake_case_		) as cl:
                      logger.warning_advice(snake_case_		)
                self.assertEqual(cl.out					,	msg + """\n"""		)





def 		__snake_case(							)						->					List[str]:
      disable_progress_bar()
      assert are_progress_bars_disabled()

      enable_progress_bar()
      assert not are_progress_bars_disabled()
