"""simple docstring"""







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	(    lowercase_		,   unittest.TestCase						):
   lowercase							=						KandinskyVaaPriorPipeline
   lowercase							=						['prompt']
   lowercase							=						['prompt', 'negative_prompt']
   lowercase							=						[
       'num_images_per_prompt',
       'generator',
       'num_inference_steps',
       'latents',
       'negative_prompt',
       'guidance_scale',
       'output_type',
       'return_dict',
   ]
   lowercase							=						False
   @property
   def     _UpperCAmelCase     (    self    )			->   Any:


        '''simple docstring'''


        return 32
   @property
   def     _UpperCAmelCase     (    self    )			->   Dict:


        '''simple docstring'''


        return 32
   @property
   def     _UpperCAmelCase     (    self    )			->   int:


        '''simple docstring'''


        return self.time_input_dim
   @property
   def     _UpperCAmelCase     (    self    )			->   List[str]:


        '''simple docstring'''


        return self.time_input_dim * 4
   @property
   def     _UpperCAmelCase     (    self    )			->   Dict:


        '''simple docstring'''


        return 100
   @property
   def     _UpperCAmelCase     (    self    )			->   Any:


        '''simple docstring'''


        lowercase_			:       str									=			CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'    )
        return tokenizer
   @property
   def     _UpperCAmelCase     (    self    )			->   Tuple:


        '''simple docstring'''


        torch.manual_seed(0    )
        lowercase_			:       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=37  ,layer_norm_eps=1e-05  ,num_attention_heads=4  ,num_hidden_layers=5  ,pad_token_id=1  ,vocab_size=1000  ,)
        return CLIPTextModelWithProjection(__UpperCamelCase    )
   @property
   def     _UpperCAmelCase     (    self    )			->   Any:


        '''simple docstring'''


        torch.manual_seed(0    )

        lowercase_			:       List[str]									=			{
            'num_attention_heads': 2,
            'attention_head_dim': 12,
            'embedding_dim': self.text_embedder_hidden_size,
            'num_layers': 1,
        }

        lowercase_			:       List[str]									=			PriorTransformer(**__UpperCamelCase    )
        # 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
        lowercase_			:       List[str]									=			nn.Parameter(torch.ones(model.clip_std.shape    )    )
        return model
   @property
   def     _UpperCAmelCase     (    self    )			->   Any:


        '''simple docstring'''


        torch.manual_seed(0    )
        lowercase_			:       Union[str, Any]									=			CLIPVisionConfig(
            hidden_size=self.text_embedder_hidden_size  ,image_size=224  ,projection_dim=self.text_embedder_hidden_size  ,intermediate_size=37  ,num_attention_heads=4  ,num_channels=3  ,num_hidden_layers=5  ,patch_size=14  ,)

        lowercase_			:       List[Any]									=			CLIPVisionModelWithProjection(__UpperCamelCase    )
        return model
   @property
   def     _UpperCAmelCase     (    self    )			->   Tuple:


        '''simple docstring'''


        lowercase_			:       Union[str, Any]									=			CLIPImageProcessor(
            crop_size=224  ,do_center_crop=__UpperCamelCase  ,do_normalize=__UpperCamelCase  ,do_resize=__UpperCamelCase  ,image_mean=[0.4814_5466, 0.457_8275, 0.4082_1073]  ,image_std=[0.2686_2954, 0.2613_0258, 0.2757_7711]  ,resample=3  ,size=224  ,)

        return image_processor
   def     _UpperCAmelCase     (    self    )			->   int:


        '''simple docstring'''


        lowercase_			:       Union[str, Any]									=			self.dummy_prior
        lowercase_			:       Optional[int]									=			self.dummy_image_encoder
        lowercase_			:       Any									=			self.dummy_text_encoder
        lowercase_			:       Any									=			self.dummy_tokenizer
        lowercase_			:       Dict									=			self.dummy_image_processor

        lowercase_			:       List[Any]									=			UnCLIPScheduler(
            variance_type='fixed_small_log'  ,prediction_type='sample'  ,num_train_timesteps=1000  ,clip_sample=__UpperCamelCase  ,clip_sample_range=10.0  ,)

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

        return components
   def     _UpperCAmelCase     (    self  ,__UpperCamelCase  ,__UpperCamelCase=0    )			->   Tuple:


        '''simple docstring'''


        if str(__UpperCamelCase    ).startswith('mps'    ):
             lowercase_			:       int									=			torch.manual_seed(__UpperCamelCase    )
        else:
             lowercase_			:       Optional[int]									=			torch.Generator(device=__UpperCamelCase    ).manual_seed(__UpperCamelCase    )
        lowercase_			:       int									=			{
            'prompt': 'horse',
            'generator': generator,
            'guidance_scale': 4.0,
            'num_inference_steps': 2,
            'output_type': 'np',
        }
        return inputs
   def     _UpperCAmelCase     (    self    )			->   Tuple:


        '''simple docstring'''


        lowercase_			:       List[Any]									=			'cpu'

        lowercase_			:       int									=			self.get_dummy_components()

        lowercase_			:       List[str]									=			self.pipeline_class(**__UpperCamelCase    )
        lowercase_			:       Tuple									=			pipe.to(__UpperCamelCase    )

        pipe.set_progress_bar_config(disable=__UpperCamelCase    )

        lowercase_			:       str									=			pipe(**self.get_dummy_inputs(__UpperCamelCase    )    )
        lowercase_			:       Optional[int]									=			output.image_embeds

        lowercase_			:       Any									=			pipe(
            **self.get_dummy_inputs(__UpperCamelCase    )  ,return_dict=__UpperCamelCase  ,)[0]

        lowercase_			:       Optional[Any]									=			image[0, -10:]
        lowercase_			:       Union[str, Any]									=			image_from_tuple[0, -10:]

        assert image.shape == (1, 32)

        lowercase_			:       Dict									=			np.array(
            [-0.0532, 1.7120, 0.3656, -1.0852, -0.8946, -1.1756, 0.4348, 0.2482, 0.5146, -0.1156]    )

        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     _UpperCAmelCase     (    self    )			->   Optional[int]:


        '''simple docstring'''


        lowercase_			:       Optional[Any]									=			torch_device == 'cpu'
        lowercase_			:       Any									=			True
        lowercase_			:       Optional[int]									=			False

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







   @skip_mps
   def     _UpperCAmelCase     (    self    )			->   str:


        '''simple docstring'''


        lowercase_			:       List[str]									=			torch_device == 'cpu'
        lowercase_			:       Optional[int]									=			False

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


