




import argparse
import json
import os

import fairseq
import torch
from fairseq.data import Dictionary

# Register SEW's fairseq modules
from sew_asapp import tasks  # noqa: F401

from transformers import (
    SEWConfig,
    SEWForCTC,
    SEWModel,
    WavaVecaCTCTokenizer,
    WavaVecaFeatureExtractor,
    WavaVecaProcessor,
    logging,
)


logging.set_verbosity_info()
snake_case_  :      Any		=		logging.get_logger(__name__)

snake_case_  :      List[Any]		=		{
    """post_extract_proj""": """feature_projection""",
    """encoder.pos_conv.0""": """encoder.pos_conv_embed.conv""",
    """self_attn.k_proj""": """encoder.layers.*.attention.k_proj""",
    """self_attn.v_proj""": """encoder.layers.*.attention.v_proj""",
    """self_attn.q_proj""": """encoder.layers.*.attention.q_proj""",
    """self_attn.out_proj""": """encoder.layers.*.attention.out_proj""",
    """self_attn_layer_norm""": """encoder.layers.*.layer_norm""",
    """fc1""": """encoder.layers.*.feed_forward.intermediate_dense""",
    """fc2""": """encoder.layers.*.feed_forward.output_dense""",
    """final_layer_norm""": """encoder.layers.*.final_layer_norm""",
    """encoder.upsample.0""": """encoder.upsample.projection""",
    """encoder.layer_norm""": """encoder.layer_norm""",
    """w2v_model.layer_norm""": """layer_norm""",
    """w2v_encoder.proj""": """lm_head""",
    """mask_emb""": """masked_spec_embed""",
}



def        __a							(  __UpperCAmelCase		:						int   ,		__UpperCAmelCase		:						Tuple   ,		__UpperCAmelCase		:						Any   ,		__UpperCAmelCase		:						Optional[Any]   ,		__UpperCAmelCase		:						str      ) ->					List[Any]:
							"""simple docstring"""




							for attribute in key.split("."      ):
														lowerCamelCase_			:    Optional[int]   						=      getattr(__UpperCAmelCase   ,		__UpperCAmelCase      )

							if weight_type is not None:
														lowerCamelCase_			:    Any   						=      getattr(__UpperCAmelCase   ,		__UpperCAmelCase      ).shape
							else:
														lowerCamelCase_			:    Optional[Any]   						=      hf_pointer.shape

							assert hf_shape == value.shape, (
							    f"Shape of hf {key + '.' + weight_type if weight_type is not None else ''} is {hf_shape}, but should be"
							    f" {value.shape} for {full_name}"
							)

							if weight_type == "weight":
														lowerCamelCase_			:    Union[str, Any]   						=      value
							elif weight_type == "weight_g":
														lowerCamelCase_			:    Union[str, Any]   						=      value
							elif weight_type == "weight_v":
														lowerCamelCase_			:    List[str]   						=      value
							elif weight_type == "bias":
														lowerCamelCase_			:    Optional[Any]   						=      value
							else:
														lowerCamelCase_			:    List[str]   						=      value

							logger.info(f"{key + '.' + weight_type if weight_type is not None else ''} was initialized from {full_name}."      )



