




'''simple docstring'''





import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    EulerAncestralDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionModelEditingPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()




@skip_mps
class      UpperCamelCase       (	snake_case__      ,		snake_case__      ,		snake_case__      ,		unittest.TestCase       ):
  a     :Any          =					StableDiffusionModelEditingPipeline
  a     :str          =					TEXT_TO_IMAGE_PARAMS
  a     :int          =					TEXT_TO_IMAGE_BATCH_PARAMS
  a     :Optional[int]          =					TEXT_TO_IMAGE_IMAGE_PARAMS
  a     :int          =					TEXT_TO_IMAGE_IMAGE_PARAMS



  def     a_							(			self ):







      '''simple docstring'''


      torch.manual_seed(0 )
      _a				:							Dict							       =      UNetaDConditionModel(
          block_out_channels=(32, 64)		,     layers_per_block=2		,     sample_size=32		,     in_channels=4		,     out_channels=4		,     down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")		,     up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")		,     cross_attention_dim=32		,     )
      _a				:							Tuple							       =      DDIMScheduler()
      torch.manual_seed(0 )
      _a				:							str							       =      AutoencoderKL(
          block_out_channels=[32, 64]		,     in_channels=3		,     out_channels=3		,     down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]		,     up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]		,     latent_channels=4		,     )
      torch.manual_seed(0 )
      _a				:							Tuple							       =      CLIPTextConfig(
          bos_token_id=0		,     eos_token_id=2		,     hidden_size=32		,     intermediate_size=37		,     layer_norm_eps=1e-05		,     num_attention_heads=4		,     num_hidden_layers=5		,     pad_token_id=1		,     vocab_size=1000		,     )
      _a				:							Any							       =      CLIPTextModel(__lowerCAmelCase )
      _a				:							str							       =      CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""" )

      _a				:							Union[str, Any]							       =      {
          """unet""": unet,
          """scheduler""": scheduler,
          """vae""": vae,
          """text_encoder""": text_encoder,
          """tokenizer""": tokenizer,
          """safety_checker""": None,
          """feature_extractor""": None,
      }
      return components



  def     a_							(			self		,     __lowerCAmelCase		,     __lowerCAmelCase=0 ):







      '''simple docstring'''


      _a				:							Optional[Any]							       =      torch.manual_seed(__lowerCAmelCase )
      _a				:							str							       =      {
          """prompt""": """A field of roses""",
          """generator""": generator,
          # Setting height and width to None to prevent OOMs on CPU.
          """height""": None,
          """width""": None,
          """num_inference_steps""": 2,
          """guidance_scale""": 6.0,
          """output_type""": """numpy""",
      }
      return inputs



  def     a_							(			self ):







      '''simple docstring'''


      _a				:							Tuple							       =      """cpu"""  # ensure determinism for the device-dependent torch.Generator
      _a				:							List[str]							       =      self.get_dummy_components()
      _a				:							List[str]							       =      StableDiffusionModelEditingPipeline(**__lowerCAmelCase )
      _a				:							Any							       =      sd_pipe.to(__lowerCAmelCase )
      sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase )

      _a				:							Dict							       =      self.get_dummy_inputs(__lowerCAmelCase )
      _a				:							List[Any]							       =      sd_pipe(**__lowerCAmelCase ).images
      _a				:							Optional[Any]							       =      image[0, -3:, -3:, -1]
      assert image.shape == (1, 64, 64, 3)

      _a				:							int							       =      np.array([0.4755, 0.5132, 0.4976, 0.3904, 0.3554, 0.4765, 0.5139, 0.5158, 0.4889] )

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



  def     a_							(			self ):







      '''simple docstring'''


      _a				:							Tuple							       =      """cpu"""  # ensure determinism for the device-dependent torch.Generator
      _a				:							Dict							       =      self.get_dummy_components()
      _a				:							int							       =      StableDiffusionModelEditingPipeline(**__lowerCAmelCase )
      _a				:							Optional[Any]							       =      sd_pipe.to(__lowerCAmelCase )
      sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase )

      _a				:							Optional[Any]							       =      self.get_dummy_inputs(__lowerCAmelCase )
      _a				:							List[str]							       =      """french fries"""
      _a				:							List[str]							       =      sd_pipe(**__lowerCAmelCase		,     negative_prompt=__lowerCAmelCase )
      _a				:							List[Any]							       =      output.images
      _a				:							Any							       =      image[0, -3:, -3:, -1]

      assert image.shape == (1, 64, 64, 3)

      _a				:							Dict							       =      np.array([0.4992, 0.5101, 0.5004, 0.3949, 0.3604, 0.4735, 0.5216, 0.5204, 0.4913] )

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



  def     a_							(			self ):







      '''simple docstring'''


      _a				:							Optional[Any]							       =      """cpu"""  # ensure determinism for the device-dependent torch.Generator
      _a				:							Tuple							       =      self.get_dummy_components()
      _a				:							int							       =      EulerAncestralDiscreteScheduler(
          beta_start=0.00085		,     beta_end=0.012		,     beta_schedule="""scaled_linear""" )
      _a				:							List[Any]							       =      StableDiffusionModelEditingPipeline(**__lowerCAmelCase )
      _a				:							List[str]							       =      sd_pipe.to(__lowerCAmelCase )
      sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase )

      _a				:							List[str]							       =      self.get_dummy_inputs(__lowerCAmelCase )
      _a				:							str							       =      sd_pipe(**__lowerCAmelCase ).images
      _a				:							str							       =      image[0, -3:, -3:, -1]

      assert image.shape == (1, 64, 64, 3)

      _a				:							Dict							       =      np.array([0.4747, 0.5372, 0.4779, 0.4982, 0.5543, 0.4816, 0.5238, 0.4904, 0.5027] )

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



  def     a_							(			self ):







      '''simple docstring'''


      _a				:							Tuple							       =      """cpu"""  # ensure determinism for the device-dependent torch.Generator
      _a				:							Any							       =      self.get_dummy_components()
      _a				:							Union[str, Any]							       =      PNDMScheduler()
      _a				:							List[str]							       =      StableDiffusionModelEditingPipeline(**__lowerCAmelCase )
      _a				:							List[Any]							       =      sd_pipe.to(__lowerCAmelCase )
      sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase )

      _a				:							List[Any]							       =      self.get_dummy_inputs(__lowerCAmelCase )
      # the pipeline does not expect pndm so test if it raises error.
      with self.assertRaises(__lowerCAmelCase ):
          _a				:							int							       =      sd_pipe(**__lowerCAmelCase ).images



  def     a_							(			self ):







      '''simple docstring'''


      super().test_inference_batch_single_identical(expected_max_diff=5e-3 )



  def     a_							(			self ):







      '''simple docstring'''


      super().test_attention_slicing_forward_pass(expected_max_diff=5e-3 )






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



  def     a_							(			self ):







      '''simple docstring'''


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



  def     a_							(			self		,     __lowerCAmelCase=0 ):







      '''simple docstring'''


      _a				:							Optional[Any]							       =      torch.manual_seed(__lowerCAmelCase )
      _a				:							List[str]							       =      {
          """prompt""": """A field of roses""",
          """generator""": generator,
          """num_inference_steps""": 3,
          """guidance_scale""": 7.5,
          """output_type""": """numpy""",
      }
      return inputs



  def     a_							(			self ):







      '''simple docstring'''


      _a				:							Union[str, Any]							       =      """CompVis/stable-diffusion-v1-4"""
      _a				:							Tuple							       =      StableDiffusionModelEditingPipeline.from_pretrained(__lowerCAmelCase		,     safety_checker=__lowerCAmelCase )
      pipe.to(__lowerCAmelCase )
      pipe.set_progress_bar_config(disable=__lowerCAmelCase )
      pipe.enable_attention_slicing()

      _a				:							List[str]							       =      self.get_inputs()
      _a				:							str							       =      pipe(**__lowerCAmelCase ).images
      _a				:							str							       =      image[0, -3:, -3:, -1].flatten()

      assert image.shape == (1, 512, 512, 3)

      _a				:							Tuple							       =      np.array(
          [0.6749496, 0.6386453, 0.51443267, 0.66094905, 0.61921215, 0.5491332, 0.5744417, 0.58075106, 0.5174658] )

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

      # make sure image changes after editing
      pipe.edit_model("""A pack of roses"""		,     """A pack of blue roses""" )

      _a				:							Union[str, Any]							       =      pipe(**__lowerCAmelCase ).images
      _a				:							Tuple							       =      image[0, -3:, -3:, -1].flatten()

      assert image.shape == (1, 512, 512, 3)

      assert np.abs(expected_slice - image_slice ).max() > 1e-1



  def     a_							(			self ):







      '''simple docstring'''


      torch.cuda.empty_cache()
      torch.cuda.reset_max_memory_allocated()
      torch.cuda.reset_peak_memory_stats()

      _a				:							Union[str, Any]							       =      """CompVis/stable-diffusion-v1-4"""
      _a				:							Any							       =      DDIMScheduler.from_pretrained(__lowerCAmelCase		,     subfolder="""scheduler""" )
      _a				:							List[Any]							       =      StableDiffusionModelEditingPipeline.from_pretrained(
          __lowerCAmelCase		,     scheduler=__lowerCAmelCase		,     safety_checker=__lowerCAmelCase )
      _a				:							Optional[int]							       =      pipe.to(__lowerCAmelCase )
      pipe.set_progress_bar_config(disable=__lowerCAmelCase )
      pipe.enable_attention_slicing(1 )
      pipe.enable_sequential_cpu_offload()

      _a				:							Dict							       =      self.get_inputs()
      _a				:							Tuple							       =      pipe(**__lowerCAmelCase )

      _a				:							Any							       =      torch.cuda.max_memory_allocated()
      # make sure that less than 4.4 GB is allocated
      assert mem_bytes < 4.4 * 10**9









"""simple docstring"""
from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_flax_available,
    is_tf_available,
    is_torch_available,
)


_snake_case	:						Any					=					{
    'configuration_unispeech_sat': ['UNISPEECH_SAT_PRETRAINED_CONFIG_ARCHIVE_MAP', 'UniSpeechSatConfig'],
}

try:
		if not is_torch_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		_snake_case	:						str					=					[
		    'UNISPEECH_SAT_PRETRAINED_MODEL_ARCHIVE_LIST',
		    'UniSpeechSatForAudioFrameClassification',
		    'UniSpeechSatForCTC',
		    'UniSpeechSatForPreTraining',
		    'UniSpeechSatForSequenceClassification',
		    'UniSpeechSatForXVector',
		    'UniSpeechSatModel',
		    'UniSpeechSatPreTrainedModel',
		]

if TYPE_CHECKING:
		from .configuration_unispeech_sat import UNISPEECH_SAT_PRETRAINED_CONFIG_ARCHIVE_MAP, UniSpeechSatConfig

		try:
				if not is_torch_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .modeling_unispeech_sat import (
				    UNISPEECH_SAT_PRETRAINED_MODEL_ARCHIVE_LIST,
				    UniSpeechSatForAudioFrameClassification,
				    UniSpeechSatForCTC,
				    UniSpeechSatForPreTraining,
				    UniSpeechSatForSequenceClassification,
				    UniSpeechSatForXVector,
				    UniSpeechSatModel,
				    UniSpeechSatPreTrainedModel,
				)

else:
		import sys

		_snake_case	:						Union[str, Any]					=					_LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)


