

import argparse

import torch

from diffusers import UNetaDConditionModel





def    _snake_case		(snake_case_				:    Tuple						,					snake_case_				:    Union[str, Any]						,					snake_case_				:    Union[str, Any]						,					snake_case_				:    Any=None						,					snake_case_				:    Tuple=None						,					snake_case_				:    Optional[int]=None     )       ->		Union[str, Any]:
		assert isinstance(snake_case_						,					snake_case_     ), "Paths should be a list of dicts containing 'old' and 'new' keys."

		# Splits the attention layers into three variables.
		if attention_paths_to_split is not None:
				for path, path_map in attention_paths_to_split.items():
						__UpperCAmelCase  :     Dict		=    old_checkpoint[path]
						__UpperCAmelCase  :     Tuple		=    old_tensor.shape[0] // 3

						__UpperCAmelCase  :     List[str]		=    (-1, channels) if len(old_tensor.shape     ) == 3 else (-1)

						__UpperCAmelCase  :     List[str]		=    old_tensor.shape[0] // config['num_head_channels'] // 3

						__UpperCAmelCase  :     Optional[Any]		=    old_tensor.reshape((num_heads, 3 * channels // num_heads) + old_tensor.shape[1:]     )
						__UpperCAmelCase  :     Optional[int]		=    old_tensor.split(channels // num_heads						,					dim=1     )

						__UpperCAmelCase  :     List[Any]		=    query.reshape(snake_case_     )
						__UpperCAmelCase  :     Tuple		=    key.reshape(snake_case_     )
						__UpperCAmelCase  :     List[str]		=    value.reshape(snake_case_     )

		for path in paths:
				__UpperCAmelCase  :     Tuple		=    path['new']

				# These have already been assigned
				if attention_paths_to_split is not None and new_path in attention_paths_to_split:
						continue

				if additional_replacements is not None:
						for replacement in additional_replacements:
								__UpperCAmelCase  :     Dict		=    new_path.replace(replacement['old']						,					replacement['new']     )

								# proj_attn.weight has to be converted from conv 1D to linear
				__UpperCAmelCase  :     Union[str, Any]		=    old_checkpoint[path['old']]
				__UpperCAmelCase  :     Dict		=    ['proj_attn.weight']
				__UpperCAmelCase  :     Any		=    ['proj_out.weight', 'proj_in.weight']
				if any(k in new_path for k in names     ):
						__UpperCAmelCase  :     Tuple		=    weight[:, :, 0]
				elif any(k in new_path for k in names_a     ) and len(weight.shape     ) > 2 and ".attentions." not in new_path:
						__UpperCAmelCase  :     Optional[Any]		=    weight[:, :, 0]
				else:
						__UpperCAmelCase  :     str		=    weight





def    _snake_case		(snake_case_				:    Optional[Any]						,					snake_case_				:    Union[str, Any]=0     )       ->		Dict:
		__UpperCAmelCase  :     Dict		=    []
		for old_item in old_list:
				__UpperCAmelCase  :     Dict		=    old_item

				#         new_item = new_item.replace('norm.weight', 'group_norm.weight')
				#         new_item = new_item.replace('norm.bias', 'group_norm.bias')

				#         new_item = new_item.replace('proj_out.weight', 'proj_attn.weight')
				#         new_item = new_item.replace('proj_out.bias', 'proj_attn.bias')

				#         new_item = shave_segments(new_item, n_shave_prefix_segments=n_shave_prefix_segments)

				mapping.append({'old': old_item, 'new': new_item}     )

		return mapping





def    _snake_case		(snake_case_				:    List[Any]						,					snake_case_				:    int=1     )       ->		Tuple:
		if n_shave_prefix_segments >= 0:
				return ".".join(path.split('.'     )[n_shave_prefix_segments:]     )
		else:
				return ".".join(path.split('.'     )[:n_shave_prefix_segments]     )





def    _snake_case		(snake_case_				:    str						,					snake_case_				:    Optional[int]=0     )       ->		Optional[Any]:
		__UpperCAmelCase  :     Union[str, Any]		=    []
		for old_item in old_list:
				mapping.append({'old': old_item, 'new': old_item}     )

		return mapping





def    _snake_case		(snake_case_				:    Optional[Any]						,					snake_case_				:    int=0     )       ->		int:
		__UpperCAmelCase  :     Dict		=    []
		for old_item in old_list:
				__UpperCAmelCase  :     Union[str, Any]		=    old_item.replace('in_layers.0'						,					'norm1'     )
				__UpperCAmelCase  :     Tuple		=    new_item.replace('in_layers.2'						,					'conv1'     )

				__UpperCAmelCase  :     Union[str, Any]		=    new_item.replace('out_layers.0'						,					'norm2'     )
				__UpperCAmelCase  :     Dict		=    new_item.replace('out_layers.3'						,					'conv2'     )

				__UpperCAmelCase  :     List[str]		=    new_item.replace('emb_layers.1'						,					'time_emb_proj'     )
				__UpperCAmelCase  :     int		=    new_item.replace('skip_connection'						,					'conv_shortcut'     )

				__UpperCAmelCase  :     Optional[Any]		=    shave_segments(snake_case_						,					n_shave_prefix_segments=snake_case_     )

				if "temopral_conv" not in old_item:
						mapping.append({'old': old_item, 'new': new_item}     )

		return mapping





def    _snake_case		(snake_case_				:    int						,					snake_case_				:    int						,					snake_case_				:    Union[str, Any]=None						,					snake_case_				:    Dict=False     )       ->		Optional[Any]:
		__UpperCAmelCase  :     str		=    {}
		__UpperCAmelCase  :     List[str]		=    list(checkpoint.keys()     )

		__UpperCAmelCase  :     Dict		=    'model.diffusion_model.'

		# at least a 100 parameters have to start with `model_ema` in order for the checkpoint to be EMA
		if sum(k.startswith('model_ema'     ) for k in keys     ) > 100 and extract_ema:
				print(F"Checkpoint {path} has both EMA and non-EMA weights."     )
				print(
				    'In this conversion only the EMA weights are extracted. If you want to instead extract the non-EMA'
				    ' weights (useful to continue fine-tuning), please make sure to remove the `--extract_ema` flag.'     )
				for key in keys:
						if key.startswith('model.diffusion_model'     ):
								__UpperCAmelCase  :     str		=    'model_ema.' + ''.join(key.split('.'     )[1:]     )
								__UpperCAmelCase  :     Optional[int]		=    checkpoint.pop(snake_case_     )
		else:
				if sum(k.startswith('model_ema'     ) for k in keys     ) > 100:
						print(
						    'In this conversion only the non-EMA weights are extracted. If you want to instead extract the EMA'
						    ' weights (usually better for inference), please make sure to add the `--extract_ema` flag.'     )

				for key in keys:
						__UpperCAmelCase  :     int		=    checkpoint.pop(snake_case_     )

		__UpperCAmelCase  :     Dict		=    {}

		__UpperCAmelCase  :     List[str]		=    unet_state_dict['time_embed.0.weight']
		__UpperCAmelCase  :     List[str]		=    unet_state_dict['time_embed.0.bias']
		__UpperCAmelCase  :     List[Any]		=    unet_state_dict['time_embed.2.weight']
		__UpperCAmelCase  :     List[str]		=    unet_state_dict['time_embed.2.bias']

		if config["class_embed_type"] is None:
				# No parameters to port
				...
		elif config["class_embed_type"] == "timestep" or config["class_embed_type"] == "projection":
				__UpperCAmelCase  :     int		=    unet_state_dict['label_emb.0.0.weight']
				__UpperCAmelCase  :     Any		=    unet_state_dict['label_emb.0.0.bias']
				__UpperCAmelCase  :     Optional[Any]		=    unet_state_dict['label_emb.0.2.weight']
				__UpperCAmelCase  :     Optional[Any]		=    unet_state_dict['label_emb.0.2.bias']
		else:
				raise NotImplementedError(F"Not implemented `class_embed_type`: {config['class_embed_type']}"     )

		__UpperCAmelCase  :     Optional[Any]		=    unet_state_dict['input_blocks.0.0.weight']
		__UpperCAmelCase  :     List[Any]		=    unet_state_dict['input_blocks.0.0.bias']

		__UpperCAmelCase  :     Optional[int]		=    [v for v in unet_state_dict if v.startswith('input_blocks.0.1'     )]
		__UpperCAmelCase  :     Tuple		=    renew_attention_paths(snake_case_     )
		__UpperCAmelCase  :     Union[str, Any]		=    {'old': 'input_blocks.0.1', 'new': 'transformer_in'}
		assign_to_checkpoint(snake_case_						,					snake_case_						,					snake_case_						,					additional_replacements=[meta_path]						,					config=snake_case_     )

		__UpperCAmelCase  :     List[Any]		=    unet_state_dict['out.0.weight']
		__UpperCAmelCase  :     int		=    unet_state_dict['out.0.bias']
		__UpperCAmelCase  :     List[Any]		=    unet_state_dict['out.2.weight']
		__UpperCAmelCase  :     Tuple		=    unet_state_dict['out.2.bias']

		# Retrieves the keys for the input blocks only
		__UpperCAmelCase  :     Union[str, Any]		=    len({'.'.join(layer.split('.'     )[:2]     ) for layer in unet_state_dict if 'input_blocks' in layer}     )
		__UpperCAmelCase  :     Optional[Any]		=    {
		    layer_id: [key for key in unet_state_dict if F"input_blocks.{layer_id}" in key]
		    for layer_id in range(snake_case_     )
		}

		# Retrieves the keys for the middle blocks only
		__UpperCAmelCase  :     Any		=    len({'.'.join(layer.split('.'     )[:2]     ) for layer in unet_state_dict if 'middle_block' in layer}     )
		__UpperCAmelCase  :     Optional[Any]		=    {
		    layer_id: [key for key in unet_state_dict if F"middle_block.{layer_id}" in key]
		    for layer_id in range(snake_case_     )
		}

		# Retrieves the keys for the output blocks only
		__UpperCAmelCase  :     Optional[int]		=    len({'.'.join(layer.split('.'     )[:2]     ) for layer in unet_state_dict if 'output_blocks' in layer}     )
		__UpperCAmelCase  :     List[str]		=    {
		    layer_id: [key for key in unet_state_dict if F"output_blocks.{layer_id}" in key]
		    for layer_id in range(snake_case_     )
		}

		for i in range(1						,					snake_case_     ):
				__UpperCAmelCase  :     Optional[int]		=    (i - 1) // (config['layers_per_block'] + 1)
				__UpperCAmelCase  :     Union[str, Any]		=    (i - 1) % (config['layers_per_block'] + 1)

				__UpperCAmelCase  :     Optional[int]		=    [
				    key for key in input_blocks[i] if F"input_blocks.{i}.0" in key and F"input_blocks.{i}.0.op" not in key
				]
				__UpperCAmelCase  :     str		=    [key for key in input_blocks[i] if F"input_blocks.{i}.1" in key]
				__UpperCAmelCase  :     Any		=    [key for key in input_blocks[i] if F"input_blocks.{i}.2" in key]

				if F"input_blocks.{i}.op.weight" in unet_state_dict:
						__UpperCAmelCase  :     Dict		=    unet_state_dict.pop(
						    F"input_blocks.{i}.op.weight"     )
						__UpperCAmelCase  :     Optional[int]		=    unet_state_dict.pop(
						    F"input_blocks.{i}.op.bias"     )

				__UpperCAmelCase  :     Optional[int]		=    renew_resnet_paths(snake_case_     )
				__UpperCAmelCase  :     List[str]		=    {'old': F"input_blocks.{i}.0", 'new': F"down_blocks.{block_id}.resnets.{layer_in_block_id}"}
				assign_to_checkpoint(
				    snake_case_						,					snake_case_						,					snake_case_						,					additional_replacements=[meta_path]						,					config=snake_case_     )

				__UpperCAmelCase  :     List[Any]		=    [key for key in resnets if 'temopral_conv' in key]
				__UpperCAmelCase  :     Any		=    renew_temp_conv_paths(snake_case_     )
				__UpperCAmelCase  :     Tuple		=    {
				    'old': F"input_blocks.{i}.0.temopral_conv",
				    'new': F"down_blocks.{block_id}.temp_convs.{layer_in_block_id}",
				}
				assign_to_checkpoint(
				    snake_case_						,					snake_case_						,					snake_case_						,					additional_replacements=[meta_path]						,					config=snake_case_     )

				if len(snake_case_     ):
						__UpperCAmelCase  :     List[str]		=    renew_attention_paths(snake_case_     )
						__UpperCAmelCase  :     Optional[Any]		=    {'old': F"input_blocks.{i}.1", 'new': F"down_blocks.{block_id}.attentions.{layer_in_block_id}"}
						assign_to_checkpoint(
						    snake_case_						,					snake_case_						,					snake_case_						,					additional_replacements=[meta_path]						,					config=snake_case_     )

				if len(snake_case_     ):
						__UpperCAmelCase  :     Optional[Any]		=    renew_attention_paths(snake_case_     )
						__UpperCAmelCase  :     Tuple		=    {
						    'old': F"input_blocks.{i}.2",
						    'new': F"down_blocks.{block_id}.temp_attentions.{layer_in_block_id}",
						}
						assign_to_checkpoint(
						    snake_case_						,					snake_case_						,					snake_case_						,					additional_replacements=[meta_path]						,					config=snake_case_     )

		__UpperCAmelCase  :     Optional[Any]		=    middle_blocks[0]
		__UpperCAmelCase  :     List[str]		=    [key for key in resnet_a if 'temopral_conv' in key]
		__UpperCAmelCase  :     Optional[int]		=    middle_blocks[1]
		__UpperCAmelCase  :     Tuple		=    middle_blocks[2]
		__UpperCAmelCase  :     int		=    middle_blocks[3]
		__UpperCAmelCase  :     Union[str, Any]		=    [key for key in resnet_a if 'temopral_conv' in key]

		__UpperCAmelCase  :     Optional[Any]		=    renew_resnet_paths(snake_case_     )
		__UpperCAmelCase  :     Optional[int]		=    {'old': 'middle_block.0', 'new': 'mid_block.resnets.0'}
		assign_to_checkpoint(
		    snake_case_						,					snake_case_						,					snake_case_						,					config=snake_case_						,					additional_replacements=[meta_path]     )

		__UpperCAmelCase  :     Union[str, Any]		=    renew_temp_conv_paths(snake_case_     )
		__UpperCAmelCase  :     List[str]		=    {'old': 'middle_block.0.temopral_conv', 'new': 'mid_block.temp_convs.0'}
		assign_to_checkpoint(
		    snake_case_						,					snake_case_						,					snake_case_						,					config=snake_case_						,					additional_replacements=[meta_path]     )

		__UpperCAmelCase  :     Union[str, Any]		=    renew_resnet_paths(snake_case_     )
		__UpperCAmelCase  :     Any		=    {'old': 'middle_block.3', 'new': 'mid_block.resnets.1'}
		assign_to_checkpoint(
		    snake_case_						,					snake_case_						,					snake_case_						,					config=snake_case_						,					additional_replacements=[meta_path]     )

		__UpperCAmelCase  :     int		=    renew_temp_conv_paths(snake_case_     )
		__UpperCAmelCase  :     Optional[int]		=    {'old': 'middle_block.3.temopral_conv', 'new': 'mid_block.temp_convs.1'}
		assign_to_checkpoint(
		    snake_case_						,					snake_case_						,					snake_case_						,					config=snake_case_						,					additional_replacements=[meta_path]     )

		__UpperCAmelCase  :     Optional[Any]		=    renew_attention_paths(snake_case_     )
		__UpperCAmelCase  :     List[str]		=    {'old': 'middle_block.1', 'new': 'mid_block.attentions.0'}
		assign_to_checkpoint(
		    snake_case_						,					snake_case_						,					snake_case_						,					additional_replacements=[meta_path]						,					config=snake_case_     )

		__UpperCAmelCase  :     Optional[Any]		=    renew_attention_paths(snake_case_     )
		__UpperCAmelCase  :     Optional[Any]		=    {'old': 'middle_block.2', 'new': 'mid_block.temp_attentions.0'}
		assign_to_checkpoint(
		    snake_case_						,					snake_case_						,					snake_case_						,					additional_replacements=[meta_path]						,					config=snake_case_     )

		for i in range(snake_case_     ):
				__UpperCAmelCase  :     Union[str, Any]		=    i // (config['layers_per_block'] + 1)
				__UpperCAmelCase  :     List[str]		=    i % (config['layers_per_block'] + 1)
				__UpperCAmelCase  :     Dict		=    [shave_segments(snake_case_						,					2     ) for name in output_blocks[i]]
				__UpperCAmelCase  :     List[Any]		=    {}

				for layer in output_block_layers:
						__UpperCAmelCase  :     str		=    layer.split('.'     )[0], shave_segments(snake_case_						,					1     )
						if layer_id in output_block_list:
								output_block_list[layer_id].append(snake_case_     )
						else:
								__UpperCAmelCase  :     Dict		=    [layer_name]

				if len(snake_case_     ) > 1:
						__UpperCAmelCase  :     List[Any]		=    [key for key in output_blocks[i] if F"output_blocks.{i}.0" in key]
						__UpperCAmelCase  :     Dict		=    [key for key in output_blocks[i] if F"output_blocks.{i}.1" in key]
						__UpperCAmelCase  :     Optional[int]		=    [key for key in output_blocks[i] if F"output_blocks.{i}.2" in key]

						__UpperCAmelCase  :     Optional[Any]		=    renew_resnet_paths(snake_case_     )
						__UpperCAmelCase  :     Any		=    renew_resnet_paths(snake_case_     )

						__UpperCAmelCase  :     Any		=    {'old': F"output_blocks.{i}.0", 'new': F"up_blocks.{block_id}.resnets.{layer_in_block_id}"}
						assign_to_checkpoint(
						    snake_case_						,					snake_case_						,					snake_case_						,					additional_replacements=[meta_path]						,					config=snake_case_     )

						__UpperCAmelCase  :     Union[str, Any]		=    [key for key in resnets if 'temopral_conv' in key]
						__UpperCAmelCase  :     Optional[Any]		=    renew_temp_conv_paths(snake_case_     )
						__UpperCAmelCase  :     Optional[Any]		=    {
						    'old': F"output_blocks.{i}.0.temopral_conv",
						    'new': F"up_blocks.{block_id}.temp_convs.{layer_in_block_id}",
						}
						assign_to_checkpoint(
						    snake_case_						,					snake_case_						,					snake_case_						,					additional_replacements=[meta_path]						,					config=snake_case_     )

						__UpperCAmelCase  :     Optional[int]		=    {k: sorted(snake_case_     ) for k, v in output_block_list.items()}
						if ["conv.bias", "conv.weight"] in output_block_list.values():
								__UpperCAmelCase  :     int		=    list(output_block_list.values()     ).index(['conv.bias', 'conv.weight']     )
								__UpperCAmelCase  :     Tuple		=    unet_state_dict[
								    F"output_blocks.{i}.{index}.conv.weight"
								]
								__UpperCAmelCase  :     List[Any]		=    unet_state_dict[
								    F"output_blocks.{i}.{index}.conv.bias"
								]

								# Clear attentions as they have been attributed above.
								if len(snake_case_     ) == 2:
										__UpperCAmelCase  :     List[str]		=    []

						if len(snake_case_     ):
								__UpperCAmelCase  :     Dict		=    renew_attention_paths(snake_case_     )
								__UpperCAmelCase  :     int		=    {
								    'old': F"output_blocks.{i}.1",
								    'new': F"up_blocks.{block_id}.attentions.{layer_in_block_id}",
								}
								assign_to_checkpoint(
								    snake_case_						,					snake_case_						,					snake_case_						,					additional_replacements=[meta_path]						,					config=snake_case_     )

						if len(snake_case_     ):
								__UpperCAmelCase  :     Tuple		=    renew_attention_paths(snake_case_     )
								__UpperCAmelCase  :     Dict		=    {
								    'old': F"output_blocks.{i}.2",
								    'new': F"up_blocks.{block_id}.temp_attentions.{layer_in_block_id}",
								}
								assign_to_checkpoint(
								    snake_case_						,					snake_case_						,					snake_case_						,					additional_replacements=[meta_path]						,					config=snake_case_     )
				else:
						__UpperCAmelCase  :     int		=    renew_resnet_paths(snake_case_						,					n_shave_prefix_segments=1     )
						for path in resnet_0_paths:
								__UpperCAmelCase  :     Union[str, Any]		=    '.'.join(['output_blocks', str(snake_case_     ), path['old']]     )
								__UpperCAmelCase  :     Union[str, Any]		=    '.'.join(['up_blocks', str(snake_case_     ), 'resnets', str(snake_case_     ), path['new']]     )
								__UpperCAmelCase  :     List[str]		=    unet_state_dict[old_path]

						__UpperCAmelCase  :     Union[str, Any]		=    [l for l in output_block_layers if 'temopral_conv' in l]
						for path in temopral_conv_paths:
								__UpperCAmelCase  :     str		=    path.split('temopral_conv.'     )[-1]
								__UpperCAmelCase  :     str		=    '.'.join(['output_blocks', str(snake_case_     ), str(snake_case_     ), 'temopral_conv', pruned_path]     )
								__UpperCAmelCase  :     Union[str, Any]		=    '.'.join(['up_blocks', str(snake_case_     ), 'temp_convs', str(snake_case_     ), pruned_path]     )
								__UpperCAmelCase  :     str		=    unet_state_dict[old_path]

		return new_checkpoint


if __name__ == "__main__":
						_SCREAMING_SNAKE_CASE          = argparse.ArgumentParser()

						parser.add_argument(
						    '''--checkpoint_path''', default=None, type=str, required=True, help='''Path to the checkpoint to convert.'''
						)
						parser.add_argument('''--dump_path''', default=None, type=str, required=True, help='''Path to the output model.''')
						_SCREAMING_SNAKE_CASE          = parser.parse_args()

						_SCREAMING_SNAKE_CASE          = torch.load(args.checkpoint_path, map_location='''cpu''')
						_SCREAMING_SNAKE_CASE          = UNetaDConditionModel()

						_SCREAMING_SNAKE_CASE          = convert_ldm_unet_checkpoint(unet_checkpoint, unet.config)

						_SCREAMING_SNAKE_CASE          = set(unet.state_dict().keys()) - set(converted_ckpt.keys())
						_SCREAMING_SNAKE_CASE          = set(converted_ckpt.keys()) - set(unet.state_dict().keys())

						assert len(diff_a) == len(diff_a) == 0, "Converted weights don't match"

						# load state_dict
						unet.load_state_dict(converted_ckpt)

						unet.save_pretrained(args.dump_path)

						# -- finish converting the unet --








import gc
import tempfile
import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionDualGuidedPipeline
from diffusers.utils.testing_utils import load_image, nightly, require_torch_gpu, torch_device


_SCREAMING_SNAKE_CASE          = False


@nightly
@require_torch_gpu
class     lowerCamelCase_   ( unittest.TestCase):




 def 						a					( self       ):
   # clean up the VRAM after each test
   super().tearDown()
   gc.collect()
   torch.cuda.empty_cache()




 def 						a					( self       ):
   __UpperCAmelCase  :     Dict		=    VersatileDiffusionDualGuidedPipeline.from_pretrained('shi-labs/versatile-diffusion'       )
   # remove text_unet
   pipe.remove_unused_weights()
   pipe.to(_UpperCAmelCase       )
   pipe.set_progress_bar_config(disable=_UpperCAmelCase       )

   __UpperCAmelCase  :     Any		=    load_image(
       'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg'       )

   __UpperCAmelCase  :     Optional[Any]		=    torch.manual_seed(0       )
   __UpperCAmelCase  :     Union[str, Any]		=    pipe(
       prompt='first prompt'     ,     image=_UpperCAmelCase     ,     text_to_image_strength=0.7_5     ,     generator=_UpperCAmelCase     ,     guidance_scale=7.5     ,     num_inference_steps=2     ,     output_type='numpy'     ,     ).images

   with tempfile.TemporaryDirectory() as tmpdirname:
     pipe.save_pretrained(_UpperCAmelCase       )
     __UpperCAmelCase  :     List[str]		=    VersatileDiffusionDualGuidedPipeline.from_pretrained(_UpperCAmelCase       )

   pipe.to(_UpperCAmelCase       )
   pipe.set_progress_bar_config(disable=_UpperCAmelCase       )

   __UpperCAmelCase  :     List[str]		=    generator.manual_seed(0       )
   __UpperCAmelCase  :     Optional[int]		=    pipe(
       prompt='first prompt'     ,     image=_UpperCAmelCase     ,     text_to_image_strength=0.7_5     ,     generator=_UpperCAmelCase     ,     guidance_scale=7.5     ,     num_inference_steps=2     ,     output_type='numpy'     ,     ).images

   assert np.abs(image - new_image       ).sum() < 1E-5, "Models don't have the same forward pass"




 def 						a					( self       ):
   __UpperCAmelCase  :     Optional[Any]		=    VersatileDiffusionDualGuidedPipeline.from_pretrained('shi-labs/versatile-diffusion'       )
   pipe.remove_unused_weights()
   pipe.to(_UpperCAmelCase       )
   pipe.set_progress_bar_config(disable=_UpperCAmelCase       )

   __UpperCAmelCase  :     Tuple		=    'cyberpunk 2077'
   __UpperCAmelCase  :     Dict		=    load_image(
       'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg'       )
   __UpperCAmelCase  :     Optional[Any]		=    torch.manual_seed(0       )
   __UpperCAmelCase  :     List[Any]		=    pipe(
       prompt=_UpperCAmelCase     ,     image=_UpperCAmelCase     ,     text_to_image_strength=0.7_5     ,     generator=_UpperCAmelCase     ,     guidance_scale=7.5     ,     num_inference_steps=50     ,     output_type='numpy'     ,     ).images

   __UpperCAmelCase  :     Tuple		=    image[0, 253:256, 253:256, -1]

   assert image.shape == (1, 512, 512, 3)
   __UpperCAmelCase  :     Tuple		=    np.array([0.0_7_8_7, 0.0_8_4_9, 0.0_8_2_6, 0.0_8_1_2, 0.0_8_0_7, 0.0_7_9_5, 0.0_8_1_8, 0.0_7_9_8, 0.0_7_7_9]       )

   assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1E-2





