code: 
  
  
  '''simple docstring'''
  
  
  
  
  
  import argparse
  import os
  from functools import reduce
  
  import fairseq
  import torch
  from datasets import load_dataset
  
  from transformers import WavaVecaProcessor, logging
  from transformers.models.dataavec.configuration_dataavec_audio import DataaVecAudioConfig
  
  # Copied from https://github.com/pytorch/fairseq/blob/main/examples/data2vec/models/data2vec_audio.py
  from transformers.models.dataavec.dataavec_audio import DataaVecAudioModel as Dummy  # noqa: F401
  from transformers.models.dataavec.modeling_dataavec_audio import DataaVecAudioForCTC, DataaVecAudioModel
  
  
  logging.set_verbosity_info()
  UpperCAmelCase__      =							logging.get_logger(__name__)
  
  UpperCAmelCase__      =							{
      '''post_extract_proj''': '''feature_projection.projection''',
      '''models.0.layer_norm''': '''feature_projection.layer_norm''',
      '''self_attn.k_proj''': '''encoder.layers.*.attention.k_proj''',
      '''self_attn.v_proj''': '''encoder.layers.*.attention.v_proj''',
      '''self_attn.q_proj''': '''encoder.layers.*.attention.q_proj''',
      '''self_attn.out_proj''': '''encoder.layers.*.attention.out_proj''',
      '''self_attn_layer_norm''': '''encoder.layers.*.layer_norm''',
      '''fc1''': '''encoder.layers.*.feed_forward.intermediate_dense''',
      '''fc2''': '''encoder.layers.*.feed_forward.output_dense''',
      '''final_layer_norm''': '''encoder.layers.*.final_layer_norm''',
      '''encoder.layer_norm''': '''encoder.layer_norm''',
      '''w2v_model.layer_norm''': '''feature_projection.layer_norm''',
      '''w2v_encoder.proj''': '''lm_head''',
      '''mask_emb''': '''masked_spec_embed''',
  }
  UpperCAmelCase__      =							[
      '''lm_head''',
  ]
  
  
  
  
  
  
  def 	UpperCAmelCase__(							_SCREAMING_SNAKE_CASE				:   Dict,_SCREAMING_SNAKE_CASE				:   List[str],_SCREAMING_SNAKE_CASE				:   Any,_SCREAMING_SNAKE_CASE				:   Optional[Any],_SCREAMING_SNAKE_CASE				:   Optional[Any]   ):
  
  
  
  
  
  
  			"""simple docstring"""
  
  
  
  
  			for attribute in key.split('.'   ):
  						__A=  getattr(_SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE   )
  
  			if weight_type is not None:
  						__A=  getattr(_SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE   ).shape
  			else:
  						__A=  hf_pointer.shape
  
  			if hf_shape != value.shape:
  						raise ValueError(
  						    f"""Shape of hf {key + "." + weight_type if weight_type is not None else ""} is {hf_shape}, but should be"""
  						    f""" {value.shape} for {full_name}"""   )
  
  			if weight_type == "weight":
  						__A=  value
  			elif weight_type == "weight_g":
  						__A=  value
  			elif weight_type == "weight_v":
  						__A=  value
  			elif weight_type == "bias":
  						__A=  value
  			else:
  						__A=  value
  
  			logger.info(f"""{key + "." + weight_type if weight_type is not None else ""} was initialized from {full_name}."""   )
  
  
  
  
  
  
  def 	UpperCAmelCase__(							_SCREAMING_SNAKE_CASE				:   Optional[Any],_SCREAMING_SNAKE_CASE				:   Dict,_SCREAMING_SNAKE_CASE				:   int   ):
  
  
  
  
  
  
  			"""simple docstring"""
  
  
  
  
  			__A=  []
  			__A=  fairseq_model.state_dict()
  
  			if not is_headless:
  						__A=  hf_model.dataavec_audio.feature_extractor
  						__A=  hf_model.dataavec_audio.encoder.pos_conv_embed
  
  			else:
  						__A=  hf_model.feature_extractor
  						__A=  hf_model.encoder.pos_conv_embed
  
  			for name, value in fairseq_dict.items():
  						__A=  False
  						if "conv_layers" in name:
  									load_conv_layer(
  									    _SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE,)
  									__A=  True
  						elif "pos_conv" in name:
  									load_pos_conv_layer(
  									    _SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE,)
  									__A=  True
  						else:
  									for key, mapped_key in MAPPING.items():
  												if not is_headless:
  															__A=  'data2vec_audio.' + mapped_key if mapped_key not in TOP_LEVEL_KEYS else mapped_key
  												if key in name or key.split('w2v_model.'   )[-1] == name.split('.'   )[0]:
  															__A=  True
  															if "*" in mapped_key:
  																		__A=  name.split(_SCREAMING_SNAKE_CASE   )[0].split('.'   )[-2]
  																		__A=  mapped_key.replace('*',_SCREAMING_SNAKE_CASE   )
  															if "weight_g" in name:
  																		__A=  'weight_g'
  															elif "weight_v" in name:
  																		__A=  'weight_v'
  															elif "bias" in name:
  																		__A=  'bias'
  															elif "weight" in name:
  																		# TODO: don't match quantizer.weight_proj
  																		__A=  'weight'
  															else:
  																		__A=  None
  															set_recursively(_SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE   )
  												continue
  						if not is_used:
  									unused_weights.append(_SCREAMING_SNAKE_CASE   )
  
  			logger.warning(f"""Unused weights: {unused_weights}"""   )
  
  
  
  
  
  
  def 	UpperCAmelCase__(							_SCREAMING_SNAKE_CASE				:   Any,_SCREAMING_SNAKE_CASE				:   List[Any]   ):
  
  
  
  
  
  
  			"""simple docstring"""
  
  
  
  
  			__A=  path.split('.'   )
  			return reduce(_SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE   )
  
  
  
  
  
  
  def 	UpperCAmelCase__(							_SCREAMING_SNAKE_CASE				:   Optional[Any],_SCREAMING_SNAKE_CASE				:   Optional[Any],_SCREAMING_SNAKE_CASE				:   str,_SCREAMING_SNAKE_CASE				:   int   ):
  
  
  
  
  
  
  			"""simple docstring"""
  
  
  
  
  			__A=  access_by_string(_SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE   )
  			__A=  hf_weight.data
  
  			if fsq_value.shape != hf_value.shape:
  						raise ValueError(f"""{full_name} has size {fsq_value.shape}, but {hf_value.shape} was found."""   )
  			__A=  fsq_value
  			logger.info(f"""{full_name} was correctly initialized from {hf_weight_path}."""   )
  
  
  
  
  
  
  def 	UpperCAmelCase__(							_SCREAMING_SNAKE_CASE				:   Any,_SCREAMING_SNAKE_CASE				:   Union[str, Any],_SCREAMING_SNAKE_CASE				:   Optional[Any],_SCREAMING_SNAKE_CASE				:   List[Any]   ):
  
  
  
  
  
  
  			"""simple docstring"""
  
  
  
  
  			__A=  full_name.split('conv_layers.'   )[-1]
  			__A=  name.split('.'   )
  			__A=  int(items[0]   )
  			__A=  int(items[1]   )
  
  			__A=  name.split('.'   )[-1]
  			if type_id == 0:
  						__A=  'conv'
  			elif type_id == 2:
  						__A=  'layer_norm'
  			else:
  						unused_weights.append(_SCREAMING_SNAKE_CASE   )
  						return
  
  			set_weights(_SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE,f"""conv_layers.{layer_id}.{layer_type}.{weight_type}"""   )
  
  
  
  
  
  
  def 	UpperCAmelCase__(							_SCREAMING_SNAKE_CASE				:   int,_SCREAMING_SNAKE_CASE				:   Union[str, Any],_SCREAMING_SNAKE_CASE				:   Any,_SCREAMING_SNAKE_CASE				:   List[Any]   ):
  
  
  
  
  
  
  			"""simple docstring"""
  
  
  
  
  			__A=  full_name.split('pos_conv.'   )[-1]
  			__A=  name.split('.'   )
  			__A=  int(items[0]   )
  			__A=  int(items[1]   )
  
  			__A=  name.split('.'   )[-1]
  			if type_id != 0:
  						unused_weights.append(_SCREAMING_SNAKE_CASE   )
  						return
  			else:
  						__A=  'conv'
  
  			set_weights(_SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE,f"""layers.{layer_id}.{layer_type}.{weight_type}"""   )
  
  
  
  
  
  
  @torch.no_grad()
  def 	UpperCAmelCase__(							_SCREAMING_SNAKE_CASE				:   Tuple,_SCREAMING_SNAKE_CASE				:   str,_SCREAMING_SNAKE_CASE				:   List[str]=None,_SCREAMING_SNAKE_CASE				:   Union[str, Any]=None,_SCREAMING_SNAKE_CASE				:   str=True   ):
  
  
  
  
  
  
  			"""simple docstring"""
  
  
  
  
  			if config_path is not None:
  						__A=  DataaVecAudioConfig.from_pretrained(_SCREAMING_SNAKE_CASE   )
  			else:
  						__A=  DataaVecAudioConfig()
  
  			if not is_finetuned:
  						# Modify final_proj layer name
  						__A=  DataaVecAudioModel(_SCREAMING_SNAKE_CASE   )
  						__A=  os.path.dirname(_SCREAMING_SNAKE_CASE   )
  
  						__A=  torch.load(_SCREAMING_SNAKE_CASE   )
  						__A=  state_dict['model'].pop('final_proj.0.weight'   )
  						__A=  state_dict['model'].pop('final_proj.0.bias'   )
  						__A=  os.path.join(_SCREAMING_SNAKE_CASE,'converted.pt'   )
  						torch.save(_SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE   )
  			else:
  						__A=  DataaVecAudioForCTC(_SCREAMING_SNAKE_CASE   )
  						__A=  checkpoint_path
  
  			def load_dataavec(_SCREAMING_SNAKE_CASE				:   Tuple   ):
  						__A,  __A,  __A=  fairseq.checkpoint_utils.load_model_ensemble_and_task([path]   )
  						return model[0].eval()
  
  			__A=  load_dataavec(_SCREAMING_SNAKE_CASE   )
  
  			recursively_load_weights(_SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE,not is_finetuned   )
  
  			__A=  WavaVecaProcessor.from_pretrained('facebook/wav2vec2-large-lv60'   )
  
  			__A=  load_dataset('patrickvonplaten/librispeech_asr_dummy','clean',split='validation'   )
  			__A=  [x['array'] for x in ds[:4]['audio']]
  
  			__A=  processor(_SCREAMING_SNAKE_CASE,return_tensors='pt',padding=_SCREAMING_SNAKE_CASE   )
  
  			__A=  inputs.input_values
  			__A=  inputs.attention_mask
  			#    input_values = inputs.input_values[:, :-1]
  			#    attention_mask = inputs.attention_mask[:, :-1]
  
  			hf_wavavec.eval()
  			model.eval()
  			if is_finetuned:
  						__A=  model(source=_SCREAMING_SNAKE_CASE,padding_mask=(1 - attention_mask),mask=_SCREAMING_SNAKE_CASE,features_only=_SCREAMING_SNAKE_CASE   )[
  						    'encoder_out'
  						].transpose(0,1   )
  						__A=  hf_wavavec(_SCREAMING_SNAKE_CASE,attention_mask=_SCREAMING_SNAKE_CASE   )['logits']
  
  						__A=  torch.argmax(_SCREAMING_SNAKE_CASE,dim=-1   )
  						__A=  processor.batch_decode(_SCREAMING_SNAKE_CASE   )
  
  						print(f"""Expected Output: {ds[:4]["text"]}, Pred: {output_string}"""   )
  			else:
  						__A=  model(source=_SCREAMING_SNAKE_CASE,padding_mask=(1 - attention_mask),mask=_SCREAMING_SNAKE_CASE,features_only=_SCREAMING_SNAKE_CASE   )[
  						    'layer_results'
  						][-1][0].transpose(0,1   )
  						__A=  hf_wavavec(_SCREAMING_SNAKE_CASE,attention_mask=_SCREAMING_SNAKE_CASE   )['last_hidden_state']
  
  			print(our_output.shape,their_output.shape   )
  			__A=  torch.max(torch.abs(our_output - their_output   )   ).item()
  			print(f"""max_absolute_diff = {max_absolute_diff}"""   )  # ~ 1e-7
  			__A=  torch.allclose(_SCREAMING_SNAKE_CASE,_SCREAMING_SNAKE_CASE,atol=1e-3   )
  			print('Do both models output the same tensors?','🔥' if success else '💩'   )
  			if not success:
  						raise Exception('Something went wRoNg'   )
  
  			hf_wavavec.save_pretrained(_SCREAMING_SNAKE_CASE   )
  
  			if is_finetuned:
  						processor.save_pretrained(_SCREAMING_SNAKE_CASE   )
  			else:
  						processor.feature_extractor.save_pretrained(_SCREAMING_SNAKE_CASE   )
  
  
  if __name__ == "__main__":
  					UpperCAmelCase__      =							argparse.ArgumentParser()
  					parser.add_argument('''--pytorch_dump_folder_path''', default=None, type=str, help='''Path to the output PyTorch model.''')
  					parser.add_argument('''--checkpoint_path''', default=None, type=str, help='''Path to fairseq checkpoint''')
  					parser.add_argument('''--dict_path''', default=None, type=str, help='''Path to dict of fine-tuned model''')
  					parser.add_argument('''--config_path''', default=None, type=str, help='''Path to hf config.json of model to convert''')
  					parser.add_argument(
  					    '''--not_finetuned''', action='''store_true''', help='''Whether the model to convert is a fine-tuned model or not'''
  					)
  					UpperCAmelCase__      =							parser.parse_args()
  					convert_wavaveca_checkpoint(
  					    args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.dict_path, not args.not_finetuned
  					)
  
  
  
  
