



'''simple docstring'''






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()
lowerCAmelCase: str							=						logging.get_logger(__name__)

lowerCAmelCase: Optional[Any]							=						{
    '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 				lowerCamelCase__	(						_A			,      _A			,      _A			,      _A			,      _A    ):
  for attribute in key.split('.'    ):
    a						: Optional[int]					       =					getattr(_A			,      _A    )

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

  logger.info(f"""{key + "." + weight_type if weight_type is not None else ""} was initialized from {full_name}."""    )
def 				lowerCamelCase__	(						_A			,      _A			,      _A    ):
  a						: str					       =					[]
  a						: int					       =					fairseq_model.state_dict()

  a						: int					       =					hf_model.hubert.feature_extractor if is_finetuned else hf_model.feature_extractor

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

  if is_finetuned:
    if dict_path:
      a						: str					       =					Dictionary.load(_A    )

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

    a						: Optional[int]					       =					HubertForCTC(_A    )
  else:
    a						: Union[str, Any]					       =					HubertModel(_A    )

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

  a						: Union[str, Any]					       =					model[0].eval()

  recursively_load_weights(_A			,      _A			,      _A    )

  hf_wavavec.save_pretrained(_A    )


if __name__ == "__main__":
     lowerCAmelCase: Optional[int]							=						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: 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
     )




'''simple docstring'''






import argparse
import os
import re

import packaging.version


lowerCAmelCase: List[str]							=						'examples/'
lowerCAmelCase: List[Any]							=						{
    'examples': (re.compile(r'^check_min_version\("[^"]+"\)\s*$', re.MULTILINE), 'check_min_version("VERSION")\n'),
    'init': (re.compile(r'^__version__\s+=\s+"([^"]+)"\s*$', re.MULTILINE), '__version__ = "VERSION"\n'),
    'setup': (re.compile(r'^(\s*)version\s*=\s*"[^"]+",', re.MULTILINE), r'\1version="VERSION",'),
    'doc': (re.compile(r'^(\s*)release\s*=\s*"[^"]+"$', re.MULTILINE), 'release = "VERSION"\n'),
}
lowerCAmelCase: str							=						{
    'init': 'src/transformers/__init__.py',
    'setup': 'setup.py',
}
lowerCAmelCase: str							=						'README.md'
def 				lowerCamelCase__	(						_A			,      _A			,      _A    ):
  with open(_A			,      'r'			,      encoding='utf-8'			,      newline='\n'    ) as f:
    a						: Tuple					       =					f.read()
  a       ,     a						: Tuple					       =					REPLACE_PATTERNS[pattern]
  a						: Dict					       =					replace.replace('VERSION'			,      _A    )
  a						: Dict					       =					re_pattern.sub(_A			,      _A    )
  with open(_A			,      'w'			,      encoding='utf-8'			,      newline='\n'    ) as f:
    f.write(_A    )
def 				lowerCamelCase__	(						_A    ):
  for folder, directories, fnames in os.walk(_A    ):
    # Removing some of the folders with non-actively maintained examples from the walk
    if "research_projects" in directories:
      directories.remove('research_projects'    )
    if "legacy" in directories:
      directories.remove('legacy'    )
    for fname in fnames:
      if fname.endswith('.py'    ):
        update_version_in_file(os.path.join(_A			,      _A    )			,      _A			,      pattern='examples'    )
def 				lowerCamelCase__	(						_A			,      _A=False    ):
  for pattern, fname in REPLACE_FILES.items():
    update_version_in_file(_A			,      _A			,      _A    )
  if not patch:
    update_version_in_examples(_A    )
def 				lowerCamelCase__	(						):
  a						: Tuple					       =					'🤗 Transformers currently provides the following architectures'
  a						: Any					       =					'1. Want to contribute a new model?'
  with open(_A			,      'r'			,      encoding='utf-8'			,      newline='\n'    ) as f:
    a						: Tuple					       =					f.readlines()

  # Find the start of the list.
  a						: Optional[int]					       =					0
  while not lines[start_index].startswith(_start_prompt    ):
    start_index += 1
  start_index += 1

  a						: Optional[int]					       =					start_index
  # Update the lines in the model list.
  while not lines[index].startswith(_end_prompt    ):
    if lines[index].startswith('1.'    ):
      a						: List[Any]					       =					lines[index].replace(
          'https://huggingface.co/docs/transformers/main/model_doc'			,      'https://huggingface.co/docs/transformers/model_doc'			,      )
    index += 1

  with open(_A			,      'w'			,      encoding='utf-8'			,      newline='\n'    ) as f:
    f.writelines(_A    )
def 				lowerCamelCase__	(						):
  with open(REPLACE_FILES['init']			,      'r'    ) as f:
    a						: Union[str, Any]					       =					f.read()
  a						: Tuple					       =					REPLACE_PATTERNS['init'][0].search(_A    ).groups()[0]
  return packaging.version.parse(_A    )
def 				lowerCamelCase__	(						_A=False    ):
  a						: int					       =					get_version()
  if patch and default_version.is_devrelease:
    raise ValueError('Can\'t create a patch version from the dev branch, checkout a released version!'    )
  if default_version.is_devrelease:
    a						: Any					       =					default_version.base_version
  elif patch:
    a						: Dict					       =					f"""{default_version.major}.{default_version.minor}.{default_version.micro + 1}"""
  else:
    a						: Union[str, Any]					       =					f"""{default_version.major}.{default_version.minor + 1}.0"""

  # Now let's ask nicely if that's the right one.
  a						: List[Any]					       =					input(f"""Which version are you releasing? [{default_version}]"""    )
  if len(_A    ) == 0:
    a						: Union[str, Any]					       =					default_version

  print(f"""Updating version to {version}."""    )
  global_version_update(_A			,      patch=_A    )
  if not patch:
    print('Cleaning main README, don\'t forget to run `make fix-copies`.'    )
    clean_main_ref_in_model_list()
def 				lowerCamelCase__	(						):
  a						: int					       =					get_version()
  a						: Any					       =					f"""{current_version.major}.{current_version.minor + 1}.0.dev0"""
  a						: int					       =					current_version.base_version

  # Check with the user we got that right.
  a						: Tuple					       =					input(f"""Which version are we developing now? [{dev_version}]"""    )
  if len(_A    ) == 0:
    a						: Optional[int]					       =					dev_version

  print(f"""Updating version to {version}."""    )
  global_version_update(_A    )
  print('Cleaning main README, don\'t forget to run `make fix-copies`.'    )
  clean_main_ref_in_model_list()


if __name__ == "__main__":
     lowerCAmelCase: Tuple							=						argparse.ArgumentParser()
     parser.add_argument('--post_release', action='store_true', help='Whether this is pre or post release.')
     parser.add_argument('--patch', action='store_true', help='Whether or not this is a patch release.')
     lowerCAmelCase: Optional[Any]							=						parser.parse_args()
     if not args.post_release:
          pre_release_work(patch=args.patch)
     elif args.patch:
          print('Nothing to do after a patch :-)')
     else:
          post_release_work()