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 	__SCREAMING_SNAKE_CASE  (					__UpperCamelCase    :       Optional[int] ,     __UpperCamelCase    :       List[str]	)							->					int:







  """simple docstring"""





  SCREAMING_SNAKE_CASE__										=      checkpoint

  SCREAMING_SNAKE_CASE__										=      {}

  SCREAMING_SNAKE_CASE__										=      vae_state_dict["""encoder.conv_in.weight"""]
  SCREAMING_SNAKE_CASE__										=      vae_state_dict["""encoder.conv_in.bias"""]
  SCREAMING_SNAKE_CASE__										=      vae_state_dict["""encoder.conv_out.weight"""]
  SCREAMING_SNAKE_CASE__										=      vae_state_dict["""encoder.conv_out.bias"""]
  SCREAMING_SNAKE_CASE__										=      vae_state_dict["""encoder.norm_out.weight"""]
  SCREAMING_SNAKE_CASE__										=      vae_state_dict["""encoder.norm_out.bias"""]

  SCREAMING_SNAKE_CASE__										=      vae_state_dict["""decoder.conv_in.weight"""]
  SCREAMING_SNAKE_CASE__										=      vae_state_dict["""decoder.conv_in.bias"""]
  SCREAMING_SNAKE_CASE__										=      vae_state_dict["""decoder.conv_out.weight"""]
  SCREAMING_SNAKE_CASE__										=      vae_state_dict["""decoder.conv_out.bias"""]
  SCREAMING_SNAKE_CASE__										=      vae_state_dict["""decoder.norm_out.weight"""]
  SCREAMING_SNAKE_CASE__										=      vae_state_dict["""decoder.norm_out.bias"""]

  SCREAMING_SNAKE_CASE__										=      vae_state_dict["""quant_conv.weight"""]
  SCREAMING_SNAKE_CASE__										=      vae_state_dict["""quant_conv.bias"""]
  SCREAMING_SNAKE_CASE__										=      vae_state_dict["""post_quant_conv.weight"""]
  SCREAMING_SNAKE_CASE__										=      vae_state_dict["""post_quant_conv.bias"""]

  # Retrieves the keys for the encoder down blocks only
  SCREAMING_SNAKE_CASE__										=      len({""".""".join(layer.split("""."""	)[:3]	) for layer in vae_state_dict if """encoder.down""" in layer}	)
  SCREAMING_SNAKE_CASE__										=      {
      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
  SCREAMING_SNAKE_CASE__										=      len({""".""".join(layer.split("""."""	)[:3]	) for layer in vae_state_dict if """decoder.up""" in layer}	)
  SCREAMING_SNAKE_CASE__										=      {
      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	):
    SCREAMING_SNAKE_CASE__										=      [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:
      SCREAMING_SNAKE_CASE__										=      vae_state_dict.pop(
          f"""encoder.down.{i}.downsample.conv.weight"""	)
      SCREAMING_SNAKE_CASE__										=      vae_state_dict.pop(
          f"""encoder.down.{i}.downsample.conv.bias"""	)

    SCREAMING_SNAKE_CASE__										=      renew_vae_resnet_paths(__UpperCamelCase	)
    SCREAMING_SNAKE_CASE__										=      {"""old""": f"""down.{i}.block""", """new""": f"""down_blocks.{i}.resnets"""}
    assign_to_checkpoint(__UpperCamelCase ,     __UpperCamelCase ,     __UpperCamelCase ,     additional_replacements=[meta_path] ,     config=__UpperCamelCase	)

  SCREAMING_SNAKE_CASE__										=      [key for key in vae_state_dict if """encoder.mid.block""" in key]
  SCREAMING_SNAKE_CASE__										=      2
  for i in range(1 ,     num_mid_res_blocks + 1	):
    SCREAMING_SNAKE_CASE__										=      [key for key in mid_resnets if f"""encoder.mid.block_{i}""" in key]

    SCREAMING_SNAKE_CASE__										=      renew_vae_resnet_paths(__UpperCamelCase	)
    SCREAMING_SNAKE_CASE__										=      {"""old""": f"""mid.block_{i}""", """new""": f"""mid_block.resnets.{i - 1}"""}
    assign_to_checkpoint(__UpperCamelCase ,     __UpperCamelCase ,     __UpperCamelCase ,     additional_replacements=[meta_path] ,     config=__UpperCamelCase	)

  SCREAMING_SNAKE_CASE__										=      [key for key in vae_state_dict if """encoder.mid.attn""" in key]
  SCREAMING_SNAKE_CASE__										=      renew_vae_attention_paths(__UpperCamelCase	)
  SCREAMING_SNAKE_CASE__										=      {"""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	):
    SCREAMING_SNAKE_CASE__										=      num_up_blocks - 1 - i
    SCREAMING_SNAKE_CASE__										=      [
        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:
      SCREAMING_SNAKE_CASE__										=      vae_state_dict[
          f"""decoder.up.{block_id}.upsample.conv.weight"""
      ]
      SCREAMING_SNAKE_CASE__										=      vae_state_dict[
          f"""decoder.up.{block_id}.upsample.conv.bias"""
      ]

    SCREAMING_SNAKE_CASE__										=      renew_vae_resnet_paths(__UpperCamelCase	)
    SCREAMING_SNAKE_CASE__										=      {"""old""": f"""up.{block_id}.block""", """new""": f"""up_blocks.{i}.resnets"""}
    assign_to_checkpoint(__UpperCamelCase ,     __UpperCamelCase ,     __UpperCamelCase ,     additional_replacements=[meta_path] ,     config=__UpperCamelCase	)

  SCREAMING_SNAKE_CASE__										=      [key for key in vae_state_dict if """decoder.mid.block""" in key]
  SCREAMING_SNAKE_CASE__										=      2
  for i in range(1 ,     num_mid_res_blocks + 1	):
    SCREAMING_SNAKE_CASE__										=      [key for key in mid_resnets if f"""decoder.mid.block_{i}""" in key]

    SCREAMING_SNAKE_CASE__										=      renew_vae_resnet_paths(__UpperCamelCase	)
    SCREAMING_SNAKE_CASE__										=      {"""old""": f"""mid.block_{i}""", """new""": f"""mid_block.resnets.{i - 1}"""}
    assign_to_checkpoint(__UpperCamelCase ,     __UpperCamelCase ,     __UpperCamelCase ,     additional_replacements=[meta_path] ,     config=__UpperCamelCase	)

  SCREAMING_SNAKE_CASE__										=      [key for key in vae_state_dict if """decoder.mid.attn""" in key]
  SCREAMING_SNAKE_CASE__										=      renew_vae_attention_paths(__UpperCamelCase	)
  SCREAMING_SNAKE_CASE__										=      {"""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 	__SCREAMING_SNAKE_CASE  (					__UpperCamelCase    :       str ,     __UpperCamelCase    :       str ,     )							->					Tuple:







  """simple docstring"""





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

  SCREAMING_SNAKE_CASE__										=      OmegaConf.load(__UpperCamelCase	)
  SCREAMING_SNAKE_CASE__										=      5_12
  SCREAMING_SNAKE_CASE__										=      """cuda""" if torch.cuda.is_available() else """cpu"""
  if checkpoint_path.endswith("""safetensors"""	):
    from safetensors import safe_open

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

  # Convert the VAE model.
  SCREAMING_SNAKE_CASE__										=      create_vae_diffusers_config(__UpperCamelCase ,     image_size=__UpperCamelCase	)
  SCREAMING_SNAKE_CASE__										=      custom_convert_ldm_vae_checkpoint(__UpperCamelCase ,     __UpperCamelCase	)

  SCREAMING_SNAKE_CASE__										=      AutoencoderKL(**__UpperCamelCase	)
  vae.load_state_dict(__UpperCamelCase	)
  vae.save_pretrained(__UpperCamelCase	)


if __name__ == "__main__":
   __lowerCamelCase			: Optional[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.''')

   __lowerCamelCase			: List[Any]							   =		parser.parse_args()

   vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)

import re

from flax.core.frozen_dict import freeze
from flax.traverse_util import flatten_dict, unflatten_dict
from jax.experimental import PartitionSpec as P


# Sentinels
__lowerCamelCase			: List[str]							   =		object()

# For specifying empty leaf dict `{}`
__lowerCamelCase			: Optional[int]							   =		object()



def 	__SCREAMING_SNAKE_CASE  (					__UpperCamelCase    :       List[Any] ,     __UpperCamelCase    :       List[str]	)							->					Optional[Any]:







  """simple docstring"""





  SCREAMING_SNAKE_CASE__										=      tuple((re.compile(x + """$"""	) for x in qs)	)
  for i in range(len(__UpperCamelCase	) - len(__UpperCamelCase	) + 1	):
    SCREAMING_SNAKE_CASE__										=      [x.match(__UpperCamelCase	) for x, y in zip(__UpperCamelCase ,     ks[i:]	)]
    if matches and all(__UpperCamelCase	):
      return True
  return False



def 	__SCREAMING_SNAKE_CASE  (					__UpperCamelCase    :       Optional[int]	)							->					List[str]:







  """simple docstring"""





  def replace(__UpperCamelCase    :       Tuple ,     __UpperCamelCase    :       Any	):
    for rule, replacement in rules:
      if _match(__UpperCamelCase ,     __UpperCamelCase	):
        return replacement
    return val

  return replace



def 	__SCREAMING_SNAKE_CASE  (					)							->					List[Any]:







  """simple docstring"""





  return [
      # embeddings
      (("transformer", "wpe", "embedding"), P("""mp""" ,     __UpperCamelCase	)),
      (("transformer", "wte", "embedding"), P("""mp""" ,     __UpperCamelCase	)),
      # atention
      (("attention", "(q_proj|k_proj|v_proj)", "kernel"), P(__UpperCamelCase ,     """mp"""	)),
      (("attention", "out_proj", "kernel"), P("""mp""" ,     __UpperCamelCase	)),
      (("attention", "out_proj", "bias"), None),
      # mlp
      (("mlp", "c_fc", "kernel"), P(__UpperCamelCase ,     """mp"""	)),
      (("mlp", "c_fc", "bias"), P("""mp"""	)),
      (("mlp", "c_proj", "kernel"), P("""mp""" ,     __UpperCamelCase	)),
      (("mlp", "c_proj", "bias"), None),
      # layer norms
      ((r"ln_\d+", "bias"), None),
      ((r"\d+", r"ln_\d+", "scale"), None),
      (("ln_f", "bias"), None),
      (("ln_f", "scale"), None),
  ]






def 	__SCREAMING_SNAKE_CASE  (					__UpperCamelCase    :       Union[str, Any]	)							->					Dict:







  """simple docstring"""





  SCREAMING_SNAKE_CASE__										=      _get_partition_rules()
  SCREAMING_SNAKE_CASE__										=      _replacement_rules(__UpperCamelCase	)
  SCREAMING_SNAKE_CASE__										=      {k: _unmatched for k in flatten_dict(__UpperCamelCase	)}
  SCREAMING_SNAKE_CASE__										=      {k: replace(__UpperCamelCase ,     __UpperCamelCase	) for k, v in initd.items()}
  assert _unmatched not in result.values(), "Incomplete partition spec."
  return freeze(unflatten_dict(__UpperCamelCase	)	)
