





import gc
import unittest

import torch
from parameterized import parameterized

from diffusers import AutoencoderKL
from diffusers.utils import floats_tensor, load_hf_numpy, require_torch_gpu, slow, torch_all_close, torch_device
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import enable_full_determinism

from .test_modeling_common import ModelTesterMixin, UNetTesterMixin


enable_full_determinism()

class    _UpperCAmelCase							(		lowercase_			,       lowercase_			,       unittest.TestCase     ):
      a						       =							AutoencoderKL
      a						       =							'''sample'''
      a						       =							1e-2
      @property
      def 		_lowerCamelCase      (		self       ):
         A_     :				Optional[int]			      =	4
         A_     :				Optional[int]			      =	3
         A_     :				Tuple			      =	(32, 32)

         A_     :				Tuple			      =	floats_tensor((batch_size, num_channels) + sizes       ).to(a__       )

         return {"sample": image}
      @property
      def 		_lowerCamelCase      (		self       ):
         return (3, 32, 32)
      @property
      def 		_lowerCamelCase      (		self       ):
         return (3, 32, 32)
      def 		_lowerCamelCase      (		self       ):
         A_     :				str			      =	{
             """block_out_channels""": [32, 64],
             """in_channels""": 3,
             """out_channels""": 3,
             """down_block_types""": ["""DownEncoderBlock2D""", """DownEncoderBlock2D"""],
             """up_block_types""": ["""UpDecoderBlock2D""", """UpDecoderBlock2D"""],
             """latent_channels""": 4,
         }
         A_     :				Optional[Any]			      =	self.dummy_input
         return init_dict, inputs_dict
      def 		_lowerCamelCase      (		self       ):
         pass
      def 		_lowerCamelCase      (		self       ):
         pass
      @unittest.skipIf(torch_device == """mps"""      ,       """Gradient checkpointing skipped on MPS"""       )
      def 		_lowerCamelCase      (		self       ):
         # enable deterministic behavior for gradient checkpointing
         A_      ,				A_     :				int			      =	self.prepare_init_args_and_inputs_for_common()
         A_     :				Dict			      =	self.model_class(**a__       )
         model.to(a__       )

         assert not model.is_gradient_checkpointing and model.training

         A_     :				Any			      =	model(**a__       ).sample
         # run the backwards pass on the model. For backwards pass, for simplicity purpose,
         # we won't calculate the loss and rather backprop on out.sum()
         model.zero_grad()

         A_     :				List[str]			      =	torch.randn_like(a__       )
         A_     :				int			      =	(out - labels).mean()
         loss.backward()

         # re-instantiate the model now enabling gradient checkpointing
         A_     :				int			      =	self.model_class(**a__       )
         # clone model
         model_a.load_state_dict(model.state_dict()       )
         model_a.to(a__       )
         model_a.enable_gradient_checkpointing()

         assert model_a.is_gradient_checkpointing and model_a.training

         A_     :				Optional[int]			      =	model_a(**a__       ).sample
         # run the backwards pass on the model. For backwards pass, for simplicity purpose,
         # we won't calculate the loss and rather backprop on out.sum()
         model_a.zero_grad()
         A_     :				Union[str, Any]			      =	(out_a - labels).mean()
         loss_a.backward()

         # compare the output and parameters gradients
         self.assertTrue((loss - loss_a).abs() < 1E-5       )
         A_     :				Optional[int]			      =	dict(model.named_parameters()       )
         A_     :				int			      =	dict(model_a.named_parameters()       )
         for name, param in named_params.items():
            self.assertTrue(torch_all_close(param.grad.data      ,       named_params_a[name].grad.data      ,       atol=5E-5       )       )
      def 		_lowerCamelCase      (		self       ):
         A_      ,				A_     :				Any			      =	AutoencoderKL.from_pretrained("""fusing/autoencoder-kl-dummy"""      ,       output_loading_info=a__       )
         self.assertIsNotNone(a__       )
         self.assertEqual(len(loading_info["""missing_keys"""]       )      ,       0       )

         model.to(a__       )
         A_     :				Tuple			      =	model(**self.dummy_input       )

         assert image is not None, "Make sure output is not None"






      def 		_lowerCamelCase      (		self       ):
         A_     :				str			      =	AutoencoderKL.from_pretrained("""fusing/autoencoder-kl-dummy"""       )
         A_     :				List[Any]			      =	model.to(a__       )
         model.eval()

         if torch_device == "mps":
            A_     :				List[Any]			      =	torch.manual_seed(0       )
         else:
            A_     :				Optional[int]			      =	torch.Generator(device=a__       ).manual_seed(0       )

         A_     :				str			      =	torch.randn(
             1      ,       model.config.in_channels      ,       model.config.sample_size      ,       model.config.sample_size      ,       generator=torch.manual_seed(0       )      ,       )
         A_     :				str			      =	image.to(a__       )
         with torch.no_grad():
            A_     :				Optional[int]			      =	model(a__      ,       sample_posterior=a__      ,       generator=a__       ).sample

         A_     :				Any			      =	output[0, -1, -3:, -3:].flatten().cpu()

         # Since the VAE Gaussian prior's generator is seeded on the appropriate device,
         # the expected output slices are not the same for CPU and GPU.
         if torch_device == "mps":
            A_     :				List[Any]			      =	torch.tensor(
                [
                    -4.0078E-01,
                    -3.8323E-04,
                    -1.2681E-01,
                    -1.1462E-01,
                    2.0095E-01,
                    1.0893E-01,
                    -8.8247E-02,
                    -3.0361E-01,
                    -9.8644E-03,
                ]       )
         elif torch_device == "cpu":
            A_     :				List[str]			      =	torch.tensor(
                [-0.1352, 0.0878, 0.0419, -0.0818, -0.1069, 0.0688, -0.1458, -0.4446, -0.0026]       )
         else:
            A_     :				List[str]			      =	torch.tensor(
                [-0.2421, 0.4642, 0.2507, -0.0438, 0.0682, 0.3160, -0.2018, -0.0727, 0.2485]       )

         self.assertTrue(torch_all_close(a__      ,       a__      ,       rtol=1E-2       )       )






