
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 					_SCREAMING_SNAKE_CASE  (      __UpperCamelCase       ,					__UpperCamelCase       ,					__UpperCamelCase       ,					unittest.TestCase ):
      _A						:		str    	=     StableDiffusionControlNetImgaImgPipeline
      _A						:		int    	=     TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'height', 'width'}
      _A						:		List[str]    	=     TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
      _A						:		List[str]    	=     IMAGE_TO_IMAGE_IMAGE_PARAMS.union({'control_image'} )
      _A						:		List[str]    	=     IMAGE_TO_IMAGE_IMAGE_PARAMS
      def A_			(			self ):
          torch.manual_seed(0 )
          snake_case__						=				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 )
          snake_case__						=				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 )
          snake_case__						=				DDIMScheduler(
              beta_start=0.0_0_0_8_5		,	beta_end=0.0_1_2		,	beta_schedule="scaled_linear"		,	clip_sample=UpperCAmelCase__		,	set_alpha_to_one=UpperCAmelCase__		,	)
          torch.manual_seed(0 )
          snake_case__						=				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 )
          snake_case__						=				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=10_00		,	)
          snake_case__						=				CLIPTextModel(UpperCAmelCase__ )
          snake_case__						=				CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip" )

          snake_case__						=				{
              "unet": unet,
              "controlnet": controlnet,
              "scheduler": scheduler,
              "vae": vae,
              "text_encoder": text_encoder,
              "tokenizer": tokenizer,
              "safety_checker": None,
              "feature_extractor": None,
          }
          return components
      def A_			(			self		,	lowerCamelCase		,	lowerCamelCase=0 ):
          if str(UpperCAmelCase__ ).startswith("mps" ):
              snake_case__						=				torch.manual_seed(UpperCAmelCase__ )
          else:
              snake_case__						=				torch.Generator(device=UpperCAmelCase__ ).manual_seed(UpperCAmelCase__ )

          snake_case__						=				2
          snake_case__						=				randn_tensor(
              (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)		,	generator=UpperCAmelCase__		,	device=torch.device(UpperCAmelCase__ )		,	)
          snake_case__						=				floats_tensor(control_image.shape		,	rng=random.Random(UpperCAmelCase__ ) ).to(UpperCAmelCase__ )
          snake_case__						=				image.cpu().permute(0		,	2		,	3		,	1 )[0]
          snake_case__						=				Image.fromarray(np.uinta(UpperCAmelCase__ ) ).convert("RGB" ).resize((64, 64) )
          snake_case__						=				{
              "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 A_			(			self ):
          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 A_			(			self ):
          self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3 )



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





