
'''simple docstring'''

import gc
import math
import unittest

import torch

from diffusers import UNetaDModel
from diffusers.utils import floats_tensor, logging, slow, torch_all_close, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from .test_modeling_common import ModelTesterMixin, UNetTesterMixin


__snake_case:  str								=			logging.get_logger(__name__)

enable_full_determinism()


class  _UpperCAmelCase ( lowerCAmelCase__	,lowerCAmelCase__	,unittest.TestCase    ):

     """simple docstring"""

     a_    	=					UNetaDModel
     a_    	=					"sample"


     @property
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	Union[str, Any]      =							4
       a_      :	Any      =							3
       a_      :	Tuple      =							(32, 32)

       a_      :	str      =							floats_tensor((batch_size, num_channels) + sizes ).to(lowerCAmelCase_ )
       a_      :	List[Any]      =							torch.tensor([10] ).to(lowerCAmelCase_ )

       return {"sample": noise, "timestep": time_step}


     @property
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       return (3, 32, 32)


     @property
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       return (3, 32, 32)






     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	Optional[int]      =							{
           """block_out_channels""": (32, 64),
           """down_block_types""": ("""DownBlock2D""", """AttnDownBlock2D"""),
           """up_block_types""": ("""AttnUpBlock2D""", """UpBlock2D"""),
           """attention_head_dim""": 3,
           """out_channels""": 3,
           """in_channels""": 3,
           """layers_per_block""": 2,
           """sample_size""": 32,
       }
       a_      :	Tuple      =							self.dummy_input
       return init_dict, inputs_dict







class  _UpperCAmelCase ( lowerCAmelCase__	,lowerCAmelCase__	,unittest.TestCase    ):

     """simple docstring"""

     a_    	=					UNetaDModel
     a_    	=					"sample"


     @property
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	List[str]      =							4
       a_      :	str      =							4
       a_      :	str      =							(32, 32)

       a_      :	List[str]      =							floats_tensor((batch_size, num_channels) + sizes ).to(lowerCAmelCase_ )
       a_      :	Dict      =							torch.tensor([10] ).to(lowerCAmelCase_ )

       return {"sample": noise, "timestep": time_step}


     @property
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       return (4, 32, 32)


     @property
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       return (4, 32, 32)


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	Optional[int]      =							{
           """sample_size""": 32,
           """in_channels""": 4,
           """out_channels""": 4,
           """layers_per_block""": 2,
           """block_out_channels""": (32, 64),
           """attention_head_dim""": 32,
           """down_block_types""": ("""DownBlock2D""", """DownBlock2D"""),
           """up_block_types""": ("""UpBlock2D""", """UpBlock2D"""),
       }
       a_      :	List[str]      =							self.dummy_input
       return init_dict, inputs_dict


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      , a_      :	List[str]      =							UNetaDModel.from_pretrained("""fusing/unet-ldm-dummy-update"""							,		output_loading_info=lowerCAmelCase_ )

       self.assertIsNotNone(lowerCAmelCase_ )
       self.assertEqual(len(loading_info["""missing_keys"""] )							,		0 )

       model.to(lowerCAmelCase_ )
       a_      :	Dict      =							model(**self.dummy_input ).sample

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


     @unittest.skipIf(torch_device != """cuda"""							,		"""This test is supposed to run on GPU""" )
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      , a_      :	Dict      =							UNetaDModel.from_pretrained("""fusing/unet-ldm-dummy-update"""							,		output_loading_info=lowerCAmelCase_ )
       model.to(lowerCAmelCase_ )
       a_      :	List[Any]      =							model(**self.dummy_input ).sample

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


     @unittest.skipIf(torch_device != """cuda"""							,		"""This test is supposed to run on GPU""" )
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      , a_      :	str      =							UNetaDModel.from_pretrained("""fusing/unet-ldm-dummy-update"""							,		output_loading_info=lowerCAmelCase_ )
       model_accelerate.to(lowerCAmelCase_ )
       model_accelerate.eval()

       a_      :	List[Any]      =							torch.randn(
           1							,		model_accelerate.config.in_channels							,		model_accelerate.config.sample_size							,		model_accelerate.config.sample_size							,		generator=torch.manual_seed(0 )							,		)
       a_      :	List[str]      =							noise.to(lowerCAmelCase_ )
       a_      :	Union[str, Any]      =							torch.tensor([10] * noise.shape[0] ).to(lowerCAmelCase_ )

       a_      :	List[Any]      =							model_accelerate(lowerCAmelCase_							,		lowerCAmelCase_ )["""sample"""]

       # two models don't need to stay in the device at the same time
       del model_accelerate
       torch.cuda.empty_cache()
       gc.collect()

       a_      , a_      :	int      =							UNetaDModel.from_pretrained(
           """fusing/unet-ldm-dummy-update"""							,		output_loading_info=lowerCAmelCase_							,		low_cpu_mem_usage=lowerCAmelCase_ )
       model_normal_load.to(lowerCAmelCase_ )
       model_normal_load.eval()
       a_      :	int      =							model_normal_load(lowerCAmelCase_							,		lowerCAmelCase_ )["""sample"""]

       assert torch_all_close(lowerCAmelCase_							,		lowerCAmelCase_							,		rtol=1E-3 )






     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	Dict      =							UNetaDModel.from_pretrained("""fusing/unet-ldm-dummy-update""" )
       model.eval()
       model.to(lowerCAmelCase_ )

       a_      :	List[Any]      =							torch.randn(
           1							,		model.config.in_channels							,		model.config.sample_size							,		model.config.sample_size							,		generator=torch.manual_seed(0 )							,		)
       a_      :	Optional[Any]      =							noise.to(lowerCAmelCase_ )
       a_      :	List[str]      =							torch.tensor([10] * noise.shape[0] ).to(lowerCAmelCase_ )

       with torch.no_grad():
         a_      :	Optional[int]      =							model(lowerCAmelCase_							,		lowerCAmelCase_ ).sample

       a_      :	Union[str, Any]      =							output[0, -1, -3:, -3:].flatten().cpu()
       # fmt: off
       a_      :	Dict      =							torch.tensor([-13.3258, -20.1100, -15.9873, -17.6617, -23.0596, -17.9419, -13.3675, -16.1889, -12.3800] )
       # fmt: on

       self.assertTrue(torch_all_close(lowerCAmelCase_							,		lowerCAmelCase_							,		rtol=1E-3 ) )







