
import argparse
import glob
import importlib.util
import os
import re

import black
from doc_builder.style_doc import style_docstrings_in_code


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/check_copies.py
lowercase_       :						Optional[int]     					=	'''src/diffusers'''
lowercase_       :						int     					=	'''.'''


# This is to make sure the diffusers module imported is the one in the repo.
lowercase_       :						List[Any]     					=	importlib.util.spec_from_file_location(
    '''diffusers''',
    os.path.join(DIFFUSERS_PATH, '''__init__.py'''),
    submodule_search_locations=[DIFFUSERS_PATH],
)
lowercase_       :						Dict     					=	spec.loader.load_module()





def    _lowerCAmelCase    (     lowerCamelCase__							:							Any,						lowerCamelCase__							:							Dict				)  ->    Dict:
  return line.startswith(lowerCamelCase__				) or len(lowerCamelCase__				) <= 1 or re.search(R"^\s*\)(\s*->.*:|:)\s*$",						lowerCamelCase__				) is not None





def    _lowerCAmelCase    (     lowerCamelCase__							:							int				)  ->    Optional[int]:
  _SCREAMING_SNAKE_CASE   :				int							    = object_name.split("."				)
  _SCREAMING_SNAKE_CASE   :				str							    = 0

  # First let's find the module where our object lives.
  _SCREAMING_SNAKE_CASE   :				str							    = parts[i]
  while i < len(lowerCamelCase__				) and not os.path.isfile(os.path.join(lowerCamelCase__,						f'''{module}.py'''				)				):
    i += 1
    if i < len(lowerCamelCase__				):
      _SCREAMING_SNAKE_CASE   :				str							    = os.path.join(lowerCamelCase__,						parts[i]				)
  if i >= len(lowerCamelCase__				):
    raise ValueError(f'''`object_name` should begin with the name of a module of diffusers but got {object_name}.'''				)

  with open(os.path.join(lowerCamelCase__,						f'''{module}.py'''				),						"r",						encoding="utf-8",						newline="\n"				) as f:
    _SCREAMING_SNAKE_CASE   :				Any							    = f.readlines()

  # Now let's find the class / func in the code!
  _SCREAMING_SNAKE_CASE   :				Tuple							    = ""
  _SCREAMING_SNAKE_CASE   :				Optional[int]							    = 0
  for name in parts[i + 1 :]:
    while (
        line_index < len(lowerCamelCase__				) and re.search(Rf'''^{indent}(class|def)\s+{name}(\(|\:)''',						lines[line_index]				) is None
    ):
      line_index += 1
    indent += "    "
    line_index += 1

  if line_index >= len(lowerCamelCase__				):
    raise ValueError(f''' {object_name} does not match any function or class in {module}.'''				)

  # We found the beginning of the class / func, now let's find the end (when the indent diminishes).
  _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = line_index
  while line_index < len(lowerCamelCase__				) and _should_continue(lines[line_index],						lowerCamelCase__				):
    line_index += 1
  # Clean up empty lines at the end (if any).
  while len(lines[line_index - 1]				) <= 1:
    line_index -= 1

  _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = lines[start_index:line_index]
  return "".join(lowerCamelCase__				)


lowercase_       :						Optional[int]     					=	re.compile(R'''^(\s*)#\s*Copied from\s+diffusers\.(\S+\.\S+)\s*($|\S.*$)''')
lowercase_       :						Dict     					=	re.compile(R'''^\s*(\S+)->(\S+)(\s+.*|$)''')
lowercase_       :						Optional[int]     					=	re.compile(R'''<FILL\s+[^>]*>''')





def    _lowerCAmelCase    (     lowerCamelCase__							:							List[str]				)  ->    Dict:
  _SCREAMING_SNAKE_CASE   :				Tuple							    = code.split("\n"				)
  _SCREAMING_SNAKE_CASE   :				List[str]							    = 0
  while idx < len(lowerCamelCase__				) and len(lines[idx]				) == 0:
    idx += 1
  if idx < len(lowerCamelCase__				):
    return re.search(R"^(\s*)\S",						lines[idx]				).groups()[0]
  return ""





