


import argparse
import json
import os

import fairseq
import torch
from fairseq.data import Dictionary

from transformers import (
    WavaVecaConformerConfig,
    WavaVecaConformerForCTC,
    WavaVecaConformerForPreTraining,
    WavaVecaCTCTokenizer,
    WavaVecaFeatureExtractor,
    WavaVecaProcessor,
    logging,
)


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

__lowerCamelCase        =     {
    "post_extract_proj": "feature_projection.projection",
    "encoder.pos_conv.0": "encoder.pos_conv_embed.conv",
    "self_attn.linear_k": "encoder.layers.*.self_attn.linear_k",
    "self_attn.linear_v": "encoder.layers.*.self_attn.linear_v",
    "self_attn.linear_q": "encoder.layers.*.self_attn.linear_q",
    "self_attn.pos_bias_u": "encoder.layers.*.self_attn.pos_bias_u",
    "self_attn.pos_bias_v": "encoder.layers.*.self_attn.pos_bias_v",
    "self_attn.linear_out": "encoder.layers.*.self_attn.linear_out",
    "self_attn.linear_pos": "encoder.layers.*.self_attn.linear_pos",
    "self_attn.rotary_emb": "encoder.embed_positions",
    "self_attn_layer_norm": "encoder.layers.*.self_attn_layer_norm",
    "conv_module.pointwise_conv1": "encoder.layers.*.conv_module.pointwise_conv1",
    "conv_module.pointwise_conv2": "encoder.layers.*.conv_module.pointwise_conv2",
    "conv_module.depthwise_conv": "encoder.layers.*.conv_module.depthwise_conv",
    "conv_module.batch_norm": "encoder.layers.*.conv_module.batch_norm",
    "conv_module.layer_norm": "encoder.layers.*.conv_module.layer_norm",
    "ffn1.w_1": "encoder.layers.*.ffn1.intermediate_dense",
    "ffn1.w_2": "encoder.layers.*.ffn1.output_dense",
    "ffn1.layer_norm": "encoder.layers.*.ffn1_layer_norm",
    "ffn2.w_1": "encoder.layers.*.ffn2.intermediate_dense",
    "ffn2.w_2": "encoder.layers.*.ffn2.output_dense",
    "ffn2.layer_norm": "encoder.layers.*.ffn2_layer_norm",
    "final_layer_norm": "encoder.layers.*.final_layer_norm",
    "encoder.layer_norm": "encoder.layer_norm",
    "w2v_model.layer_norm": "feature_projection.layer_norm",
    "quantizer.weight_proj": "quantizer.weight_proj",
    "quantizer.vars": "quantizer.codevectors",
    "project_q": "project_q",
    "final_proj": "project_hid",
    "w2v_encoder.proj": "lm_head",
    "mask_emb": "masked_spec_embed",
}
__lowerCamelCase        =     [
    "lm_head",
    "quantizer.weight_proj",
    "quantizer.codevectors",
    "project_q",
    "project_hid",
]



