





'''simple docstring'''







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       _lowercase    (							SCREAMING_SNAKE_CASE_       ):
     _UpperCAmelCase					    =       0
     _UpperCAmelCase					    =       False
     _UpperCAmelCase					    =       3.0




class       _lowercase    (							unittest.TestCase       ):







     def      UpperCamelCase      (			self     )			->						Tuple:
      self.assertDictEqual(MockClass().to_kwargs()     , {}     )
      self.assertDictEqual(MockClass(a=2     ).to_kwargs()     , {'''a''': 2}     )
      self.assertDictEqual(MockClass(a=2     , b=snake_case__     ).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      UpperCamelCase      (			self     )			->						Optional[int]:
      snake_case					=      GradScalerKwargs(init_scale=10_24     , growth_factor=2     )
      AcceleratorState._reset_state()
      snake_case					=      Accelerator(mixed_precision='''fp16'''     , kwargs_handlers=[scaler_handler]     )
      print(accelerator.use_fpaa     )
      snake_case					=      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     , 20_00     )
      self.assertEqual(scaler._enabled     , snake_case__     )







     @require_multi_gpu
     def      UpperCamelCase      (			self     )			->						int:
      snake_case					=      ['torchrun', F"""--nproc_per_node={torch.cuda.device_count()}""", inspect.getfile(self.__class__     )]
      execute_subprocess_async(snake_case__     , 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 // (1_024 * 1_024)
   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)





'''simple docstring'''







import gc
import tempfile
import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionTextToImagePipeline
from diffusers.utils.testing_utils import nightly, require_torch_gpu, torch_device


lowerCAmelCase     :  str											=    False
class 						UpperCamelCase__       (			unittest.TestCase						):



     """simple docstring"""






     pass
@nightly
@require_torch_gpu
class 						UpperCamelCase__       (			unittest.TestCase						):



     """simple docstring"""






     def a						(			self ):

            '''simple docstring'''

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







     def a						(			self ):

            '''simple docstring'''

            _lowerCAmelCase     :				List[str]       =							VersatileDiffusionTextToImagePipeline.from_pretrained('shi-labs/versatile-diffusion' )
            # remove text_unet
            pipe.remove_unused_weights()
            pipe.to(snake_case__ )
            pipe.set_progress_bar_config(disable=snake_case__ )

            _lowerCAmelCase     :				Any       =							'A painting of a squirrel eating a burger '
            _lowerCAmelCase     :				List[Any]       =							torch.manual_seed(0 )
            _lowerCAmelCase     :				Any       =							pipe(
                prompt=snake_case__   ,						generator=snake_case__   ,						guidance_scale=7.5   ,						num_inference_steps=2   ,						output_type='numpy' ).images

            with tempfile.TemporaryDirectory() as tmpdirname:
                   pipe.save_pretrained(snake_case__ )
                   _lowerCAmelCase     :				Any       =							VersatileDiffusionTextToImagePipeline.from_pretrained(snake_case__ )
            pipe.to(snake_case__ )
            pipe.set_progress_bar_config(disable=snake_case__ )

            _lowerCAmelCase     :				Optional[int]       =							generator.manual_seed(0 )
            _lowerCAmelCase     :				Tuple       =							pipe(
                prompt=snake_case__   ,						generator=snake_case__   ,						guidance_scale=7.5   ,						num_inference_steps=2   ,						output_type='numpy' ).images

            assert np.abs(image - new_image ).sum() < 1E-5, "Models don't have the same forward pass"







     def a						(			self ):

            '''simple docstring'''

            _lowerCAmelCase     :				Union[str, Any]       =							VersatileDiffusionTextToImagePipeline.from_pretrained(
                'shi-labs/versatile-diffusion'   ,						torch_dtype=torch.floataa )
            pipe.to(snake_case__ )
            pipe.set_progress_bar_config(disable=snake_case__ )

            _lowerCAmelCase     :				Dict       =							'A painting of a squirrel eating a burger '
            _lowerCAmelCase     :				Any       =							torch.manual_seed(0 )
            _lowerCAmelCase     :				int       =							pipe(
                prompt=snake_case__   ,						generator=snake_case__   ,						guidance_scale=7.5   ,						num_inference_steps=50   ,						output_type='numpy' ).images

            _lowerCAmelCase     :				Optional[int]       =							image[0, 253:256, 253:256, -1]

            assert image.shape == (1, 512, 512, 3)
            _lowerCAmelCase     :				Dict       =							np.array([0.3367, 0.3169, 0.2656, 0.3870, 0.4790, 0.3796, 0.4009, 0.4878, 0.4778] )

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

