
"""simple docstring"""




import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    require_torch_gpu,
)

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()







class 						lowercase(_lowercase						,	_lowercase						,	unittest.TestCase					):
     __snake_case:  Tuple  				=    StableDiffusionParadigmsPipeline
     __snake_case:  str  				=    TEXT_TO_IMAGE_PARAMS
     __snake_case:  List[str]  				=    TEXT_TO_IMAGE_BATCH_PARAMS
     __snake_case:  str  				=    TEXT_TO_IMAGE_IMAGE_PARAMS
     __snake_case:  Tuple  				=    TEXT_TO_IMAGE_IMAGE_PARAMS



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







         """simple docstring"""



         torch.manual_seed(0      )
         a__    = UNetaDConditionModel(
             block_out_channels=(3_2, 6_4)    ,							layers_per_block=2    ,							sample_size=3_2    ,							in_channels=4    ,							out_channels=4    ,							down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')    ,							up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')    ,							cross_attention_dim=3_2    ,							attention_head_dim=(2, 4)    ,							use_linear_projection=__SCREAMING_SNAKE_CASE    ,							)
         a__    = DDIMParallelScheduler(
             beta_start=0.0_00_85    ,							beta_end=0.0_12    ,							beta_schedule='scaled_linear'    ,							clip_sample=__SCREAMING_SNAKE_CASE    ,							set_alpha_to_one=__SCREAMING_SNAKE_CASE    ,							)
         torch.manual_seed(0      )
         a__    = AutoencoderKL(
             block_out_channels=[3_2, 6_4]    ,							in_channels=3    ,							out_channels=3    ,							down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']    ,							up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']    ,							latent_channels=4    ,							sample_size=1_2_8    ,							)
         torch.manual_seed(0      )
         a__    = CLIPTextConfig(
             bos_token_id=0    ,							eos_token_id=2    ,							hidden_size=3_2    ,							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    ,							hidden_act='gelu'    ,							projection_dim=5_1_2    ,							)
         a__    = CLIPTextModel(__SCREAMING_SNAKE_CASE      )
         a__    = CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'      )

         a__    = {
             'unet': unet,
             'scheduler': scheduler,
             'vae': vae,
             'text_encoder': text_encoder,
             'tokenizer': tokenizer,
             'safety_checker': None,
             'feature_extractor': None,
         }
         return components



     def  lowercase__     (       self    ,							__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE=0      )		->    List[Any]:







         """simple docstring"""



         if str(__SCREAMING_SNAKE_CASE      ).startswith('mps'      ):
             a__    = torch.manual_seed(__SCREAMING_SNAKE_CASE      )
         else:
             a__    = torch.Generator(device=__SCREAMING_SNAKE_CASE      ).manual_seed(__SCREAMING_SNAKE_CASE      )
         a__    = {
             'prompt': 'a photograph of an astronaut riding a horse',
             'generator': generator,
             'num_inference_steps': 1_0,
             'guidance_scale': 6.0,
             'output_type': 'numpy',
             'parallel': 3,
             'debug': True,
         }
         return inputs



     def  lowercase__     (       self      )		->    Tuple:







         """simple docstring"""



         a__    = 'cpu'  # ensure determinism for the device-dependent torch.Generator
         a__    = self.get_dummy_components()
         a__    = StableDiffusionParadigmsPipeline(**__SCREAMING_SNAKE_CASE      )
         a__    = sd_pipe.to(__SCREAMING_SNAKE_CASE      )
         sd_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE      )

         a__    = self.get_dummy_inputs(__SCREAMING_SNAKE_CASE      )
         a__    = sd_pipe(**__SCREAMING_SNAKE_CASE      ).images
         a__    = image[0, -3:, -3:, -1]
         assert image.shape == (1, 6_4, 6_4, 3)

         a__    = np.array([0.47_73, 0.54_17, 0.47_23, 0.49_25, 0.56_31, 0.47_52, 0.52_40, 0.49_35, 0.50_23]      )

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



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







         """simple docstring"""



         a__    = 'cpu'  # ensure determinism for the device-dependent torch.Generator
         a__    = self.get_dummy_components()
         torch.manual_seed(0      )
         a__    = DDPMParallelScheduler()
         torch.manual_seed(0      )
         a__    = StableDiffusionParadigmsPipeline(**__SCREAMING_SNAKE_CASE      )
         a__    = sd_pipe.to(__SCREAMING_SNAKE_CASE      )
         sd_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE      )

         a__    = self.get_dummy_inputs(__SCREAMING_SNAKE_CASE      )
         a__    = sd_pipe(**__SCREAMING_SNAKE_CASE      ).images
         a__    = image[0, -3:, -3:, -1]
         assert image.shape == (1, 6_4, 6_4, 3)

         a__    = np.array([0.35_73, 0.44_20, 0.49_60, 0.47_99, 0.37_96, 0.38_79, 0.48_19, 0.43_65, 0.44_68]      )

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



     def  lowercase__     (       self      )		->    List[str]:







         """simple docstring"""



         super().test_inference_batch_consistent(batch_sizes=[1, 2]      )



     def  lowercase__     (       self      )		->    str:







         """simple docstring"""



         super().test_inference_batch_single_identical(batch_size=2    ,							expected_max_diff=3e-3      )




     def  lowercase__     (       self      )		->    Optional[int]:







         """simple docstring"""



         a__    = 'cpu'  # ensure determinism for the device-dependent torch.Generator
         a__    = self.get_dummy_components()
         a__    = StableDiffusionParadigmsPipeline(**__SCREAMING_SNAKE_CASE      )
         a__    = sd_pipe.to(__SCREAMING_SNAKE_CASE      )
         sd_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE      )

         a__    = self.get_dummy_inputs(__SCREAMING_SNAKE_CASE      )
         a__    = 'french fries'
         a__    = sd_pipe(**__SCREAMING_SNAKE_CASE    ,							negative_prompt=__SCREAMING_SNAKE_CASE      )
         a__    = output.images
         a__    = image[0, -3:, -3:, -1]

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

         a__    = np.array([0.47_71, 0.54_20, 0.46_83, 0.49_18, 0.56_36, 0.47_25, 0.52_30, 0.49_23, 0.50_15]      )

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







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



     def  lowercase__     (       self      )		->    Tuple:







         """simple docstring"""



         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()



     def  lowercase__     (       self    ,							__SCREAMING_SNAKE_CASE=0      )		->    Any:







         """simple docstring"""



         a__    = torch.Generator(device=__SCREAMING_SNAKE_CASE      ).manual_seed(__SCREAMING_SNAKE_CASE      )
         a__    = {
             'prompt': 'a photograph of an astronaut riding a horse',
             'generator': generator,
             'num_inference_steps': 1_0,
             'guidance_scale': 7.5,
             'output_type': 'numpy',
             'parallel': 3,
             'debug': True,
         }
         return inputs




     def  lowercase__     (       self      )		->    Dict:







         """simple docstring"""



         a__    = 'stabilityai/stable-diffusion-2-base'
         a__    = DDIMParallelScheduler.from_pretrained(__SCREAMING_SNAKE_CASE    ,							subfolder='scheduler'      )
         a__    = StableDiffusionParadigmsPipeline.from_pretrained(__SCREAMING_SNAKE_CASE    ,							scheduler=__SCREAMING_SNAKE_CASE    ,							safety_checker=__SCREAMING_SNAKE_CASE      )
         pipe.to(__SCREAMING_SNAKE_CASE      )
         pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE      )
         pipe.enable_attention_slicing()

         a__    = self.get_inputs()
         a__    = pipe(**__SCREAMING_SNAKE_CASE      ).images
         a__    = image[0, -3:, -3:, -1].flatten()

         assert image.shape == (1, 5_1_2, 5_1_2, 3)

         a__    = np.array([0.96_22, 0.96_02, 0.97_48, 0.95_91, 0.96_30, 0.96_91, 0.96_61, 0.96_31, 0.97_41]      )

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








