
'''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_				(	snake_case__    ,						snake_case__    ,						snake_case__    ,						unittest.TestCase					):
 UpperCAmelCase_		    =						StableDiffusionControlNetImgaImgPipeline
 UpperCAmelCase_		    =						TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""height""", """width"""}
 UpperCAmelCase_		    =						TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
 UpperCAmelCase_		    =						IMAGE_TO_IMAGE_IMAGE_PARAMS.union({"""control_image"""}					)
 UpperCAmelCase_		    =						IMAGE_TO_IMAGE_IMAGE_PARAMS





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

     snake_case_ :    Optional[int]     			=						{
         "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 ,       lowercase_ ,       lowercase_=0):
     if str(lowercase_).startswith("mps"):
         snake_case_ :    Dict     			=						torch.manual_seed(lowercase_)
     else:
         snake_case_ :    List[str]     			=						torch.Generator(device=lowercase_).manual_seed(lowercase_)

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





 def        snake_case__		(    self):
     self._test_inference_batch_single_identical(expected_max_diff=2E-3)






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





 def        snake_case__		(    self):
     torch.manual_seed(0)
     snake_case_ :    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)

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

     snake_case_ :    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(lowercase_)

     torch.manual_seed(0)
     snake_case_ :    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(lowercase_)

     torch.manual_seed(0)
     snake_case_ :    Optional[Any]     			=						DDIMScheduler(
         beta_start=0.00_085 ,       beta_end=0.012 ,       beta_schedule="scaled_linear" ,       clip_sample=lowercase_ ,       set_alpha_to_one=lowercase_ ,       )
     torch.manual_seed(0)
     snake_case_ :    int     			=						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_ :    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=10_00 ,       )
     snake_case_ :    Any     			=						CLIPTextModel(lowercase_)
     snake_case_ :    Tuple     			=						CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip")

     snake_case_ :    Union[str, Any]     			=						MultiControlNetModel([controlneta, controlneta])

     snake_case_ :    int     			=						{
         "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 ,       lowercase_ ,       lowercase_=0):
     if str(lowercase_).startswith("mps"):
         snake_case_ :    Optional[int]     			=						torch.manual_seed(lowercase_)
     else:
         snake_case_ :    Tuple     			=						torch.Generator(device=lowercase_).manual_seed(lowercase_)

     snake_case_ :    List[Any]     			=						2

     snake_case_ :    Union[str, Any]     			=						[
         randn_tensor(
             (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor) ,       generator=lowercase_ ,       device=torch.device(lowercase_) ,       ),
         randn_tensor(
             (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor) ,       generator=lowercase_ ,       device=torch.device(lowercase_) ,       ),
     ]

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

     snake_case_ :    Tuple     			=						10.0
     snake_case_ :    List[Any]     			=						4

     snake_case_ :    str     			=						self.get_dummy_inputs(lowercase_)
     snake_case_ :    Any     			=						steps
     snake_case_ :    Tuple     			=						scale
     snake_case_ :    Dict     			=						pipe(**lowercase_)[0]

     snake_case_ :    int     			=						self.get_dummy_inputs(lowercase_)
     snake_case_ :    Optional[Any]     			=						steps
     snake_case_ :    Tuple     			=						scale
     snake_case_ :    Optional[int]     			=						pipe(**lowercase_ ,       control_guidance_start=0.1 ,       control_guidance_end=0.2)[0]

     snake_case_ :    int     			=						self.get_dummy_inputs(lowercase_)
     snake_case_ :    str     			=						steps
     snake_case_ :    Dict     			=						scale
     snake_case_ :    int     			=						pipe(**lowercase_ ,       control_guidance_start=[0.1, 0.3] ,       control_guidance_end=[0.2, 0.7])[0]

     snake_case_ :    List[str]     			=						self.get_dummy_inputs(lowercase_)
     snake_case_ :    Tuple     			=						steps
     snake_case_ :    int     			=						scale
     snake_case_ :    Union[str, Any]     			=						pipe(**lowercase_ ,       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):
     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):
     self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2E-3)





 def        snake_case__		(    self):
     self._test_inference_batch_single_identical(expected_max_diff=2E-3)





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






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





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





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

     snake_case_ :    Optional[Any]     			=						StableDiffusionControlNetImgaImgPipeline.from_pretrained(
         "runwayml/stable-diffusion-v1-5" ,       safety_checker=lowercase_ ,       controlnet=lowercase_)
     pipe.enable_model_cpu_offload()
     pipe.set_progress_bar_config(disable=lowercase_)

     snake_case_ :    str     			=						torch.Generator(device="cpu").manual_seed(0)
     snake_case_ :    List[Any]     			=						"evil space-punk bird"
     snake_case_ :    Dict     			=						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_ :    Optional[int]     			=						load_image(
         "https://huggingface.co/lllyasviel/sd-controlnet-canny/resolve/main/images/bird.png").resize((5_12, 5_12))

     snake_case_ :    List[str]     			=						pipe(
         lowercase_ ,       lowercase_ ,       control_image=lowercase_ ,       generator=lowercase_ ,       output_type="np" ,       num_inference_steps=50 ,       strength=0.6 ,       )

     snake_case_ :    str     			=						output.images[0]

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

     snake_case_ :    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'''
import unittest
from pathlib import Path
from tempfile import TemporaryDirectory

