






'''simple docstring'''



import argparse
import json
import os

import fairseq
import torch
from fairseq.data import Dictionary

from transformers import (
    WavaVecaConfig,
    WavaVecaCTCTokenizer,
    WavaVecaFeatureExtractor,
    WavaVecaForCTC,
    WavaVecaForPreTraining,
    WavaVecaProcessor,
    logging,
)
from transformers.models.wavaveca.modeling_wavaveca import WavaVecaForSequenceClassification


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

SCREAMING_SNAKE_CASE							=						{
    'post_extract_proj': 'feature_projection.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.layer_norm': 'encoder.layer_norm',
    'adapter_layer': 'encoder.layers.*.adapter_layer',
    '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',
    'pooling_layer.linear': 'projector',
    'pooling_layer.projection': 'classifier',
}
SCREAMING_SNAKE_CASE							=						[
    'lm_head',
    'quantizer.weight_proj',
    'quantizer.codevectors',
    'project_q',
    'project_hid',
    'projector',
    'classifier',
]


def 				lowercase_					(     __A			:							Optional[int]      )							->   Optional[int]:



 """simple docstring"""




 lowercase       : Optional[int]						={}
 with open(__A       ,   '''r'''      ) as file:
  for line_number, line in enumerate(__A      ):
   lowercase       : List[Any]						=line.strip()
   if line:
    lowercase       : int						=line.split()
    lowercase       : int						=line_number
    lowercase       : str						=words[0]
    lowercase       : Dict						=value
 return result


def 				lowercase_					(     __A			:							str       ,   __A			:							Any       ,   __A			:							Tuple       ,   __A			:							Dict       ,   __A			:							Union[str, Any]      )							->   Optional[Any]:



 """simple docstring"""




 for attribute in key.split('''.'''      ):
  lowercase       : Optional[Any]						=getattr(__A       ,   __A      )

 lowercase       : Dict						=None
 for param_key in PARAM_MAPPING.keys():
  if full_name.endswith(__A      ):
   lowercase       : Optional[int]						=PARAM_MAPPING[full_name.split('''.'''      )[-1]]
   lowercase       : int						='''param'''

 if weight_type is not None and weight_type != "param":
  lowercase       : Union[str, Any]						=getattr(__A       ,   __A      ).shape
 elif weight_type is not None and weight_type == "param":
  lowercase       : Dict						=hf_pointer
  for attribute in hf_param_name.split('''.'''      ):
   lowercase       : Optional[int]						=getattr(__A       ,   __A      )
  lowercase       : List[str]						=shape_pointer.shape

  # let's reduce dimension
  lowercase       : Any						=value[0]
 else:
  lowercase       : int						=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":
  lowercase       : int						=value
 elif weight_type == "weight_g":
  lowercase       : Optional[int]						=value
 elif weight_type == "weight_v":
  lowercase       : str						=value
 elif weight_type == "bias":
  lowercase       : str						=value
 elif weight_type == "param":
  for attribute in hf_param_name.split('''.'''      ):
   lowercase       : Union[str, Any]						=getattr(__A       ,   __A      )
  lowercase       : List[Any]						=value
 else:
  lowercase       : List[str]						=value

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


def 				lowercase_					(     __A			:							Union[str, Any]       ,   __A			:							int       ,   __A			:							Optional[int]       ,   __A			:							str       ,   __A			:							List[Any]      )							->   Tuple:



 """simple docstring"""




 lowercase       : Optional[Any]						=None
 for param_key in PARAM_MAPPING.keys():
  if full_name.endswith(__A      ):
   lowercase       : Tuple						=PARAM_MAPPING[full_name.split('''.'''      )[-1]]
   lowercase       : List[str]						='''param'''

 if weight_type is not None and weight_type != "param":
  lowercase       : List[str]						='''.'''.join([key, weight_type]      )
 elif weight_type is not None and weight_type == "param":
  lowercase       : List[Any]						='''.'''.join([key, hf_param_name]      )
 else:
  lowercase       : List[Any]						=key

 lowercase       : Any						=value if '''lm_head''' in full_key else value[0]


SCREAMING_SNAKE_CASE							=						{
    'W_a': 'linear_1.weight',
    'W_b': 'linear_2.weight',
    'b_a': 'linear_1.bias',
    'b_b': 'linear_2.bias',
    'ln_W': 'norm.weight',
    'ln_b': 'norm.bias',
}