def  _a			(						__UpperCamelCase  ,  __UpperCamelCase  ,  __UpperCamelCase  ,  __UpperCamelCase  ,  __UpperCamelCase ):

     for attribute in key.split(""".""" ):
          a_    :     List[str]										=    getattr(__UpperCamelCase  ,  __UpperCamelCase )

     if weight_type is not None:
          a_    :     Tuple										=    getattr(__UpperCamelCase  ,  __UpperCamelCase ).shape
     else:
          a_    :     Optional[Any]										=    hf_pointer.shape

     if hf_shape != value.shape:
          raise ValueError(
              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":
          a_    :     Union[str, Any]										=    value
     elif weight_type == "weight_g":
          a_    :     int										=    value
     elif weight_type == "weight_v":
          a_    :     Any										=    value
     elif weight_type == "bias":
          a_    :     Any										=    value
     elif weight_type == "running_mean":
          a_    :     Optional[int]										=    value
     elif weight_type == "running_var":
          a_    :     str										=    value
     elif weight_type == "num_batches_tracked":
          a_    :     str										=    value
     elif weight_type == "inv_freq":
          a_    :     Tuple										=    value
     else:
          a_    :     Any										=    value

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



def  _a			(						__UpperCamelCase  ,  __UpperCamelCase  ,  __UpperCamelCase ):
     a_    :     Optional[int]										=    []
     a_    :     Tuple										=    fairseq_model.state_dict()

     a_    :     int										=    hf_model.wavaveca_conformer.feature_extractor

     for name, value in fairseq_dict.items():
          a_    :     Dict										=    False
          if "conv_layers" in name:
               load_conv_layer(
                   __UpperCamelCase  ,  __UpperCamelCase  ,  __UpperCamelCase  ,  __UpperCamelCase  ,  hf_model.config.feat_extract_norm == """group"""  ,  )
               a_    :     Tuple										=    True
          else:
               for key, mapped_key in MAPPING.items():
                    a_    :     List[Any]										=    """wav2vec2_conformer.""" + mapped_key if mapped_key not in TOP_LEVEL_KEYS else mapped_key
                    if key in name or key.split("""w2v_model.""" )[-1] == name.split(""".""" )[0]:
                         a_    :     Dict										=    True
                         if "*" in mapped_key:
                              a_    :     Union[str, Any]										=    name.split(__UpperCamelCase )[0].split(""".""" )[-2]
                              a_    :     Tuple										=    mapped_key.replace("""*"""  ,  __UpperCamelCase )
                         if "pos_bias_u" in name:
                              a_    :     Union[str, Any]										=    None
                         elif "pos_bias_v" in name:
                              a_    :     Union[str, Any]										=    None
                         elif "weight_g" in name:
                              a_    :     int										=    """weight_g"""
                         elif "weight_v" in name:
                              a_    :     List[Any]										=    """weight_v"""
                         elif "bias" in name:
                              a_    :     Any										=    """bias"""
                         elif "weight" in name:
                              # TODO: don't match quantizer.weight_proj
                              a_    :     Optional[int]										=    """weight"""
                         elif "running_mean" in name:
                              a_    :     int										=    """running_mean"""
                         elif "inv_freq" in name:
                              a_    :     int										=    """inv_freq"""
                         elif "running_var" in name:
                              a_    :     List[str]										=    """running_var"""
                         elif "num_batches_tracked" in name:
                              a_    :     List[str]										=    """num_batches_tracked"""
                         else:
                              a_    :     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  ,  __UpperCamelCase  ,  __UpperCamelCase  ,  __UpperCamelCase  ,  __UpperCamelCase ):
     a_    :     Optional[Any]										=    full_name.split("""conv_layers.""" )[-1]
     a_    :     Any										=    name.split(""".""" )
     a_    :     str										=    int(items[0] )
     a_    :     List[str]										=    int(items[1] )

     if type_id == 0:
          if "bias" in name:
               if value.shape != feature_extractor.conv_layers[layer_id].conv.bias.data.shape:
                    raise ValueError(
                        F'''{full_name} has size {value.shape}, but'''
                        F''' {feature_extractor.conv_layers[layer_id].conv.bias.data.shape} was found.''' )
               a_    :     Union[str, Any]										=    value
               logger.info(F'''Feat extract conv layer {layer_id} was initialized from {full_name}.''' )
          elif "weight" in name:
               if value.shape != feature_extractor.conv_layers[layer_id].conv.weight.data.shape:
                    raise ValueError(
                        F'''{full_name} has size {value.shape}, but'''
                        F''' {feature_extractor.conv_layers[layer_id].conv.weight.data.shape} was found.''' )
               a_    :     Optional[Any]										=    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:
               if value.shape != feature_extractor.conv_layers[layer_id].layer_norm.bias.data.shape:
                    raise ValueError(
                        F'''{full_name} has size {value.shape}, but'''
                        F''' {feature_extractor.conv_layers[layer_id].layer_norm.bias.data.shape} was found.''' )
               a_    :     Union[str, Any]										=    value
               logger.info(F'''Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}.''' )
          elif "weight" in name:
               if value.shape != feature_extractor.conv_layers[layer_id].layer_norm.weight.data.shape:
                    raise ValueError(
                        F'''{full_name} has size {value.shape}, but'''
                        F''' {feature_extractor.conv_layers[layer_id].layer_norm.weight.data.shape} was found.''' )
               a_    :     Optional[Any]										=    value
               logger.info(F'''Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}.''' )
     else:
          unused_weights.append(__UpperCamelCase )