@slow
class    _UpperCAmelCase							(		unittest.TestCase     ):
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__       ):
         return F"""gaussian_noise_s={seed}_shape={'_'.join([str(a__       ) for s in shape]       )}.npy"""
      def 		_lowerCamelCase      (		self       ):
         # clean up the VRAM after each test
         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()
      def 		_lowerCamelCase      (		self      ,       a__=0      ,       a__=(4, 3, 512, 512)      ,       a__=False       ):
         A_     :				Union[str, Any]			      =	torch.floataa if fpaa else torch.floataa
         A_     :				Optional[Any]			      =	torch.from_numpy(load_hf_numpy(self.get_file_format(a__      ,       a__       )       )       ).to(a__       ).to(a__       )
         return image
      def 		_lowerCamelCase      (		self      ,       a__="CompVis/stable-diffusion-v1-4"      ,       a__=False       ):
         A_     :				List[str]			      =	"""fp16""" if fpaa else None
         A_     :				Any			      =	torch.floataa if fpaa else torch.floataa

         A_     :				Any			      =	AutoencoderKL.from_pretrained(
             a__      ,       subfolder="""vae"""      ,       torch_dtype=a__      ,       revision=a__      ,       )
         model.to(a__       ).eval()

         return model
      def 		_lowerCamelCase      (		self      ,       a__=0       ):
         if torch_device == "mps":
            return torch.manual_seed(a__       )
         return torch.Generator(device=a__       ).manual_seed(a__       )
      @parameterized.expand(
          [
              # fmt: off
              [33, [-0.1603, 0.9878, -0.0495, -0.0790, -0.2709, 0.8375, -0.2060, -0.0824], [-0.2395, 0.0098, 0.0102, -0.0709, -0.2840, -0.0274, -0.0718, -0.1824]],
              [47, [-0.2376, 0.1168, 0.1332, -0.4840, -0.2508, -0.0791, -0.0493, -0.4089], [0.0350, 0.0847, 0.0467, 0.0344, -0.0842, -0.0547, -0.0633, -0.1131]],
              # fmt: on
          ]       )
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__      ,       a__       ):
         A_     :				str			      =	self.get_sd_vae_model()
         A_     :				Any			      =	self.get_sd_image(a__       )
         A_     :				Dict			      =	self.get_generator(a__       )

         with torch.no_grad():
            A_     :				List[str]			      =	model(a__      ,       generator=a__      ,       sample_posterior=a__       ).sample

         assert sample.shape == image.shape

         A_     :				List[Any]			      =	sample[-1, -2:, -2:, :2].flatten().float().cpu()
         A_     :				Optional[int]			      =	torch.tensor(expected_slice_mps if torch_device == """mps""" else expected_slice       )

         assert torch_all_close(a__      ,       a__      ,       atol=3E-3       )
      @parameterized.expand(
          [
              # fmt: off
              [33, [-0.0513, 0.0289, 1.3799, 0.2166, -0.2573, -0.0871, 0.5103, -0.0999]],
              [47, [-0.4128, -0.1320, -0.3704, 0.1965, -0.4116, -0.2332, -0.3340, 0.2247]],
              # fmt: on
          ]       )
      @require_torch_gpu
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__       ):
         A_     :				Optional[Any]			      =	self.get_sd_vae_model(fpaa=a__       )
         A_     :				Optional[Any]			      =	self.get_sd_image(a__      ,       fpaa=a__       )
         A_     :				Tuple			      =	self.get_generator(a__       )

         with torch.no_grad():
            A_     :				List[Any]			      =	model(a__      ,       generator=a__      ,       sample_posterior=a__       ).sample

         assert sample.shape == image.shape

         A_     :				Tuple			      =	sample[-1, -2:, :2, -2:].flatten().float().cpu()
         A_     :				Dict			      =	torch.tensor(a__       )

         assert torch_all_close(a__      ,       a__      ,       atol=1E-2       )
      @parameterized.expand(
          [
              # fmt: off
              [33, [-0.1609, 0.9866, -0.0487, -0.0777, -0.2716, 0.8368, -0.2055, -0.0814], [-0.2395, 0.0098, 0.0102, -0.0709, -0.2840, -0.0274, -0.0718, -0.1824]],
              [47, [-0.2377, 0.1147, 0.1333, -0.4841, -0.2506, -0.0805, -0.0491, -0.4085], [0.0350, 0.0847, 0.0467, 0.0344, -0.0842, -0.0547, -0.0633, -0.1131]],
              # fmt: on
          ]       )
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__      ,       a__       ):
         A_     :				Tuple			      =	self.get_sd_vae_model()
         A_     :				int			      =	self.get_sd_image(a__       )

         with torch.no_grad():
            A_     :				Any			      =	model(a__       ).sample

         assert sample.shape == image.shape

         A_     :				int			      =	sample[-1, -2:, -2:, :2].flatten().float().cpu()
         A_     :				List[str]			      =	torch.tensor(expected_slice_mps if torch_device == """mps""" else expected_slice       )

         assert torch_all_close(a__      ,       a__      ,       atol=3E-3       )
      @parameterized.expand(
          [
              # fmt: off
              [13, [-0.2051, -0.1803, -0.2311, -0.2114, -0.3292, -0.3574, -0.2953, -0.3323]],
              [37, [-0.2632, -0.2625, -0.2199, -0.2741, -0.4539, -0.4990, -0.3720, -0.4925]],
              # fmt: on
          ]       )
      @require_torch_gpu
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__       ):
         A_     :				Optional[int]			      =	self.get_sd_vae_model()
         A_     :				Union[str, Any]			      =	self.get_sd_image(a__      ,       shape=(3, 4, 64, 64)       )

         with torch.no_grad():
            A_     :				List[str]			      =	model.decode(a__       ).sample

         assert list(sample.shape       ) == [3, 3, 512, 512]

         A_     :				int			      =	sample[-1, -2:, :2, -2:].flatten().cpu()
         A_     :				int			      =	torch.tensor(a__       )

         assert torch_all_close(a__      ,       a__      ,       atol=1E-3       )
      @parameterized.expand(
          [
              # fmt: off
              [27, [-0.0369, 0.0207, -0.0776, -0.0682, -0.1747, -0.1930, -0.1465, -0.2039]],
              [16, [-0.1628, -0.2134, -0.2747, -0.2642, -0.3774, -0.4404, -0.3687, -0.4277]],
              # fmt: on
          ]       )
      @require_torch_gpu
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__       ):
         A_     :				List[Any]			      =	self.get_sd_vae_model(fpaa=a__       )
         A_     :				str			      =	self.get_sd_image(a__      ,       shape=(3, 4, 64, 64)      ,       fpaa=a__       )

         with torch.no_grad():
            A_     :				Tuple			      =	model.decode(a__       ).sample

         assert list(sample.shape       ) == [3, 3, 512, 512]

         A_     :				Union[str, Any]			      =	sample[-1, -2:, :2, -2:].flatten().float().cpu()
         A_     :				Union[str, Any]			      =	torch.tensor(a__       )

         assert torch_all_close(a__      ,       a__      ,       atol=5E-3       )
      @parameterized.expand([(13,), (16,), (27,)]       )
      @require_torch_gpu
      @unittest.skipIf(not is_xformers_available()      ,       reason="""xformers is not required when using PyTorch 2.0."""       )
      def 		_lowerCamelCase      (		self      ,       a__       ):
         A_     :				Tuple			      =	self.get_sd_vae_model(fpaa=a__       )
         A_     :				Optional[int]			      =	self.get_sd_image(a__      ,       shape=(3, 4, 64, 64)      ,       fpaa=a__       )

         with torch.no_grad():
            A_     :				str			      =	model.decode(a__       ).sample

         model.enable_xformers_memory_efficient_attention()
         with torch.no_grad():
            A_     :				Any			      =	model.decode(a__       ).sample

         assert list(sample.shape       ) == [3, 3, 512, 512]

         assert torch_all_close(a__      ,       a__      ,       atol=1E-1       )
      @parameterized.expand([(13,), (16,), (37,)]       )
      @require_torch_gpu
      @unittest.skipIf(not is_xformers_available()      ,       reason="""xformers is not required when using PyTorch 2.0."""       )
      def 		_lowerCamelCase      (		self      ,       a__       ):
         A_     :				Tuple			      =	self.get_sd_vae_model()
         A_     :				Optional[Any]			      =	self.get_sd_image(a__      ,       shape=(3, 4, 64, 64)       )

         with torch.no_grad():
            A_     :				int			      =	model.decode(a__       ).sample

         model.enable_xformers_memory_efficient_attention()
         with torch.no_grad():
            A_     :				Tuple			      =	model.decode(a__       ).sample

         assert list(sample.shape       ) == [3, 3, 512, 512]

         assert torch_all_close(a__      ,       a__      ,       atol=1E-2       )






      @parameterized.expand(
          [
              # fmt: off
              [33, [-0.3001, 0.0918, -2.6984, -3.9720, -3.2099, -5.0353, 1.7338, -0.2065, 3.4267]],
              [47, [-1.5030, -4.3871, -6.0355, -9.1157, -1.6661, -2.7853, 2.1607, -5.0823, 2.5633]],
              # fmt: on
          ]       )
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__       ):
         A_     :				int			      =	self.get_sd_vae_model()
         A_     :				Tuple			      =	self.get_sd_image(a__       )
         A_     :				Dict			      =	self.get_generator(a__       )

         with torch.no_grad():
            A_     :				Tuple			      =	model.encode(a__       ).latent_dist
            A_     :				List[Any]			      =	dist.sample(generator=a__       )

         assert list(sample.shape       ) == [image.shape[0], 4] + [i // 8 for i in image.shape[2:]]

         A_     :				Optional[int]			      =	sample[0, -1, -3:, -3:].flatten().cpu()
         A_     :				List[Any]			      =	torch.tensor(a__       )

         A_     :				Optional[Any]			      =	3E-3 if torch_device != """mps""" else 1E-2
         assert torch_all_close(a__      ,       a__      ,       atol=a__       )









import baseaa
import io
import json
import os
from copy import deepcopy

from ..optimizer import AcceleratedOptimizer
from ..scheduler import AcceleratedScheduler

class    _UpperCAmelCase							:
      def __init__(		self      ,       a__       ):
         if isinstance(a__      ,       a__       ):
            # Don't modify user's data should they want to reuse it (e.g. in tests), because once we
            # modified it, it will not be accepted here again, since `auto` values would have been overridden
            A_     :				Optional[Any]			      =	deepcopy(a__       )
         elif os.path.exists(a__       ):
            with io.open(a__      ,       """r"""      ,       encoding="""utf-8"""       ) as f:
               A_     :				str			      =	json.load(a__       )
         else:
            try:
               A_     :				Dict			      =	baseaa.urlsafe_baadecode(a__       ).decode("""utf-8"""       )
               A_     :				List[Any]			      =	json.loads(a__       )
            except (UnicodeDecodeError, AttributeError, ValueError):
               raise ValueError(
                   F"""Expected a string path to an existing deepspeed config, or a dictionary, or a base64 encoded string. Received: {config_file_or_dict}"""       )

         A_     :				Any			      =	config

         self.set_stage_and_offload()
      def 		_lowerCamelCase      (		self       ):
         # zero stage - this is done as early as possible, before model is created, to allow
         # ``is_deepspeed_zero3_enabled`` query and getting to the early deepspeed config object
         # during ``zero.Init()`` which needs to know the dtype, and some other hparams.
         A_     :				List[str]			      =	self.get_value("""zero_optimization.stage"""      ,       -1       )

         # offload
         A_     :				Any			      =	False
         if self.is_zeroa() or self.is_zeroa():
            A_     :				Optional[int]			      =	set(["""cpu""", """nvme"""]       )
            A_     :				Dict			      =	set(
                [
                    self.get_value("""zero_optimization.offload_optimizer.device"""       ),
                    self.get_value("""zero_optimization.offload_param.device"""       ),
                ]       )
            if len(offload_devices & offload_devices_valid       ) > 0:
               A_     :				Tuple			      =	True
      def 		_lowerCamelCase      (		self      ,       a__       ):
         A_     :				List[Any]			      =	self.config

         # find the config node of interest if it exists
         A_     :				Optional[Any]			      =	ds_key_long.split("""."""       )
         A_     :				Union[str, Any]			      =	nodes.pop()
         for node in nodes:
            A_     :				List[str]			      =	config.get(a__       )
            if config is None:
               return None, ds_key

         return config, ds_key
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__=None       ):
         A_      ,				A_     :				Union[str, Any]			      =	self.find_config_node(a__       )
         if config is None:
            return default
         return config.get(a__      ,       a__       )
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__=False       ):
         A_     :				Union[str, Any]			      =	self.config

         # find the config node of interest if it exists
         A_     :				str			      =	ds_key_long.split("""."""       )
         for node in nodes:
            A_     :				int			      =	config
            A_     :				int			      =	config.get(a__       )
            if config is None:
               if must_exist:
                  raise ValueError(F"""Can't find {ds_key_long} entry in the config: {self.config}"""       )
               else:
                  return

        # if found remove it
         if parent_config is not None:
            parent_config.pop(a__       )
      def 		_lowerCamelCase      (		self      ,       a__       ):
         A_     :				Optional[Any]			      =	self.get_value(a__       )
         return False if value is None else bool(a__       )
      def 		_lowerCamelCase      (		self      ,       a__       ):
         A_     :				Optional[Any]			      =	self.get_value(a__       )
         return False if value is None else not bool(a__       )
      def 		_lowerCamelCase      (		self       ):
         return self._stage == 2
      def 		_lowerCamelCase      (		self       ):
         return self._stage == 3






      def 		_lowerCamelCase      (		self       ):
         return self._offload