"""simple docstring"""







import os
import unittest
from tempfile import TemporaryDirectory

import torch
import torch.nn as nn

from accelerate.utils import (
    OffloadedWeightsLoader,
    extract_submodules_state_dict,
    load_offloaded_weight,
    offload_state_dict,
    offload_weight,
)







class  UpperCamelCase	(    nn.Module						):
   def __init__(    self    )			->   Any:


        '''simple docstring'''


        super().__init__()
        lowercase_			:       Dict									=			nn.Linear(3  ,4    )
        lowercase_			:       Union[str, Any]									=			nn.BatchNormad(4    )
        lowercase_			:       Union[str, Any]									=			nn.Linear(4  ,5    )







   def     _UpperCAmelCase     (    self  ,__UpperCamelCase    )			->   int:


        '''simple docstring'''


        return self.lineara(self.batchnorm(self.lineara(__UpperCamelCase    )    )    )







class  UpperCamelCase	(    unittest.TestCase						):
   def     _UpperCAmelCase     (    self    )			->   Dict:


        '''simple docstring'''


        lowercase_			:       int									=			ModelForTest()
        with TemporaryDirectory() as tmp_dir:
             offload_state_dict(__UpperCamelCase  ,model.state_dict()    )
             lowercase_			:       str									=			os.path.join(__UpperCamelCase  ,'index.json'    )
             self.assertTrue(os.path.isfile(__UpperCamelCase    )    )
             # TODO: add tests on what is inside the index

             for key in ["linear1.weight", "linear1.bias", "linear2.weight", "linear2.bias"]:
                  lowercase_			:       int									=			os.path.join(__UpperCamelCase  ,f'''{key}.dat'''    )
                  self.assertTrue(os.path.isfile(__UpperCamelCase    )    )
                  # TODO: add tests on the fact weights are properly loaded
   def     _UpperCAmelCase     (    self    )			->   Tuple:


        '''simple docstring'''


        lowercase_			:       List[str]									=			[torch.floataa, torch.floataa, torch.bfloataa]

        for dtype in dtypes:
             lowercase_			:       Tuple									=			torch.randn(2  ,3  ,dtype=__UpperCamelCase    )
             with TemporaryDirectory() as tmp_dir:
                  lowercase_			:       List[str]									=			offload_weight(__UpperCamelCase  ,'weight'  ,__UpperCamelCase  ,{}    )
                  lowercase_			:       str									=			os.path.join(__UpperCamelCase  ,'weight.dat'    )
                  self.assertTrue(os.path.isfile(__UpperCamelCase    )    )
                  self.assertDictEqual(__UpperCamelCase  ,{'weight': {'shape': [2, 3], 'dtype': str(__UpperCamelCase    ).split('.'    )[1]}}    )

                  lowercase_			:       Optional[Any]									=			load_offloaded_weight(__UpperCamelCase  ,index['weight']    )
                  self.assertTrue(torch.equal(__UpperCamelCase  ,__UpperCamelCase    )    )
   def     _UpperCAmelCase     (    self    )			->   List[Any]:


        '''simple docstring'''


        lowercase_			:       List[str]									=			ModelForTest()
        lowercase_			:       Optional[int]									=			model.state_dict()
        lowercase_			:       int									=			{k: v for k, v in state_dict.items() if 'linear2' not in k}
        lowercase_			:       Optional[Any]									=			{k: v for k, v in state_dict.items() if 'linear2' in k}

        with TemporaryDirectory() as tmp_dir:
             offload_state_dict(__UpperCamelCase  ,__UpperCamelCase    )
             lowercase_			:       Tuple									=			OffloadedWeightsLoader(state_dict=__UpperCamelCase  ,save_folder=__UpperCamelCase    )

             # Every key is there with the right value
             self.assertEqual(sorted(__UpperCamelCase    )  ,sorted(state_dict.keys()    )    )
             for key, param in state_dict.items():
                  self.assertTrue(torch.allclose(__UpperCamelCase  ,weight_map[key]    )    )

        lowercase_			:       int									=			{k: v for k, v in state_dict.items() if 'weight' in k}
        lowercase_			:       int									=			{k: v for k, v in state_dict.items() if 'weight' not in k}

        with TemporaryDirectory() as tmp_dir:
             offload_state_dict(__UpperCamelCase  ,__UpperCamelCase    )
             lowercase_			:       int									=			OffloadedWeightsLoader(state_dict=__UpperCamelCase  ,save_folder=__UpperCamelCase    )

             # Every key is there with the right value
             self.assertEqual(sorted(__UpperCamelCase    )  ,sorted(state_dict.keys()    )    )
             for key, param in state_dict.items():
                  self.assertTrue(torch.allclose(__UpperCamelCase  ,weight_map[key]    )    )

        with TemporaryDirectory() as tmp_dir:
             offload_state_dict(__UpperCamelCase  ,__UpperCamelCase    )
             # Duplicates are removed
             lowercase_			:       Tuple									=			OffloadedWeightsLoader(state_dict=__UpperCamelCase  ,save_folder=__UpperCamelCase    )

             # Every key is there with the right value
             self.assertEqual(sorted(__UpperCamelCase    )  ,sorted(state_dict.keys()    )    )
             for key, param in state_dict.items():
                  self.assertTrue(torch.allclose(__UpperCamelCase  ,weight_map[key]    )    )







   def     _UpperCAmelCase     (    self    )			->   Union[str, Any]:


        '''simple docstring'''


        lowercase_			:       Optional[int]									=			{'a.1': 0, 'a.10': 1, 'a.2': 2}
        lowercase_			:       int									=			extract_submodules_state_dict(__UpperCamelCase  ,['a.1', 'a.2']    )
        self.assertDictEqual(__UpperCamelCase  ,{'a.1': 0, 'a.2': 2}    )

        lowercase_			:       Union[str, Any]									=			{'a.1.a': 0, 'a.10.a': 1, 'a.2.a': 2}
        lowercase_			:       Optional[int]									=			extract_submodules_state_dict(__UpperCamelCase  ,['a.1', 'a.2']    )
        self.assertDictEqual(__UpperCamelCase  ,{'a.1.a': 0, 'a.2.a': 2}    )

