




"""simple docstring"""

import gc
import unittest

import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    PriorTransformer,
    StableUnCLIPPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.testing_utils import enable_full_determinism, load_numpy, require_torch_gpu, slow, torch_device

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,
    assert_mean_pixel_difference,
)


enable_full_determinism()



class     lowercase_	(					__lowerCAmelCase      ,     __lowerCAmelCase      ,     __lowerCAmelCase      ,     unittest.TestCase							):



 '''simple docstring'''
 UpperCAmelCase   :					List[str]				  =     StableUnCLIPPipeline
 UpperCAmelCase   :					int				  =     TEXT_TO_IMAGE_PARAMS
 UpperCAmelCase   :					Any				  =     TEXT_TO_IMAGE_BATCH_PARAMS
 UpperCAmelCase   :					int				  =     TEXT_TO_IMAGE_IMAGE_PARAMS
 UpperCAmelCase   :					List[str]				  =     TEXT_TO_IMAGE_IMAGE_PARAMS

 # TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
 UpperCAmelCase   :					str				  =     False

 def       lowerCAmelCase_							(			self	:	Dict		):
       _A        =       32
       _A        =       embedder_hidden_size

       # prior components

       torch.manual_seed(0		)
       _A        =       CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'		)

       torch.manual_seed(0		)
       _A        =       CLIPTextModelWithProjection(
           CLIPTextConfig(
               bos_token_id=0   ,							eos_token_id=2   ,							hidden_size=_UpperCAmelCase   ,							projection_dim=_UpperCAmelCase   ,							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   ,							)		)

       torch.manual_seed(0		)
       _A        =       PriorTransformer(
           num_attention_heads=2   ,							attention_head_dim=12   ,							embedding_dim=_UpperCAmelCase   ,							num_layers=1   ,							)

       torch.manual_seed(0		)
       _A        =       DDPMScheduler(
           variance_type='fixed_small_log'   ,							prediction_type='sample'   ,							num_train_timesteps=1_000   ,							clip_sample=_UpperCAmelCase   ,							clip_sample_range=5.0   ,							beta_schedule='squaredcos_cap_v2'   ,							)

       # regular denoising components

       torch.manual_seed(0		)
       _A        =       StableUnCLIPImageNormalizer(embedding_dim=_UpperCAmelCase		)
       _A        =       DDPMScheduler(beta_schedule='squaredcos_cap_v2'		)

       torch.manual_seed(0		)
       _A        =       CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'		)

       torch.manual_seed(0		)
       _A        =       CLIPTextModel(
           CLIPTextConfig(
               bos_token_id=0   ,							eos_token_id=2   ,							hidden_size=_UpperCAmelCase   ,							projection_dim=32   ,							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   ,							)		)

       torch.manual_seed(0		)
       _A        =       UNetaDConditionModel(
           sample_size=32   ,							in_channels=4   ,							out_channels=4   ,							down_block_types=('CrossAttnDownBlock2D', 'DownBlock2D')   ,							up_block_types=('UpBlock2D', 'CrossAttnUpBlock2D')   ,							block_out_channels=(32, 64)   ,							attention_head_dim=(2, 4)   ,							class_embed_type='projection'   ,							projection_class_embeddings_input_dim=embedder_projection_dim * 2   ,							cross_attention_dim=_UpperCAmelCase   ,							layers_per_block=1   ,							upcast_attention=_UpperCAmelCase   ,							use_linear_projection=_UpperCAmelCase   ,							)

       torch.manual_seed(0		)
       _A        =       DDIMScheduler(
           beta_schedule='scaled_linear'   ,							beta_start=0.0_0085   ,							beta_end=0.012   ,							prediction_type='v_prediction'   ,							set_alpha_to_one=_UpperCAmelCase   ,							steps_offset=1   ,							)

       torch.manual_seed(0		)
       _A        =       AutoencoderKL()

       _A        =       {
           # prior components
           'prior_tokenizer': prior_tokenizer,
           'prior_text_encoder': prior_text_encoder,
           'prior': prior,
           'prior_scheduler': prior_scheduler,
           # image noising components
           'image_normalizer': image_normalizer,
           'image_noising_scheduler': image_noising_scheduler,
           # regular denoising components
           'tokenizer': tokenizer,
           'text_encoder': text_encoder,
           'unet': unet,
           'scheduler': scheduler,
           'vae': vae,
       }

       return components

 def       lowerCAmelCase_							(			self	:	Tuple   ,							_UpperCAmelCase	:	Dict   ,							_UpperCAmelCase	:	List[Any]=0		):
       if str(_UpperCAmelCase		).startswith('mps'		):
             _A        =       torch.manual_seed(_UpperCAmelCase		)
       else:
             _A        =       torch.Generator(device=_UpperCAmelCase		).manual_seed(_UpperCAmelCase		)
       _A        =       {
           'prompt': 'A painting of a squirrel eating a burger',
           'generator': generator,
           'num_inference_steps': 2,
           'prior_num_inference_steps': 2,
           'output_type': 'numpy',
       }
       return inputs

 def       lowerCAmelCase_							(			self	:	str		):
       _A        =       torch_device == 'cpu'

       self._test_attention_slicing_forward_pass(test_max_difference=_UpperCAmelCase		)



 def       lowerCAmelCase_							(			self	:	Any		):
       _A        =       torch_device in ['cpu', 'mps']

       self._test_inference_batch_single_identical(test_max_difference=_UpperCAmelCase		)



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



 '''simple docstring'''

 def       lowerCAmelCase_							(			self	:	Optional[int]		):
       # clean up the VRAM after each test
       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()

 def       lowerCAmelCase_							(			self	:	List[Any]		):
       _A        =       load_numpy(
           'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy'		)

       _A        =       StableUnCLIPPipeline.from_pretrained('fusing/stable-unclip-2-1-l'   ,							torch_dtype=torch.floataa		)
       pipe.to(_UpperCAmelCase		)
       pipe.set_progress_bar_config(disable=_UpperCAmelCase		)
       # stable unclip will oom when integration tests are run on a V100,
       # so turn on memory savings
       pipe.enable_attention_slicing()
       pipe.enable_sequential_cpu_offload()

       _A        =       torch.Generator(device='cpu'		).manual_seed(0		)
       _A        =       pipe('anime turle'   ,							generator=_UpperCAmelCase   ,							output_type='np'		)

       _A        =       output.images[0]

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

       assert_mean_pixel_difference(_UpperCAmelCase   ,							_UpperCAmelCase		)



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

       _A        =       StableUnCLIPPipeline.from_pretrained('fusing/stable-unclip-2-1-l'   ,							torch_dtype=torch.floataa		)
       _A        =       pipe.to(_UpperCAmelCase		)
       pipe.set_progress_bar_config(disable=_UpperCAmelCase		)
       pipe.enable_attention_slicing()
       pipe.enable_sequential_cpu_offload()

       _A        =       pipe(
           'anime turtle'   ,							prior_num_inference_steps=2   ,							num_inference_steps=2   ,							output_type='np'   ,							)

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







