
'''simple docstring'''

import gc
import random
import tempfile
import unittest

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

from diffusers import (
    AutoencoderKL,
    ControlNetModel,
    DDIMScheduler,
    StableDiffusionControlNetImgaImgPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion_controlnet import MultiControlNetModel
from diffusers.utils import floats_tensor, load_image, load_numpy, randn_tensor, slow, torch_device
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
)


enable_full_determinism()


class  _UpperCAmelCase ( lowerCAmelCase__	,lowerCAmelCase__	,lowerCAmelCase__	,unittest.TestCase    ):

     """simple docstring"""

     a_    	=					StableDiffusionControlNetImgaImgPipeline
     a_    	=					TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"height", "width"}
     a_    	=					TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
     a_    	=					IMAGE_TO_IMAGE_IMAGE_PARAMS.union({"control_image"}    )
     a_    	=					IMAGE_TO_IMAGE_IMAGE_PARAMS


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       torch.manual_seed(0 )
       a_      :	Optional[int]      =							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							,		)
       torch.manual_seed(0 )
       a_      :	Any      =							ControlNetModel(
           block_out_channels=(32, 64)							,		layers_per_block=2							,		in_channels=4							,		down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")							,		cross_attention_dim=32							,		conditioning_embedding_out_channels=(16, 32)							,		)
       torch.manual_seed(0 )
       a_      :	List[str]      =							DDIMScheduler(
           beta_start=0.00085							,		beta_end=0.012							,		beta_schedule="""scaled_linear"""							,		clip_sample=lowerCAmelCase_							,		set_alpha_to_one=lowerCAmelCase_							,		)
       torch.manual_seed(0 )
       a_      :	List[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_      :	Dict      =							CLIPTextConfig(
           bos_token_id=0							,		eos_token_id=2							,		hidden_size=32							,		intermediate_size=37							,		layer_norm_eps=1E-0_5							,		num_attention_heads=4							,		num_hidden_layers=5							,		pad_token_id=1							,		vocab_size=10_00							,		)
       a_      :	Union[str, Any]      =							CLIPTextModel(lowerCAmelCase_ )
       a_      :	int      =							CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""" )

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


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_=0 ):

       '''simple docstring'''

       if str(lowerCAmelCase_ ).startswith("""mps""" ):
         a_      :	int      =							torch.manual_seed(lowerCAmelCase_ )
       else:
         a_      :	List[Any]      =							torch.Generator(device=lowerCAmelCase_ ).manual_seed(lowerCAmelCase_ )

       a_      :	Dict      =							2
       a_      :	int      =							randn_tensor(
           (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)							,		generator=lowerCAmelCase_							,		device=torch.device(lowerCAmelCase_ )							,		)
       a_      :	Dict      =							floats_tensor(control_image.shape							,		rng=random.Random(lowerCAmelCase_ ) ).to(lowerCAmelCase_ )
       a_      :	List[Any]      =							image.cpu().permute(0							,		2							,		3							,		1 )[0]
       a_      :	List[Any]      =							Image.fromarray(np.uinta(lowerCAmelCase_ ) ).convert("""RGB""" ).resize((64, 64) )
       a_      :	Union[str, Any]      =							{
           """prompt""": """A painting of a squirrel eating a burger""",
           """generator""": generator,
           """num_inference_steps""": 2,
           """guidance_scale""": 6.0,
           """output_type""": """numpy""",
           """image""": image,
           """control_image""": control_image,
       }

       return inputs


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       return self._test_attention_slicing_forward_pass(expected_max_diff=2E-3 )


     @unittest.skipIf(
         torch_device != """cuda""" or not is_xformers_available()							,		reason="""XFormers attention is only available with CUDA and `xformers` installed"""							,		)
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2E-3 )






     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       self._test_inference_batch_single_identical(expected_max_diff=2E-3 )







class  _UpperCAmelCase ( lowerCAmelCase__	,lowerCAmelCase__	,unittest.TestCase    ):

     """simple docstring"""

     a_    	=					StableDiffusionControlNetImgaImgPipeline
     a_    	=					TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"height", "width"}
     a_    	=					TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
     a_    	=					frozenset([]    )  # TO_DO: add image_params once refactored VaeImageProcessor.preprocess


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       torch.manual_seed(0 )
       a_      :	List[Any]      =							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							,		)
       torch.manual_seed(0 )

       def init_weights(lowerCAmelCase_ ):
         if isinstance(lowerCAmelCase_							,		torch.nn.Convad ):
           torch.nn.init.normal(m.weight )
           m.bias.data.fill_(1.0 )

       a_      :	Tuple      =							ControlNetModel(
           block_out_channels=(32, 64)							,		layers_per_block=2							,		in_channels=4							,		down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")							,		cross_attention_dim=32							,		conditioning_embedding_out_channels=(16, 32)							,		)
       controlneta.controlnet_down_blocks.apply(lowerCAmelCase_ )

       torch.manual_seed(0 )
       a_      :	List[str]      =							ControlNetModel(
           block_out_channels=(32, 64)							,		layers_per_block=2							,		in_channels=4							,		down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")							,		cross_attention_dim=32							,		conditioning_embedding_out_channels=(16, 32)							,		)
       controlneta.controlnet_down_blocks.apply(lowerCAmelCase_ )

       torch.manual_seed(0 )
       a_      :	Union[str, Any]      =							DDIMScheduler(
           beta_start=0.00085							,		beta_end=0.012							,		beta_schedule="""scaled_linear"""							,		clip_sample=lowerCAmelCase_							,		set_alpha_to_one=lowerCAmelCase_							,		)
       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_      :	List[Any]      =							CLIPTextConfig(
           bos_token_id=0							,		eos_token_id=2							,		hidden_size=32							,		intermediate_size=37							,		layer_norm_eps=1E-0_5							,		num_attention_heads=4							,		num_hidden_layers=5							,		pad_token_id=1							,		vocab_size=10_00							,		)
       a_      :	Optional[int]      =							CLIPTextModel(lowerCAmelCase_ )
       a_      :	Any      =							CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""" )

       a_      :	Tuple      =							MultiControlNetModel([controlneta, controlneta] )

       a_      :	Any      =							{
           """unet""": unet,
           """controlnet""": controlnet,
           """scheduler""": scheduler,
           """vae""": vae,
           """text_encoder""": text_encoder,
           """tokenizer""": tokenizer,
           """safety_checker""": None,
           """feature_extractor""": None,
       }
       return components


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_=0 ):

       '''simple docstring'''

       if str(lowerCAmelCase_ ).startswith("""mps""" ):
         a_      :	Any      =							torch.manual_seed(lowerCAmelCase_ )
       else:
         a_      :	Optional[Any]      =							torch.Generator(device=lowerCAmelCase_ ).manual_seed(lowerCAmelCase_ )

       a_      :	Optional[Any]      =							2

       a_      :	int      =							[
           randn_tensor(
               (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)							,		generator=lowerCAmelCase_							,		device=torch.device(lowerCAmelCase_ )							,		),
           randn_tensor(
               (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)							,		generator=lowerCAmelCase_							,		device=torch.device(lowerCAmelCase_ )							,		),
       ]

       a_      :	Tuple      =							floats_tensor(control_image[0].shape							,		rng=random.Random(lowerCAmelCase_ ) ).to(lowerCAmelCase_ )
       a_      :	Optional[Any]      =							image.cpu().permute(0							,		2							,		3							,		1 )[0]
       a_      :	List[Any]      =							Image.fromarray(np.uinta(lowerCAmelCase_ ) ).convert("""RGB""" ).resize((64, 64) )
       a_      :	Union[str, Any]      =							{
           """prompt""": """A painting of a squirrel eating a burger""",
           """generator""": generator,
           """num_inference_steps""": 2,
           """guidance_scale""": 6.0,
           """output_type""": """numpy""",
           """image""": image,
           """control_image""": control_image,
       }

       return inputs


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	List[Any]      =							self.get_dummy_components()
       a_      :	List[str]      =							self.pipeline_class(**lowerCAmelCase_ )
       pipe.to(lowerCAmelCase_ )

       a_      :	Optional[int]      =							10.0
       a_      :	List[str]      =							4

       a_      :	List[str]      =							self.get_dummy_inputs(lowerCAmelCase_ )
       a_      :	Optional[int]      =							steps
       a_      :	Any      =							scale
       a_      :	Optional[Any]      =							pipe(**lowerCAmelCase_ )[0]

       a_      :	int      =							self.get_dummy_inputs(lowerCAmelCase_ )
       a_      :	Dict      =							steps
       a_      :	Optional[Any]      =							scale
       a_      :	Optional[Any]      =							pipe(**lowerCAmelCase_							,		control_guidance_start=0.1							,		control_guidance_end=0.2 )[0]

       a_      :	Optional[int]      =							self.get_dummy_inputs(lowerCAmelCase_ )
       a_      :	Union[str, Any]      =							steps
       a_      :	List[Any]      =							scale
       a_      :	List[Any]      =							pipe(**lowerCAmelCase_							,		control_guidance_start=[0.1, 0.3]							,		control_guidance_end=[0.2, 0.7] )[0]

       a_      :	List[str]      =							self.get_dummy_inputs(lowerCAmelCase_ )
       a_      :	Optional[Any]      =							steps
       a_      :	str      =							scale
       a_      :	Optional[Any]      =							pipe(**lowerCAmelCase_							,		control_guidance_start=0.4							,		control_guidance_end=[0.5, 0.8] )[0]

       # make sure that all outputs are different
       assert np.sum(np.abs(output_a - output_a ) ) > 1E-3
       assert np.sum(np.abs(output_a - output_a ) ) > 1E-3
       assert np.sum(np.abs(output_a - output_a ) ) > 1E-3


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       return self._test_attention_slicing_forward_pass(expected_max_diff=2E-3 )


     @unittest.skipIf(
         torch_device != """cuda""" or not is_xformers_available()							,		reason="""XFormers attention is only available with CUDA and `xformers` installed"""							,		)
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2E-3 )


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       self._test_inference_batch_single_identical(expected_max_diff=2E-3 )






     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	int      =							self.get_dummy_components()
       a_      :	Optional[Any]      =							self.pipeline_class(**lowerCAmelCase_ )
       pipe.to(lowerCAmelCase_ )
       pipe.set_progress_bar_config(disable=lowerCAmelCase_ )
       with tempfile.TemporaryDirectory() as tmpdir:
         try:
           # save_pretrained is not implemented for Multi-ControlNet
           pipe.save_pretrained(lowerCAmelCase_ )
         except NotImplementedError:
           pass







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

     """simple docstring"""

     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

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






     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	List[str]      =							ControlNetModel.from_pretrained("""lllyasviel/sd-controlnet-canny""" )

       a_      :	int      =							StableDiffusionControlNetImgaImgPipeline.from_pretrained(
           """runwayml/stable-diffusion-v1-5"""							,		safety_checker=lowerCAmelCase_							,		controlnet=lowerCAmelCase_ )
       pipe.enable_model_cpu_offload()
       pipe.set_progress_bar_config(disable=lowerCAmelCase_ )

       a_      :	Optional[int]      =							torch.Generator(device="""cpu""" ).manual_seed(0 )
       a_      :	Dict      =							"""evil space-punk bird"""
       a_      :	Union[str, Any]      =							load_image(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/bird_canny.png""" ).resize((5_12, 5_12) )
       a_      :	Union[str, Any]      =							load_image(
           """https://huggingface.co/lllyasviel/sd-controlnet-canny/resolve/main/images/bird.png""" ).resize((5_12, 5_12) )

       a_      :	Optional[int]      =							pipe(
           lowerCAmelCase_							,		lowerCAmelCase_							,		control_image=lowerCAmelCase_							,		generator=lowerCAmelCase_							,		output_type="""np"""							,		num_inference_steps=50							,		strength=0.6							,		)

       a_      :	Any      =							output.images[0]

       assert image.shape == (5_12, 5_12, 3)

       a_      :	List[str]      =							load_numpy(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/img2img.npy""" )

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




'''simple docstring'''

from datasets.utils.patching import _PatchedModuleObj, patch_submodule

from . import _test_patching






def 	_snake_case						(      ):
  """simple docstring"""







  import os as original_os
  from os import path as original_path
  from os import rename as original_rename
  from os.path import dirname as original_dirname
  from os.path import join as original_join

  assert _test_patching.os is original_os
  assert _test_patching.path is original_path
  assert _test_patching.join is original_join

  assert _test_patching.renamed_os is original_os
  assert _test_patching.renamed_path is original_path
  assert _test_patching.renamed_join is original_join

  a_      :	List[str]      =							"""__test_patch_submodule_mock__"""
  with patch_submodule(_test_patching     ,		"""os.path.join"""     ,		A_   ):
    # Every way to access os.path.join must be patched, and the rest must stay untouched

    # check os.path.join
    assert isinstance(_test_patching.os     ,		_PatchedModuleObj   )
    assert isinstance(_test_patching.os.path     ,		_PatchedModuleObj   )
    assert _test_patching.os.path.join is mock

    # check path.join
    assert isinstance(_test_patching.path     ,		_PatchedModuleObj   )
    assert _test_patching.path.join is mock

    # check join
    assert _test_patching.join is mock

    # check that the other attributes are untouched
    assert _test_patching.os.rename is original_rename
    assert _test_patching.path.dirname is original_dirname
    assert _test_patching.os.path.dirname is original_dirname

    # Even renamed modules or objects must be patched

    # check renamed_os.path.join
    assert isinstance(_test_patching.renamed_os     ,		_PatchedModuleObj   )
    assert isinstance(_test_patching.renamed_os.path     ,		_PatchedModuleObj   )
    assert _test_patching.renamed_os.path.join is mock

    # check renamed_path.join
    assert isinstance(_test_patching.renamed_path     ,		_PatchedModuleObj   )
    assert _test_patching.renamed_path.join is mock

    # check renamed_join
    assert _test_patching.renamed_join is mock

    # check that the other attributes are untouched
    assert _test_patching.renamed_os.rename is original_rename
    assert _test_patching.renamed_path.dirname is original_dirname
    assert _test_patching.renamed_os.path.dirname is original_dirname

  # check that everthing is back to normal when the patch is over

  assert _test_patching.os is original_os
  assert _test_patching.path is original_path
  assert _test_patching.join is original_join

  assert _test_patching.renamed_os is original_os
  assert _test_patching.renamed_path is original_path
  assert _test_patching.renamed_join is original_join






def 	_snake_case						(      ):
  """simple docstring"""







  assert _test_patching.open is open

  a_      :	Tuple      =							"""__test_patch_submodule_builtin_mock__"""
  # _test_patching has "open" in its globals
  assert _test_patching.open is open
  with patch_submodule(_test_patching     ,		"""open"""     ,		A_   ):
    assert _test_patching.open is mock

  # check that everthing is back to normal when the patch is over

  assert _test_patching.open is open






def 	_snake_case						(      ):
  """simple docstring"""







  a_      :	Any      =							"""__test_patch_submodule_missing_mock__"""
  with patch_submodule(_test_patching     ,		"""pandas.read_csv"""     ,		A_   ):
    pass






def 	_snake_case						(      ):
  """simple docstring"""







  a_      :	int      =							"""__test_patch_submodule_missing_builtin_mock__"""
  # _test_patching doesn't have "len" in its globals
  assert getattr(_test_patching     ,		"""len"""     ,		A_   ) is None
  with patch_submodule(_test_patching     ,		"""len"""     ,		A_   ):
    assert _test_patching.len is mock
  assert _test_patching.len is len






def 	_snake_case						(      ):
  """simple docstring"""







  a_      :	List[str]      =							"""__test_patch_submodule_start_and_stop_mock__"""
  a_      :	int      =							patch_submodule(_test_patching     ,		"""open"""     ,		A_   )
  assert _test_patching.open is open
  patch.start()
  assert _test_patching.open is mock
  patch.stop()
  assert _test_patching.open is open






def 	_snake_case						(      ):
  """simple docstring"""







  from os import rename as original_rename
  from os.path import dirname as original_dirname
  from os.path import join as original_join

  a_      :	Any      =							"""__test_patch_submodule_successive_join__"""
  a_      :	Optional[Any]      =							"""__test_patch_submodule_successive_dirname__"""
  a_      :	Optional[int]      =							"""__test_patch_submodule_successive_rename__"""
  assert _test_patching.os.path.join is original_join
  assert _test_patching.os.path.dirname is original_dirname
  assert _test_patching.os.rename is original_rename

  with patch_submodule(_test_patching     ,		"""os.path.join"""     ,		A_   ):
    with patch_submodule(_test_patching     ,		"""os.rename"""     ,		A_   ):
      with patch_submodule(_test_patching     ,		"""os.path.dirname"""     ,		A_   ):
        assert _test_patching.os.path.join is mock_join
        assert _test_patching.os.path.dirname is mock_dirname
        assert _test_patching.os.rename is mock_rename

    # try another order
  with patch_submodule(_test_patching     ,		"""os.rename"""     ,		A_   ):
    with patch_submodule(_test_patching     ,		"""os.path.join"""     ,		A_   ):
      with patch_submodule(_test_patching     ,		"""os.path.dirname"""     ,		A_   ):
        assert _test_patching.os.path.join is mock_join
        assert _test_patching.os.path.dirname is mock_dirname
        assert _test_patching.os.rename is mock_rename

  assert _test_patching.os.path.join is original_join
  assert _test_patching.os.path.dirname is original_dirname
  assert _test_patching.os.rename is original_rename






def 	_snake_case						(      ):
  """simple docstring"""







  a_      :	Optional[int]      =							"""__test_patch_submodule_doesnt_exist_mock__"""
  with patch_submodule(_test_patching     ,		"""__module_that_doesn_exist__.__attribute_that_doesn_exist__"""     ,		A_   ):
    pass
  with patch_submodule(_test_patching     ,		"""os.__attribute_that_doesn_exist__"""     ,		A_   ):
    pass


