





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


SCREAMING_SNAKE_CASE__       :			str			      =						logging.get_logger(__name__)

enable_full_determinism()







class   __lowerCAmelCase			(					_UpperCamelCase		,_UpperCamelCase		,unittest.TestCase  ):
   _UpperCamelCase :						Union[str, Any]     		=				UNetaDModel
   _UpperCamelCase :						Any     		=				"""sample"""





   @property
   def 	_snake_case ( self      )				->  List[Any]:
      """simple docstring"""





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

      a__	:						Dict      		=  floats_tensor((batch_size, num_channels) + sizes      ).to(snake_case      )
      a__	:						Tuple      		=  torch.tensor([10]      ).to(snake_case      )

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





   @property
   def 	_snake_case ( self      )				->  str:
      """simple docstring"""





      return (3, 32, 32)





   @property
   def 	_snake_case ( self      )				->  List[Any]:
      """simple docstring"""





      return (3, 32, 32)





   def 	_snake_case ( self      )				->  Union[str, Any]:
      """simple docstring"""





      a__	:						List[str]      		=  {
          "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__	:						Optional[int]      		=  self.dummy_input
      return init_dict, inputs_dict







class   __lowerCAmelCase			(					_UpperCamelCase		,_UpperCamelCase		,unittest.TestCase  ):
   _UpperCamelCase :						Dict     		=				UNetaDModel
   _UpperCamelCase :						Union[str, Any]     		=				"""sample"""





   @property
   def 	_snake_case ( self      )				->  List[Any]:
      """simple docstring"""





      a__	:						Union[str, Any]      		=  4
      a__	:						Union[str, Any]      		=  4
      a__	:						Dict      		=  (32, 32)

      a__	:						Tuple      		=  floats_tensor((batch_size, num_channels) + sizes      ).to(snake_case      )
      a__	:						List[Any]      		=  torch.tensor([10]      ).to(snake_case      )

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





   @property
   def 	_snake_case ( self      )				->  Optional[int]:
      """simple docstring"""





      return (4, 32, 32)





   @property
   def 	_snake_case ( self      )				->  int:
      """simple docstring"""





      return (4, 32, 32)





   def 	_snake_case ( self      )				->  List[str]:
      """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__	:						Optional[Any]      		=  self.dummy_input
      return init_dict, inputs_dict





   def 	_snake_case ( self      )				->  Dict:
      """simple docstring"""





      a__		,     a__	:						Dict      		=  UNetaDModel.from_pretrained("fusing/unet-ldm-dummy-update"					,			output_loading_info=snake_case      )

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

      model.to(snake_case      )
      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 	_snake_case ( self      )				->  str:
      """simple docstring"""





      a__		,     a__	:						Optional[int]      		=  UNetaDModel.from_pretrained("fusing/unet-ldm-dummy-update"					,			output_loading_info=snake_case      )
      model.to(snake_case      )
      a__	:						Optional[int]      		=  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 	_snake_case ( self      )				->  List[Any]:
      """simple docstring"""





      a__		,     a__	:						int      		=  UNetaDModel.from_pretrained("fusing/unet-ldm-dummy-update"					,			output_loading_info=snake_case      )
      model_accelerate.to(snake_case      )
      model_accelerate.eval()

      a__	:						Optional[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[Any]      		=  noise.to(snake_case      )
      a__	:						Optional[int]      		=  torch.tensor([10] * noise.shape[0]      ).to(snake_case      )

      a__	:						Any      		=  model_accelerate(snake_case					,			snake_case      )["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__	:						Tuple      		=  UNetaDModel.from_pretrained(
          "fusing/unet-ldm-dummy-update"					,			output_loading_info=snake_case					,			low_cpu_mem_usage=snake_case      )
      model_normal_load.to(snake_case      )
      model_normal_load.eval()
      a__	:						Any      		=  model_normal_load(snake_case					,			snake_case      )["sample"]

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





   def 	_snake_case ( self      )				->  Optional[int]:
      """simple docstring"""





      a__	:						str      		=  UNetaDModel.from_pretrained("fusing/unet-ldm-dummy-update"      )
      model.eval()
      model.to(snake_case      )

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

      with torch.no_grad():
         a__	:						Dict      		=  model(snake_case					,			snake_case      ).sample

      a__	:						str      		=  output[0, -1, -3:, -3:].flatten().cpu()
      # fmt: off
      a__	:						Optional[Any]      		=  torch.tensor([-13.3_258, -20.1_100, -15.9_873, -17.6_617, -23.0_596, -17.9_419, -13.3_675, -16.1_889, -12.3_800]      )
      # fmt: on

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







class   __lowerCAmelCase			(					_UpperCamelCase		,_UpperCamelCase		,unittest.TestCase  ):
   _UpperCamelCase :						Dict     		=				UNetaDModel
   _UpperCamelCase :						List[Any]     		=				"""sample"""





   @property
   def 	_snake_case ( self					,			snake_case=(32, 32)      )				->  List[Any]:
      """simple docstring"""





      a__	:						Union[str, Any]      		=  4
      a__	:						Optional[Any]      		=  3

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

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





   @property
   def 	_snake_case ( self      )				->  Dict:
      """simple docstring"""





      return (3, 32, 32)





   @property
   def 	_snake_case ( self      )				->  Union[str, Any]:
      """simple docstring"""





      return (3, 32, 32)





   def 	_snake_case ( self      )				->  int:
      """simple docstring"""





      a__	:						Tuple      		=  {
          "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 	_snake_case ( self      )				->  str:
      """simple docstring"""





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

      model.to(snake_case      )
      a__	:						int      		=  self.dummy_input
      a__	:						Union[str, Any]      		=  floats_tensor((4, 3) + (256, 256)      ).to(snake_case      )
      a__	:						List[str]      		=  noise
      a__	:						Any      		=  model(**snake_case      )

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





   @slow
   def 	_snake_case ( self      )				->  Union[str, Any]:
      """simple docstring"""





      a__	:						Dict      		=  UNetaDModel.from_pretrained("google/ncsnpp-celebahq-256"      )
      model.to(snake_case      )

      a__	:						Optional[Any]      		=  4
      a__	:						Any      		=  3
      a__	:						Union[str, Any]      		=  (256, 256)

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

      with torch.no_grad():
         a__	:						Optional[int]      		=  model(snake_case					,			snake_case      ).sample

      a__	:						Tuple      		=  output[0, -3:, -3:, -1].flatten().cpu()
      # fmt: off
      a__	:						Dict      		=  torch.tensor([-4_842.8_691, -6_499.6_631, -3_800.1_953, -7_978.2_686, -10_980.7_129, -20_028.8_535, 8_148.2_822, 2_342.2_905, 567.7_608]      )
      # fmt: on

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





   def 	_snake_case ( self      )				->  Tuple:
      """simple docstring"""





      a__	:						List[Any]      		=  UNetaDModel.from_pretrained("fusing/ncsnpp-ffhq-ve-dummy-update"      )
      model.to(snake_case      )

      a__	:						List[Any]      		=  4
      a__	:						Union[str, Any]      		=  3
      a__	:						str      		=  (32, 32)

      a__	:						Optional[int]      		=  torch.ones((batch_size, num_channels) + sizes      ).to(snake_case      )
      a__	:						str      		=  torch.tensor(batch_size * [1E-4]      ).to(snake_case      )

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

      a__	:						int      		=  output[0, -3:, -3:, -1].flatten().cpu()
      # fmt: off
      a__	:						Tuple      		=  torch.tensor([-0.0_325, -0.0_900, -0.0_869, -0.0_332, -0.0_725, -0.0_270, -0.0_101, 0.0_227, 0.0_256]      )
      # fmt: on

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





   def 	_snake_case ( self      )				->  Tuple:
      """simple docstring"""





      pass








import PIL.Image
import PIL.ImageOps
from packaging import version
from PIL import Image


if version.parse(version.parse(PIL.__version__).base_version) >= version.parse("""9.1.0"""):
  SCREAMING_SNAKE_CASE__       :			int			      =						{
      """linear""": PIL.Image.Resampling.BILINEAR,
      """bilinear""": PIL.Image.Resampling.BILINEAR,
      """bicubic""": PIL.Image.Resampling.BICUBIC,
      """lanczos""": PIL.Image.Resampling.LANCZOS,
      """nearest""": PIL.Image.Resampling.NEAREST,
  }
else:
  SCREAMING_SNAKE_CASE__       :			Dict			      =						{
      """linear""": PIL.Image.LINEAR,
      """bilinear""": PIL.Image.BILINEAR,
      """bicubic""": PIL.Image.BICUBIC,
      """lanczos""": PIL.Image.LANCZOS,
      """nearest""": PIL.Image.NEAREST,
  }






def       _A  (      lowerCamelCase   ):
   a__	:						Optional[int]      		=  (images / 2 + 0.5).clamp(0							, 1   )
   a__	:						List[Any]      		=  images.cpu().permute(0							, 2							, 3							, 1   ).float().numpy()
   a__	:						int      		=  numpy_to_pil(lowerCamelCase   )
   return images






def       _A  (      lowerCamelCase   ):
   if images.ndim == 3:
      a__	:						Tuple      		=  images[None, ...]
   a__	:						Dict      		=  (images * 255).round().astype("uint8"   )
   if images.shape[-1] == 1:
      # special case for grayscale (single channel) images
      a__	:						str      		=  [Image.fromarray(image.squeeze()							, mode="L"   ) for image in images]
   else:
      a__	:						List[Any]      		=  [Image.fromarray(lowerCamelCase   ) for image in images]

   return pil_images

