



import unittest

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

from diffusers import KandinskyVaaPriorPipeline, PriorTransformer, UnCLIPScheduler
from diffusers.utils import torch_device
from diffusers.utils.testing_utils import enable_full_determinism, skip_mps

from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()



class      _UpperCAmelCase      (							A__      ,   unittest.TestCase  ):
     UpperCamelCase__		     =					KandinskyVaaPriorPipeline
     UpperCamelCase__		     =					['''prompt''']
     UpperCamelCase__		     =					['''prompt''', '''negative_prompt''']
     UpperCamelCase__		     =					[
         '''num_images_per_prompt''',
         '''generator''',
         '''num_inference_steps''',
         '''latents''',
         '''negative_prompt''',
         '''guidance_scale''',
         '''output_type''',
         '''return_dict''',
     ]
     UpperCamelCase__		     =					False




     @property
     def 		snake_case_		(					self):
         return 3_2




     @property
     def 		snake_case_		(					self):
         return 3_2




     @property
     def 		snake_case_		(					self):
         return self.time_input_dim




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




     @property
     def 		snake_case_		(					self):
         return 1_0_0




     @property
     def 		snake_case_		(					self):
         A__					=       CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip''')
         return tokenizer




     @property
     def 		snake_case_		(					self):
         torch.manual_seed(0)
         A__					=       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(a__)




     @property
     def 		snake_case_		(					self):
         torch.manual_seed(0)

         A__					=       {
             '''num_attention_heads''': 2,
             '''attention_head_dim''': 1_2,
             '''embedding_dim''': self.text_embedder_hidden_size,
             '''num_layers''': 1,
         }

         A__					=       PriorTransformer(**a__)
         # clip_std and clip_mean is initialized to be 0 so PriorTransformer.post_process_latents will always return 0 - set clip_std to be 1 so it won't return 0
         A__					=       nn.Parameter(torch.ones(model.clip_std.shape))
         return model




     @property
     def 		snake_case_		(					self):
         torch.manual_seed(0)
         A__					=       CLIPVisionConfig(
             hidden_size=self.text_embedder_hidden_size			,    image_size=2_2_4			,    projection_dim=self.text_embedder_hidden_size			,    intermediate_size=3_7			,    num_attention_heads=4			,    num_channels=3			,    num_hidden_layers=5			,    patch_size=1_4			,    )

         A__					=       CLIPVisionModelWithProjection(a__)
         return model




     @property
     def 		snake_case_		(					self):
         A__					=       CLIPImageProcessor(
             crop_size=2_2_4			,    do_center_crop=a__			,    do_normalize=a__			,    do_resize=a__			,    image_mean=[0.4_8_1_4_5_4_6_6, 0.4_5_7_8_2_7_5, 0.4_0_8_2_1_0_7_3]			,    image_std=[0.2_6_8_6_2_9_5_4, 0.2_6_1_3_0_2_5_8, 0.2_7_5_7_7_7_1_1]			,    resample=3			,    size=2_2_4			,    )

         return image_processor




     def 		snake_case_		(					self):
         A__					=       self.dummy_prior
         A__					=       self.dummy_image_encoder
         A__					=       self.dummy_text_encoder
         A__					=       self.dummy_tokenizer
         A__					=       self.dummy_image_processor

         A__					=       UnCLIPScheduler(
             variance_type='''fixed_small_log'''			,    prediction_type='''sample'''			,    num_train_timesteps=1_0_0_0			,    clip_sample=a__			,    clip_sample_range=1_0.0			,    )

         A__					=       {
             '''prior''': prior,
             '''image_encoder''': image_encoder,
             '''text_encoder''': text_encoder,
             '''tokenizer''': tokenizer,
             '''scheduler''': scheduler,
             '''image_processor''': image_processor,
         }

         return components




     def 		snake_case_		(					self			,    a__			,    a__=0):
         if str(a__).startswith('''mps'''):
             A__					=       torch.manual_seed(a__)
         else:
             A__					=       torch.Generator(device=a__).manual_seed(a__)
         A__					=       {
             '''prompt''': '''horse''',
             '''generator''': generator,
             '''guidance_scale''': 4.0,
             '''num_inference_steps''': 2,
             '''output_type''': '''np''',
         }
         return inputs




     def 		snake_case_		(					self):
         A__					=       '''cpu'''

         A__					=       self.get_dummy_components()

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

         pipe.set_progress_bar_config(disable=a__)

         A__					=       pipe(**self.get_dummy_inputs(a__))
         A__					=       output.image_embeds

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

         A__					=       image[0, -1_0:]
         A__					=       image_from_tuple[0, -1_0:]

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

         A__					=       np.array(
             [-0.0_5_3_2, 1.7_1_2_0, 0.3_6_5_6, -1.0_8_5_2, -0.8_9_4_6, -1.1_7_5_6, 0.4_3_4_8, 0.2_4_8_2, 0.5_1_4_6, -0.1_1_5_6])

         assert np.abs(image_slice.flatten() - expected_slice).max() < 1e-2
         assert np.abs(image_from_tuple_slice.flatten() - expected_slice).max() < 1e-2




     @skip_mps
     def 		snake_case_		(					self):
         A__					=       torch_device == '''cpu'''
         A__					=       True
         A__					=       False

         self._test_inference_batch_single_identical(
             test_max_difference=a__			,    relax_max_difference=a__			,    test_mean_pixel_difference=a__			,    )




     @skip_mps
     def 		snake_case_		(					self):
         A__					=       torch_device == '''cpu'''
         A__					=       False

         self._test_attention_slicing_forward_pass(
             test_max_difference=a__			,    test_mean_pixel_difference=a__			,    )





import inspect
import os
import unittest
from dataclasses import dataclass

import torch

from accelerate import Accelerator, DistributedDataParallelKwargs, GradScalerKwargs
from accelerate.state import AcceleratorState
from accelerate.test_utils import execute_subprocess_async, require_cuda, require_multi_gpu
from accelerate.utils import KwargsHandler



@dataclass
class      _UpperCAmelCase      (							A__  ):
     UpperCamelCase__		     =					0
     UpperCamelCase__		     =					False
     UpperCamelCase__		     =					3.0







class      _UpperCAmelCase      (							unittest.TestCase  ):




     def 		snake_case_		(					self):
         # If no defaults are changed, `to_kwargs` returns an empty dict.
         self.assertDictEqual(MockClass().to_kwargs()			,    {})
         self.assertDictEqual(MockClass(a=2).to_kwargs()			,    {'''a''': 2})
         self.assertDictEqual(MockClass(a=2			,    b=a__).to_kwargs()			,    {'''a''': 2, '''b''': True})
         self.assertDictEqual(MockClass(a=2			,    c=2.2_5).to_kwargs()			,    {'''a''': 2, '''c''': 2.2_5})




     @require_cuda
     def 		snake_case_		(					self):
         # If no defaults are changed, `to_kwargs` returns an empty dict.
         A__					=       GradScalerKwargs(init_scale=1_0_2_4			,    growth_factor=2)
         AcceleratorState._reset_state()
         A__					=       Accelerator(mixed_precision='''fp16'''			,    kwargs_handlers=[scaler_handler])
         print(accelerator.use_fpaa)
         A__					=       accelerator.scaler

         # Check the kwargs have been applied
         self.assertEqual(scaler._init_scale			,    1_0_2_4.0)
         self.assertEqual(scaler._growth_factor			,    2.0)

         # Check the other values are at the default
         self.assertEqual(scaler._backoff_factor			,    0.5)
         self.assertEqual(scaler._growth_interval			,    2_0_0_0)
         self.assertEqual(scaler._enabled			,    a__)




     @require_multi_gpu
     def 		snake_case_		(					self):
         A__					=       ['''torchrun''', F"--nproc_per_node={torch.cuda.device_count()}", inspect.getfile(self.__class__)]
         execute_subprocess_async(a__			,    env=os.environ.copy())


if __name__ == "__main__":
   _lowercase    	=		DistributedDataParallelKwargs(bucket_cap_mb=15, find_unused_parameters=True)
   _lowercase    	=		Accelerator(kwargs_handlers=[ddp_scaler])
   _lowercase    	=		torch.nn.Linear(100, 200)
   _lowercase    	=		accelerator.prepare(model)

   # Check the values changed in kwargs
   _lowercase    	=		""
   _lowercase    	=		model.bucket_bytes_cap // (1024 * 1024)
   if observed_bucket_cap_map != 15:
      error_msg += F"Kwargs badly passed, should have `15` but found {observed_bucket_cap_map}.\n"
   if model.find_unused_parameters is not True:
      error_msg += F"Kwargs badly passed, should have `True` but found {model.find_unused_parameters}.\n"

   # Check the values of the defaults
   if model.dim != 0:
      error_msg += F"Default value not respected, should have `0` but found {model.dim}.\n"
   if model.broadcast_buffers is not True:
      error_msg += F"Default value not respected, should have `True` but found {model.broadcast_buffers}.\n"
   if model.gradient_as_bucket_view is not False:
      error_msg += F"Default value not respected, should have `False` but found {model.gradient_as_bucket_view}.\n"

   # Raise error at the end to make sure we don't stop at the first failure.
   if len(error_msg) > 0:
      raise ValueError(error_msg)
