



"""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 	_SCREAMING_SNAKE_CASE  (__lowerCAmelCase       ,     __lowerCAmelCase							)       ->  Union[str, Any]:







  '''simple docstring'''


  lowercase_	    =       checkpoint

  lowercase_	    =       {}

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

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

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

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

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

  for i in range(__snake_case							):
    lowercase_	    =       [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_	    =       vae_state_dict.pop(
          F'''encoder.down.{i}.downsample.conv.weight'''							)
      lowercase_	    =       vae_state_dict.pop(
          F'''encoder.down.{i}.downsample.conv.bias'''							)

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

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

    lowercase_	    =       renew_vae_resnet_paths(__snake_case							)
    lowercase_	    =       {"""old""": F'''mid.block_{i}''', """new""": F'''mid_block.resnets.{i - 1}'''}
    assign_to_checkpoint(__snake_case       ,     __snake_case       ,     __snake_case       ,     additional_replacements=[meta_path]       ,     config=__snake_case							)

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

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

    lowercase_	    =       renew_vae_resnet_paths(__snake_case							)
    lowercase_	    =       {"""old""": F'''up.{block_id}.block''', """new""": F'''up_blocks.{i}.resnets'''}
    assign_to_checkpoint(__snake_case       ,     __snake_case       ,     __snake_case       ,     additional_replacements=[meta_path]       ,     config=__snake_case							)

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

    lowercase_	    =       renew_vae_resnet_paths(__snake_case							)
    lowercase_	    =       {"""old""": F'''mid.block_{i}''', """new""": F'''mid_block.resnets.{i - 1}'''}
    assign_to_checkpoint(__snake_case       ,     __snake_case       ,     __snake_case       ,     additional_replacements=[meta_path]       ,     config=__snake_case							)

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


def 	_SCREAMING_SNAKE_CASE  (__lowerCAmelCase       ,     __lowerCAmelCase       ,     )       ->  Union[str, Any]:







  '''simple docstring'''


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

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

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

  # Convert the VAE model.
  lowercase_	    =       create_vae_diffusers_config(__snake_case       ,     image_size=__snake_case							)
  lowercase_	    =       custom_convert_ldm_vae_checkpoint(__snake_case       ,     __snake_case							)

  lowercase_	    =       AutoencoderKL(**__snake_case							)
  vae.load_state_dict(__snake_case							)
  vae.save_pretrained(__snake_case							)


if __name__ == "__main__":
       UpperCAmelCase  :						List[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.")

       UpperCAmelCase  :						str	=       parser.parse_args()

       vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)








import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin






class 					__A    (    lowerCAmelCase ):

			'''simple docstring'''







			lowerCAmelCase_							     =						"""Speech2TextFeatureExtractor"""
			lowerCAmelCase_							     =						"""Speech2TextTokenizer"""





			def __init__(    self       ,  __lowerCAmelCase       ,  __lowerCAmelCase							):



				'''simple docstring'''







				super().__init__(__lowerCAmelCase       ,  __lowerCAmelCase							)
				lowerCamelCase__								=		self.feature_extractor
				lowerCamelCase__								=		False





			def __call__(    self       ,  *__lowerCAmelCase       ,  **__lowerCAmelCase							):



				'''simple docstring'''







				if self._in_target_context_manager:
					return self.current_processor(*__lowerCAmelCase       ,  **__lowerCAmelCase							)

				if "raw_speech" in kwargs:
					warnings.warn('''Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead.'''							)
					lowerCamelCase__								=		kwargs.pop('''raw_speech'''							)
				else:
					lowerCamelCase__								=		kwargs.pop('''audio'''       ,  __lowerCAmelCase							)
				lowerCamelCase__								=		kwargs.pop('''sampling_rate'''       ,  __lowerCAmelCase							)
				lowerCamelCase__								=		kwargs.pop('''text'''       ,  __lowerCAmelCase							)
				if len(__lowerCAmelCase							) > 0:
					lowerCamelCase__								=		args[0]
					lowerCamelCase__								=		args[1:]

				if audio is None and text is None:
					raise ValueError('''You need to specify either an `audio` or `text` input to process.'''							)

				if audio is not None:
					lowerCamelCase__								=		self.feature_extractor(__lowerCAmelCase       ,  *__lowerCAmelCase       ,  sampling_rate=__lowerCAmelCase       ,  **__lowerCAmelCase							)
				if text is not None:
					lowerCamelCase__								=		self.tokenizer(__lowerCAmelCase       ,  **__lowerCAmelCase							)

				if text is None:
					return inputs
				elif audio is None:
					return encodings
				else:
					lowerCamelCase__								=		encodings['''input_ids''']
					return inputs





			def 		__lowerCamelCase				(    self       ,  *__lowerCAmelCase       ,  **__lowerCAmelCase							):



				'''simple docstring'''







				return self.tokenizer.batch_decode(*__lowerCAmelCase       ,  **__lowerCAmelCase							)





			def 		__lowerCamelCase				(    self       ,  *__lowerCAmelCase       ,  **__lowerCAmelCase							):



				'''simple docstring'''







				return self.tokenizer.decode(*__lowerCAmelCase       ,  **__lowerCAmelCase							)





			@contextmanager
			def 		__lowerCamelCase				(    self							):



				'''simple docstring'''







				warnings.warn(
				    '''`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your '''
				    '''labels by using the argument `text` of the regular `__call__` method (either in the same call as '''
				    '''your audio inputs, or in a separate call.'''							)
				lowerCamelCase__								=		True
				lowerCamelCase__								=		self.tokenizer
				yield
				lowerCamelCase__								=		self.feature_extractor
				lowerCamelCase__								=		False