def 				lowercase_					(     __A			:							List[str]       ,   __A			:							Optional[int]       ,   __A			:							int=None       ,   __A			:							str=None      )							->   Optional[int]:



 """simple docstring"""




 lowercase       : Any						=False
 for key, mapped_key in MAPPING.items():
  lowercase       : Union[str, Any]						='''wav2vec2.''' + 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]:
   lowercase       : Dict						=True
   if "*" in mapped_key:
    lowercase       : Union[str, Any]						=name.split(__A      )[0].split('''.'''      )[-2]
    lowercase       : Union[str, Any]						=mapped_key.replace('''*'''       ,   __A      )
   if "weight_g" in name:
    lowercase       : Optional[int]						='''weight_g'''
   elif "weight_v" in name:
    lowercase       : Optional[Any]						='''weight_v'''
   elif "bias" in name:
    lowercase       : Optional[int]						='''bias'''
   elif "weight" in name:
    # TODO: don't match quantizer.weight_proj
    lowercase       : str						='''weight'''
   else:
    lowercase       : str						=None
   if hf_dict is not None:
    rename_dict(__A       ,   __A       ,   __A       ,   __A       ,   __A      )
   else:
    set_recursively(__A       ,   __A       ,   __A       ,   __A       ,   __A      )
   return is_used
 return is_used


def 				lowercase_					(     __A			:							List[str]       ,   __A			:							Dict       ,   __A			:							Optional[Any]      )							->   Union[str, Any]:



 """simple docstring"""




 lowercase       : Union[str, Any]						=[]
 lowercase       : str						=fairseq_model.state_dict()

 lowercase       : Tuple						=hf_model.wavaveca.feature_extractor

 for name, value in fairseq_dict.items():
  lowercase       : List[str]						=False
  if "conv_layers" in name:
   load_conv_layer(
       __A       ,   __A       ,   __A       ,   __A       ,   hf_model.config.feat_extract_norm == '''group'''       ,   )
   lowercase       : Tuple						=True
  else:
   lowercase       : Dict						=load_wavaveca_layer(__A       ,   __A       ,   __A      )
  if not is_used:
   unused_weights.append(__A      )

 logger.warning(F'Unused weights: {unused_weights}'      )


def 				lowercase_					(     __A			:							Optional[Any]       ,   __A			:							Tuple       ,   __A			:							List[Any]       ,   __A			:							Dict       ,   __A			:							Optional[int]      )							->   List[str]:



 """simple docstring"""




 lowercase       : List[str]						=full_name.split('''conv_layers.'''      )[-1]
 lowercase       : str						=name.split('''.'''      )
 lowercase       : Optional[int]						=int(items[0]      )
 lowercase       : Optional[int]						=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.'      )
   lowercase       : List[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.'      )
   lowercase       : str						=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.'      )
   lowercase       : 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.'      )
   lowercase       : List[Any]						=value
   logger.info(F'Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}.'      )
 else:
  unused_weights.append(__A      )


@torch.no_grad()
def 				lowercase_					(     __A			:							Union[str, Any]       ,   __A			:							Optional[int]       ,   __A			:							str=None       ,   __A			:							List[str]=None       ,   __A			:							str=True       ,   __A			:							List[Any]=False      )							->   List[Any]:



 """simple docstring"""




 if config_path is not None:
  lowercase       : Union[str, Any]						=WavaVecaConfig.from_pretrained(__A      )
 else:
  lowercase       : Optional[Any]						=WavaVecaConfig()

 if is_seq_class:
  lowercase       : Optional[Any]						=read_txt_into_dict(__A      )
  lowercase       : int						=idalabel
  lowercase       : Dict						=WavaVecaForSequenceClassification(__A      )
  lowercase       : str						=WavaVecaFeatureExtractor(
      feature_size=1       ,   sampling_rate=1_6_0_0_0       ,   padding_value=0       ,   do_normalize=__A       ,   return_attention_mask=__A       ,   )
  feature_extractor.save_pretrained(__A      )

 elif is_finetuned:
  if dict_path:
   lowercase       : int						=Dictionary.load(__A      )

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

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

  lowercase       : Optional[int]						=WavaVecaForCTC(__A      )
 else:
  lowercase       : Tuple						=WavaVecaForPreTraining(__A      )

 if is_finetuned or is_seq_class:
  lowercase	,		lowercase	,		lowercase       : Union[str, Any]						=fairseq.checkpoint_utils.load_model_ensemble_and_task(
      [checkpoint_path]       ,   arg_overrides={'''data''': '''/'''.join(dict_path.split('''/'''      )[:-1]      )}      )
 else:
  lowercase       : List[Any]						=argparse.Namespace(task='''audio_pretraining'''      )
  lowercase       : Any						=fairseq.tasks.setup_task(__A      )

  lowercase	,		lowercase	,		lowercase       : Any						=fairseq.checkpoint_utils.load_model_ensemble_and_task([checkpoint_path]       ,   task=__A      )

 lowercase       : Any						=model[0].eval()

 recursively_load_weights(__A       ,   __A       ,   not is_finetuned      )

 hf_wavavec.save_pretrained(__A      )


if __name__ == "__main__":
      SCREAMING_SNAKE_CASE							=						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'
      )
      parser.add_argument(
          '--is_seq_class',
          action='store_true',
          help='Whether the model to convert is a fine-tuned sequence classification model or not',
      )
      SCREAMING_SNAKE_CASE							=						parser.parse_args()

      SCREAMING_SNAKE_CASE							=						not args.not_finetuned and not args.is_seq_class
      convert_wavaveca_checkpoint(
          args.checkpoint_path,
          args.pytorch_dump_folder_path,
          args.config_path,
          args.dict_path,
          is_finetuned,
          args.is_seq_class,
      )