def    _lowerCAmelCase    (     lowerCamelCase__							:							Any				)  ->    Optional[int]:
  _SCREAMING_SNAKE_CASE   :				int							    = len(get_indent(lowerCamelCase__				)				) > 0
  if has_indent:
    _SCREAMING_SNAKE_CASE   :				List[Any]							    = f'''class Bla:\n{code}'''
  _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = black.Mode(target_versions={black.TargetVersion.PYaa},						line_length=1_1_9,						preview=lowerCamelCase__				)
  _SCREAMING_SNAKE_CASE   :				Dict							    = black.format_str(lowerCamelCase__,						mode=lowerCamelCase__				)
  _SCREAMING_SNAKE_CASE   :				Any							    = style_docstrings_in_code(lowerCamelCase__				)
  return result[len("class Bla:\n"				) :] if has_indent else result





def    _lowerCAmelCase    (     lowerCamelCase__							:							Any,						lowerCamelCase__							:							int=False				)  ->    List[str]:
  with open(lowerCamelCase__,						"r",						encoding="utf-8",						newline="\n"				) as f:
    _SCREAMING_SNAKE_CASE   :				str							    = f.readlines()
  _SCREAMING_SNAKE_CASE   :				str							    = []
  _SCREAMING_SNAKE_CASE   :				int							    = 0
  # Not a for loop cause `lines` is going to change (if `overwrite=True`).
  while line_index < len(lowerCamelCase__				):
    _SCREAMING_SNAKE_CASE   :				int							    = _re_copy_warning.search(lines[line_index]				)
    if search is None:
      line_index += 1
      continue

    # There is some copied code here, let's retrieve the original.
    _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = search.groups()
    _SCREAMING_SNAKE_CASE   :				List[str]							    = find_code_in_diffusers(lowerCamelCase__				)
    _SCREAMING_SNAKE_CASE   :				List[str]							    = get_indent(lowerCamelCase__				)

    _SCREAMING_SNAKE_CASE   :				List[Any]							    = line_index + 1 if indent == theoretical_indent else line_index + 2
    _SCREAMING_SNAKE_CASE   :				List[str]							    = theoretical_indent
    _SCREAMING_SNAKE_CASE   :				int							    = start_index

    # Loop to check the observed code, stop when indentation diminishes or if we see a End copy comment.
    _SCREAMING_SNAKE_CASE   :				Any							    = True
    while line_index < len(lowerCamelCase__				) and should_continue:
      line_index += 1
      if line_index >= len(lowerCamelCase__				):
        break
      _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = lines[line_index]
      _SCREAMING_SNAKE_CASE   :				List[Any]							    = _should_continue(lowerCamelCase__,						lowerCamelCase__				) and re.search(f'''^{indent}# End copy''',						lowerCamelCase__				) is None
    # Clean up empty lines at the end (if any).
    while len(lines[line_index - 1]				) <= 1:
      line_index -= 1

    _SCREAMING_SNAKE_CASE   :				Any							    = lines[start_index:line_index]
    _SCREAMING_SNAKE_CASE   :				int							    = "".join(lowerCamelCase__				)

    # Remove any nested `Copied from` comments to avoid circular copies
    _SCREAMING_SNAKE_CASE   :				Optional[int]							    = [line for line in theoretical_code.split("\n"				) if _re_copy_warning.search(lowerCamelCase__				) is None]
    _SCREAMING_SNAKE_CASE   :				List[Any]							    = "\n".join(lowerCamelCase__				)

    # Before comparing, use the `replace_pattern` on the original code.
    if len(lowerCamelCase__				) > 0:
      _SCREAMING_SNAKE_CASE   :				Any							    = replace_pattern.replace("with",						""				).split(","				)
      _SCREAMING_SNAKE_CASE   :				int							    = [_re_replace_pattern.search(lowerCamelCase__				) for p in patterns]
      for pattern in patterns:
        if pattern is None:
          continue
        _SCREAMING_SNAKE_CASE   :				Dict							    = pattern.groups()
        _SCREAMING_SNAKE_CASE   :				Optional[int]							    = re.sub(lowerCamelCase__,						lowerCamelCase__,						lowerCamelCase__				)
        if option.strip() == "all-casing":
          _SCREAMING_SNAKE_CASE   :				str							    = re.sub(obja.lower(),						obja.lower(),						lowerCamelCase__				)
          _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = re.sub(obja.upper(),						obja.upper(),						lowerCamelCase__				)

            # Blackify after replacement. To be able to do that, we need the header (class or function definition)
            # from the previous line
      _SCREAMING_SNAKE_CASE   :				List[Any]							    = blackify(lines[start_index - 1] + theoretical_code				)
      _SCREAMING_SNAKE_CASE   :				Tuple							    = theoretical_code[len(lines[start_index - 1]				) :]

    # Test for a diff and act accordingly.
    if observed_code != theoretical_code:
      diffs.append([object_name, start_index]				)
      if overwrite:
        _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = lines[:start_index] + [theoretical_code] + lines[line_index:]
        _SCREAMING_SNAKE_CASE   :				Optional[int]							    = start_index + 1

  if overwrite and len(lowerCamelCase__				) > 0:
    # Warn the user a file has been modified.
    print(f'''Detected changes, rewriting {filename}.'''				)
    with open(lowerCamelCase__,						"w",						encoding="utf-8",						newline="\n"				) as f:
      f.writelines(lowerCamelCase__				)
  return diffs





