





'''simple docstring'''


import importlib
import inspect
import json
import os
from collections import OrderedDict

# Build the list of all feature extractors
from ...configuration_utils import PretrainedConfig
from ...dynamic_module_utils import get_class_from_dynamic_module, resolve_trust_remote_code
from ...feature_extraction_utils import FeatureExtractionMixin
from ...image_processing_utils import ImageProcessingMixin
from ...tokenization_utils import TOKENIZER_CONFIG_FILE
from ...utils import FEATURE_EXTRACTOR_NAME, get_file_from_repo, logging
from .auto_factory import _LazyAutoMapping
from .configuration_auto import (
    CONFIG_MAPPING_NAMES,
    AutoConfig,
    model_type_to_module_name,
    replace_list_option_in_docstrings,
)
from .feature_extraction_auto import AutoFeatureExtractor
from .image_processing_auto import AutoImageProcessor
from .tokenization_auto import AutoTokenizer


_lowerCamelCase					:						List[Any]					     =							logging.get_logger(__name__)

_lowerCamelCase					:						int					     =							OrderedDict(
    [
        ('align', 'AlignProcessor'),
        ('altclip', 'AltCLIPProcessor'),
        ('bark', 'BarkProcessor'),
        ('blip', 'BlipProcessor'),
        ('blip-2', 'Blip2Processor'),
        ('bridgetower', 'BridgeTowerProcessor'),
        ('chinese_clip', 'ChineseCLIPProcessor'),
        ('clap', 'ClapProcessor'),
        ('clip', 'CLIPProcessor'),
        ('clipseg', 'CLIPSegProcessor'),
        ('flava', 'FlavaProcessor'),
        ('git', 'GitProcessor'),
        ('groupvit', 'CLIPProcessor'),
        ('hubert', 'Wav2Vec2Processor'),
        ('instructblip', 'InstructBlipProcessor'),
        ('layoutlmv2', 'LayoutLMv2Processor'),
        ('layoutlmv3', 'LayoutLMv3Processor'),
        ('markuplm', 'MarkupLMProcessor'),
        ('mctct', 'MCTCTProcessor'),
        ('mgp-str', 'MgpstrProcessor'),
        ('oneformer', 'OneFormerProcessor'),
        ('owlvit', 'OwlViTProcessor'),
        ('pix2struct', 'Pix2StructProcessor'),
        ('sam', 'SamProcessor'),
        ('sew', 'Wav2Vec2Processor'),
        ('sew-d', 'Wav2Vec2Processor'),
        ('speech_to_text', 'Speech2TextProcessor'),
        ('speech_to_text_2', 'Speech2Text2Processor'),
        ('speecht5', 'SpeechT5Processor'),
        ('trocr', 'TrOCRProcessor'),
        ('tvlt', 'TvltProcessor'),
        ('unispeech', 'Wav2Vec2Processor'),
        ('unispeech-sat', 'Wav2Vec2Processor'),
        ('vilt', 'ViltProcessor'),
        ('vision-text-dual-encoder', 'VisionTextDualEncoderProcessor'),
        ('wav2vec2', 'Wav2Vec2Processor'),
        ('wav2vec2-conformer', 'Wav2Vec2Processor'),
        ('wavlm', 'Wav2Vec2Processor'),
        ('whisper', 'WhisperProcessor'),
        ('xclip', 'XCLIPProcessor'),
    ]
)

_lowerCamelCase					:						str					     =							_LazyAutoMapping(CONFIG_MAPPING_NAMES, PROCESSOR_MAPPING_NAMES)




def  __a			(  UpperCAmelCase     )   ->Optional[int]:



	"""simple docstring"""







	for module_name, processors in PROCESSOR_MAPPING_NAMES.items():
		if class_name in processors:
			A					=   model_type_to_module_name(a__     )

			A					=   importlib.import_module(f""".{module_name}"""							,      """transformers.models"""     )
			try:
				return getattr(a__							,      a__     )
			except AttributeError:
				continue

	for processor in PROCESSOR_MAPPING._extra_content.values():
		if getattr(a__							,      """__name__"""							,      a__     ) == class_name:
			return processor

    # We did not fine the class, but maybe it's because a dep is missing. In that case, the class will be in the main
    # init and we return the proper dummy to get an appropriate error message.
	A					=   importlib.import_module("""transformers"""     )
	if hasattr(a__							,      a__     ):
		return getattr(a__							,      a__     )

	return None



