






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        snake_case_				(							__lowerCAmelCase   ,      __lowerCAmelCase   ,      __lowerCAmelCase   ,      unittest.TestCase   ):







   """simple docstring"""



   __lowerCAmelCase  =    StableDiffusionModelEditingPipeline
   __lowerCAmelCase  =    TEXT_TO_IMAGE_PARAMS
   __lowerCAmelCase  =    TEXT_TO_IMAGE_BATCH_PARAMS
   __lowerCAmelCase  =    TEXT_TO_IMAGE_IMAGE_PARAMS
   __lowerCAmelCase  =    TEXT_TO_IMAGE_IMAGE_PARAMS

   def       SCREAMING_SNAKE_CASE__					(     self:						str						)	->		Optional[int]:


       """simple docstring"""



       torch.manual_seed(0						)
       _lowercase     			=						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 ,							)
       _lowercase     			=						DDIMScheduler()
       torch.manual_seed(0						)
       _lowercase     			=						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						)
       _lowercase     			=						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=1_000 ,							)
       _lowercase     			=						CLIPTextModel(lowerCamelCase__						)
       _lowercase     			=						CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'						)

       _lowercase     			=						{
           '''unet''': unet,
           '''scheduler''': scheduler,
           '''vae''': vae,
           '''text_encoder''': text_encoder,
           '''tokenizer''': tokenizer,
           '''safety_checker''': None,
           '''feature_extractor''': None,
       }
       return components

   def       SCREAMING_SNAKE_CASE__					(     self:						Dict ,							snake_case__:						Optional[int] ,							snake_case__:						str=0						)	->		Optional[int]:


       """simple docstring"""



       _lowercase     			=						torch.manual_seed(lowerCamelCase__						)
       _lowercase     			=						{
           '''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       SCREAMING_SNAKE_CASE__					(     self:						Dict						)	->		Any:


       """simple docstring"""



       _lowercase     			=						'''cpu'''  # ensure determinism for the device-dependent torch.Generator
       _lowercase     			=						self.get_dummy_components()
       _lowercase     			=						StableDiffusionModelEditingPipeline(**lowerCamelCase__						)
       _lowercase     			=						sd_pipe.to(lowerCamelCase__						)
       sd_pipe.set_progress_bar_config(disable=lowerCamelCase__						)

       _lowercase     			=						self.get_dummy_inputs(lowerCamelCase__						)
       _lowercase     			=						sd_pipe(**lowerCamelCase__						).images
       _lowercase     			=						image[0, -3:, -3:, -1]
       assert image.shape == (1, 64, 64, 3)

       _lowercase     			=						np.array([0.47_55, 0.51_32, 0.49_76, 0.39_04, 0.35_54, 0.47_65, 0.51_39, 0.51_58, 0.48_89]						)

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

   def       SCREAMING_SNAKE_CASE__					(     self:						Optional[Any]						)	->		Tuple:


       """simple docstring"""



       _lowercase     			=						'''cpu'''  # ensure determinism for the device-dependent torch.Generator
       _lowercase     			=						self.get_dummy_components()
       _lowercase     			=						StableDiffusionModelEditingPipeline(**lowerCamelCase__						)
       _lowercase     			=						sd_pipe.to(lowerCamelCase__						)
       sd_pipe.set_progress_bar_config(disable=lowerCamelCase__						)

       _lowercase     			=						self.get_dummy_inputs(lowerCamelCase__						)
       _lowercase     			=						'''french fries'''
       _lowercase     			=						sd_pipe(**lowerCamelCase__ ,							negative_prompt=lowerCamelCase__						)
       _lowercase     			=						output.images
       _lowercase     			=						image[0, -3:, -3:, -1]

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

       _lowercase     			=						np.array([0.49_92, 0.51_01, 0.50_04, 0.39_49, 0.36_04, 0.47_35, 0.52_16, 0.52_04, 0.49_13]						)

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

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


       """simple docstring"""



       _lowercase     			=						'''cpu'''  # ensure determinism for the device-dependent torch.Generator
       _lowercase     			=						self.get_dummy_components()
       _lowercase     			=						EulerAncestralDiscreteScheduler(
           beta_start=0.0_00_85 ,							beta_end=0.0_12 ,							beta_schedule='scaled_linear'						)
       _lowercase     			=						StableDiffusionModelEditingPipeline(**lowerCamelCase__						)
       _lowercase     			=						sd_pipe.to(lowerCamelCase__						)
       sd_pipe.set_progress_bar_config(disable=lowerCamelCase__						)

       _lowercase     			=						self.get_dummy_inputs(lowerCamelCase__						)
       _lowercase     			=						sd_pipe(**lowerCamelCase__						).images
       _lowercase     			=						image[0, -3:, -3:, -1]

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

       _lowercase     			=						np.array([0.47_47, 0.53_72, 0.47_79, 0.49_82, 0.55_43, 0.48_16, 0.52_38, 0.49_04, 0.50_27]						)

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

   def       SCREAMING_SNAKE_CASE__					(     self:						int						)	->		Optional[int]:


       """simple docstring"""



       _lowercase     			=						'''cpu'''  # ensure determinism for the device-dependent torch.Generator
       _lowercase     			=						self.get_dummy_components()
       _lowercase     			=						PNDMScheduler()
       _lowercase     			=						StableDiffusionModelEditingPipeline(**lowerCamelCase__						)
       _lowercase     			=						sd_pipe.to(lowerCamelCase__						)
       sd_pipe.set_progress_bar_config(disable=lowerCamelCase__						)

       _lowercase     			=						self.get_dummy_inputs(lowerCamelCase__						)
       # the pipeline does not expect pndm so test if it raises error.
       with self.assertRaises(lowerCamelCase__						):
           _lowercase     			=						sd_pipe(**lowerCamelCase__						).images

   def       SCREAMING_SNAKE_CASE__					(     self:						Union[str, Any]						)	->		Dict:


       """simple docstring"""



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

   def       SCREAMING_SNAKE_CASE__					(     self:						str						)	->		Optional[Any]:


       """simple docstring"""



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






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







   """simple docstring"""



   def       SCREAMING_SNAKE_CASE__					(     self:						Optional[int]						)	->		Tuple:


       """simple docstring"""



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

   def       SCREAMING_SNAKE_CASE__					(     self:						Union[str, Any] ,							snake_case__:						List[Any]=0						)	->		Optional[int]:


       """simple docstring"""



       _lowercase     			=						torch.manual_seed(lowerCamelCase__						)
       _lowercase     			=						{
           '''prompt''': '''A field of roses''',
           '''generator''': generator,
           '''num_inference_steps''': 3,
           '''guidance_scale''': 7.5,
           '''output_type''': '''numpy''',
       }
       return inputs

   def       SCREAMING_SNAKE_CASE__					(     self:						Dict						)	->		Optional[Any]:


       """simple docstring"""



       _lowercase     			=						'''CompVis/stable-diffusion-v1-4'''
       _lowercase     			=						StableDiffusionModelEditingPipeline.from_pretrained(lowerCamelCase__ ,							safety_checker=lowerCamelCase__						)
       pipe.to(lowerCamelCase__						)
       pipe.set_progress_bar_config(disable=lowerCamelCase__						)
       pipe.enable_attention_slicing()

       _lowercase     			=						self.get_inputs()
       _lowercase     			=						pipe(**lowerCamelCase__						).images
       _lowercase     			=						image[0, -3:, -3:, -1].flatten()

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

       _lowercase     			=						np.array(
           [0.6_74_94_96, 0.6_38_64_53, 0.51_44_32_67, 0.66_09_49_05, 0.61_92_12_15, 0.5_49_13_32, 0.5_74_44_17, 0.58_07_51_06, 0.5_17_46_58]						)

       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'						)

       _lowercase     			=						pipe(**lowerCamelCase__						).images
       _lowercase     			=						image[0, -3:, -3:, -1].flatten()

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

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

   def       SCREAMING_SNAKE_CASE__					(     self:						Any						)	->		str:


       """simple docstring"""



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

       _lowercase     			=						'''CompVis/stable-diffusion-v1-4'''
       _lowercase     			=						DDIMScheduler.from_pretrained(lowerCamelCase__ ,							subfolder='scheduler'						)
       _lowercase     			=						StableDiffusionModelEditingPipeline.from_pretrained(
           lowerCamelCase__ ,							scheduler=lowerCamelCase__ ,							safety_checker=lowerCamelCase__						)
       _lowercase     			=						pipe.to(lowerCamelCase__						)
       pipe.set_progress_bar_config(disable=lowerCamelCase__						)
       pipe.enable_attention_slicing(1						)
       pipe.enable_sequential_cpu_offload()

       _lowercase     			=						self.get_inputs()
       _lowercase     			=						pipe(**lowerCamelCase__						)

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




import argparse
import os
import re


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/check_dummies.py
__UpperCamelCase							:							Optional[Any]    							=					"src/transformers"

# Matches is_xxx_available()
__UpperCamelCase							:							str    							=					re.compile(r"is\_([a-z_]*)_available()")
# Matches from xxx import bla
__UpperCamelCase							:							Union[str, Any]    							=					re.compile(r"\s+from\s+\S*\s+import\s+([^\(\s].*)\n")
__UpperCamelCase							:							Dict    							=					re.compile(r"^\s+if\s+not\s+\(?is\_[a-z_]*\_available\(\)")


__UpperCamelCase							:							Any    							=					"\n{0} = None\n"

__UpperCamelCase							:							Tuple    							=					"\nclass {0}(metaclass=DummyObject):\n    _backends = {1}\n\n    def __init__(self, *args, **kwargs):\n        requires_backends(self, {1})\n"


__UpperCamelCase							:							Optional[Any]    							=					"\ndef {0}(*args, **kwargs):\n    requires_backends({0}, {1})\n"

def       _a	(					SCREAMING_SNAKE_CASE						:					str  ):







 """simple docstring"""



 if _re_test_backend.search(SCREAMING_SNAKE_CASE  ) is None:
  return None
 UpperCamelCase__       :     List[str] 							= [b[0] for b in _re_backend.findall(SCREAMING_SNAKE_CASE  )]
 backends.sort()
 return "_and_".join(SCREAMING_SNAKE_CASE  )

def       _a	(					):







 """simple docstring"""



 with open(os.path.join(SCREAMING_SNAKE_CASE     ,			'''__init__.py'''  )     ,			'''r'''     ,			encoding='''utf-8'''     ,			newline='''\n'''  ) as f:
  UpperCamelCase__       :     Union[str, Any] 							= f.readlines()

 # Get to the point we do the actual imports for type checking
 UpperCamelCase__       :     int 							= 0
 while not lines[line_index].startswith('''if TYPE_CHECKING'''  ):
  line_index += 1

 UpperCamelCase__       :     Optional[Any] 							= {}
 # Go through the end of the file
 while line_index < len(SCREAMING_SNAKE_CASE  ):
  # If the line is an if is_backend_available, we grab all objects associated.
  UpperCamelCase__       :     Any 							= find_backend(lines[line_index]  )
  if backend is not None:
   while not lines[line_index].startswith('''    else:'''  ):
    line_index += 1
   line_index += 1

   UpperCamelCase__       :     List[str] 							= []
   # Until we unindent, add backend objects to the list
   while len(lines[line_index]  ) <= 1 or lines[line_index].startswith(''' ''' * 8  ):
    UpperCamelCase__       :     str 							= lines[line_index]
    UpperCamelCase__       :     Union[str, Any] 							= _re_single_line_import.search(SCREAMING_SNAKE_CASE  )
    if single_line_import_search is not None:
     objects.extend(single_line_import_search.groups()[0].split(''', '''  )  )
    elif line.startswith(''' ''' * 12  ):
     objects.append(line[12:-2]  )
    line_index += 1

   UpperCamelCase__       :     Optional[int] 							= objects
  else:
   line_index += 1

 return backend_specific_objects

def       _a	(					SCREAMING_SNAKE_CASE						:					str     ,			SCREAMING_SNAKE_CASE						:					List[str]  ):







 """simple docstring"""



 if name.isupper():
  return DUMMY_CONSTANT.format(SCREAMING_SNAKE_CASE  )
 elif name.islower():
  return DUMMY_FUNCTION.format(SCREAMING_SNAKE_CASE     ,			SCREAMING_SNAKE_CASE  )
 else:
  return DUMMY_CLASS.format(SCREAMING_SNAKE_CASE     ,			SCREAMING_SNAKE_CASE  )

def       _a	(					SCREAMING_SNAKE_CASE						:					Dict=None  ):







 """simple docstring"""



 if backend_specific_objects is None:
  UpperCamelCase__       :     Any 							= read_init()
 # For special correspondence backend to module name as used in the function requires_modulename
 UpperCamelCase__       :     Optional[Any] 							= {}

 for backend, objects in backend_specific_objects.items():
  UpperCamelCase__       :     int 							= '''[''' + ''', '''.join(F"\"{b}\"" for b in backend.split('''_and_'''  )  ) + ''']'''
  UpperCamelCase__       :     Any 							= '''# This file is autogenerated by the command `make fix-copies`, do not edit.\n'''
  dummy_file += "from ..utils import DummyObject, requires_backends\n\n"
  dummy_file += "\n".join([create_dummy_object(SCREAMING_SNAKE_CASE     ,			SCREAMING_SNAKE_CASE  ) for o in objects]  )
  UpperCamelCase__       :     List[str] 							= dummy_file

 return dummy_files

def       _a	(					SCREAMING_SNAKE_CASE						:					Union[str, Any]=False  ):







 """simple docstring"""



 UpperCamelCase__       :     Dict 							= create_dummy_files()
 # For special correspondence backend to shortcut as used in utils/dummy_xxx_objects.py
 UpperCamelCase__       :     Tuple 							= {'''torch''': '''pt'''}

 # Locate actual dummy modules and read their content.
 UpperCamelCase__       :     Optional[Any] 							= os.path.join(SCREAMING_SNAKE_CASE     ,			'''utils'''  )
 UpperCamelCase__       :     int 							= {
     backend: os.path.join(SCREAMING_SNAKE_CASE     ,			F"dummy_{short_names.get(SCREAMING_SNAKE_CASE     ,			SCREAMING_SNAKE_CASE  )}_objects.py"  )
     for backend in dummy_files.keys()
 }

 UpperCamelCase__       :     Optional[Any] 							= {}
 for backend, file_path in dummy_file_paths.items():
  if os.path.isfile(SCREAMING_SNAKE_CASE  ):
   with open(SCREAMING_SNAKE_CASE     ,			'''r'''     ,			encoding='''utf-8'''     ,			newline='''\n'''  ) as f:
    UpperCamelCase__       :     List[Any] 							= f.read()
  else:
   UpperCamelCase__       :     List[Any] 							= ''''''

 for backend in dummy_files.keys():
  if dummy_files[backend] != actual_dummies[backend]:
   if overwrite:
    print(
        F"Updating transformers.utils.dummy_{short_names.get(SCREAMING_SNAKE_CASE     ,			SCREAMING_SNAKE_CASE  )}_objects.py as the main "
        '''__init__ has new objects.'''  )
    with open(dummy_file_paths[backend]     ,			'''w'''     ,			encoding='''utf-8'''     ,			newline='''\n'''  ) as f:
     f.write(dummy_files[backend]  )
   else:
    raise ValueError(
        '''The main __init__ has objects that are not present in '''
        F"transformers.utils.dummy_{short_names.get(SCREAMING_SNAKE_CASE     ,			SCREAMING_SNAKE_CASE  )}_objects.py. Run `make fix-copies` "
        '''to fix this.'''  )


if __name__ == "__main__":
       __UpperCamelCase							:							Optional[int]    							=					argparse.ArgumentParser()
       parser.add_argument("--fix_and_overwrite", action="store_true", help="Whether to fix inconsistencies.")
       __UpperCamelCase							:							str    							=					parser.parse_args()

       check_dummies(args.fix_and_overwrite)