def        __a							(  __UpperCAmelCase		:						str   ,		__UpperCAmelCase		:						str   ,		__UpperCAmelCase		:						Any      ) ->					List[str]:
							"""simple docstring"""




							lowerCamelCase_			:    List[Any]   						=      []
							lowerCamelCase_			:    str   						=      fairseq_model.state_dict()

							lowerCamelCase_			:    Optional[int]   						=      hf_model.sew.feature_extractor if is_finetuned else hf_model.feature_extractor

							for name, value in fairseq_dict.items():
														lowerCamelCase_			:    int   						=      False
														if "conv_layers" in name:
																					load_conv_layer(
																					    __UpperCAmelCase   ,		__UpperCAmelCase   ,		__UpperCAmelCase   ,		__UpperCAmelCase   ,		hf_model.config.feat_extract_norm == "group"   ,		)
																					lowerCamelCase_			:    Any   						=      True
														else:
																					for key, mapped_key in MAPPING.items():
																												lowerCamelCase_			:    Union[str, Any]   						=      "sew." + mapped_key if (is_finetuned and mapped_key != "lm_head") else mapped_key

																												if key in name or key.split("w2v_model."      )[-1] == name.split("."      )[0]:
																																			lowerCamelCase_			:    Optional[Any]   						=      True
																																			if "*" in mapped_key:
																																										lowerCamelCase_			:    Optional[int]   						=      name.split(__UpperCAmelCase      )[0].split("."      )[-2]
																																										lowerCamelCase_			:    List[Any]   						=      mapped_key.replace("*"   ,		__UpperCAmelCase      )
																																			if "weight_g" in name:
																																										lowerCamelCase_			:    Union[str, Any]   						=      "weight_g"
																																			elif "weight_v" in name:
																																										lowerCamelCase_			:    int   						=      "weight_v"
																																			elif "weight" in name:
																																										lowerCamelCase_			:    Dict   						=      "weight"
																																			elif "bias" in name:
																																										lowerCamelCase_			:    Optional[int]   						=      "bias"
																																			else:
																																										lowerCamelCase_			:    str   						=      None
																																			set_recursively(__UpperCAmelCase   ,		__UpperCAmelCase   ,		__UpperCAmelCase   ,		__UpperCAmelCase   ,		__UpperCAmelCase      )
																												continue
														if not is_used:
																					unused_weights.append(__UpperCAmelCase      )

							logger.warning(f"Unused weights: {unused_weights}"      )



def        __a							(  __UpperCAmelCase		:						Union[str, Any]   ,		__UpperCAmelCase		:						Tuple   ,		__UpperCAmelCase		:						Any   ,		__UpperCAmelCase		:						List[Any]   ,		__UpperCAmelCase		:						List[str]      ) ->					List[Any]:
							"""simple docstring"""




							lowerCamelCase_			:    Union[str, Any]   						=      full_name.split("conv_layers."      )[-1]
							lowerCamelCase_			:    Union[str, Any]   						=      name.split("."      )
							lowerCamelCase_			:    Optional[Any]   						=      int(items[0]      )
							lowerCamelCase_			:    Dict   						=      int(items[1]      )

							if type_id == 0:
														if "bias" in name:
																					assert value.shape == feature_extractor.conv_layers[layer_id].conv.bias.data.shape, (
																					    f"{full_name} has size {value.shape}, but"
																					    f" {feature_extractor.conv_layers[layer_id].conv.bias.data.shape} was found."
																					)
																					lowerCamelCase_			:    Union[str, Any]   						=      value
																					logger.info(f"Feat extract conv layer {layer_id} was initialized from {full_name}."      )
														elif "weight" in name:
																					assert value.shape == feature_extractor.conv_layers[layer_id].conv.weight.data.shape, (
																					    f"{full_name} has size {value.shape}, but"
																					    f" {feature_extractor.conv_layers[layer_id].conv.weight.data.shape} was found."
																					)
																					lowerCamelCase_			:    Dict   						=      value
																					logger.info(f"Feat extract conv layer {layer_id} was initialized from {full_name}."      )
							elif (type_id == 2 and not use_group_norm) or (type_id == 2 and layer_id == 0 and use_group_norm):
														if "bias" in name:
																					assert value.shape == feature_extractor.conv_layers[layer_id].layer_norm.bias.data.shape, (
																					    f"{full_name} has size {value.shape}, but {feature_extractor[layer_id].layer_norm.bias.data.shape} was"
																					    " found."
																					)
																					lowerCamelCase_			:    Any   						=      value
																					logger.info(f"Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}."      )
														elif "weight" in name:
																					assert value.shape == feature_extractor.conv_layers[layer_id].layer_norm.weight.data.shape, (
																					    f"{full_name} has size {value.shape}, but"
																					    f" {feature_extractor[layer_id].layer_norm.weight.data.shape} was found."
																					)
																					lowerCamelCase_			:    List[str]   						=      value
																					logger.info(f"Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}."      )
							else:
														unused_weights.append(__UpperCAmelCase      )