def    _lowerCAmelCase    (     lowerCamelCase__							:							bool = False				)  ->    Any:
  _SCREAMING_SNAKE_CASE   :				Optional[int]							    = glob.glob(os.path.join(lowerCamelCase__,						"**/*.py"				),						recursive=lowerCamelCase__				)
  _SCREAMING_SNAKE_CASE   :				Optional[int]							    = []
  for filename in all_files:
    _SCREAMING_SNAKE_CASE   :				int							    = is_copy_consistent(lowerCamelCase__,						lowerCamelCase__				)
    diffs += [f'''- {filename}: copy does not match {d[0]} at line {d[1]}''' for d in new_diffs]
  if not overwrite and len(lowerCamelCase__				) > 0:
    _SCREAMING_SNAKE_CASE   :				Tuple							    = "\n".join(lowerCamelCase__				)
    raise Exception(
        "Found the following copy inconsistencies:\n"
        + diff
        + "\nRun `make fix-copies` or `python utils/check_copies.py --fix_and_overwrite` to fix them."				)


if __name__ == "__main__":
      lowercase_       :						int     					=	argparse.ArgumentParser()
      parser.add_argument('''--fix_and_overwrite''', action='''store_true''', help='''Whether to fix inconsistencies.''')
      lowercase_       :						List[Any]     					=	parser.parse_args()

      check_copies(args.fix_and_overwrite)





"""simple docstring"""
import gc
import random
import unittest

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

import diffusers
from diffusers import (
    AutoencoderKL,
    EulerDiscreteScheduler,
    StableDiffusionLatentUpscalePipeline,
    StableDiffusionPipeline,
    UNetaDConditionModel,
)
from diffusers.schedulers import KarrasDiffusionSchedulers
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()





def    _lowerCAmelCase    (     lowerCamelCase__							:							Optional[Any]				)  ->    Optional[Any]:
  _SCREAMING_SNAKE_CASE   :				Any							    = [tensor.shape for tensor in tensor_list]
  return all(shape == shapes[0] for shape in shapes[1:]				)
