





import argparse
import io

import requests
import torch
from omegaconf import OmegaConf

from diffusers import AutoencoderKL
from diffusers.pipelines.stable_diffusion.convert_from_ckpt import (
    assign_to_checkpoint,
    conv_attn_to_linear,
    create_vae_diffusers_config,
    renew_vae_attention_paths,
    renew_vae_resnet_paths,
)

def    A__(		__lowerCAmelCase						,      __lowerCAmelCase			):
				_snake_case			:      str	    = checkpoint

				_snake_case			:      List[Any]	    = {}

				_snake_case			:      Dict	    = vae_state_dict["""encoder.conv_in.weight"""]
				_snake_case			:      Tuple	    = vae_state_dict["""encoder.conv_in.bias"""]
				_snake_case			:      Union[str, Any]	    = vae_state_dict["""encoder.conv_out.weight"""]
				_snake_case			:      Union[str, Any]	    = vae_state_dict["""encoder.conv_out.bias"""]
				_snake_case			:      int	    = vae_state_dict["""encoder.norm_out.weight"""]
				_snake_case			:      List[Any]	    = vae_state_dict["""encoder.norm_out.bias"""]

				_snake_case			:      List[str]	    = vae_state_dict["""decoder.conv_in.weight"""]
				_snake_case			:      Any	    = vae_state_dict["""decoder.conv_in.bias"""]
				_snake_case			:      List[str]	    = vae_state_dict["""decoder.conv_out.weight"""]
				_snake_case			:      Dict	    = vae_state_dict["""decoder.conv_out.bias"""]
				_snake_case			:      Dict	    = vae_state_dict["""decoder.norm_out.weight"""]
				_snake_case			:      Union[str, Any]	    = vae_state_dict["""decoder.norm_out.bias"""]

				_snake_case			:      int	    = vae_state_dict["""quant_conv.weight"""]
				_snake_case			:      Any	    = vae_state_dict["""quant_conv.bias"""]
				_snake_case			:      str	    = vae_state_dict["""post_quant_conv.weight"""]
				_snake_case			:      Dict	    = vae_state_dict["""post_quant_conv.bias"""]

				# Retrieves the keys for the encoder down blocks only
				_snake_case			:      int	    = len({'.'.join(layer.split('.'			)[:3]			) for layer in vae_state_dict if 'encoder.down' in layer}			)
				_snake_case			:      Optional[int]	    = {
				    layer_id: [key for key in vae_state_dict if F'''down.{layer_id}''' in key] for layer_id in range(__lowerCAmelCase			)
				}

				# Retrieves the keys for the decoder up blocks only
				_snake_case			:      Optional[Any]	    = len({'.'.join(layer.split('.'			)[:3]			) for layer in vae_state_dict if 'decoder.up' in layer}			)
				_snake_case			:      str	    = {
				    layer_id: [key for key in vae_state_dict if F'''up.{layer_id}''' in key] for layer_id in range(__lowerCAmelCase			)
				}

				for i in range(__lowerCAmelCase			):
								_snake_case			:      Union[str, Any]	    = [key for key in down_blocks[i] if F'''down.{i}''' in key and F'''down.{i}.downsample''' not in key]

								if F'''encoder.down.{i}.downsample.conv.weight''' in vae_state_dict:
												_snake_case			:      Dict	    = vae_state_dict.pop(
												    F'''encoder.down.{i}.downsample.conv.weight'''			)
												_snake_case			:      Union[str, Any]	    = vae_state_dict.pop(
												    F'''encoder.down.{i}.downsample.conv.bias'''			)

								_snake_case			:      Dict	    = renew_vae_resnet_paths(__lowerCAmelCase			)
								_snake_case			:      str	    = {"""old""": F'''down.{i}.block''', """new""": F'''down_blocks.{i}.resnets'''}
								assign_to_checkpoint(__lowerCAmelCase						,      __lowerCAmelCase						,      __lowerCAmelCase						,      additional_replacements=[meta_path]						,      config=__lowerCAmelCase			)

				_snake_case			:      Any	    = [key for key in vae_state_dict if """encoder.mid.block""" in key]
				_snake_case			:      List[Any]	    = 2
				for i in range(1						,      num_mid_res_blocks + 1			):
								_snake_case			:      str	    = [key for key in mid_resnets if F'''encoder.mid.block_{i}''' in key]

								_snake_case			:      Any	    = renew_vae_resnet_paths(__lowerCAmelCase			)
								_snake_case			:      Union[str, Any]	    = {"""old""": F'''mid.block_{i}''', """new""": F'''mid_block.resnets.{i - 1}'''}
								assign_to_checkpoint(__lowerCAmelCase						,      __lowerCAmelCase						,      __lowerCAmelCase						,      additional_replacements=[meta_path]						,      config=__lowerCAmelCase			)

				_snake_case			:      Optional[int]	    = [key for key in vae_state_dict if """encoder.mid.attn""" in key]
				_snake_case			:      str	    = renew_vae_attention_paths(__lowerCAmelCase			)
				_snake_case			:      Any	    = {"""old""": """mid.attn_1""", """new""": """mid_block.attentions.0"""}
				assign_to_checkpoint(__lowerCAmelCase						,      __lowerCAmelCase						,      __lowerCAmelCase						,      additional_replacements=[meta_path]						,      config=__lowerCAmelCase			)
				conv_attn_to_linear(__lowerCAmelCase			)

				for i in range(__lowerCAmelCase			):
								_snake_case			:      Any	    = num_up_blocks - 1 - i
								_snake_case			:      List[str]	    = [
								    key for key in up_blocks[block_id] if F'''up.{block_id}''' in key and F'''up.{block_id}.upsample''' not in key
								]

								if F'''decoder.up.{block_id}.upsample.conv.weight''' in vae_state_dict:
												_snake_case			:      Tuple	    = vae_state_dict[
												    F'''decoder.up.{block_id}.upsample.conv.weight'''
												]
												_snake_case			:      List[str]	    = vae_state_dict[
												    F'''decoder.up.{block_id}.upsample.conv.bias'''
												]

								_snake_case			:      Optional[Any]	    = renew_vae_resnet_paths(__lowerCAmelCase			)
								_snake_case			:      str	    = {"""old""": F'''up.{block_id}.block''', """new""": F'''up_blocks.{i}.resnets'''}
								assign_to_checkpoint(__lowerCAmelCase						,      __lowerCAmelCase						,      __lowerCAmelCase						,      additional_replacements=[meta_path]						,      config=__lowerCAmelCase			)

				_snake_case			:      Any	    = [key for key in vae_state_dict if """decoder.mid.block""" in key]
				_snake_case			:      Optional[int]	    = 2
				for i in range(1						,      num_mid_res_blocks + 1			):
								_snake_case			:      Optional[int]	    = [key for key in mid_resnets if F'''decoder.mid.block_{i}''' in key]

								_snake_case			:      Union[str, Any]	    = renew_vae_resnet_paths(__lowerCAmelCase			)
								_snake_case			:      Tuple	    = {"""old""": F'''mid.block_{i}''', """new""": F'''mid_block.resnets.{i - 1}'''}
								assign_to_checkpoint(__lowerCAmelCase						,      __lowerCAmelCase						,      __lowerCAmelCase						,      additional_replacements=[meta_path]						,      config=__lowerCAmelCase			)

				_snake_case			:      Dict	    = [key for key in vae_state_dict if """decoder.mid.attn""" in key]
				_snake_case			:      Optional[int]	    = renew_vae_attention_paths(__lowerCAmelCase			)
				_snake_case			:      Any	    = {"""old""": """mid.attn_1""", """new""": """mid_block.attentions.0"""}
				assign_to_checkpoint(__lowerCAmelCase						,      __lowerCAmelCase						,      __lowerCAmelCase						,      additional_replacements=[meta_path]						,      config=__lowerCAmelCase			)
				conv_attn_to_linear(__lowerCAmelCase			)
				return new_checkpoint