'''simple docstring'''



# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import copy
import importlib.metadata
import json
import os
from dataclasses import dataclass
from typing import Any, Dict, Union

from packaging import version

from ..utils import is_torch_available, logging


if is_torch_available():
      import torch


SCREAMING_SNAKE_CASE							=						logging.get_logger(__name__)
@dataclass
class UpperCAmelCase_       :




      """simple docstring"""




      def __init__(			self      :		List[Any]  ,   UpperCAmelCase      :		List[Any]=False  ,   UpperCAmelCase      :		Union[str, Any]=False  ,   UpperCAmelCase      :		Union[str, Any]=6.0  ,   UpperCAmelCase      :		Any=None  ,   UpperCAmelCase      :		Optional[Any]=False  ,   UpperCAmelCase      :		List[Any]=False  ,   UpperCAmelCase      :		Dict=None  ,   UpperCAmelCase      :		Tuple="fp4"  ,   UpperCAmelCase      :		str=False  ,   **UpperCAmelCase      :		Union[str, Any]  ,   )	->	str:


       '''simple docstring'''

       lowercase       : Optional[int]						=load_in_abit
       lowercase       : Union[str, Any]						=load_in_abit
       lowercase       : Tuple						=llm_inta_threshold
       lowercase       : Optional[Any]						=llm_inta_skip_modules
       lowercase       : int						=llm_inta_enable_fpaa_cpu_offload
       lowercase       : Dict						=llm_inta_has_fpaa_weight
       lowercase       : str						=bnb_abit_quant_type
       lowercase       : int						=bnb_abit_use_double_quant

       if bnb_abit_compute_dtype is None:
        lowercase       : str						=torch.floataa
       elif isinstance(UpperCAmelCase  ,   UpperCAmelCase		):
        lowercase       : Tuple						=getattr(UpperCAmelCase  ,   UpperCAmelCase		)
       elif isinstance(UpperCAmelCase  ,   torch.dtype		):
        lowercase       : Optional[int]						=bnb_abit_compute_dtype
       else:
        raise ValueError('''bnb_4bit_compute_dtype must be a string or a torch.dtype'''		)

       self.post_init()






      def 					A__   (			self      :		List[Any]		)	->	Union[str, Any]:


       '''simple docstring'''

       if not isinstance(self.llm_inta_threshold  ,   UpperCAmelCase		):
        raise ValueError('''llm_int8_threshold must be a float'''		)

       if self.llm_inta_skip_modules is not None and not isinstance(self.llm_inta_skip_modules  ,   UpperCAmelCase		):
        raise ValueError('''llm_int8_skip_modules must be a list of strings'''		)
       if not isinstance(self.llm_inta_enable_fpaa_cpu_offload  ,   UpperCAmelCase		):
        raise ValueError('''llm_int8_enable_fp32_cpu_offload must be a boolean'''		)

       if not isinstance(self.llm_inta_has_fpaa_weight  ,   UpperCAmelCase		):
        raise ValueError('''llm_int8_has_fp16_weight must be a boolean'''		)

       if self.bnb_abit_compute_dtype is not None and not isinstance(self.bnb_abit_compute_dtype  ,   torch.dtype		):
        raise ValueError('''bnb_4bit_compute_dtype must be torch.dtype'''		)

       if not isinstance(self.bnb_abit_quant_type  ,   UpperCAmelCase		):
        raise ValueError('''bnb_4bit_quant_type must be a string'''		)

       if not isinstance(self.bnb_abit_use_double_quant  ,   UpperCAmelCase		):
        raise ValueError('''bnb_4bit_use_double_quant must be a boolean'''		)

       if self.load_in_abit and not version.parse(importlib.metadata.version('''bitsandbytes'''		)		) >= version.parse(
           '''0.39.0'''		):
        raise ValueError(
            '''4 bit quantization requires bitsandbytes>=0.39.0 - please upgrade your bitsandbytes version'''		)






      def 					A__   (			self      :		Any		)	->	Tuple:


       '''simple docstring'''

       return self.load_in_abit or self.load_in_abit






      def 					A__   (			self      :		Tuple		)	->	Any:


       '''simple docstring'''

       if self.load_in_abit:
        return "llm_int8"
       elif self.load_in_abit and self.bnb_abit_quant_type == "fp4":
        return "fp4"
       elif self.load_in_abit and self.bnb_abit_quant_type == "nf4":
        return "nf4"
       else:
        return None






      @classmethod
      def 					A__   (			cls      :		Union[str, Any]  ,   UpperCAmelCase      :		Optional[Any]  ,   UpperCAmelCase      :		Any  ,   **UpperCAmelCase      :		List[str]		)	->	Tuple:


       '''simple docstring'''

       lowercase       : Dict						=cls(**UpperCAmelCase		)

       lowercase       : Dict						=[]
       for key, value in kwargs.items():
        if hasattr(UpperCAmelCase  ,   UpperCAmelCase		):
         setattr(UpperCAmelCase  ,   UpperCAmelCase  ,   UpperCAmelCase		)
         to_remove.append(UpperCAmelCase		)
       for key in to_remove:
        kwargs.pop(UpperCAmelCase  ,   UpperCAmelCase		)

       if return_unused_kwargs:
        return config, kwargs
       else:
        return config






      def 					A__   (			self      :		List[Any]  ,   UpperCAmelCase      :		Union[str, os.PathLike]		)	->	Dict:


       '''simple docstring'''

       with open(UpperCAmelCase  ,   '''w'''  ,   encoding='''utf-8'''		) as writer:
        lowercase       : Any						=self.to_dict()
        lowercase       : List[Any]						=json.dumps(UpperCAmelCase  ,   indent=2  ,   sort_keys=UpperCAmelCase		) + '''\n'''

        writer.write(UpperCAmelCase		)






      def 					A__   (			self      :		Optional[Any]		)	->	Dict[str, Any]:


       '''simple docstring'''

       lowercase       : Tuple						=copy.deepcopy(self.__dict__		)
       lowercase       : Optional[Any]						=str(output['''bnb_4bit_compute_dtype''']		).split('''.'''		)[1]

       return output






      def __repr__(			self      :		Optional[Any]		)	->	Union[str, Any]:


       '''simple docstring'''

       return f'{self.__class__.__name__} {self.to_json_string()}'






      def 					A__   (			self      :		List[Any]  ,   UpperCAmelCase      :		bool = True		)	->	str:


       '''simple docstring'''

       if use_diff is True:
        lowercase       : int						=self.to_diff_dict()
       else:
        lowercase       : List[str]						=self.to_dict()
       return json.dumps(UpperCAmelCase  ,   indent=2  ,   sort_keys=UpperCAmelCase		) + "\n"






      def 					A__   (			self      :		Any		)	->	Dict[str, Any]:


       '''simple docstring'''

       lowercase       : Any						=self.to_dict()

       # get the default config dict
       lowercase       : Union[str, Any]						=BitsAndBytesConfig().to_dict()

       lowercase       : int						={}

       # only serialize values that differ from the default config
       for key, value in config_dict.items():
        if value != default_config_dict[key]:
         lowercase       : Dict						=value

       return serializable_config_dict