class  _UpperCAmelCase ( lowerCAmelCase__	,lowerCAmelCase__	,unittest.TestCase    ):

     """simple docstring"""

     a_    	=					UNetaDModel
     a_    	=					"sample"


     @property
     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_=(32, 32) ):

       '''simple docstring'''

       a_      :	Any      =							4
       a_      :	Union[str, Any]      =							3

       a_      :	Optional[Any]      =							floats_tensor((batch_size, num_channels) + sizes ).to(lowerCAmelCase_ )
       a_      :	Dict      =							torch.tensor(batch_size * [10] ).to(dtype=torch.intaa							,		device=lowerCAmelCase_ )

       return {"sample": noise, "timestep": time_step}


     @property
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       return (3, 32, 32)


     @property
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       return (3, 32, 32)


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	Optional[Any]      =							{
           """block_out_channels""": [32, 64, 64, 64],
           """in_channels""": 3,
           """layers_per_block""": 1,
           """out_channels""": 3,
           """time_embedding_type""": """fourier""",
           """norm_eps""": 1E-6,
           """mid_block_scale_factor""": math.sqrt(2.0 ),
           """norm_num_groups""": None,
           """down_block_types""": [
               """SkipDownBlock2D""",
               """AttnSkipDownBlock2D""",
               """SkipDownBlock2D""",
               """SkipDownBlock2D""",
           ],
           """up_block_types""": [
               """SkipUpBlock2D""",
               """SkipUpBlock2D""",
               """AttnSkipUpBlock2D""",
               """SkipUpBlock2D""",
           ],
       }
       a_      :	Optional[Any]      =							self.dummy_input
       return init_dict, inputs_dict


     @slow
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      , a_      :	Optional[Any]      =							UNetaDModel.from_pretrained("""google/ncsnpp-celebahq-256"""							,		output_loading_info=lowerCAmelCase_ )
       self.assertIsNotNone(lowerCAmelCase_ )
       self.assertEqual(len(loading_info["""missing_keys"""] )							,		0 )

       model.to(lowerCAmelCase_ )
       a_      :	List[Any]      =							self.dummy_input
       a_      :	Tuple      =							floats_tensor((4, 3) + (2_56, 2_56) ).to(lowerCAmelCase_ )
       a_      :	str      =							noise
       a_      :	Dict      =							model(**lowerCAmelCase_ )

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


     @slow
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	List[str]      =							UNetaDModel.from_pretrained("""google/ncsnpp-celebahq-256""" )
       model.to(lowerCAmelCase_ )

       a_      :	str      =							4
       a_      :	str      =							3
       a_      :	Optional[Any]      =							(2_56, 2_56)

       a_      :	str      =							torch.ones((batch_size, num_channels) + sizes ).to(lowerCAmelCase_ )
       a_      :	Any      =							torch.tensor(batch_size * [1E-4] ).to(lowerCAmelCase_ )

       with torch.no_grad():
         a_      :	Tuple      =							model(lowerCAmelCase_							,		lowerCAmelCase_ ).sample

       a_      :	Any      =							output[0, -3:, -3:, -1].flatten().cpu()
       # fmt: off
       a_      :	Optional[Any]      =							torch.tensor([-4842.8691, -6499.6631, -3800.1953, -7978.2686, -10980.7129, -20028.8535, 8148.2822, 2342.2905, 567.7608] )
       # fmt: on

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


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	Tuple      =							UNetaDModel.from_pretrained("""fusing/ncsnpp-ffhq-ve-dummy-update""" )
       model.to(lowerCAmelCase_ )

       a_      :	str      =							4
       a_      :	Optional[Any]      =							3
       a_      :	Optional[Any]      =							(32, 32)

       a_      :	Any      =							torch.ones((batch_size, num_channels) + sizes ).to(lowerCAmelCase_ )
       a_      :	Dict      =							torch.tensor(batch_size * [1E-4] ).to(lowerCAmelCase_ )

       with torch.no_grad():
         a_      :	Union[str, Any]      =							model(lowerCAmelCase_							,		lowerCAmelCase_ ).sample

       a_      :	Any      =							output[0, -3:, -3:, -1].flatten().cpu()
       # fmt: off
       a_      :	int      =							torch.tensor([-0.0325, -0.0900, -0.0869, -0.0332, -0.0725, -0.0270, -0.0101, 0.0227, 0.0256] )
       # fmt: on

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






     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       pass