class 					_SCREAMING_SNAKE_CASE  (      __UpperCamelCase       ,					__UpperCamelCase       ,					unittest.TestCase ):
      _A						:		Optional[Any]    	=     StableDiffusionControlNetImgaImgPipeline
      _A						:		Union[str, Any]    	=     TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'height', 'width'}
      _A						:		Tuple    	=     TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
      _A						:		str    	=     frozenset([] )  # TO_DO: add image_params once refactored VaeImageProcessor.preprocess
      def A_			(			self ):
          torch.manual_seed(0 )
          snake_case__						=				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(UpperCAmelCase__		,	torch.nn.Convad ):
                  torch.nn.init.normal(m.weight )
                  m.bias.data.fill_(1.0 )

          snake_case__						=				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 )
          snake_case__						=				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 )
          snake_case__						=				DDIMScheduler(
              beta_start=0.0_0_0_8_5		,	beta_end=0.0_1_2		,	beta_schedule="scaled_linear"		,	clip_sample=UpperCAmelCase__		,	set_alpha_to_one=UpperCAmelCase__		,	)
          torch.manual_seed(0 )
          snake_case__						=				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 )
          snake_case__						=				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=10_00		,	)
          snake_case__						=				CLIPTextModel(UpperCAmelCase__ )
          snake_case__						=				CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip" )

          snake_case__						=				MultiControlNetModel([controlneta, controlneta] )

          snake_case__						=				{
              "unet": unet,
              "controlnet": controlnet,
              "scheduler": scheduler,
              "vae": vae,
              "text_encoder": text_encoder,
              "tokenizer": tokenizer,
              "safety_checker": None,
              "feature_extractor": None,
          }
          return components
      def A_			(			self		,	lowerCamelCase		,	lowerCamelCase=0 ):
          if str(UpperCAmelCase__ ).startswith("mps" ):
              snake_case__						=				torch.manual_seed(UpperCAmelCase__ )
          else:
              snake_case__						=				torch.Generator(device=UpperCAmelCase__ ).manual_seed(UpperCAmelCase__ )

          snake_case__						=				2

          snake_case__						=				[
              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__ )		,	),
          ]

          snake_case__						=				floats_tensor(control_image[0].shape		,	rng=random.Random(UpperCAmelCase__ ) ).to(UpperCAmelCase__ )
          snake_case__						=				image.cpu().permute(0		,	2		,	3		,	1 )[0]
          snake_case__						=				Image.fromarray(np.uinta(UpperCAmelCase__ ) ).convert("RGB" ).resize((64, 64) )
          snake_case__						=				{
              "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 A_			(			self ):
          snake_case__						=				self.get_dummy_components()
          snake_case__						=				self.pipeline_class(**UpperCAmelCase__ )
          pipe.to(UpperCAmelCase__ )

          snake_case__						=				1_0.0
          snake_case__						=				4

          snake_case__						=				self.get_dummy_inputs(UpperCAmelCase__ )
          snake_case__						=				steps
          snake_case__						=				scale
          snake_case__						=				pipe(**UpperCAmelCase__ )[0]

          snake_case__						=				self.get_dummy_inputs(UpperCAmelCase__ )
          snake_case__						=				steps
          snake_case__						=				scale
          snake_case__						=				pipe(**UpperCAmelCase__		,	control_guidance_start=0.1		,	control_guidance_end=0.2 )[0]

          snake_case__						=				self.get_dummy_inputs(UpperCAmelCase__ )
          snake_case__						=				steps
          snake_case__						=				scale
          snake_case__						=				pipe(**UpperCAmelCase__		,	control_guidance_start=[0.1, 0.3]		,	control_guidance_end=[0.2, 0.7] )[0]

          snake_case__						=				self.get_dummy_inputs(UpperCAmelCase__ )
          snake_case__						=				steps
          snake_case__						=				scale
          snake_case__						=				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 A_			(			self ):
          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 A_			(			self ):
          self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3 )
      def A_			(			self ):
          self._test_inference_batch_single_identical(expected_max_diff=2e-3 )



      def A_			(			self ):
          snake_case__						=				self.get_dummy_components()
          snake_case__						=				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 					_SCREAMING_SNAKE_CASE  (      unittest.TestCase ):
      def A_			(			self ):
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()



      def A_			(			self ):
          snake_case__						=				ControlNetModel.from_pretrained("lllyasviel/sd-controlnet-canny" )

          snake_case__						=				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__ )

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

          snake_case__						=				pipe(
              UpperCAmelCase__		,	UpperCAmelCase__		,	control_image=UpperCAmelCase__		,	generator=UpperCAmelCase__		,	output_type="np"		,	num_inference_steps=50		,	strength=0.6		,	)

          snake_case__						=				output.images[0]

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

          snake_case__						=				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





import os
from shutil import copyfile
from typing import List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


__magic_name__            =  logging.get_logger(__name__)

__magic_name__            =  {'''vocab_file''': '''sentencepiece.model'''}

__magic_name__            =  {
    '''vocab_file''': {
        '''google/rembert''': '''https://huggingface.co/google/rembert/resolve/main/sentencepiece.model''',
    },
}

__magic_name__            =  {
    '''google/rembert''': 256,
}