class 			UpperCamelCase      (							__SCREAMING_SNAKE_CASE      ,       __SCREAMING_SNAKE_CASE      ,       __SCREAMING_SNAKE_CASE      ,       unittest.TestCase  ):
     A__      =			StableDiffusionLatentUpscalePipeline
     A__      =			TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
         """height""",
         """width""",
         """cross_attention_kwargs""",
         """negative_prompt_embeds""",
         """prompt_embeds""",
     }
     A__      =			PipelineTesterMixin.required_optional_params - {"""num_images_per_prompt"""}
     A__      =			TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
     A__      =			frozenset(
         []  )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
     A__      =			frozenset([]  )

     A__      =			True
     @property
     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       _SCREAMING_SNAKE_CASE   :				List[str]							    = 1
       _SCREAMING_SNAKE_CASE   :				int							    = 4
       _SCREAMING_SNAKE_CASE   :				str							    = (16, 16)

       _SCREAMING_SNAKE_CASE   :				Optional[int]							    = floats_tensor((batch_size, num_channels) + sizes  ,     rng=random.Random(0				)				).to(snake_case__				)
       return image
     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       torch.manual_seed(0				)
       _SCREAMING_SNAKE_CASE   :				List[str]							    = UNetaDConditionModel(
           act_fn="gelu"  ,     attention_head_dim=8  ,     norm_num_groups=snake_case__  ,     block_out_channels=[32, 32, 64, 64]  ,     time_cond_proj_dim=160  ,     conv_in_kernel=1  ,     conv_out_kernel=1  ,     cross_attention_dim=32  ,     down_block_types=(
               "KDownBlock2D",
               "KCrossAttnDownBlock2D",
               "KCrossAttnDownBlock2D",
               "KCrossAttnDownBlock2D",
           )  ,     in_channels=8  ,     mid_block_type=snake_case__  ,     only_cross_attention=snake_case__  ,     out_channels=5  ,     resnet_time_scale_shift="scale_shift"  ,     time_embedding_type="fourier"  ,     timestep_post_act="gelu"  ,     up_block_types=("KCrossAttnUpBlock2D", "KCrossAttnUpBlock2D", "KCrossAttnUpBlock2D", "KUpBlock2D")  ,     )
       _SCREAMING_SNAKE_CASE   :				Any							    = AutoencoderKL(
           block_out_channels=[32, 32, 64, 64]  ,     in_channels=3  ,     out_channels=3  ,     down_block_types=[
               "DownEncoderBlock2D",
               "DownEncoderBlock2D",
               "DownEncoderBlock2D",
               "DownEncoderBlock2D",
           ]  ,     up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D", "UpDecoderBlock2D", "UpDecoderBlock2D"]  ,     latent_channels=4  ,     )
       _SCREAMING_SNAKE_CASE   :				Optional[int]							    = EulerDiscreteScheduler(prediction_type="sample"				)
       _SCREAMING_SNAKE_CASE   :				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  ,     hidden_act="quick_gelu"  ,     projection_dim=512  ,     )
       _SCREAMING_SNAKE_CASE   :				List[Any]							    = CLIPTextModel(snake_case__				)
       _SCREAMING_SNAKE_CASE   :				str							    = CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"				)

       _SCREAMING_SNAKE_CASE   :				int							    = {
           "unet": model.eval(),
           "vae": vae.eval(),
           "scheduler": scheduler,
           "text_encoder": text_encoder,
           "tokenizer": tokenizer,
       }

       return components
     def 				__SCREAMING_SNAKE_CASE							(     self  ,     snake_case__  ,     snake_case__=0				):




       """simple docstring"""


       if str(snake_case__				).startswith("mps"				):
         _SCREAMING_SNAKE_CASE   :				Optional[int]							    = torch.manual_seed(snake_case__				)
       else:
         _SCREAMING_SNAKE_CASE   :				Dict							    = torch.Generator(device=snake_case__				).manual_seed(snake_case__				)
       _SCREAMING_SNAKE_CASE   :				int							    = {
           "prompt": "A painting of a squirrel eating a burger",
           "image": self.dummy_image.cpu(),
           "generator": generator,
           "num_inference_steps": 2,
           "output_type": "numpy",
       }
       return inputs
     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       _SCREAMING_SNAKE_CASE   :				str							    = "cpu"

       _SCREAMING_SNAKE_CASE   :				Any							    = self.get_dummy_components()
       _SCREAMING_SNAKE_CASE   :				str							    = self.pipeline_class(**snake_case__				)
       pipe.to(snake_case__				)
       pipe.set_progress_bar_config(disable=snake_case__				)

       _SCREAMING_SNAKE_CASE   :				str							    = self.get_dummy_inputs(snake_case__				)
       _SCREAMING_SNAKE_CASE   :				Tuple							    = pipe(**snake_case__				).images
       _SCREAMING_SNAKE_CASE   :				str							    = image[0, -3:, -3:, -1]

       self.assertEqual(image.shape  ,     (1, 256, 256, 3)				)
       _SCREAMING_SNAKE_CASE   :				Tuple							    = np.array(
           [0.47_222_412, 0.41_921_633, 0.44_717_434, 0.46_874_192, 0.42_588_258, 0.46_150_726, 0.4_677_534, 0.45_583_832, 0.48_579_055]				)
       _SCREAMING_SNAKE_CASE   :				Tuple							    = np.abs(image_slice.flatten() - expected_slice				).max()
       self.assertLessEqual(snake_case__  ,     1E-3				)
     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       super().test_attention_slicing_forward_pass(expected_max_diff=7E-3				)
     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       super().test_cpu_offload_forward_pass(expected_max_diff=3E-3				)
     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       super().test_dict_tuple_outputs_equivalent(expected_max_difference=3E-3				)
     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       super().test_inference_batch_single_identical(expected_max_diff=7E-3				)
     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       super().test_pt_np_pil_outputs_equivalent(expected_max_diff=3E-3				)
     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       super().test_save_load_local(expected_max_difference=3E-3				)
     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       super().test_save_load_optional_components(expected_max_difference=3E-3				)




     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       _SCREAMING_SNAKE_CASE   :				Any							    = [
           "DDIMScheduler",
           "DDPMScheduler",
           "PNDMScheduler",
           "HeunDiscreteScheduler",
           "EulerAncestralDiscreteScheduler",
           "KDPM2DiscreteScheduler",
           "KDPM2AncestralDiscreteScheduler",
           "DPMSolverSDEScheduler",
       ]
       _SCREAMING_SNAKE_CASE   :				Any							    = self.get_dummy_components()
       _SCREAMING_SNAKE_CASE   :				Dict							    = self.pipeline_class(**snake_case__				)

       # make sure that PNDM does not need warm-up
       pipe.scheduler.register_to_config(skip_prk_steps=snake_case__				)

       pipe.to(snake_case__				)
       pipe.set_progress_bar_config(disable=snake_case__				)
       _SCREAMING_SNAKE_CASE   :				str							    = self.get_dummy_inputs(snake_case__				)
       _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = 2

       _SCREAMING_SNAKE_CASE   :				Optional[int]							    = []
       for scheduler_enum in KarrasDiffusionSchedulers:
         if scheduler_enum.name in skip_schedulers:
           # no sigma schedulers are not supported
           # no schedulers
           continue

         _SCREAMING_SNAKE_CASE   :				List[str]							    = getattr(snake_case__  ,     scheduler_enum.name				)
         _SCREAMING_SNAKE_CASE   :				Dict							    = scheduler_cls.from_config(pipe.scheduler.config				)
         _SCREAMING_SNAKE_CASE   :				int							    = pipe(**snake_case__				)[0]
         outputs.append(snake_case__				)

       assert check_same_shape(snake_case__				)