class 				__UpperCAmelCase	:







		'''simple docstring'''







		def __init__(self    :   Union[str, Any]   ):
			raise EnvironmentError(
			    """AutoProcessor is designed to be instantiated """
			    """using the `AutoProcessor.from_pretrained(pretrained_model_name_or_path)` method."""   )







		@classmethod
		@replace_list_option_in_docstrings(_lowerCAmelCase   )
		def 							A  (cls    :   Optional[int]   ,   _lowerCAmelCase    :   Union[str, Any]   ,   **_lowerCAmelCase    :   Any   ):
			A					=   kwargs.pop("""config"""   ,   _lowerCAmelCase   )
			A					=   kwargs.pop("""trust_remote_code"""   ,   _lowerCAmelCase   )
			A					=   True

			A					=   None
			A					=   None

			# First, let's see if we have a preprocessor config.
			# Filter the kwargs for `get_file_from_repo`.
			A					=   {
			    key: kwargs[key] for key in inspect.signature(_lowerCAmelCase   ).parameters.keys() if key in kwargs
			}
			# Let's start by checking whether the processor class is saved in an image processor
			A					=   get_file_from_repo(
			    _lowerCAmelCase   ,   _lowerCAmelCase   ,   **_lowerCAmelCase   )
			if preprocessor_config_file is not None:
				A	,     A					=   ImageProcessingMixin.get_image_processor_dict(_lowerCAmelCase   ,   **_lowerCAmelCase   )
				A					=   config_dict.get("""processor_class"""   ,   _lowerCAmelCase   )
				if "AutoProcessor" in config_dict.get("""auto_map"""   ,   {}   ):
					A					=   config_dict["""auto_map"""]["""AutoProcessor"""]

        # If not found, let's check whether the processor class is saved in a feature extractor config
			if preprocessor_config_file is not None and processor_class is None:
				A	,     A					=   FeatureExtractionMixin.get_feature_extractor_dict(_lowerCAmelCase   ,   **_lowerCAmelCase   )
				A					=   config_dict.get("""processor_class"""   ,   _lowerCAmelCase   )
				if "AutoProcessor" in config_dict.get("""auto_map"""   ,   {}   ):
					A					=   config_dict["""auto_map"""]["""AutoProcessor"""]

			if processor_class is None:
				# Next, let's check whether the processor class is saved in a tokenizer
				A					=   get_file_from_repo(
				    _lowerCAmelCase   ,   _lowerCAmelCase   ,   **_lowerCAmelCase   )
				if tokenizer_config_file is not None:
					with open(_lowerCAmelCase   ,   encoding="""utf-8"""   ) as reader:
						A					=   json.load(_lowerCAmelCase   )

					A					=   config_dict.get("""processor_class"""   ,   _lowerCAmelCase   )
					if "AutoProcessor" in config_dict.get("""auto_map"""   ,   {}   ):
						A					=   config_dict["""auto_map"""]["""AutoProcessor"""]

			if processor_class is None:
				# Otherwise, load config, if it can be loaded.
				if not isinstance(_lowerCAmelCase   ,   _lowerCAmelCase   ):
					A					=   AutoConfig.from_pretrained(
					    _lowerCAmelCase   ,   trust_remote_code=_lowerCAmelCase   ,   **_lowerCAmelCase   )

				# And check if the config contains the processor class.
				A					=   getattr(_lowerCAmelCase   ,   """processor_class"""   ,   _lowerCAmelCase   )
				if hasattr(_lowerCAmelCase   ,   """auto_map"""   ) and "AutoProcessor" in config.auto_map:
					A					=   config.auto_map["""AutoProcessor"""]

			if processor_class is not None:
				A					=   processor_class_from_name(_lowerCAmelCase   )

			A					=   processor_auto_map is not None
			A					=   processor_class is not None or type(_lowerCAmelCase   ) in PROCESSOR_MAPPING
			A					=   resolve_trust_remote_code(
			    _lowerCAmelCase   ,   _lowerCAmelCase   ,   _lowerCAmelCase   ,   _lowerCAmelCase   )

			if has_remote_code and trust_remote_code:
				A					=   get_class_from_dynamic_module(
				    _lowerCAmelCase   ,   _lowerCAmelCase   ,   **_lowerCAmelCase   )
				A					=   kwargs.pop("""code_revision"""   ,   _lowerCAmelCase   )
				if os.path.isdir(_lowerCAmelCase   ):
					processor_class.register_for_auto_class()
				return processor_class.from_pretrained(
				    _lowerCAmelCase   ,   trust_remote_code=_lowerCAmelCase   ,   **_lowerCAmelCase   )
			elif processor_class is not None:
				return processor_class.from_pretrained(
				    _lowerCAmelCase   ,   trust_remote_code=_lowerCAmelCase   ,   **_lowerCAmelCase   )
			# Last try: we use the PROCESSOR_MAPPING.
			elif type(_lowerCAmelCase   ) in PROCESSOR_MAPPING:
				return PROCESSOR_MAPPING[type(_lowerCAmelCase   )].from_pretrained(_lowerCAmelCase   ,   **_lowerCAmelCase   )

			# At this stage, there doesn't seem to be a `Processor` class available for this model, so let's try a
			# tokenizer.
			try:
				return AutoTokenizer.from_pretrained(
				    _lowerCAmelCase   ,   trust_remote_code=_lowerCAmelCase   ,   **_lowerCAmelCase   )
			except Exception:
				try:
					return AutoImageProcessor.from_pretrained(
					    _lowerCAmelCase   ,   trust_remote_code=_lowerCAmelCase   ,   **_lowerCAmelCase   )
				except Exception:
					pass

				try:
					return AutoFeatureExtractor.from_pretrained(
					    _lowerCAmelCase   ,   trust_remote_code=_lowerCAmelCase   ,   **_lowerCAmelCase   )
				except Exception:
					pass

			raise ValueError(
			    F"""Unrecognized processing class in {pretrained_model_name_or_path}. Can't instantiate a processor, a """
			    """tokenizer, an image processor or a feature extractor for this model. Make sure the repository contains"""
			    """the files of at least one of those processing classes."""   )







		@staticmethod
		def 							A  (_lowerCAmelCase    :   Dict   ,   _lowerCAmelCase    :   Any   ):
			PROCESSOR_MAPPING.register(_lowerCAmelCase   ,   _lowerCAmelCase   )