"""simple docstring"""

import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional, Union

from .generation.configuration_utils import GenerationConfig
from .training_args import TrainingArguments
from .utils import add_start_docstrings


a         =      logging.getLogger(__name__)





@dataclass
@add_start_docstrings(TrainingArguments.__doc__							)
class     lowercase_	(					__lowerCAmelCase							):



 '''simple docstring'''
 UpperCAmelCase   :					bool				  =     field(default=__lowerCAmelCase      ,     metadata={'''help''': '''Whether to use SortishSampler or not.'''}							)
 UpperCAmelCase   :					bool				  =     field(
     default=__lowerCAmelCase      ,     metadata={'''help''': '''Whether to use generate to calculate generative metrics (ROUGE, BLEU).'''}							)
 UpperCAmelCase   :					Optional[int]				  =     field(
     default=__lowerCAmelCase      ,     metadata={
         '''help''': (
             '''The `max_length` to use on each evaluation loop when `predict_with_generate=True`. Will default '''
             '''to the `max_length` value of the model configuration.'''
         )
     }      ,     )
 UpperCAmelCase   :					Optional[int]				  =     field(
     default=__lowerCAmelCase      ,     metadata={
         '''help''': (
             '''The `num_beams` to use on each evaluation loop when `predict_with_generate=True`. Will default '''
             '''to the `num_beams` value of the model configuration.'''
         )
     }      ,     )
 UpperCAmelCase   :					Optional[Union[str, Path, GenerationConfig]]				  =     field(
     default=__lowerCAmelCase      ,     metadata={
         '''help''': '''Model id, file path or url pointing to a GenerationConfig json file, to use during prediction.'''
     }      ,     )

 def       lowerCAmelCase_							(			self	:	int		):
       _A        =       super().to_dict()
       for k, v in d.items():
             if isinstance(_UpperCAmelCase   ,							_UpperCAmelCase		):
                   _A        =       v.to_dict()
       return d

