


"""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 						_A  (  _a    ,_a    ,_a    ,unittest.TestCase  ):







    """simple docstring"""




    UpperCAmelCase       :							Union[str, Any]	=       StableDiffusionControlNetImgaImgPipeline
    UpperCAmelCase       :							List[str]	=       TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""height""", """width"""}
    UpperCAmelCase       :							str	=       TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
    UpperCAmelCase       :							Union[str, Any]	=       IMAGE_TO_IMAGE_IMAGE_PARAMS.union({"""control_image"""}  )
    UpperCAmelCase       :							Dict	=       IMAGE_TO_IMAGE_IMAGE_PARAMS






    def 	__snake_case							(					self				:    Tuple):
      torch.manual_seed(0)
      a						:		Tuple      							=			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						:		Dict      							=			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						:		Union[str, Any]      							=			DDIMScheduler(
          beta_start=0.00_085				, beta_end=0.012				, beta_schedule="scaled_linear"				, clip_sample=__UpperCAmelCase				, set_alpha_to_one=__UpperCAmelCase				, )
      torch.manual_seed(0)
      a						:		Tuple      							=			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						:		Union[str, Any]      							=			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						:		str      							=			CLIPTextModel(__UpperCAmelCase)
      a						:		List[str]      							=			CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip")

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






    def 	__snake_case							(					self				:    Union[str, Any]				, __UpperCAmelCase				:    Optional[int]				, __UpperCAmelCase				:    Dict=0):
      if str(__UpperCAmelCase).startswith("mps"):
        a						:		str      							=			torch.manual_seed(__UpperCAmelCase)
      else:
        a						:		Union[str, Any]      							=			torch.Generator(device=__UpperCAmelCase).manual_seed(__UpperCAmelCase)

      a						:		List[Any]      							=			2
      a						:		List[str]      							=			randn_tensor(
          (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)				, generator=__UpperCAmelCase				, device=torch.device(__UpperCAmelCase)				, )
      a						:		List[str]      							=			floats_tensor(control_image.shape				, rng=random.Random(__UpperCAmelCase)).to(__UpperCAmelCase)
      a						:		Dict      							=			image.cpu().permute(0				, 2				, 3				, 1)[0]
      a						:		List[Any]      							=			Image.fromarray(np.uinta(__UpperCAmelCase)).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 	__snake_case							(					self				:    str):
      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 	__snake_case							(					self				:    Optional[int]):
      self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3)






    def 	__snake_case							(					self				:    str):
      self._test_inference_batch_single_identical(expected_max_diff=2e-3)