'''simple docstring'''







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 (
    SEWDConfig,
    SEWDForCTC,
    SEWDModel,
    WavaVecaCTCTokenizer,
    WavaVecaFeatureExtractor,
    WavaVecaProcessor,
    logging,
)


logging.set_verbosity_info()
lowerCAmelCase_      =   logging.get_logger(__name__)

lowerCAmelCase_      =   {
    "post_extract_proj": "feature_projection",
    "encoder.pos_conv.0": "encoder.pos_conv_embed.conv",
    "attention.self.query_proj": "encoder.encoder.layer.*.attention.self.query_proj",
    "attention.self.key_proj": "encoder.encoder.layer.*.attention.self.key_proj",
    "attention.self.value_proj": "encoder.encoder.layer.*.attention.self.value_proj",
    "attention.output.dense": "encoder.encoder.layer.*.attention.output.dense",
    "attention.output.LayerNorm": "encoder.encoder.layer.*.attention.output.LayerNorm",
    "intermediate.dense": "encoder.encoder.layer.*.intermediate.dense",
    "output.dense": "encoder.encoder.layer.*.output.dense",
    "output.LayerNorm": "encoder.encoder.layer.*.output.LayerNorm",
    "encoder.encoder.rel_embeddings": "encoder.encoder.rel_embeddings",
    "encoder.encoder.LayerNorm": "encoder.encoder.LayerNorm",
    "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   lowerCAmelCase(  a__    :  List[Any]      , a__    :  List[str]      , a__    :  Optional[int]      , a__    :  Optional[Any]      , a__    :  List[Any]      ):







       '''simple docstring'''






       for attribute in key.split("."      ):
              lowerCamelCase__				   =      getattr(a__      , a__      )

       if weight_type is not None:
              lowerCamelCase__				   =      getattr(a__      , a__      ).shape
       else:
              lowerCamelCase__				   =      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__				   =      value
       elif weight_type == "weight_g":
              lowerCamelCase__				   =      value
       elif weight_type == "weight_v":
              lowerCamelCase__				   =      value
       elif weight_type == "bias":
              lowerCamelCase__				   =      value
       else:
              lowerCamelCase__				   =      value

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







def   lowerCAmelCase(  a__    :  Any      , a__    :  List[str]      , a__    :  str      ):







       '''simple docstring'''






       lowerCamelCase__				   =      []
       lowerCamelCase__				   =      fairseq_model.state_dict()

       lowerCamelCase__				   =      hf_model.sew_d.feature_extractor if is_finetuned else hf_model.feature_extractor

       for name, value in fairseq_dict.items():
              lowerCamelCase__				   =      False
              if "conv_layers" in name:
                     load_conv_layer(
                         a__      , a__      , a__      , a__      , hf_model.config.feat_extract_norm == "group"      , )
                     lowerCamelCase__				   =      True
              else:
                     for key, mapped_key in MAPPING.items():
                            lowerCamelCase__				   =      "sew_d." + 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__				   =      True
                                   if "*" in mapped_key:
                                          lowerCamelCase__				   =      name.split(a__      )[0].split("."      )[-2]
                                          if not layer_index.isnumeric():
                                                 continue
                                          lowerCamelCase__				   =      mapped_key.replace("*"      , a__      )
                                   if "weight_g" in name:
                                          lowerCamelCase__				   =      "weight_g"
                                   elif "weight_v" in name:
                                          lowerCamelCase__				   =      "weight_v"
                                   elif "weight" in name:
                                          lowerCamelCase__				   =      "weight"
                                   elif "bias" in name:
                                          lowerCamelCase__				   =      "bias"
                                   else:
                                          lowerCamelCase__				   =      None
                                   set_recursively(a__      , a__      , a__      , a__      , a__      )
                            continue
              if not is_used:
                     unused_weights.append(a__      )

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







def   lowerCAmelCase(  a__    :  Optional[Any]      , a__    :  Dict      , a__    :  List[Any]      , a__    :  Union[str, Any]      , a__    :  List[str]      ):







       '''simple docstring'''






       lowerCamelCase__				   =      full_name.split("conv_layers."      )[-1]
       lowerCamelCase__				   =      name.split("."      )
       lowerCamelCase__				   =      int(items[0]      )
       lowerCamelCase__				   =      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__				   =      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__				   =      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__				   =      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__				   =      value
                     logger.info(f"""Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}."""      )
       else:
              unused_weights.append(a__      )







def   lowerCAmelCase(  a__    :  Optional[int]      , a__    :  Tuple      ):







       '''simple docstring'''






       lowerCamelCase__				   =      SEWDConfig()
       if is_finetuned:
              lowerCamelCase__				   =      model.wav_encoder.wav_model.cfg
       else:
              lowerCamelCase__				   =      model.cfg

       lowerCamelCase__				   =      fs_config.conv_bias
       lowerCamelCase__				   =      eval(fs_config.conv_feature_layers      )
       lowerCamelCase__				   =      [x[0] for x in conv_layers]
       lowerCamelCase__				   =      [x[1] for x in conv_layers]
       lowerCamelCase__				   =      [x[2] for x in conv_layers]
       lowerCamelCase__				   =      "gelu"
       lowerCamelCase__				   =      "layer" if fs_config.extractor_mode == "layer_norm" else "group"
       lowerCamelCase__				   =      0.0
       lowerCamelCase__				   =      fs_config.activation_fn.name
       lowerCamelCase__				   =      fs_config.encoder_embed_dim
       lowerCamelCase__				   =      0.0_2
       lowerCamelCase__				   =      fs_config.encoder_ffn_embed_dim
       lowerCamelCase__				   =      1E-5
       lowerCamelCase__				   =      fs_config.encoder_layerdrop
       lowerCamelCase__				   =      fs_config.encoder_attention_heads
       lowerCamelCase__				   =      fs_config.conv_pos_groups
       lowerCamelCase__				   =      fs_config.conv_pos
       lowerCamelCase__				   =      len(a__      )
       lowerCamelCase__				   =      fs_config.encoder_layers
       lowerCamelCase__				   =      fs_config.squeeze_factor
       # DeBERTa-specific parameters:
       lowerCamelCase__				   =      fs_config.max_position_embeddings
       lowerCamelCase__				   =      fs_config.position_buckets
       lowerCamelCase__				   =      fs_config.share_att_key
       lowerCamelCase__				   =      fs_config.relative_attention
       lowerCamelCase__				   =      fs_config.position_biased_input
       lowerCamelCase__				   =      tuple(fs_config.pos_att_type.split("|"      )      )
       lowerCamelCase__				   =      fs_config.norm_rel_ebd

       # take care of any params that are overridden by the Wav2VecCtc model
       if is_finetuned:
              lowerCamelCase__				   =      model.cfg
              lowerCamelCase__				   =      fs_config.final_dropout
              lowerCamelCase__				   =      fs_config.layerdrop
       lowerCamelCase__				   =      fs_config.activation_dropout
       lowerCamelCase__				   =      fs_config.mask_prob > 0 or fs_config.mask_channel_prob > 0
       lowerCamelCase__				   =      fs_config.attention_dropout
       lowerCamelCase__				   =      fs_config.dropout_input
       lowerCamelCase__				   =      fs_config.dropout
       lowerCamelCase__				   =      fs_config.mask_channel_length
       lowerCamelCase__				   =      fs_config.mask_channel_prob
       lowerCamelCase__				   =      fs_config.mask_length
       lowerCamelCase__				   =      fs_config.mask_prob

       lowerCamelCase__				   =      "Wav2Vec2FeatureExtractor"
       lowerCamelCase__				   =      "Wav2Vec2CTCTokenizer"

       return config







@torch.no_grad()
def   lowerCAmelCase(  a__    :  List[Any]      , a__    :  int      , a__    :  Any=None      , a__    :  Any=None      , a__    :  Any=True      ):







       '''simple docstring'''






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

       if config_path is not None:
              lowerCamelCase__				   =      SEWDConfig.from_pretrained(a__      )
       else:
              lowerCamelCase__				   =      convert_config(model[0]      , a__      )
       lowerCamelCase__				   =      model[0].eval()

       lowerCamelCase__				   =      True if config.feat_extract_norm == "layer" else False
       lowerCamelCase__				   =      WavaVecaFeatureExtractor(
           feature_size=1      , sampling_rate=1_6000      , padding_value=0      , do_normalize=a__      , return_attention_mask=a__      , )

       if is_finetuned:
              if dict_path:
                     lowerCamelCase__				   =      Dictionary.load(a__      )

                     # important change bos & pad token id since CTC symbol is <pad> and
                     # not <s> as in fairseq
                     lowerCamelCase__				   =      target_dict.pad_index
                     lowerCamelCase__				   =      target_dict.bos_index
                     lowerCamelCase__				   =      target_dict.pad_index
                     lowerCamelCase__				   =      target_dict.bos_index
                     lowerCamelCase__				   =      target_dict.eos_index
                     lowerCamelCase__				   =      len(target_dict.symbols      )
                     lowerCamelCase__				   =      os.path.join(a__      , "vocab.json"      )
                     if not os.path.isdir(a__      ):
                            logger.error("--pytorch_dump_folder_path ({}) should be a directory".format(a__      )      )
                            return
                     os.makedirs(a__      , exist_ok=a__      )
                     with open(a__      , "w"      , encoding="utf-8"      ) as vocab_handle:
                            json.dump(target_dict.indices      , a__      )
                     lowerCamelCase__				   =      WavaVecaCTCTokenizer(
                         a__      , 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=a__      , )
                     lowerCamelCase__				   =      WavaVecaProcessor(feature_extractor=a__      , tokenizer=a__      )
                     processor.save_pretrained(a__      )

              lowerCamelCase__				   =      SEWDForCTC(a__      )
       else:
              lowerCamelCase__				   =      SEWDModel(a__      )
              feature_extractor.save_pretrained(a__      )

       recursively_load_weights(a__      , a__      , a__      )

       hf_model.save_pretrained(a__      )


if __name__ == "__main__":
    lowerCAmelCase_      =   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"
    )
    lowerCAmelCase_      =   parser.parse_args()
    convert_sew_checkpoint(
        args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.dict_path, args.is_finetuned
    )