@torch.no_grad()
def  _a			(						__UpperCamelCase  ,  __UpperCamelCase  ,  __UpperCamelCase=None  ,  __UpperCamelCase=None  ,  __UpperCamelCase=True ):

     if config_path is not None:
          a_    :     List[str]										=    WavaVecaConformerConfig.from_pretrained(__UpperCamelCase  ,  hidden_act="""swish""" )
     else:
          a_    :     Tuple										=    WavaVecaConformerConfig()

     if "rope" in checkpoint_path:
          a_    :     int										=    """rotary"""

     if is_finetuned:
          if dict_path:
               a_    :     Any										=    Dictionary.load(__UpperCamelCase )

               # important change bos & pad token id since CTC symbol is <pad> and
               # not <s> as in fairseq
               a_    :     Dict										=    target_dict.pad_index
               a_    :     Any										=    target_dict.bos_index
               a_    :     List[Any]										=    target_dict.eos_index
               a_    :     str										=    len(target_dict.symbols )
               a_    :     Union[str, Any]										=    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 )
               a_    :     List[Any]										=    target_dict.indices

               # fairseq has the <pad> and <s> switched
               a_    :     Tuple										=    0
               a_    :     str										=    1
               with open(__UpperCamelCase  ,  """w"""  ,  encoding="""utf-8""" ) as vocab_handle:
                    json.dump(__UpperCamelCase  ,  __UpperCamelCase )
               a_    :     Optional[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  ,  )
               a_    :     Union[str, Any]										=    True if config.feat_extract_norm == """layer""" else False
               a_    :     Any										=    WavaVecaFeatureExtractor(
                   feature_size=1  ,  sampling_rate=1_6_0_0_0  ,  padding_value=0  ,  do_normalize=__UpperCamelCase  ,  return_attention_mask=__UpperCamelCase  ,  )
               a_    :     Tuple										=    WavaVecaProcessor(feature_extractor=__UpperCamelCase  ,  tokenizer=__UpperCamelCase )
               processor.save_pretrained(__UpperCamelCase )

          a_    :     Union[str, Any]										=    WavaVecaConformerForCTC(__UpperCamelCase )
     else:
          a_    :     str										=    WavaVecaConformerForPreTraining(__UpperCamelCase )

     if is_finetuned:
          a_			,			a_			,			a_    :     List[str]										=    fairseq.checkpoint_utils.load_model_ensemble_and_task(
              [checkpoint_path]  ,  arg_overrides={"""data""": """/""".join(dict_path.split("""/""" )[:-1] )} )
     else:
          a_    :     Optional[int]										=    argparse.Namespace(task="""audio_pretraining""" )
          a_    :     Union[str, Any]										=    fairseq.tasks.setup_task(__UpperCamelCase )

          a_			,			a_			,			a_    :     Optional[Any]										=    fairseq.checkpoint_utils.load_model_ensemble_and_task([checkpoint_path]  ,  task=__UpperCamelCase )

     a_    :     List[str]										=    model[0].eval()

     recursively_load_weights(__UpperCamelCase  ,  __UpperCamelCase  ,  not is_finetuned )

     hf_wavavec.save_pretrained(__UpperCamelCase )


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(
           '''--not_finetuned''', action='''store_true''', help='''Whether the model to convert is a fine-tuned model or not'''
       )
       __lowerCamelCase        =     parser.parse_args()
       convert_wavaveca_conformer_checkpoint(
           args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.dict_path, not args.not_finetuned
       )







from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tokenizers_available, is_torch_available


__lowerCamelCase        =     {
    '''configuration_altclip''': [
        '''ALTCLIP_PRETRAINED_CONFIG_ARCHIVE_MAP''',
        '''AltCLIPConfig''',
        '''AltCLIPTextConfig''',
        '''AltCLIPVisionConfig''',
    ],
    '''processing_altclip''': ['''AltCLIPProcessor'''],
}

try:
       if not is_torch_available():
              raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
       pass
else:
       __lowerCamelCase        =     [
           '''ALTCLIP_PRETRAINED_MODEL_ARCHIVE_LIST''',
           '''AltCLIPPreTrainedModel''',
           '''AltCLIPModel''',
           '''AltCLIPTextModel''',
           '''AltCLIPVisionModel''',
       ]


if TYPE_CHECKING:
       from .configuration_altclip import (
           ALTCLIP_PRETRAINED_CONFIG_ARCHIVE_MAP,
           AltCLIPConfig,
           AltCLIPTextConfig,
           AltCLIPVisionConfig,
       )
       from .processing_altclip import AltCLIPProcessor

       try:
              if not is_torch_available():
                     raise OptionalDependencyNotAvailable()
       except OptionalDependencyNotAvailable:
              pass
       else:
              from .modeling_altclip import (
                  ALTCLIP_PRETRAINED_MODEL_ARCHIVE_LIST,
                  AltCLIPModel,
                  AltCLIPPreTrainedModel,
                  AltCLIPTextModel,
                  AltCLIPVisionModel,
              )


else:
       import sys

       __lowerCamelCase        =     _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)