def    A__(		__lowerCAmelCase						,      __lowerCAmelCase						,      ):
				_snake_case			:      Optional[int]	    = requests.get(
				    ' https://raw.githubusercontent.com/CompVis/stable-diffusion/main/configs/stable-diffusion/v1-inference.yaml'			)
				_snake_case			:      int	    = io.BytesIO(r.content			)

				_snake_case			:      int	    = OmegaConf.load(__lowerCAmelCase			)
				_snake_case			:      int	    = 5_12
				_snake_case			:      Dict	    = """cuda""" if torch.cuda.is_available() else """cpu"""
				if checkpoint_path.endswith('safetensors'			):
								from safetensors import safe_open

								_snake_case			:      str	    = {}
								with safe_open(__lowerCAmelCase						,      framework='pt'						,      device='cpu'			) as f:
												for key in f.keys():
																_snake_case			:      Any	    = f.get_tensor(__lowerCAmelCase			)
				else:
								_snake_case			:      Optional[int]	    = torch.load(__lowerCAmelCase						,      map_location=__lowerCAmelCase			)["""state_dict"""]

				# Convert the VAE model.
				_snake_case			:      List[str]	    = create_vae_diffusers_config(__lowerCAmelCase						,      image_size=__lowerCAmelCase			)
				_snake_case			:      str	    = custom_convert_ldm_vae_checkpoint(__lowerCAmelCase						,      __lowerCAmelCase			)

				_snake_case			:      int	    = AutoencoderKL(**__lowerCAmelCase			)
				vae.load_state_dict(__lowerCAmelCase			)
				vae.save_pretrained(__lowerCAmelCase			)