def        __a							(  __UpperCAmelCase		:						List[Any]   ,		__UpperCAmelCase		:						Tuple      ) ->					Any:
							"""simple docstring"""




							lowerCamelCase_			:    Dict   						=      SEWConfig()
							if is_finetuned:
														lowerCamelCase_			:    Optional[Any]   						=      model.wav_encoder.wav_model.cfg
							else:
														lowerCamelCase_			:    int   						=      model.cfg

							lowerCamelCase_			:    Any   						=      fs_config.conv_bias
							lowerCamelCase_			:    List[str]   						=      eval(fs_config.conv_feature_layers      )
							lowerCamelCase_			:    Any   						=      [x[0] for x in conv_layers]
							lowerCamelCase_			:    Union[str, Any]   						=      [x[1] for x in conv_layers]
							lowerCamelCase_			:    Optional[Any]   						=      [x[2] for x in conv_layers]
							lowerCamelCase_			:    Any   						=      "gelu"
							lowerCamelCase_			:    Any   						=      "layer" if fs_config.extractor_mode == "layer_norm" else "group"
							lowerCamelCase_			:    Tuple   						=      0.0
							lowerCamelCase_			:    List[str]   						=      fs_config.activation_fn.name
							lowerCamelCase_			:    List[str]   						=      fs_config.encoder_embed_dim
							lowerCamelCase_			:    Tuple   						=      0.0_2
							lowerCamelCase_			:    Optional[Any]   						=      fs_config.encoder_ffn_embed_dim
							lowerCamelCase_			:    Optional[int]   						=      1e-5
							lowerCamelCase_			:    Dict   						=      fs_config.encoder_layerdrop
							lowerCamelCase_			:    Any   						=      fs_config.encoder_attention_heads
							lowerCamelCase_			:    Any   						=      fs_config.conv_pos_groups
							lowerCamelCase_			:    List[Any]   						=      fs_config.conv_pos
							lowerCamelCase_			:    Tuple   						=      len(__UpperCAmelCase      )
							lowerCamelCase_			:    Optional[Any]   						=      fs_config.encoder_layers
							lowerCamelCase_			:    str   						=      fs_config.squeeze_factor

							# take care of any params that are overridden by the Wav2VecCtc model
							if is_finetuned:
														lowerCamelCase_			:    List[str]   						=      model.cfg
														lowerCamelCase_			:    List[str]   						=      fs_config.final_dropout
														lowerCamelCase_			:    int   						=      fs_config.layerdrop
							lowerCamelCase_			:    int   						=      fs_config.activation_dropout
							lowerCamelCase_			:    Tuple   						=      fs_config.mask_prob > 0 or fs_config.mask_channel_prob > 0
							lowerCamelCase_			:    Tuple   						=      fs_config.attention_dropout
							lowerCamelCase_			:    Tuple   						=      fs_config.dropout_input
							lowerCamelCase_			:    int   						=      fs_config.dropout
							lowerCamelCase_			:    int   						=      fs_config.mask_channel_length
							lowerCamelCase_			:    int   						=      fs_config.mask_channel_prob
							lowerCamelCase_			:    List[Any]   						=      fs_config.mask_length
							lowerCamelCase_			:    Optional[Any]   						=      fs_config.mask_prob

							lowerCamelCase_			:    Any   						=      "Wav2Vec2FeatureExtractor"
							lowerCamelCase_			:    List[Any]   						=      "Wav2Vec2CTCTokenizer"

							return config