"""simple docstring"""




import os
import pickle
import unittest
from collections import UserDict, namedtuple

import torch

from accelerate.test_utils.testing import require_cuda, require_torch_min_version
from accelerate.test_utils.training import RegressionModel
from accelerate.utils import (
    convert_outputs_to_fpaa,
    extract_model_from_parallel,
    find_device,
    listify,
    patch_environment,
    recursively_apply,
    send_to_device,
)


a		:					Tuple    	=					namedtuple('ExampleNamedTuple', 'a b c')







class 						lowercase(unittest.TestCase					):



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







         """simple docstring"""



         a__    = torch.randn(5    ,							2      )
         a__    = torch.device('cuda'      ) if torch.cuda.is_available() else torch.device('cpu'      )

         a__    = send_to_device(__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE      )
         self.assertTrue(torch.equal(resulta.cpu()    ,							__SCREAMING_SNAKE_CASE      )      )

         a__    = send_to_device((tensor, [tensor, tensor], 1)    ,							__SCREAMING_SNAKE_CASE      )
         self.assertIsInstance(__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE      )
         self.assertTrue(torch.equal(resulta[0].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertIsInstance(resulta[1]    ,							__SCREAMING_SNAKE_CASE      )
         self.assertTrue(torch.equal(resulta[1][0].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertTrue(torch.equal(resulta[1][1].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertEqual(resulta[2]    ,							1      )

         a__    = send_to_device({'a': tensor, 'b': [tensor, tensor], 'c': 1}    ,							__SCREAMING_SNAKE_CASE      )
         self.assertIsInstance(__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE      )
         self.assertTrue(torch.equal(resulta['a'].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertIsInstance(resulta['b']    ,							__SCREAMING_SNAKE_CASE      )
         self.assertTrue(torch.equal(resulta['b'][0].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertTrue(torch.equal(resulta['b'][1].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertEqual(resulta['c']    ,							1      )

         a__    = send_to_device(ExampleNamedTuple(a=__SCREAMING_SNAKE_CASE    ,							b=[tensor, tensor]    ,							c=1      )    ,							__SCREAMING_SNAKE_CASE      )
         self.assertIsInstance(__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE      )
         self.assertTrue(torch.equal(resulta.a.cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertIsInstance(resulta.b    ,							__SCREAMING_SNAKE_CASE      )
         self.assertTrue(torch.equal(resulta.b[0].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertTrue(torch.equal(resulta.b[1].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertEqual(resulta.c    ,							1      )

         a__    = send_to_device(UserDict({'a': tensor, 'b': [tensor, tensor], 'c': 1}      )    ,							__SCREAMING_SNAKE_CASE      )
         self.assertIsInstance(__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE      )
         self.assertTrue(torch.equal(resulta['a'].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertIsInstance(resulta['b']    ,							__SCREAMING_SNAKE_CASE      )
         self.assertTrue(torch.equal(resulta['b'][0].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertTrue(torch.equal(resulta['b'][1].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertEqual(resulta['c']    ,							1      )



     def  lowercase__     (       self      )		->    Dict:







         """simple docstring"""



         with self.assertRaises(__SCREAMING_SNAKE_CASE      ) as cm:
             a__    = recursively_apply(torch.tensor    ,							(torch.tensor(1      ), 1)    ,							error_on_other_type=__SCREAMING_SNAKE_CASE      )
         self.assertEqual(
             str(cm.exception      )    ,							'Unsupported types (<class \'int\'>) passed to `tensor`. Only nested list/tuple/dicts of objects that are valid for `is_torch_tensor` should be passed.'    ,							)



     def  lowercase__     (       self      )		->    Any:







         """simple docstring"""



         a__    = torch.tensor([1, 2, 3, 4, 5]      )
         self.assertEqual(listify(__SCREAMING_SNAKE_CASE      )    ,							[1, 2, 3, 4, 5]      )

         a__    = torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]]      )
         self.assertEqual(listify(__SCREAMING_SNAKE_CASE      )    ,							[[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]]      )

         a__    = torch.tensor([[[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]], [[1_1, 1_2, 1_3, 1_4, 1_5], [1_6, 1_7, 1_8, 1_9, 2_0]]]      )
         self.assertEqual(
             listify(__SCREAMING_SNAKE_CASE      )    ,							[[[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]], [[1_1, 1_2, 1_3, 1_4, 1_5], [1_6, 1_7, 1_8, 1_9, 2_0]]]      )



     def  lowercase__     (       self      )		->    str:







         """simple docstring"""



         with patch_environment(aa=1    ,							BB=2      ):
             self.assertEqual(os.environ.get('AA'      )    ,							'1'      )
             self.assertEqual(os.environ.get('BB'      )    ,							'2'      )

         self.assertNotIn('AA'    ,							os.environ      )
         self.assertNotIn('BB'    ,							os.environ      )



     def  lowercase__     (       self      )		->    Tuple:







         """simple docstring"""



         a__    = RegressionModel()
         a__    = model.forward
         a__    = convert_outputs_to_fpaa(model.forward      )
         a__    = extract_model_from_parallel(__SCREAMING_SNAKE_CASE    ,							keep_fpaa_wrapper=__SCREAMING_SNAKE_CASE      )
         a__    = pickle.dumps(__SCREAMING_SNAKE_CASE      )



     @require_cuda
     def  lowercase__     (       self      )		->    Union[str, Any]:







         """simple docstring"""



         a__    = RegressionModel()
         a__    = model.forward
         a__    = torch.cuda.amp.autocast(dtype=torch.floataa      )(model.forward      )
         a__    = convert_outputs_to_fpaa(model.forward      )
         a__    = extract_model_from_parallel(__SCREAMING_SNAKE_CASE    ,							keep_fpaa_wrapper=__SCREAMING_SNAKE_CASE      )
         a__    = pickle.dumps(__SCREAMING_SNAKE_CASE      )



     @require_cuda
     @require_torch_min_version(version='2.0'      )
     def  lowercase__     (       self      )		->    Optional[int]:







         """simple docstring"""



         a__    = RegressionModel()
         a__    = model.forward
         a__    = torch.cuda.amp.autocast(dtype=torch.floataa      )(model.forward      )
         a__    = convert_outputs_to_fpaa(model.forward      )
         a__    = torch.compile(model.forward    ,							backend='inductor'      )
         a__    = torch.randn(4    ,							1_0      ).cuda()
         a__    = model(__SCREAMING_SNAKE_CASE      )



     def  lowercase__     (       self      )		->    str:







         """simple docstring"""



         a__    = RegressionModel()
         # could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
         a__    = torch.nn.parallel.DataParallel(__SCREAMING_SNAKE_CASE      )
         a__    = extract_model_from_parallel(__SCREAMING_SNAKE_CASE      )

         self.assertEqual(__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE      )



     @require_torch_min_version(version='2.0'      )
     def  lowercase__     (       self      )		->    str:







         """simple docstring"""



         a__    = RegressionModel()
         a__    = torch.compile(__SCREAMING_SNAKE_CASE      )

         # could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
         a__    = torch.nn.parallel.DataParallel(__SCREAMING_SNAKE_CASE      )
         a__    = torch.compile(__SCREAMING_SNAKE_CASE      )
         a__    = extract_model_from_parallel(__SCREAMING_SNAKE_CASE      )

         self.assertEqual(compiled_model._orig_mod    ,							compiled_model_unwrapped._orig_mod      )




     def  lowercase__     (       self      )		->    Optional[Any]:







         """simple docstring"""



         self.assertEqual(find_device([1, 'a', torch.tensor([1, 2, 3]      )]      )    ,							torch.device('cpu'      )      )
         self.assertEqual(find_device({'a': 1, 'b': torch.tensor([1, 2, 3]      )}      )    ,							torch.device('cpu'      )      )
         self.assertIsNone(find_device([1, 'a']      )      )