if __name__ == "__main__":
		lowercase_		:					str      						=    argparse.ArgumentParser()

		parser.add_argument('''--vae_pt_path''', default=None, type=str, required=True, help='''Path to the VAE.pt to convert.''')
		parser.add_argument('''--dump_path''', default=None, type=str, required=True, help='''Path to the VAE.pt to convert.''')

		lowercase_		:					Tuple      						=    parser.parse_args()

		vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)













from typing import Dict, List, Optional, Union

import numpy as np

from transformers.utils import is_vision_available
from transformers.utils.generic import TensorType

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import (
    center_crop,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    IMAGENET_STANDARD_MEAN,
    IMAGENET_STANDARD_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    is_valid_image,
    to_numpy_array,
    valid_images,
)
from ...utils import logging


if is_vision_available():
							import PIL

UpperCAmelCase_							=				logging.get_logger(__name__)





def  __magic_name__  (				lowercase				)				-> List[List[ImageInput]]:

							"""simple docstring"""



							if isinstance(lowercase				,       (list, tuple)				) and isinstance(videos[0]				,       (list, tuple)				) and is_valid_image(videos[0][0]				):
														return videos

							elif isinstance(lowercase				,       (list, tuple)				) and is_valid_image(videos[0]				):
														return [videos]

							elif is_valid_image(lowercase				):
														return [[videos]]

							raise ValueError(f"""Could not make batched video from {videos}"""				)