@require_torch_gpu
@slow
class 			UpperCamelCase      (							unittest.TestCase  ):
     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()
     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       _SCREAMING_SNAKE_CASE   :				Tuple							    = torch.manual_seed(33				)

       _SCREAMING_SNAKE_CASE   :				str							    = StableDiffusionPipeline.from_pretrained("CompVis/stable-diffusion-v1-4"  ,     torch_dtype=torch.floataa				)
       pipe.to("cuda"				)

       _SCREAMING_SNAKE_CASE   :				int							    = StableDiffusionLatentUpscalePipeline.from_pretrained(
           "stabilityai/sd-x2-latent-upscaler"  ,     torch_dtype=torch.floataa				)
       upscaler.to("cuda"				)

       _SCREAMING_SNAKE_CASE   :				str							    = "a photo of an astronaut high resolution, unreal engine, ultra realistic"

       _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = pipe(snake_case__  ,     generator=snake_case__  ,     output_type="latent"				).images

       _SCREAMING_SNAKE_CASE   :				Tuple							    = upscaler(
           prompt=snake_case__  ,     image=snake_case__  ,     num_inference_steps=20  ,     guidance_scale=0  ,     generator=snake_case__  ,     output_type="np"  ,     ).images[0]

       _SCREAMING_SNAKE_CASE   :				Optional[int]							    = load_numpy(
           "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/astronaut_1024.npy"				)
       assert np.abs((expected_image - image).mean()				) < 5E-2




     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       _SCREAMING_SNAKE_CASE   :				int							    = torch.manual_seed(33				)

       _SCREAMING_SNAKE_CASE   :				str							    = StableDiffusionLatentUpscalePipeline.from_pretrained(
           "stabilityai/sd-x2-latent-upscaler"  ,     torch_dtype=torch.floataa				)
       upscaler.to("cuda"				)

       _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = "the temple of fire by Ross Tran and Gerardo Dottori, oil on canvas"

       _SCREAMING_SNAKE_CASE   :				List[str]							    = load_image(
           "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/fire_temple_512.png"				)

       _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = upscaler(
           prompt=snake_case__  ,     image=snake_case__  ,     num_inference_steps=20  ,     guidance_scale=0  ,     generator=snake_case__  ,     output_type="np"  ,     ).images[0]

       _SCREAMING_SNAKE_CASE   :				int							    = load_numpy(
           "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/fire_temple_1024.npy"				)
       assert np.abs((expected_image - image).max()				) < 5E-2