@torch.no_grad()
def        __a							(  __UpperCAmelCase		:						str   ,		__UpperCAmelCase		:						Optional[int]   ,		__UpperCAmelCase		:						Any=None   ,		__UpperCAmelCase		:						Union[str, Any]=None   ,		__UpperCAmelCase		:						Union[str, Any]=True      ) ->					str:
							"""simple docstring"""




							if is_finetuned:
														lowerCamelCase_			:    Union[str, Any]   						=      fairseq.checkpoint_utils.load_model_ensemble_and_task(
														    [checkpoint_path]   ,		arg_overrides={"data": "/".join(dict_path.split("/"      )[:-1]      )}      )
							else:
														lowerCamelCase_			:    Tuple   						=      fairseq.checkpoint_utils.load_model_ensemble_and_task([checkpoint_path]      )

							if config_path is not None:
														lowerCamelCase_			:    int   						=      SEWConfig.from_pretrained(__UpperCAmelCase      )
							else:
														lowerCamelCase_			:    List[str]   						=      convert_config(model[0]   ,		__UpperCAmelCase      )
							lowerCamelCase_			:    str   						=      model[0].eval()

							lowerCamelCase_			:    List[str]   						=      True if config.feat_extract_norm == "layer" else False
							lowerCamelCase_			:    List[Any]   						=      WavaVecaFeatureExtractor(
							    feature_size=1   ,		sampling_rate=16000   ,		padding_value=0   ,		do_normalize=__UpperCAmelCase   ,		return_attention_mask=__UpperCAmelCase   ,		)

							if is_finetuned:
														if dict_path:
																					lowerCamelCase_			:    Optional[int]   						=      Dictionary.load(__UpperCAmelCase      )

																					# important change bos & pad token id since CTC symbol is <pad> and
																					# not <s> as in fairseq
																					lowerCamelCase_			:    Optional[int]   						=      target_dict.pad_index
																					lowerCamelCase_			:    Union[str, Any]   						=      target_dict.bos_index
																					lowerCamelCase_			:    Any   						=      target_dict.pad_index
																					lowerCamelCase_			:    int   						=      target_dict.bos_index
																					lowerCamelCase_			:    str   						=      target_dict.eos_index
																					lowerCamelCase_			:    Union[str, Any]   						=      len(target_dict.symbols      )
																					lowerCamelCase_			:    str   						=      os.path.join(__UpperCAmelCase   ,		"vocab.json"      )
																					if not os.path.isdir(__UpperCAmelCase      ):
																												logger.error("--pytorch_dump_folder_path ({}) should be a directory".format(__UpperCAmelCase      )      )
																												return
																					os.makedirs(__UpperCAmelCase   ,		exist_ok=__UpperCAmelCase      )
																					with open(__UpperCAmelCase   ,		"w"   ,		encoding="utf-8"      ) as vocab_handle:
																												json.dump(target_dict.indices   ,		__UpperCAmelCase      )
																					lowerCamelCase_			:    Union[str, Any]   						=      WavaVecaCTCTokenizer(
																					    __UpperCAmelCase   ,		unk_token=target_dict.unk_word   ,		pad_token=target_dict.pad_word   ,		bos_token=target_dict.bos_word   ,		eos_token=target_dict.eos_word   ,		word_delimiter_token="|"   ,		do_lower_case=__UpperCAmelCase   ,		)
																					lowerCamelCase_			:    Dict   						=      WavaVecaProcessor(feature_extractor=__UpperCAmelCase   ,		tokenizer=__UpperCAmelCase      )
																					processor.save_pretrained(__UpperCAmelCase      )

														lowerCamelCase_			:    Optional[int]   						=      SEWForCTC(__UpperCAmelCase      )
							else:
														lowerCamelCase_			:    str   						=      SEWModel(__UpperCAmelCase      )
														feature_extractor.save_pretrained(__UpperCAmelCase      )

							recursively_load_weights(__UpperCAmelCase   ,		__UpperCAmelCase   ,		__UpperCAmelCase      )

							hf_model.save_pretrained(__UpperCAmelCase      )


if __name__ == "__main__":
	snake_case_  :      List[Any]		=		argparse.ArgumentParser()
	parser.add_argument("--pytorch_dump_folder_path", default=None, type=str, help="Path to the output PyTorch model.")
	parser.add_argument("--checkpoint_path", default=None, type=str, help="Path to fairseq checkpoint")
	parser.add_argument("--dict_path", default=None, type=str, help="Path to dict of fine-tuned model")
	parser.add_argument("--config_path", default=None, type=str, help="Path to hf config.json of model to convert")
	parser.add_argument(
	    "--is_finetuned", action="store_true", help="Whether the model to convert is a fine-tuned model or not"
	)
	snake_case_  :      Optional[Any]		=		parser.parse_args()
	convert_sew_checkpoint(
	    args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.dict_path, args.is_finetuned
	)








from dataclasses import dataclass
from typing import List, Optional, Union

import numpy as np
import PIL
from PIL import Image

from ...utils import (
    BaseOutput,
    OptionalDependencyNotAvailable,
    is_flax_available,
    is_k_diffusion_available,
    is_k_diffusion_version,
    is_onnx_available,
    is_torch_available,
    is_transformers_available,
    is_transformers_version,
)







@dataclass
class 	__lowerCAmelCase      (       _UpperCamelCase	):
   _UpperCamelCase			:    Union[List[PIL.Image.Image], np.ndarray]
   _UpperCamelCase			:    Optional[List[bool]]


