import gc
import unittest

import numpy as np
import torch

from diffusers import AutoencoderKL, DDIMScheduler, DiTPipeline, DPMSolverMultistepScheduler, TransformeraDModel
from diffusers.utils import is_xformers_available, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    CLASS_CONDITIONED_IMAGE_GENERATION_BATCH_PARAMS,
    CLASS_CONDITIONED_IMAGE_GENERATION_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()




class 	__snake_case	( __lowerCAmelCase ,  unittest.TestCase		):





      '''simple docstring'''






      _snake_case    		=       DiTPipeline
      _snake_case    		=       CLASS_CONDITIONED_IMAGE_GENERATION_PARAMS
      _snake_case    		=       PipelineTesterMixin.required_optional_params - {
          'latents',
          'num_images_per_prompt',
          'callback',
          'callback_steps',
      }
      _snake_case    		=       CLASS_CONDITIONED_IMAGE_GENERATION_BATCH_PARAMS
      _snake_case    		=       False



      def 	_SCREAMING_SNAKE_CASE    (					self     :				Any)				->List[str]:


          """simple docstring"""


          torch.manual_seed(0)
          _lowerCamelCase     :		Union[str, Any] 						=						TransformeraDModel(
              sample_size=16    ,  num_layers=2    ,  patch_size=4    ,  attention_head_dim=8    ,  num_attention_heads=2    ,  in_channels=4    ,  out_channels=8    ,  attention_bias=_UpperCamelCase    ,  activation_fn="""gelu-approximate"""    ,  num_embeds_ada_norm=1000    ,  norm_type="""ada_norm_zero"""    ,  norm_elementwise_affine=_UpperCamelCase    ,  )
          _lowerCamelCase     :		Dict 						=						AutoencoderKL()
          _lowerCamelCase     :		Tuple 						=						DDIMScheduler()
          _lowerCamelCase     :		Any 						=						{"""transformer""": transformer.eval(), """vae""": vae.eval(), """scheduler""": scheduler}
          return components



      def 	_SCREAMING_SNAKE_CASE    (					self     :				Optional[int]    ,  _UpperCamelCase     :				Optional[int]    ,  _UpperCamelCase     :				Dict=0)				->Optional[Any]:


          """simple docstring"""


          if str(_UpperCamelCase).startswith("""mps"""):
              _lowerCamelCase     :		Any 						=						torch.manual_seed(_UpperCamelCase)
          else:
              _lowerCamelCase     :		Any 						=						torch.Generator(device=_UpperCamelCase).manual_seed(_UpperCamelCase)
          _lowerCamelCase     :		Optional[int] 						=						{
              """class_labels""": [1],
              """generator""": generator,
              """num_inference_steps""": 2,
              """output_type""": """numpy""",
          }
          return inputs



      def 	_SCREAMING_SNAKE_CASE    (					self     :				Dict)				->Dict:


          """simple docstring"""


          _lowerCamelCase     :		Optional[int] 						=						"""cpu"""

          _lowerCamelCase     :		List[Any] 						=						self.get_dummy_components()
          _lowerCamelCase     :		List[str] 						=						self.pipeline_class(**_UpperCamelCase)
          pipe.to(_UpperCamelCase)
          pipe.set_progress_bar_config(disable=_UpperCamelCase)

          _lowerCamelCase     :		Dict 						=						self.get_dummy_inputs(_UpperCamelCase)
          _lowerCamelCase     :		Any 						=						pipe(**_UpperCamelCase).images
          _lowerCamelCase     :		str 						=						image[0, -3:, -3:, -1]

          self.assertEqual(image.shape    ,  (1, 16, 16, 3))
          _lowerCamelCase     :		List[Any] 						=						np.array([0.2_9_4_6, 0.6_6_0_1, 0.4_3_2_9, 0.3_2_9_6, 0.4_1_4_4, 0.5_3_1_9, 0.7_2_7_3, 0.5_0_1_3, 0.4_4_5_7])
          _lowerCamelCase     :		Optional[int] 						=						np.abs(image_slice.flatten() - expected_slice).max()
          self.assertLessEqual(_UpperCamelCase    ,  1E-3)



      def 	_SCREAMING_SNAKE_CASE    (					self     :				List[str])				->Tuple:


          """simple docstring"""


          self._test_inference_batch_single_identical(relax_max_difference=_UpperCamelCase    ,  expected_max_diff=1E-3)




      @unittest.skipIf(
          torch_device != """cuda""" or not is_xformers_available()    ,  reason="""XFormers attention is only available with CUDA and `xformers` installed"""    ,  )
      def 	_SCREAMING_SNAKE_CASE    (					self     :				str)				->Optional[int]:


          """simple docstring"""


          self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1E-3)





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





      '''simple docstring'''






      def 	_SCREAMING_SNAKE_CASE    (					self     :				str)				->int:


          """simple docstring"""


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



      def 	_SCREAMING_SNAKE_CASE    (					self     :				List[Any])				->List[str]:


          """simple docstring"""


          _lowerCamelCase     :		Tuple 						=						torch.manual_seed(0)

          _lowerCamelCase     :		Dict 						=						DiTPipeline.from_pretrained("""facebook/DiT-XL-2-256""")
          pipe.to("""cuda""")

          _lowerCamelCase     :		List[str] 						=						["""vase""", """umbrella""", """white shark""", """white wolf"""]
          _lowerCamelCase     :		Any 						=						pipe.get_label_ids(_UpperCamelCase)

          _lowerCamelCase     :		List[Any] 						=						pipe(_UpperCamelCase    ,  generator=_UpperCamelCase    ,  num_inference_steps=40    ,  output_type="""np""").images

          for word, image in zip(_UpperCamelCase    ,  _UpperCamelCase):
              _lowerCamelCase     :		List[str] 						=						load_numpy(
                  F"""https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/dit/{word}.npy""")
              assert np.abs((expected_image - image).max()) < 1E-2




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


          """simple docstring"""


          _lowerCamelCase     :		Union[str, Any] 						=						DiTPipeline.from_pretrained("""facebook/DiT-XL-2-512""")
          _lowerCamelCase     :		List[str] 						=						DPMSolverMultistepScheduler.from_config(pipe.scheduler.config)
          pipe.to("""cuda""")

          _lowerCamelCase     :		Tuple 						=						["""vase""", """umbrella"""]
          _lowerCamelCase     :		Dict 						=						pipe.get_label_ids(_UpperCamelCase)

          _lowerCamelCase     :		List[Any] 						=						torch.manual_seed(0)
          _lowerCamelCase     :		List[Any] 						=						pipe(_UpperCamelCase    ,  generator=_UpperCamelCase    ,  num_inference_steps=25    ,  output_type="""np""").images

          for word, image in zip(_UpperCamelCase    ,  _UpperCamelCase):
              _lowerCamelCase     :		str 						=						load_numpy(
                  """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
                  F"""/dit/{word}_512.npy""")

              assert np.abs((expected_image - image).max()) < 1E-1





from __future__ import annotations

from math import pi
from typing import Protocol

import matplotlib.pyplot as plt
import numpy as np




class 	__snake_case	( __lowerCAmelCase		):





      '''simple docstring'''






      def 	_SCREAMING_SNAKE_CASE    (					self     :				Optional[Any]    ,  _UpperCamelCase     :				float)				->float:


          """simple docstring"""


          return 0.0







def 						A__  (					__A   , __A	):







    '''simple docstring'''
    _lowerCamelCase     :		int 						=						min([-20, np.min(fft_results[1 : samplerate // 2 - 1]	)]	)
    _lowerCamelCase     :		Tuple 						=						max([20, np.max(fft_results[1 : samplerate // 2 - 1]	)]	)
    return lowest, highest







def 						A__  (					__A   , __A	):







    '''simple docstring'''
    _lowerCamelCase     :		Tuple 						=						512
    _lowerCamelCase     :		Tuple 						=						[1] + [0] * (size - 1)
    _lowerCamelCase     :		Optional[Any] 						=						[filter_type.process(__A	) for item in inputs]

    _lowerCamelCase     :		Optional[Any] 						=						[0] * (samplerate - size)  # zero-padding
    outputs += filler
    _lowerCamelCase     :		Tuple 						=						np.abs(np.fft.fft(__A	)	)
    _lowerCamelCase     :		List[Any] 						=						20 * np.logaa(__A	)

    # Frequencies on log scale from 24 to nyquist frequency
    plt.xlim(24   , samplerate / 2 - 1	)
    plt.xlabel("""Frequency (Hz)"""	)
    plt.xscale("""log"""	)

    # Display within reasonable bounds
    _lowerCamelCase     :		Any 						=						get_bounds(__A   , __A	)
    plt.ylim(max([-80, bounds[0]]	)   , min([80, bounds[1]]	)	)
    plt.ylabel("""Gain (dB)"""	)

    plt.plot(__A	)
    plt.show()







def 						A__  (					__A   , __A	):







    '''simple docstring'''
    _lowerCamelCase     :		Tuple 						=						512
    _lowerCamelCase     :		Union[str, Any] 						=						[1] + [0] * (size - 1)
    _lowerCamelCase     :		int 						=						[filter_type.process(__A	) for item in inputs]

    _lowerCamelCase     :		Optional[Any] 						=						[0] * (samplerate - size)  # zero-padding
    outputs += filler
    _lowerCamelCase     :		Any 						=						np.angle(np.fft.fft(__A	)	)

    # Frequencies on log scale from 24 to nyquist frequency
    plt.xlim(24   , samplerate / 2 - 1	)
    plt.xlabel("""Frequency (Hz)"""	)
    plt.xscale("""log"""	)

    plt.ylim(-2 * pi   , 2 * pi	)
    plt.ylabel("""Phase shift (Radians)"""	)
    plt.plot(np.unwrap(__A   , -2 * pi	)	)
    plt.show()




