




"""simple docstring"""







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 						snake_case__       (					_lowerCamelCase,  _lowerCamelCase				)			->int:






		"""simple docstring"""


		__lowercase				:		Dict												=				checkpoint

		__lowercase				:		Union[str, Any]												=				{}

		__lowercase				:		Optional[int]												=				vae_state_dict["encoder.conv_in.weight"]
		__lowercase				:		str												=				vae_state_dict["encoder.conv_in.bias"]
		__lowercase				:		Any												=				vae_state_dict["encoder.conv_out.weight"]
		__lowercase				:		Any												=				vae_state_dict["encoder.conv_out.bias"]
		__lowercase				:		Tuple												=				vae_state_dict["encoder.norm_out.weight"]
		__lowercase				:		List[str]												=				vae_state_dict["encoder.norm_out.bias"]

		__lowercase				:		Optional[int]												=				vae_state_dict["decoder.conv_in.weight"]
		__lowercase				:		Any												=				vae_state_dict["decoder.conv_in.bias"]
		__lowercase				:		Optional[int]												=				vae_state_dict["decoder.conv_out.weight"]
		__lowercase				:		Optional[int]												=				vae_state_dict["decoder.conv_out.bias"]
		__lowercase				:		str												=				vae_state_dict["decoder.norm_out.weight"]
		__lowercase				:		str												=				vae_state_dict["decoder.norm_out.bias"]

		__lowercase				:		List[str]												=				vae_state_dict["quant_conv.weight"]
		__lowercase				:		Any												=				vae_state_dict["quant_conv.bias"]
		__lowercase				:		Dict												=				vae_state_dict["post_quant_conv.weight"]
		__lowercase				:		Tuple												=				vae_state_dict["post_quant_conv.bias"]

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

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

		for i in range(UpperCAmelCase__				):
				__lowercase				:		Tuple												=				[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:
						__lowercase				:		Optional[Any]												=				vae_state_dict.pop(
						    F'encoder.down.{i}.downsample.conv.weight'				)
						__lowercase				:		Optional[int]												=				vae_state_dict.pop(
						    F'encoder.down.{i}.downsample.conv.bias'				)

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

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

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

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

		for i in range(UpperCAmelCase__				):
				__lowercase				:		Optional[Any]												=				num_up_blocks - 1 - i
				__lowercase				:		Any												=				[
				    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:
						__lowercase				:		Tuple												=				vae_state_dict[
						    F'decoder.up.{block_id}.upsample.conv.weight'
						]
						__lowercase				:		Tuple												=				vae_state_dict[
						    F'decoder.up.{block_id}.upsample.conv.bias'
						]

				__lowercase				:		Dict												=				renew_vae_resnet_paths(UpperCAmelCase__				)
				__lowercase				:		Optional[int]												=				{"old": F'up.{block_id}.block', "new": F'up_blocks.{i}.resnets'}
				assign_to_checkpoint(UpperCAmelCase__,  UpperCAmelCase__,  UpperCAmelCase__,  additional_replacements=[meta_path],  config=UpperCAmelCase__				)

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

				__lowercase				:		Optional[int]												=				renew_vae_resnet_paths(UpperCAmelCase__				)
				__lowercase				:		Dict												=				{"old": F'mid.block_{i}', "new": F'mid_block.resnets.{i - 1}'}
				assign_to_checkpoint(UpperCAmelCase__,  UpperCAmelCase__,  UpperCAmelCase__,  additional_replacements=[meta_path],  config=UpperCAmelCase__				)

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

def 						snake_case__       (					_lowerCamelCase,  _lowerCamelCase,  )			->Tuple:






		"""simple docstring"""


		__lowercase				:		List[str]												=				requests.get(
		    " https://raw.githubusercontent.com/CompVis/stable-diffusion/main/configs/stable-diffusion/v1-inference.yaml"				)
		__lowercase				:		str												=				io.BytesIO(r.content				)

		__lowercase				:		int												=				OmegaConf.load(UpperCAmelCase__				)
		__lowercase				:		List[Any]												=				5_12
		__lowercase				:		Dict												=				"cuda" if torch.cuda.is_available() else "cpu"
		if checkpoint_path.endswith("safetensors"				):
				from safetensors import safe_open

				__lowercase				:		int												=				{}
				with safe_open(UpperCAmelCase__,  framework="pt",  device="cpu"				) as f:
						for key in f.keys():
								__lowercase				:		Dict												=				f.get_tensor(UpperCAmelCase__				)
		else:
				__lowercase				:		Dict												=				torch.load(UpperCAmelCase__,  map_location=UpperCAmelCase__				)["state_dict"]

		# Convert the VAE model.
		__lowercase				:		Optional[int]												=				create_vae_diffusers_config(UpperCAmelCase__,  image_size=UpperCAmelCase__				)
		__lowercase				:		str												=				custom_convert_ldm_vae_checkpoint(UpperCAmelCase__,  UpperCAmelCase__				)

		__lowercase				:		List[str]												=				AutoencoderKL(**UpperCAmelCase__				)
		vae.load_state_dict(UpperCAmelCase__				)
		vae.save_pretrained(UpperCAmelCase__				)


if __name__ == "__main__":
				__A   :  Any					=      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.')

				__A   :  Tuple					=      parser.parse_args()

				vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)






"""simple docstring"""







# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import re

from ..utils import cached_file


# docstyle-ignore
__A   :  int					=      '\nHuman: <<task>>\n\nAssistant: '


__A   :  List[str]					=      'huggingface-tools/default-prompts'
__A   :  Tuple					=      {'chat': 'chat_prompt_template.txt', 'run': 'run_prompt_template.txt'}

def 						snake_case__       (					_lowerCamelCase,  _lowerCamelCase,  _lowerCamelCase="run"				)			->Optional[int]:






		"""simple docstring"""


		if prompt_or_repo_id is None:
				__lowercase				:		Any												=				DEFAULT_PROMPTS_REPO

		# prompt is considered a repo ID when it does not contain any kind of space
		if re.search("\\s",  _lowerCamelCase				) is not None:
				return prompt_or_repo_id

		__lowercase				:		Optional[Any]												=				cached_file(
		    _lowerCamelCase,  PROMPT_FILES[mode],  repo_type="dataset",  user_agent={"agent": agent_name}				)
		with open(_lowerCamelCase,  "r",  encoding="utf-8"				) as f:
				return f.read()