class    _UpperCAmelCase							:
      def __init__(		self      ,       a__       ):
         A_     :				Any			      =	engine






      def 		_lowerCamelCase      (		self      ,       a__      ,       **a__       ):
         # runs backpropagation and handles mixed precision
         self.engine.backward(a__      ,       **a__       )

         # Deepspeed's `engine.step` performs the following operations:
         # - gradient accumulation check
         # - gradient clipping
         # - optimizer step
         # - zero grad
         # - checking overflow
         # - lr_scheduler step (only if engine.lr_scheduler is not None)
         self.engine.step()
         # and this plugin overrides the above calls with no-ops when Accelerate runs under
         # Deepspeed, but allows normal functionality for non-Deepspeed cases thus enabling a simple
         # training loop that works transparently under many training regimes.






class    _UpperCAmelCase							(		_lowerCamelCase     ):
      def __init__(		self      ,       a__       ):
         super().__init__(a__      ,       device_placement=a__      ,       scaler=a__       )
         A_     :				Dict			      =	hasattr(self.optimizer      ,       """overflow"""       )
      def 		_lowerCamelCase      (		self      ,       a__=None       ):
         pass  # `accelerator.backward(loss)` is doing that automatically. Therefore, its implementation is not needed
      def 		_lowerCamelCase      (		self       ):
         pass  # `accelerator.backward(loss)` is doing that automatically. Therefore, its implementation is not needed






      @property
      def 		_lowerCamelCase      (		self       ):
         if self.__has_overflow__:
            return self.optimizer.overflow
         return False






class    _UpperCAmelCase							(		_lowerCamelCase     ):
      def __init__(		self      ,       a__      ,       a__       ):
         super().__init__(a__      ,       a__       )






      def 		_lowerCamelCase      (		self       ):
         pass  # `accelerator.backward(loss)` is doing that automatically. Therefore, its implementation is not needed






class    _UpperCAmelCase							:
      def __init__(		self      ,       a__      ,       a__=0.001      ,       a__=0      ,       **a__       ):
         A_     :				List[str]			      =	params
         A_     :				Any			      =	lr
         A_     :				int			      =	weight_decay
         A_     :				Optional[int]			      =	kwargs






class    _UpperCAmelCase							:
      def __init__(		self      ,       a__      ,       a__=None      ,       a__=0      ,       **a__       ):
         A_     :				Union[str, Any]			      =	optimizer
         A_     :				int			      =	total_num_steps
         A_     :				Any			      =	warmup_num_steps
         A_     :				int			      =	kwargs