'''simple docstring'''

from datasets.utils.patching import _PatchedModuleObj, patch_submodule

from . import _test_patching






def 	_snake_case						(      ):
  """simple docstring"""







  import os as original_os
  from os import path as original_path
  from os import rename as original_rename
  from os.path import dirname as original_dirname
  from os.path import join as original_join

  assert _test_patching.os is original_os
  assert _test_patching.path is original_path
  assert _test_patching.join is original_join

  assert _test_patching.renamed_os is original_os
  assert _test_patching.renamed_path is original_path
  assert _test_patching.renamed_join is original_join

  a_      :	List[str]      =							"""__test_patch_submodule_mock__"""
  with patch_submodule(_test_patching     ,		"""os.path.join"""     ,		A_   ):
    # Every way to access os.path.join must be patched, and the rest must stay untouched

    # check os.path.join
    assert isinstance(_test_patching.os     ,		_PatchedModuleObj   )
    assert isinstance(_test_patching.os.path     ,		_PatchedModuleObj   )
    assert _test_patching.os.path.join is mock

    # check path.join
    assert isinstance(_test_patching.path     ,		_PatchedModuleObj   )
    assert _test_patching.path.join is mock

    # check join
    assert _test_patching.join is mock

    # check that the other attributes are untouched
    assert _test_patching.os.rename is original_rename
    assert _test_patching.path.dirname is original_dirname
    assert _test_patching.os.path.dirname is original_dirname

    # Even renamed modules or objects must be patched

    # check renamed_os.path.join
    assert isinstance(_test_patching.renamed_os     ,		_PatchedModuleObj   )
    assert isinstance(_test_patching.renamed_os.path     ,		_PatchedModuleObj   )
    assert _test_patching.renamed_os.path.join is mock

    # check renamed_path.join
    assert isinstance(_test_patching.renamed_path     ,		_PatchedModuleObj   )
    assert _test_patching.renamed_path.join is mock

    # check renamed_join
    assert _test_patching.renamed_join is mock

    # check that the other attributes are untouched
    assert _test_patching.renamed_os.rename is original_rename
    assert _test_patching.renamed_path.dirname is original_dirname
    assert _test_patching.renamed_os.path.dirname is original_dirname

  # check that everthing is back to normal when the patch is over

  assert _test_patching.os is original_os
  assert _test_patching.path is original_path
  assert _test_patching.join is original_join

  assert _test_patching.renamed_os is original_os
  assert _test_patching.renamed_path is original_path
  assert _test_patching.renamed_join is original_join