class     UpperCamelCase__					(     lowerCamelCase__				):






		'''simple docstring'''






		__a       :       List[Any]       =	["""pixel_values"""]






		def __init__(							self,   snake_case__ = True,   snake_case__ = None,   snake_case__ = PILImageResampling.BILINEAR,   snake_case__ = True,   snake_case__ = None,   snake_case__ = True,   snake_case__ = 1 / 2_55,   snake_case__ = True,   snake_case__ = True,   snake_case__ = None,   snake_case__ = None,   **snake_case__,   )       -> None:






									"""simple docstring"""
									super().__init__(**snake_case__  )
									lowercase_	:   str  				=     size if size is not None else {"""shortest_edge""": 2_56}
									lowercase_	:   int  				=     get_size_dict(snake_case__,   default_to_square=snake_case__  )
									lowercase_	:   int  				=     crop_size if crop_size is not None else {"""height""": 2_24, """width""": 2_24}
									lowercase_	:   Optional[int]  				=     get_size_dict(snake_case__,   param_name="""crop_size"""  )

									lowercase_	:   List[Any]  				=     do_resize
									lowercase_	:   int  				=     size
									lowercase_	:   int  				=     do_center_crop
									lowercase_	:   Optional[Any]  				=     crop_size
									lowercase_	:   str  				=     resample
									lowercase_	:   Any  				=     do_rescale
									lowercase_	:   Dict  				=     rescale_factor
									lowercase_	:   List[Any]  				=     offset
									lowercase_	:   int  				=     do_normalize
									lowercase_	:   str  				=     image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
									lowercase_	:   Tuple  				=     image_std if image_std is not None else IMAGENET_STANDARD_STD






		def 						snake_case__				(							self,   snake_case__,   snake_case__,   snake_case__ = PILImageResampling.BILINEAR,   snake_case__ = None,   **snake_case__,   )       -> np.ndarray:






									"""simple docstring"""
									lowercase_	:   int  				=     get_size_dict(snake_case__,   default_to_square=snake_case__  )
									if "shortest_edge" in size:
																lowercase_	:   Union[str, Any]  				=     get_resize_output_image_size(snake_case__,   size["""shortest_edge"""],   default_to_square=snake_case__  )
									elif "height" in size and "width" in size:
																lowercase_	:   Dict  				=     (size["""height"""], size["""width"""])
									else:
																raise ValueError(f"""Size must have 'height' and 'width' or 'shortest_edge' as keys. Got {size.keys()}"""  )
									return resize(snake_case__,   size=snake_case__,   resample=snake_case__,   data_format=snake_case__,   **snake_case__  )






		def 						snake_case__				(							self,   snake_case__,   snake_case__,   snake_case__ = None,   **snake_case__,   )       -> np.ndarray:






									"""simple docstring"""
									lowercase_	:   List[Any]  				=     get_size_dict(snake_case__  )
									if "height" not in size or "width" not in size:
																raise ValueError(f"""Size must have 'height' and 'width' as keys. Got {size.keys()}"""  )
									return center_crop(snake_case__,   size=(size["""height"""], size["""width"""]),   data_format=snake_case__,   **snake_case__  )






		def 						snake_case__				(							self,   snake_case__,   snake_case__,   snake_case__ = True,   snake_case__ = None,   **snake_case__,   )       -> str:






									"""simple docstring"""
									lowercase_	:   Dict  				=     image.astype(np.floataa  )
									if offset:
																lowercase_	:   Optional[Any]  				=     image - (scale / 2)
									return rescale(snake_case__,   scale=snake_case__,   data_format=snake_case__,   **snake_case__  )






		def 						snake_case__				(							self,   snake_case__,   snake_case__,   snake_case__,   snake_case__ = None,   **snake_case__,   )       -> np.ndarray:






									"""simple docstring"""
									return normalize(snake_case__,   mean=snake_case__,   std=snake_case__,   data_format=snake_case__,   **snake_case__  )






		def 						snake_case__				(							self,   snake_case__,   snake_case__ = None,   snake_case__ = None,   snake_case__ = None,   snake_case__ = None,   snake_case__ = None,   snake_case__ = None,   snake_case__ = None,   snake_case__ = None,   snake_case__ = None,   snake_case__ = None,   snake_case__ = None,   snake_case__ = ChannelDimension.FIRST,   )       -> np.ndarray:






									"""simple docstring"""
									if do_resize and size is None or resample is None:
																raise ValueError("""Size and resample must be specified if do_resize is True."""  )

									if do_center_crop and crop_size is None:
																raise ValueError("""Crop size must be specified if do_center_crop is True."""  )

									if do_rescale and rescale_factor is None:
																raise ValueError("""Rescale factor must be specified if do_rescale is True."""  )

									if do_normalize and (image_mean is None or image_std is None):
																raise ValueError("""Image mean and std must be specified if do_normalize is True."""  )

									if offset and not do_rescale:
																raise ValueError("""For offset, do_rescale must also be set to True."""  )

									# All transformations expect numpy arrays.
									lowercase_	:   Optional[int]  				=     to_numpy_array(snake_case__  )

									if do_resize:
																lowercase_	:   Dict  				=     self.resize(image=snake_case__,   size=snake_case__,   resample=snake_case__  )

									if do_center_crop:
																lowercase_	:   Optional[int]  				=     self.center_crop(snake_case__,   size=snake_case__  )

									if do_rescale:
																lowercase_	:   Dict  				=     self.rescale(image=snake_case__,   scale=snake_case__,   offset=snake_case__  )

									if do_normalize:
																lowercase_	:   Optional[Any]  				=     self.normalize(image=snake_case__,   mean=snake_case__,   std=snake_case__  )

									lowercase_	:   Tuple  				=     to_channel_dimension_format(snake_case__,   snake_case__  )
									return image






		def 						snake_case__				(							self,   snake_case__,   snake_case__ = None,   snake_case__ = None,   snake_case__ = None,   snake_case__ = None,   snake_case__ = None,   snake_case__ = None,   snake_case__ = None,   snake_case__ = None,   snake_case__ = None,   snake_case__ = None,   snake_case__ = None,   snake_case__ = None,   snake_case__ = ChannelDimension.FIRST,   **snake_case__,   )       -> PIL.Image.Image:






									"""simple docstring"""
									lowercase_	:   Any  				=     do_resize if do_resize is not None else self.do_resize
									lowercase_	:   str  				=     resample if resample is not None else self.resample
									lowercase_	:   List[str]  				=     do_center_crop if do_center_crop is not None else self.do_center_crop
									lowercase_	:   Optional[Any]  				=     do_rescale if do_rescale is not None else self.do_rescale
									lowercase_	:   str  				=     rescale_factor if rescale_factor is not None else self.rescale_factor
									lowercase_	:   List[Any]  				=     offset if offset is not None else self.offset
									lowercase_	:   Optional[Any]  				=     do_normalize if do_normalize is not None else self.do_normalize
									lowercase_	:   Any  				=     image_mean if image_mean is not None else self.image_mean
									lowercase_	:   Optional[Any]  				=     image_std if image_std is not None else self.image_std

									lowercase_	:   Optional[Any]  				=     size if size is not None else self.size
									lowercase_	:   Dict  				=     get_size_dict(snake_case__,   default_to_square=snake_case__  )
									lowercase_	:   Union[str, Any]  				=     crop_size if crop_size is not None else self.crop_size
									lowercase_	:   Tuple  				=     get_size_dict(snake_case__,   param_name="""crop_size"""  )

									if not valid_images(snake_case__  ):
																raise ValueError(
																    """Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, """
																    """torch.Tensor, tf.Tensor or jax.ndarray."""  )

									lowercase_	:   Optional[Any]  				=     make_batched(snake_case__  )

									lowercase_	:   Optional[Any]  				=     [
									    [
									        self._preprocess_image(
									            image=snake_case__,   do_resize=snake_case__,   size=snake_case__,   resample=snake_case__,   do_center_crop=snake_case__,   crop_size=snake_case__,   do_rescale=snake_case__,   rescale_factor=snake_case__,   offset=snake_case__,   do_normalize=snake_case__,   image_mean=snake_case__,   image_std=snake_case__,   data_format=snake_case__,   )
									        for img in video
									    ]
									    for video in videos
									]

									lowercase_	:   List[str]  				=     {"""pixel_values""": videos}
									return BatchFeature(data=snake_case__,   tensor_type=snake_case__  )