class 					_SCREAMING_SNAKE_CASE  (      __UpperCamelCase ):
      _A						:		Optional[Any]    	=     VOCAB_FILES_NAMES
      _A						:		Optional[int]    	=     PRETRAINED_VOCAB_FILES_MAP
      _A						:		Any    	=     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      def __init__(			self		,	lowerCamelCase		,	lowerCamelCase=False		,	lowerCamelCase=True		,	lowerCamelCase=True		,	lowerCamelCase="[CLS]"		,	lowerCamelCase="[SEP]"		,	lowerCamelCase="[UNK]"		,	lowerCamelCase="[SEP]"		,	lowerCamelCase="[PAD]"		,	lowerCamelCase="[CLS]"		,	lowerCamelCase="[MASK]"		,	**lowerCamelCase		,	):
          super().__init__(
              do_lower_case=lowerCamelCase		,	remove_space=lowerCamelCase		,	keep_accents=lowerCamelCase		,	bos_token=lowerCamelCase		,	eos_token=lowerCamelCase		,	unk_token=lowerCamelCase		,	sep_token=lowerCamelCase		,	pad_token=lowerCamelCase		,	cls_token=lowerCamelCase		,	mask_token=lowerCamelCase		,	**lowerCamelCase		,	)

          snake_case__						=				do_lower_case
          snake_case__						=				remove_space
          snake_case__						=				keep_accents
          snake_case__						=				vocab_file

          snake_case__						=				spm.SentencePieceProcessor()
          self.sp_model.Load(lowerCamelCase )
      @property
      def A_			(			self ):
          return len(self.sp_model )
      def A_			(			self ):
          snake_case__						=				{self.convert_ids_to_tokens(lowerCamelCase ): i for i in range(self.vocab_size )}
          vocab.update(self.added_tokens_encoder )
          return vocab
      def __getstate__(			self ):
          snake_case__						=				self.__dict__.copy()
          snake_case__						=				None
          return state
      def __setstate__(			self		,	lowerCamelCase ):
          snake_case__						=				d
          snake_case__						=				spm.SentencePieceProcessor()
          self.sp_model.Load(self.vocab_file )
      def A_			(			self		,	lowerCamelCase		,	lowerCamelCase=False ):
          snake_case__						=				self.sp_model.EncodeAsPieces(lowerCamelCase )
          return pieces
      def A_			(			self		,	lowerCamelCase ):
          return self.sp_model.PieceToId(lowerCamelCase )
      def A_			(			self		,	lowerCamelCase ):
          return self.sp_model.IdToPiece(lowerCamelCase )
      def A_			(			self		,	lowerCamelCase ):
          snake_case__						=				self.sp_model.decode_pieces(lowerCamelCase )
          return out_string
      def A_			(			self		,	lowerCamelCase		,	lowerCamelCase = None ):
          snake_case__						=				[self.sep_token_id]
          snake_case__						=				[self.cls_token_id]
          if token_ids_a is None:
              return cls + token_ids_a + sep
          return cls + token_ids_a + sep + token_ids_a + sep
      def A_			(			self		,	lowerCamelCase		,	lowerCamelCase = None		,	lowerCamelCase = False ):

          if already_has_special_tokens:
              if token_ids_a is not None:
                  raise ValueError(
                      "You should not supply a second sequence if the provided sequence of "
                      "ids is already formatted with special tokens for the model." )
              return [1 if x in [self.sep_token_id, self.cls_token_id] else 0 for x in token_ids_a]

          if token_ids_a is not None:
              return [1] + ([0] * len(lowerCamelCase )) + [1] + ([0] * len(lowerCamelCase )) + [1]
          return [1] + ([0] * len(lowerCamelCase )) + [1]
      def A_			(			self		,	lowerCamelCase		,	lowerCamelCase = None ):
          snake_case__						=				[self.sep_token_id]
          snake_case__						=				[self.cls_token_id]

          if token_ids_a is None:
              return len(cls + token_ids_a + sep ) * [0]
          return len(cls + token_ids_a + sep ) * [0] + len(token_ids_a + sep ) * [1]



      def A_			(			self		,	lowerCamelCase		,	lowerCamelCase = None ):
          if not os.path.isdir(lowerCamelCase ):
              logger.error("Vocabulary path ({}) should be a directory".format(lowerCamelCase ) )
              return
          snake_case__						=				os.path.join(
              lowerCamelCase		,	(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"] )

          if os.path.abspath(self.vocab_file ) != os.path.abspath(lowerCamelCase ):
              copyfile(self.vocab_file		,	lowerCamelCase )

          return (out_vocab_file,)