def 	_snake_case						(      ):
  """simple docstring"""







  assert _test_patching.open is open

  a_      :	Tuple      =							"""__test_patch_submodule_builtin_mock__"""
  # _test_patching has "open" in its globals
  assert _test_patching.open is open
  with patch_submodule(_test_patching     ,		"""open"""     ,		A_   ):
    assert _test_patching.open is mock

  # check that everthing is back to normal when the patch is over

  assert _test_patching.open is open






def 	_snake_case						(      ):
  """simple docstring"""







  a_      :	Any      =							"""__test_patch_submodule_missing_mock__"""
  with patch_submodule(_test_patching     ,		"""pandas.read_csv"""     ,		A_   ):
    pass






def 	_snake_case						(      ):
  """simple docstring"""







  a_      :	int      =							"""__test_patch_submodule_missing_builtin_mock__"""
  # _test_patching doesn't have "len" in its globals
  assert getattr(_test_patching     ,		"""len"""     ,		A_   ) is None
  with patch_submodule(_test_patching     ,		"""len"""     ,		A_   ):
    assert _test_patching.len is mock
  assert _test_patching.len is len






def 	_snake_case						(      ):
  """simple docstring"""







  a_      :	List[str]      =							"""__test_patch_submodule_start_and_stop_mock__"""
  a_      :	int      =							patch_submodule(_test_patching     ,		"""open"""     ,		A_   )
  assert _test_patching.open is open
  patch.start()
  assert _test_patching.open is mock
  patch.stop()
  assert _test_patching.open is open






def 	_snake_case						(      ):
  """simple docstring"""







  from os import rename as original_rename
  from os.path import dirname as original_dirname
  from os.path import join as original_join

  a_      :	Any      =							"""__test_patch_submodule_successive_join__"""
  a_      :	Optional[Any]      =							"""__test_patch_submodule_successive_dirname__"""
  a_      :	Optional[int]      =							"""__test_patch_submodule_successive_rename__"""
  assert _test_patching.os.path.join is original_join
  assert _test_patching.os.path.dirname is original_dirname
  assert _test_patching.os.rename is original_rename

  with patch_submodule(_test_patching     ,		"""os.path.join"""     ,		A_   ):
    with patch_submodule(_test_patching     ,		"""os.rename"""     ,		A_   ):
      with patch_submodule(_test_patching     ,		"""os.path.dirname"""     ,		A_   ):
        assert _test_patching.os.path.join is mock_join
        assert _test_patching.os.path.dirname is mock_dirname
        assert _test_patching.os.rename is mock_rename

    # try another order
  with patch_submodule(_test_patching     ,		"""os.rename"""     ,		A_   ):
    with patch_submodule(_test_patching     ,		"""os.path.join"""     ,		A_   ):
      with patch_submodule(_test_patching     ,		"""os.path.dirname"""     ,		A_   ):
        assert _test_patching.os.path.join is mock_join
        assert _test_patching.os.path.dirname is mock_dirname
        assert _test_patching.os.rename is mock_rename

  assert _test_patching.os.path.join is original_join
  assert _test_patching.os.path.dirname is original_dirname
  assert _test_patching.os.rename is original_rename






def 	_snake_case						(      ):
  """simple docstring"""







  a_      :	Optional[int]      =							"""__test_patch_submodule_doesnt_exist_mock__"""
  with patch_submodule(_test_patching     ,		"""__module_that_doesn_exist__.__attribute_that_doesn_exist__"""     ,		A_   ):
    pass
  with patch_submodule(_test_patching     ,		"""os.__attribute_that_doesn_exist__"""     ,		A_   ):
    pass