try:
  if not (is_transformers_available() and is_torch_available()):
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  from ...utils.dummy_torch_and_transformers_objects import *  # noqa F403
else:
  from .pipeline_cycle_diffusion import CycleDiffusionPipeline
  from .pipeline_stable_diffusion import StableDiffusionPipeline
  from .pipeline_stable_diffusion_attend_and_excite import StableDiffusionAttendAndExcitePipeline
  from .pipeline_stable_diffusion_imgaimg import StableDiffusionImgaImgPipeline
  from .pipeline_stable_diffusion_inpaint import StableDiffusionInpaintPipeline
  from .pipeline_stable_diffusion_inpaint_legacy import StableDiffusionInpaintPipelineLegacy
  from .pipeline_stable_diffusion_instruct_pixapix import StableDiffusionInstructPixaPixPipeline
  from .pipeline_stable_diffusion_latent_upscale import StableDiffusionLatentUpscalePipeline
  from .pipeline_stable_diffusion_ldmad import StableDiffusionLDMaDPipeline
  from .pipeline_stable_diffusion_model_editing import StableDiffusionModelEditingPipeline
  from .pipeline_stable_diffusion_panorama import StableDiffusionPanoramaPipeline
  from .pipeline_stable_diffusion_paradigms import StableDiffusionParadigmsPipeline
  from .pipeline_stable_diffusion_sag import StableDiffusionSAGPipeline
  from .pipeline_stable_diffusion_upscale import StableDiffusionUpscalePipeline
  from .pipeline_stable_unclip import StableUnCLIPPipeline
  from .pipeline_stable_unclip_imgaimg import StableUnCLIPImgaImgPipeline
  from .safety_checker import StableDiffusionSafetyChecker
  from .stable_unclip_image_normalizer import StableUnCLIPImageNormalizer

try:
  if not (is_transformers_available() and is_torch_available() and is_transformers_version(""">=""", """4.25.0""")):
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  from ...utils.dummy_torch_and_transformers_objects import StableDiffusionImageVariationPipeline
else:
  from .pipeline_stable_diffusion_image_variation import StableDiffusionImageVariationPipeline


try:
  if not (is_transformers_available() and is_torch_available() and is_transformers_version(""">=""", """4.26.0""")):
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  from ...utils.dummy_torch_and_transformers_objects import (
      StableDiffusionDepthaImgPipeline,
      StableDiffusionDiffEditPipeline,
      StableDiffusionPixaPixZeroPipeline,
  )
else:
  from .pipeline_stable_diffusion_depthaimg import StableDiffusionDepthaImgPipeline
  from .pipeline_stable_diffusion_diffedit import StableDiffusionDiffEditPipeline
  from .pipeline_stable_diffusion_pixapix_zero import StableDiffusionPixaPixZeroPipeline


try:
  if not (
      is_torch_available()
      and is_transformers_available()
      and is_k_diffusion_available()
      and is_k_diffusion_version(""">=""", """0.0.12""")
  ):
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  from ...utils.dummy_torch_and_transformers_and_k_diffusion_objects import *  # noqa F403
else:
  from .pipeline_stable_diffusion_k_diffusion import StableDiffusionKDiffusionPipeline

try:
  if not (is_transformers_available() and is_onnx_available()):
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  from ...utils.dummy_onnx_objects import *  # noqa F403
else:
  from .pipeline_onnx_stable_diffusion import OnnxStableDiffusionPipeline, StableDiffusionOnnxPipeline
  from .pipeline_onnx_stable_diffusion_imgaimg import OnnxStableDiffusionImgaImgPipeline
  from .pipeline_onnx_stable_diffusion_inpaint import OnnxStableDiffusionInpaintPipeline
  from .pipeline_onnx_stable_diffusion_inpaint_legacy import OnnxStableDiffusionInpaintPipelineLegacy
  from .pipeline_onnx_stable_diffusion_upscale import OnnxStableDiffusionUpscalePipeline

if is_transformers_available() and is_flax_available():
  import flax







  @flax.struct.dataclass
  class 	__lowerCAmelCase      (       _UpperCamelCase	):
     _UpperCamelCase			:    np.ndarray
     _UpperCamelCase			:    List[bool]







  from ...schedulers.scheduling_pndm_flax import PNDMSchedulerState
  from .pipeline_flax_stable_diffusion import FlaxStableDiffusionPipeline
  from .pipeline_flax_stable_diffusion_imgaimg import FlaxStableDiffusionImgaImgPipeline
  from .pipeline_flax_stable_diffusion_inpaint import FlaxStableDiffusionInpaintPipeline
  from .safety_checker_flax import FlaxStableDiffusionSafetyChecker