from transformers import AutoConfig, TFAutoModel, is_tensorflow_text_available, is_tf_available
from transformers.models.bert.tokenization_bert import BertTokenizer
from transformers.testing_utils import require_tensorflow_text, require_tf, slow


if is_tf_available():
  import tensorflow as tf

if is_tensorflow_text_available():
  from transformers.models.bert import TFBertTokenizer


a_											=						["bert-base-uncased", "bert-base-cased"]
a_											=						"hf-internal-testing/tiny-bert-tf-only"

if is_tf_available():






  class    UpperCAmelCase_				(	tf.keras.Model					):





   def __init__(    self ,       lowercase_):
       super().__init__()
       snake_case_ :    Union[str, Any]     			=						tokenizer
       snake_case_ :    List[str]     			=						AutoConfig.from_pretrained(lowercase_)
       snake_case_ :    List[str]     			=						TFAutoModel.from_config(lowercase_)





   def        snake_case__		(    self ,       lowercase_):
       snake_case_ :    List[str]     			=						self.tokenizer(lowercase_)
       snake_case_ :    str     			=						self.bert(**lowercase_)
       return out["pooler_output"]






@require_tf
@require_tensorflow_text
class    UpperCAmelCase_				(	unittest.TestCase					):





 def        snake_case__		(    self):
     super().setUp()

     snake_case_ :    Dict     			=						[
         BertTokenizer.from_pretrained(lowercase_) for checkpoint in (TOKENIZER_CHECKPOINTS * 2)
     ]  # repeat for when fast_bert_tokenizer=false
     snake_case_ :    Tuple     			=						[TFBertTokenizer.from_pretrained(lowercase_) for checkpoint in TOKENIZER_CHECKPOINTS] + [
         TFBertTokenizer.from_pretrained(lowercase_ ,       use_fast_bert_tokenizer=lowercase_)
         for checkpoint in TOKENIZER_CHECKPOINTS
     ]
     assert len(self.tokenizers) == len(self.tf_tokenizers)

     snake_case_ :    Dict     			=						[
         "This is a straightforward English test sentence.",
         "This one has some weird characters\rto\nsee\r\nif  those\u00E9break things.",
         "Now we're going to add some Chinese: 一 二 三 一二三",
         "And some much more rare Chinese: 齉 堃 齉堃",
         "Je vais aussi écrire en français pour tester les accents",
         "Classical Irish also has some unusual characters, so in they go: Gaelaċ, ꝼ",
     ]
     snake_case_ :    str     			=						list(zip(self.test_sentences ,       self.test_sentences[::-1]))





 def        snake_case__		(    self):
     for tokenizer, tf_tokenizer in zip(self.tokenizers ,       self.tf_tokenizers):
         for test_inputs in (self.test_sentences, self.paired_sentences):
             snake_case_ :    Any     			=						tokenizer(lowercase_ ,       return_tensors="tf" ,       padding="longest")
             snake_case_ :    Union[str, Any]     			=						tf_tokenizer(lowercase_)

             for key in python_outputs.keys():
                 self.assertTrue(tf.reduce_all(python_outputs[key].shape == tf_outputs[key].shape))
                 self.assertTrue(tf.reduce_all(tf.cast(python_outputs[key] ,       tf.intaa) == tf_outputs[key]))





 @slow
 def        snake_case__		(    self):
     for tf_tokenizer in self.tf_tokenizers:
         snake_case_ :    Union[str, Any]     			=						tf_tokenizer(self.paired_sentences)
         snake_case_ :    str     			=						tf_tokenizer(
             text=[sentence[0] for sentence in self.paired_sentences] ,       text_pair=[sentence[1] for sentence in self.paired_sentences] ,       )
         for key in merged_outputs.keys():
             self.assertTrue(tf.reduce_all(tf.cast(merged_outputs[key] ,       tf.intaa) == separated_outputs[key]))





 @slow
 def        snake_case__		(    self):
     for tf_tokenizer in self.tf_tokenizers:
         snake_case_ :    int     			=						tf.function(lowercase_)
         for test_inputs in (self.test_sentences, self.paired_sentences):
             snake_case_ :    Optional[int]     			=						tf.constant(lowercase_)
             snake_case_ :    Tuple     			=						compiled_tokenizer(lowercase_)
             snake_case_ :    Any     			=						tf_tokenizer(lowercase_)

             for key in eager_outputs.keys():
                 self.assertTrue(tf.reduce_all(eager_outputs[key] == compiled_outputs[key]))





 @slow
 def        snake_case__		(    self):
     for tf_tokenizer in self.tf_tokenizers:
         snake_case_ :    List[Any]     			=						ModelToSave(tokenizer=lowercase_)
         snake_case_ :    Any     			=						tf.convert_to_tensor(self.test_sentences)
         snake_case_ :    List[Any]     			=						model(lowercase_)  # Build model with some sample inputs
         with TemporaryDirectory() as tempdir:
             snake_case_ :    Optional[Any]     			=						Path(lowercase_) / "saved.model"
             model.save(lowercase_)
             snake_case_ :    Tuple     			=						tf.keras.models.load_model(lowercase_)
         snake_case_ :    str     			=						loaded_model(lowercase_)
         # We may see small differences because the loaded model is compiled, so we need an epsilon for the test
         self.assertLessEqual(tf.reduce_max(tf.abs(out - loaded_output)) ,       1E-5)




