
'''simple docstring'''

# This model implementation is heavily inspired by https://github.com/haofanwang/ControlNet-for-Diffusers/

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

def 	_snake_case						(      A_       :    Optional[int]   ):
  """simple docstring"""







  if not head:
    return True
  # split the list to two parts
  a_      , a_      :	Union[str, Any]      =							head.next, head
  while fast and fast.next:
    a_      :	Dict      =							fast.next.next
    a_      :	List[str]      =							slow.next
  a_      :	str      =							slow.next
  a_      :	Optional[int]      =							None  # Don't forget here! But forget still works!
  # reverse the second part
  a_      :	Dict      =							None
  while second:
    a_      :	Any      =							second.next
    a_      :	Any      =							node
    a_      :	Optional[Any]      =							second
    a_      :	Optional[Any]      =							nxt
  # compare two parts
  # second part has the same or one less node
  while node:
    if node.val != head.val:
      return False
    a_      :	List[str]      =							node.next
    a_      :	Dict      =							head.next
  return True






def 	_snake_case						(      A_       :    List[str]   ):
  """simple docstring"""







  if not head or not head.next:
    return True

  # 1. Get the midpoint (slow)
  a_      :	Optional[Any]      =							head
  while fast and fast.next:
    a_      , a_      :	Optional[int]      =							fast.next.next, slow.next

  # 2. Push the second half into the stack
  a_      :	List[str]      =							[slow.val]
  while slow.next:
    a_      :	List[str]      =							slow.next
    stack.append(slow.val   )

  # 3. Comparison
  while stack:
    if stack.pop() != cur.val:
      return False
    a_      :	Union[str, Any]      =							cur.next

  return True






def 	_snake_case						(      A_       :    Optional[int]   ):
  """simple docstring"""







  if not head or not head.next:
    return True
  a_      :	List[Any]      =							{}
  a_      :	int      =							0
  while head:
    if head.val in d:
      d[head.val].append(A_   )
    else:
      a_      :	int      =							[pos]
    a_      :	List[str]      =							head.next
    pos += 1
  a_      :	str      =							pos - 1
  a_      :	List[str]      =							0
  for v in d.values():
    if len(A_   ) % 2 != 0:
      middle += 1
    else:
      a_      :	Optional[int]      =							0
      for i in range(0     ,		len(A_   )   ):
        if v[i] + v[len(A_   ) - 1 - step] != checksum:
          return False
        step += 1
    if middle > 1:
      return False
  return True


