






import argparse
import json
import os

import fairseq
import torch
from fairseq.data import Dictionary

from transformers import (
    HubertConfig,
    HubertForCTC,
    HubertModel,
    WavaVecaCTCTokenizer,
    WavaVecaFeatureExtractor,
    WavaVecaProcessor,
    logging,
)


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

A_		:		Dict		     =			{
    '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',
    'w2v_model.layer_norm': 'feature_projection.layer_norm',
    'w2v_encoder.proj': 'lm_head',
    'mask_emb': 'masked_spec_embed',
}

def   snake_case (UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__  )	->    List[str]:
  for attribute in key.split('.'  ):
    UpperCamelCase_: Tuple   				=    getattr(UpperCAmelCase__     ,   UpperCAmelCase__  )

  if weight_type is not None:
    UpperCamelCase_: List[str]   				=    getattr(UpperCAmelCase__     ,   UpperCAmelCase__  ).shape
  else:
    UpperCamelCase_: Union[str, 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":
    UpperCamelCase_: int   				=    value
  elif weight_type == "weight_g":
    UpperCamelCase_: Dict   				=    value
  elif weight_type == "weight_v":
    UpperCamelCase_: List[Any]   				=    value
  elif weight_type == "bias":
    UpperCamelCase_: Tuple   				=    value
  else:
    UpperCamelCase_: Optional[int]   				=    value

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

def   snake_case (UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__  )	->    Optional[int]:
  UpperCamelCase_: Tuple   				=    []
  UpperCamelCase_: Dict   				=    fairseq_model.state_dict()

  UpperCamelCase_: Union[str, Any]   				=    hf_model.hubert.feature_extractor if is_finetuned else hf_model.feature_extractor

  for name, value in fairseq_dict.items():
    UpperCamelCase_: Tuple   				=    False
    if "conv_layers" in name:
      load_conv_layer(
          UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   hf_model.config.feat_extract_norm == 'group'     ,   )
      UpperCamelCase_: Dict   				=    True
    else:
      for key, mapped_key in MAPPING.items():
        UpperCamelCase_: Tuple   				=    'hubert.' + 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] and not is_finetuned):
          UpperCamelCase_: Union[str, Any]   				=    True
          if "*" in mapped_key:
            UpperCamelCase_: str   				=    name.split(UpperCAmelCase__  )[0].split('.'  )[-2]
            UpperCamelCase_: Dict   				=    mapped_key.replace('*'     ,   UpperCAmelCase__  )
          if "weight_g" in name:
            UpperCamelCase_: Tuple   				=    'weight_g'
          elif "weight_v" in name:
            UpperCamelCase_: Dict   				=    'weight_v'
          elif "weight" in name:
            UpperCamelCase_: List[Any]   				=    'weight'
          elif "bias" in name:
            UpperCamelCase_: List[str]   				=    'bias'
          else:
            UpperCamelCase_: int   				=    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   snake_case (UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__  )	->    Tuple:
  UpperCamelCase_: Union[str, Any]   				=    full_name.split('conv_layers.'  )[-1]
  UpperCamelCase_: int   				=    name.split('.'  )
  UpperCamelCase_: Optional[Any]   				=    int(items[0]  )
  UpperCamelCase_: Optional[int]   				=    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.'''
      )
      UpperCamelCase_: Dict   				=    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.'''
      )
      UpperCamelCase_: Tuple   				=    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."
      )
      UpperCamelCase_: List[str]   				=    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.'''
      )
      UpperCamelCase_: Optional[int]   				=    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   snake_case (UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__=None     ,   UpperCAmelCase__=None     ,   UpperCAmelCase__=True  )	->    Dict:
  if config_path is not None:
    UpperCamelCase_: Tuple   				=    HubertConfig.from_pretrained(UpperCAmelCase__  )
  else:
    UpperCamelCase_: Tuple   				=    HubertConfig()

  if is_finetuned:
    if dict_path:
      UpperCamelCase_: str   				=    Dictionary.load(UpperCAmelCase__  )

      # important change bos & pad token id since CTC symbol is <pad> and
      # not <s> as in fairseq
      UpperCamelCase_: Dict   				=    target_dict.pad_index
      UpperCamelCase_: int   				=    target_dict.bos_index
      UpperCamelCase_: Any   				=    target_dict.eos_index
      UpperCamelCase_: Union[str, Any]   				=    len(target_dict.symbols  )
      UpperCamelCase_: Dict   				=    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__  )
      UpperCamelCase_: Dict   				=    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__     ,   )
      UpperCamelCase_: Tuple   				=    True if config.feat_extract_norm == 'layer' else False
      UpperCamelCase_: Union[str, Any]   				=    WavaVecaFeatureExtractor(
          feature_size=1     ,   sampling_rate=1_6_0_0_0     ,   padding_value=0     ,   do_normalize=UpperCAmelCase__     ,   return_attention_mask=UpperCAmelCase__     ,   )
      UpperCamelCase_: List[Any]   				=    WavaVecaProcessor(feature_extractor=UpperCAmelCase__     ,   tokenizer=UpperCAmelCase__  )
      processor.save_pretrained(UpperCAmelCase__  )

    UpperCamelCase_: Union[str, Any]   				=    HubertForCTC(UpperCAmelCase__  )
  else:
    UpperCamelCase_: Tuple   				=    HubertModel(UpperCAmelCase__  )

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

  UpperCamelCase_: str   				=    model[0].eval()

  recursively_load_weights(UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__  )

  hf_wavavec.save_pretrained(UpperCAmelCase__  )


if __name__ == "__main__":
   A_		:		Optional[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(
       '--not_finetuned', action='store_true', help='Whether the model to convert is a fine-tuned model or not'
   )
   A_		:		Tuple		     =			parser.parse_args()
   convert_hubert_checkpoint(
       args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.dict_path, not args.not_finetuned
   )







# Copyright 2021 The HuggingFace 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 argparse
import os

from accelerate.utils import ComputeEnvironment

from .cluster import get_cluster_input
from .config_args import cache_dir, default_config_file, default_yaml_config_file, load_config_from_file  # noqa: F401
from .config_utils import _ask_field, _ask_options, _convert_compute_environment  # noqa: F401
from .sagemaker import get_sagemaker_input


A_		:		Any		     =			'Launches a series of prompts to create and save a `default_config.yaml` configuration file for your training system. Should always be ran first on your machine'

def   snake_case ()	->    Union[str, Any]:
  UpperCamelCase_: Tuple   				=    _ask_options(
      'In which compute environment are you running?'     ,   ['This machine', 'AWS (Amazon SageMaker)']     ,   _convert_compute_environment     ,   )
  if compute_environment == ComputeEnvironment.AMAZON_SAGEMAKER:
    UpperCamelCase_: List[str]   				=    get_sagemaker_input()
  else:
    UpperCamelCase_: List[str]   				=    get_cluster_input()
  return config

def   snake_case (UpperCAmelCase__=None  )	->    Union[str, Any]:
  if subparsers is not None:
    UpperCamelCase_: List[Any]   				=    subparsers.add_parser('config'     ,   description=UpperCAmelCase__  )
  else:
    UpperCamelCase_: List[Any]   				=    argparse.ArgumentParser('Accelerate config command'     ,   description=UpperCAmelCase__  )

  parser.add_argument(
      '--config_file'     ,   default=UpperCAmelCase__     ,   help=(
          'The path to use to store the config file. Will default to a file named default_config.yaml in the cache '
          'location, which is the content of the environment `HF_HOME` suffixed with \'accelerate\', or if you don\'t have '
          'such an environment variable, your cache directory (\'~/.cache\' or the content of `XDG_CACHE_HOME`) suffixed '
          'with \'huggingface\'.'
      )     ,   )

  if subparsers is not None:
    parser.set_defaults(func=UpperCAmelCase__  )
  return parser

def   snake_case (UpperCAmelCase__  )	->    List[Any]:
  UpperCamelCase_: Union[str, Any]   				=    get_user_input()
  if args.config_file is not None:
    UpperCamelCase_: Tuple   				=    args.config_file
  else:
    if not os.path.isdir(UpperCAmelCase__  ):
      os.makedirs(UpperCAmelCase__  )
    UpperCamelCase_: Dict   				=    default_yaml_config_file

  if config_file.endswith('.json'  ):
    config.to_json_file(UpperCAmelCase__  )
  else:
    config.to_yaml_file(UpperCAmelCase__  )
  print(F'''accelerate configuration saved at {config_file}'''  )

def   snake_case ()	->    str:
  UpperCamelCase_: Tuple   				=    config_command_parser()
  UpperCamelCase_: int   				=    parser.parse_args()
  config_command(UpperCAmelCase__  )


if __name__ == "__main__":
   main()