style_context: 
  
  
  '''simple docstring'''
  
  
  
  
  
  # This model implementation is heavily based on:
  
  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,
      StableDiffusionControlNetInpaintPipeline,
      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 (
      TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS,
      TEXT_GUIDED_IMAGE_INPAINTING_PARAMS,
      TEXT_TO_IMAGE_IMAGE_PARAMS,
  )
  from ..test_pipelines_common import (
      PipelineKarrasSchedulerTesterMixin,
      PipelineLatentTesterMixin,
      PipelineTesterMixin,
  )
  
  
  enable_full_determinism()
  class 	a__				(				a_			,     a_			,     a_			,     unittest.TestCase      ):
  
  
  
  
  
  			'''simple docstring'''
  
  			A	:							Optional[int]							= StableDiffusionControlNetInpaintPipeline
  			A	:							List[str]							= TEXT_GUIDED_IMAGE_INPAINTING_PARAMS
  			A	:							List[Any]							= TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
  			A	:							Any							= frozenset({'''control_image'''}      )  # skip `image` and `mask` for now, only test for control_image
  			A	:							List[Any]							= TEXT_TO_IMAGE_IMAGE_PARAMS
  			def 				lowerCAmelCase    (     self       : Tuple  )			->		List[str]:
  						torch.manual_seed(0  )
  						__A=  UNetaDConditionModel(
  						    block_out_channels=(32, 64) ,						layers_per_block=2 ,						sample_size=32 ,						in_channels=9 ,						out_channels=4 ,						down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D') ,						up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D') ,						cross_attention_dim=32 ,						)
  						torch.manual_seed(0  )
  						__A=  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=  DDIMScheduler(
  						    beta_start=0.0_00_85 ,						beta_end=0.0_12 ,						beta_schedule='scaled_linear' ,						clip_sample=lowerCAmelCase_ ,						set_alpha_to_one=lowerCAmelCase_ ,						)
  						torch.manual_seed(0  )
  						__A=  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=  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=1_000 ,						)
  						__A=  CLIPTextModel(lowerCAmelCase_  )
  						__A=  CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'  )
  
  						__A=  {
  						    '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       : List[str] ,						lowerCAmelCase_       : List[Any] ,						lowerCAmelCase_       : int=0  )			->		str:
  						if str(lowerCAmelCase_  ).startswith('mps'  ):
  									__A=  torch.manual_seed(lowerCAmelCase_  )
  						else:
  									__A=  torch.Generator(device=lowerCAmelCase_  ).manual_seed(lowerCAmelCase_  )
  
  						__A=  2
  						__A=  randn_tensor(
  						    (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor) ,						generator=lowerCAmelCase_ ,						device=torch.device(lowerCAmelCase_  ) ,						)
  						__A=  floats_tensor((1, 3, 32, 32) ,						rng=random.Random(lowerCAmelCase_  )  ).to(lowerCAmelCase_  )
  						__A=  init_image.cpu().permute(0 ,						2 ,						3 ,						1  )[0]
  
  						__A=  Image.fromarray(np.uinta(lowerCAmelCase_  )  ).convert('RGB'  ).resize((64, 64)  )
  						__A=  Image.fromarray(np.uinta(init_image + 4  )  ).convert('RGB'  ).resize((64, 64)  )
  
  						__A=  {
  						    'prompt': 'A painting of a squirrel eating a burger',
  						    'generator': generator,
  						    'num_inference_steps': 2,
  						    'guidance_scale': 6.0,
  						    'output_type': 'numpy',
  						    'image': image,
  						    'mask_image': mask_image,
  						    'control_image': control_image,
  						}
  
  						return inputs
  			def 				lowerCAmelCase    (     self       : Tuple  )			->		Optional[int]:
  						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       : Optional[int]  )			->		str:
  						self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2E-3  )
  
  			def 				lowerCAmelCase    (     self       : List[Any]  )			->		str:
  						self._test_inference_batch_single_identical(expected_max_diff=2E-3  )
  
  
  class 	a__				(				a_      ):
  
  
  
  
  
  			'''simple docstring'''
  
  			A	:							int							= StableDiffusionControlNetInpaintPipeline
  			A	:							List[str]							= TEXT_GUIDED_IMAGE_INPAINTING_PARAMS
  			A	:							Tuple							= TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
  			A	:							int							= frozenset([]      )
  			def 				lowerCAmelCase    (     self       : int  )			->		List[Any]:
  						torch.manual_seed(0  )
  						__A=  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=  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=  DDIMScheduler(
  						    beta_start=0.0_00_85 ,						beta_end=0.0_12 ,						beta_schedule='scaled_linear' ,						clip_sample=lowerCAmelCase_ ,						set_alpha_to_one=lowerCAmelCase_ ,						)
  						torch.manual_seed(0  )
  						__A=  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=  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=1_000 ,						)
  						__A=  CLIPTextModel(lowerCAmelCase_  )
  						__A=  CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'  )
  
  						__A=  {
  						    'unet': unet,
  						    'controlnet': controlnet,
  						    'scheduler': scheduler,
  						    'vae': vae,
  						    'text_encoder': text_encoder,
  						    'tokenizer': tokenizer,
  						    'safety_checker': None,
  						    'feature_extractor': None,
  						}
  						return components
  
  
  class 	a__				(				a_			,     a_			,     unittest.TestCase      ):
  
  
  
  
  
  			'''simple docstring'''
  
  			A	:							Tuple							= StableDiffusionControlNetInpaintPipeline
  			A	:							int							= TEXT_GUIDED_IMAGE_INPAINTING_PARAMS
  			A	:							Optional[int]							= TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
  			def 				lowerCAmelCase    (     self       : List[Any]  )			->		int:
  						torch.manual_seed(0  )
  						__A=  UNetaDConditionModel(
  						    block_out_channels=(32, 64) ,						layers_per_block=2 ,						sample_size=32 ,						in_channels=9 ,						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_       : Optional[int]  ):
  									if isinstance(lowerCAmelCase_ ,						torch.nn.Convad  ):
  												torch.nn.init.normal(m.weight  )
  												m.bias.data.fill_(1.0  )
  
  						__A=  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=  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=  DDIMScheduler(
  						    beta_start=0.0_00_85 ,						beta_end=0.0_12 ,						beta_schedule='scaled_linear' ,						clip_sample=lowerCAmelCase_ ,						set_alpha_to_one=lowerCAmelCase_ ,						)
  						torch.manual_seed(0  )
  						__A=  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=  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=1_000 ,						)
  						__A=  CLIPTextModel(lowerCAmelCase_  )
  						__A=  CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'  )
  
  						__A=  MultiControlNetModel([controlneta, controlneta]  )
  
  						__A=  {
  						    '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       : Union[str, Any] ,						lowerCAmelCase_       : str ,						lowerCAmelCase_       : List[Any]=0  )			->		List[Any]:
  						if str(lowerCAmelCase_  ).startswith('mps'  ):
  									__A=  torch.manual_seed(lowerCAmelCase_  )
  						else:
  									__A=  torch.Generator(device=lowerCAmelCase_  ).manual_seed(lowerCAmelCase_  )
  
  						__A=  2
  
  						__A=  [
  						    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=  floats_tensor((1, 3, 32, 32) ,						rng=random.Random(lowerCAmelCase_  )  ).to(lowerCAmelCase_  )
  						__A=  init_image.cpu().permute(0 ,						2 ,						3 ,						1  )[0]
  
  						__A=  Image.fromarray(np.uinta(lowerCAmelCase_  )  ).convert('RGB'  ).resize((64, 64)  )
  						__A=  Image.fromarray(np.uinta(init_image + 4  )  ).convert('RGB'  ).resize((64, 64)  )
  
  						__A=  {
  						    'prompt': 'A painting of a squirrel eating a burger',
  						    'generator': generator,
  						    'num_inference_steps': 2,
  						    'guidance_scale': 6.0,
  						    'output_type': 'numpy',
  						    'image': image,
  						    'mask_image': mask_image,
  						    'control_image': control_image,
  						}
  
  						return inputs
  			def 				lowerCAmelCase    (     self       : Union[str, Any]  )			->		Any:
  						__A=  self.get_dummy_components()
  						__A=  self.pipeline_class(**lowerCAmelCase_  )
  						pipe.to(lowerCAmelCase_  )
  
  						__A=  10.0
  						__A=  4
  
  						__A=  self.get_dummy_inputs(lowerCAmelCase_  )
  						__A=  steps
  						__A=  scale
  						__A=  pipe(**lowerCAmelCase_  )[0]
  
  						__A=  self.get_dummy_inputs(lowerCAmelCase_  )
  						__A=  steps
  						__A=  scale
  						__A=  pipe(**lowerCAmelCase_ ,						control_guidance_start=0.1 ,						control_guidance_end=0.2  )[0]
  
  						__A=  self.get_dummy_inputs(lowerCAmelCase_  )
  						__A=  steps
  						__A=  scale
  						__A=  pipe(**lowerCAmelCase_ ,						control_guidance_start=[0.1, 0.3] ,						control_guidance_end=[0.2, 0.7]  )[0]
  
  						__A=  self.get_dummy_inputs(lowerCAmelCase_  )
  						__A=  steps
  						__A=  scale
  						__A=  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       : str  )			->		Any:
  						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       : Any  )			->		int:
  						self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2E-3  )
  			def 				lowerCAmelCase    (     self       : Dict  )			->		Dict:
  						self._test_inference_batch_single_identical(expected_max_diff=2E-3  )
  
  			def 				lowerCAmelCase    (     self       : Union[str, Any]  )			->		Optional[int]:
  						__A=  self.get_dummy_components()
  						__A=  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 	a__				(				unittest.TestCase      ):
  
  
  
  
  
  			'''simple docstring'''
  
  			def 				lowerCAmelCase    (     self       : Optional[int]  )			->		Tuple:
  						super().tearDown()
  						gc.collect()
  						torch.cuda.empty_cache()
  			def 				lowerCAmelCase    (     self       : Union[str, Any]  )			->		Tuple:
  						__A=  ControlNetModel.from_pretrained('lllyasviel/sd-controlnet-canny'  )
  
  						__A=  StableDiffusionControlNetInpaintPipeline.from_pretrained(
  						    'runwayml/stable-diffusion-inpainting' ,						safety_checker=lowerCAmelCase_ ,						controlnet=lowerCAmelCase_  )
  						pipe.enable_model_cpu_offload()
  						pipe.set_progress_bar_config(disable=lowerCAmelCase_  )
  
  						__A=  torch.Generator(device='cpu'  ).manual_seed(0  )
  						__A=  load_image(
  						    'https://huggingface.co/lllyasviel/sd-controlnet-canny/resolve/main/images/bird.png'  ).resize((512, 512)  )
  
  						__A=  load_image(
  						    'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
  						    '/stable_diffusion_inpaint/input_bench_mask.png'  ).resize((512, 512)  )
  
  						__A=  'pitch black hole'
  
  						__A=  load_image(
  						    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/bird_canny.png'  ).resize((512, 512)  )
  
  						__A=  pipe(
  						    lowerCAmelCase_ ,						image=lowerCAmelCase_ ,						mask_image=lowerCAmelCase_ ,						control_image=lowerCAmelCase_ ,						generator=lowerCAmelCase_ ,						output_type='np' ,						num_inference_steps=3 ,						)
  
  						__A=  output.images[0]
  
  						assert image.shape == (512, 512, 3)
  
  						__A=  load_numpy(
  						    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/inpaint.npy'  )
  
  						assert np.abs(expected_image - image  ).max() < 9E-2
  
  			def 				lowerCAmelCase    (     self       : str  )			->		Tuple:
  						__A=  ControlNetModel.from_pretrained('lllyasviel/control_v11p_sd15_inpaint'  )
  
  						__A=  StableDiffusionControlNetInpaintPipeline.from_pretrained(
  						    'runwayml/stable-diffusion-v1-5' ,						safety_checker=lowerCAmelCase_ ,						controlnet=lowerCAmelCase_  )
  						__A=  DDIMScheduler.from_config(pipe.scheduler.config  )
  						pipe.enable_model_cpu_offload()
  						pipe.set_progress_bar_config(disable=lowerCAmelCase_  )
  
  						__A=  torch.Generator(device='cpu'  ).manual_seed(33  )
  
  						__A=  load_image(
  						    'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/stable_diffusion_inpaint/boy.png'  )
  						__A=  init_image.resize((512, 512)  )
  
  						__A=  load_image(
  						    'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/stable_diffusion_inpaint/boy_mask.png'  )
  						__A=  mask_image.resize((512, 512)  )
  
  						__A=  'a handsome man with ray-ban sunglasses'
  
  						def make_inpaint_condition(lowerCAmelCase_       : int ,						lowerCAmelCase_       : str  ):
  									__A=  np.array(image.convert('RGB'  )  ).astype(np.floataa  ) / 2_55.0
  									__A=  np.array(image_mask.convert('L'  )  ).astype(np.floataa  ) / 2_55.0
  
  									assert image.shape[0:1] == image_mask.shape[0:1], "image and image_mask must have the same image size"
  									__A=  -1.0  # set as masked pixel
  									__A=  np.expand_dims(lowerCAmelCase_ ,						0  ).transpose(0 ,						3 ,						1 ,						2  )
  									__A=  torch.from_numpy(lowerCAmelCase_  )
  									return image
  
  						__A=  make_inpaint_condition(lowerCAmelCase_ ,						lowerCAmelCase_  )
  
  						__A=  pipe(
  						    lowerCAmelCase_ ,						image=lowerCAmelCase_ ,						mask_image=lowerCAmelCase_ ,						control_image=lowerCAmelCase_ ,						guidance_scale=9.0 ,						eta=1.0 ,						generator=lowerCAmelCase_ ,						num_inference_steps=20 ,						output_type='np' ,						)
  						__A=  output.images[0]
  
  						assert image.shape == (512, 512, 3)
  
  						__A=  load_numpy(
  						    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/boy_ray_ban.npy'  )
  
  						assert np.abs(expected_image - image  ).max() < 9E-2
  
  
  
  