class 						_A  (  _a    ,_a    ,unittest.TestCase  ):







    """simple docstring"""




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






    def 	__snake_case							(					self				:    Dict):
      torch.manual_seed(0)
      a						:		str      							=			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(__UpperCAmelCase				:    Dict):
        if isinstance(__UpperCAmelCase				, torch.nn.Convad):
          torch.nn.init.normal(m.weight)
          m.bias.data.fill_(1.0)

      a						:		Union[str, 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)				, )
      controlneta.controlnet_down_blocks.apply(__UpperCAmelCase)

      torch.manual_seed(0)
      a						:		Optional[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)				, )
      controlneta.controlnet_down_blocks.apply(__UpperCAmelCase)

      torch.manual_seed(0)
      a						:		Optional[Any]      							=			DDIMScheduler(
          beta_start=0.00_085				, beta_end=0.012				, beta_schedule="scaled_linear"				, clip_sample=__UpperCAmelCase				, set_alpha_to_one=__UpperCAmelCase				, )
      torch.manual_seed(0)
      a						:		Any      							=			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						:		Union[str, Any]      							=			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						:		Optional[Any]      							=			CLIPTextModel(__UpperCAmelCase)
      a						:		str      							=			CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip")

      a						:		Any      							=			MultiControlNetModel([controlneta, controlneta])

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






    def 	__snake_case							(					self				:    Tuple				, __UpperCAmelCase				:    Optional[Any]				, __UpperCAmelCase				:    Union[str, Any]=0):
      if str(__UpperCAmelCase).startswith("mps"):
        a						:		Dict      							=			torch.manual_seed(__UpperCAmelCase)
      else:
        a						:		Dict      							=			torch.Generator(device=__UpperCAmelCase).manual_seed(__UpperCAmelCase)

      a						:		List[Any]      							=			2

      a						:		Optional[int]      							=			[
          randn_tensor(
              (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)				, generator=__UpperCAmelCase				, device=torch.device(__UpperCAmelCase)				, ),
          randn_tensor(
              (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)				, generator=__UpperCAmelCase				, device=torch.device(__UpperCAmelCase)				, ),
      ]

      a						:		Optional[int]      							=			floats_tensor(control_image[0].shape				, rng=random.Random(__UpperCAmelCase)).to(__UpperCAmelCase)
      a						:		str      							=			image.cpu().permute(0				, 2				, 3				, 1)[0]
      a						:		List[str]      							=			Image.fromarray(np.uinta(__UpperCAmelCase)).convert("RGB").resize((64, 64))
      a						:		Tuple      							=			{
          "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 	__snake_case							(					self				:    Optional[int]):
      a						:		Tuple      							=			self.get_dummy_components()
      a						:		Union[str, Any]      							=			self.pipeline_class(**__UpperCAmelCase)
      pipe.to(__UpperCAmelCase)

      a						:		int      							=			10.0
      a						:		Union[str, Any]      							=			4

      a						:		List[Any]      							=			self.get_dummy_inputs(__UpperCAmelCase)
      a						:		Any      							=			steps
      a						:		Tuple      							=			scale
      a						:		Dict      							=			pipe(**__UpperCAmelCase)[0]

      a						:		Union[str, Any]      							=			self.get_dummy_inputs(__UpperCAmelCase)
      a						:		int      							=			steps
      a						:		Optional[Any]      							=			scale
      a						:		int      							=			pipe(**__UpperCAmelCase				, control_guidance_start=0.1				, control_guidance_end=0.2)[0]

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

      a						:		str      							=			self.get_dummy_inputs(__UpperCAmelCase)
      a						:		Any      							=			steps
      a						:		Any      							=			scale
      a						:		List[str]      							=			pipe(**__UpperCAmelCase				, 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 	__snake_case							(					self				:    Tuple):
      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 	__snake_case							(					self				:    Optional[Any]):
      self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3)






    def 	__snake_case							(					self				:    List[str]):
      self._test_inference_batch_single_identical(expected_max_diff=2e-3)






    def 	__snake_case							(					self				:    Union[str, Any]):
      a						:		List[Any]      							=			self.get_dummy_components()
      a						:		Optional[Any]      							=			self.pipeline_class(**__UpperCAmelCase)
      pipe.to(__UpperCAmelCase)
      pipe.set_progress_bar_config(disable=__UpperCAmelCase)
      with tempfile.TemporaryDirectory() as tmpdir:
        try:
          # save_pretrained is not implemented for Multi-ControlNet
          pipe.save_pretrained(__UpperCAmelCase)
        except NotImplementedError:
          pass






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







    """simple docstring"""




    def 	__snake_case							(					self				:    List[str]):
      super().tearDown()
      gc.collect()
      torch.cuda.empty_cache()






    def 	__snake_case							(					self				:    Union[str, Any]):
      a						:		Optional[int]      							=			ControlNetModel.from_pretrained("lllyasviel/sd-controlnet-canny")

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

      a						:		List[str]      							=			torch.Generator(device="cpu").manual_seed(0)
      a						:		List[str]      							=			"evil space-punk bird"
      a						:		Tuple      							=			load_image(
          "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/bird_canny.png").resize((512, 512))
      a						:		int      							=			load_image(
          "https://huggingface.co/lllyasviel/sd-controlnet-canny/resolve/main/images/bird.png").resize((512, 512))

      a						:		Tuple      							=			pipe(
          __UpperCAmelCase				, __UpperCAmelCase				, control_image=__UpperCAmelCase				, generator=__UpperCAmelCase				, output_type="np"				, num_inference_steps=50				, strength=0.6				, )

      a						:		Optional[Any]      							=			output.images[0]

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

      a						:		Dict      							=			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 __future__ import annotations






class 						_A  :







    """simple docstring"""




    def __init__(					self				:    List[str]				, __UpperCAmelCase				:    int = 0):
      a						:		Tuple      							=			key






    def 	__snake_case							(					self				:    List[str]				, __UpperCAmelCase				:    str				, __UpperCAmelCase				:    int):
      assert isinstance(__UpperCAmelCase				, __UpperCAmelCase) and isinstance(__UpperCAmelCase				, __UpperCAmelCase)

      a						:		Dict      							=			key or self.__key or 1

      # make sure key is an appropriate size
      key %= 255

      return [chr(ord(__UpperCAmelCase) ^ key) for ch in content]






    def 	__snake_case							(					self				:    int				, __UpperCAmelCase				:    str				, __UpperCAmelCase				:    int):
      assert isinstance(__UpperCAmelCase				, __UpperCAmelCase) and isinstance(__UpperCAmelCase				, __UpperCAmelCase)

      a						:		Optional[Any]      							=			key or self.__key or 1

      # make sure key is an appropriate size
      key %= 255

      return [chr(ord(__UpperCAmelCase) ^ key) for ch in content]






    def 	__snake_case							(					self				:    Optional[Any]				, __UpperCAmelCase				:    str				, __UpperCAmelCase				:    int = 0):
      assert isinstance(__UpperCAmelCase				, __UpperCAmelCase) and isinstance(__UpperCAmelCase				, __UpperCAmelCase)

      a						:		List[Any]      							=			key or self.__key or 1

      # make sure key can be any size
      while key > 255:
        key -= 255

      # This will be returned
      a						:		Any      							=			""

      for ch in content:
        ans += chr(ord(__UpperCAmelCase) ^ key)

      return ans






    def 	__snake_case							(					self				:    List[str]				, __UpperCAmelCase				:    str				, __UpperCAmelCase				:    int = 0):
      assert isinstance(__UpperCAmelCase				, __UpperCAmelCase) and isinstance(__UpperCAmelCase				, __UpperCAmelCase)

      a						:		Dict      							=			key or self.__key or 1

      # make sure key can be any size
      while key > 255:
        key -= 255

      # This will be returned
      a						:		str      							=			""

      for ch in content:
        ans += chr(ord(__UpperCAmelCase) ^ key)

      return ans






    def 	__snake_case							(					self				:    int				, __UpperCAmelCase				:    str				, __UpperCAmelCase				:    int = 0):
      assert isinstance(__UpperCAmelCase				, __UpperCAmelCase) and isinstance(__UpperCAmelCase				, __UpperCAmelCase)

      try:
        with open(__UpperCAmelCase) as fin, open("encrypt.out"				, "w+") as fout:
          # actual encrypt-process
          for line in fin:
            fout.write(self.encrypt_string(__UpperCAmelCase				, __UpperCAmelCase))

      except OSError:
        return False

      return True






    def 	__snake_case							(					self				:    Any				, __UpperCAmelCase				:    str				, __UpperCAmelCase				:    int):
      assert isinstance(__UpperCAmelCase				, __UpperCAmelCase) and isinstance(__UpperCAmelCase				, __UpperCAmelCase)

      try:
        with open(__UpperCAmelCase) as fin, open("decrypt.out"				, "w+") as fout:
          # actual encrypt-process
          for line in fin:
            fout.write(self.decrypt_string(__UpperCAmelCase				, __UpperCAmelCase))

      except OSError:
        return False

      return True


# Tests
# crypt = XORCipher()
# key = 67

# # test encrypt
# print(crypt.encrypt("hallo welt",key))
# # test decrypt
# print(crypt.decrypt(crypt.encrypt("hallo welt",key), key))

# # test encrypt_string
# print(crypt.encrypt_string("hallo welt",key))

# # test decrypt_string
# print(crypt.decrypt_string(crypt.encrypt_string("hallo welt",key),key))

# if (crypt.encrypt_file("test.txt",key)):
#       print("encrypt successful")
# else:
#       print("encrypt unsuccessful")

# if (crypt.decrypt_file("encrypt.out",key)):
#       print("decrypt successful")
# else:
#       print("decrypt unsuccessful")






