




import argparse

import torch
from saprl.hub import distilhubert

from transformers import HubertConfig, HubertModel, WavaVecaFeatureExtractor, logging


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

a__      =		{
    """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""",
    """mask_emb""": """masked_spec_embed""",
}



def  lowercase    (				SCREAMING_SNAKE_CASE__    :		Dict			,    SCREAMING_SNAKE_CASE__    :		Any			,    SCREAMING_SNAKE_CASE__    :		Optional[Any]			,    SCREAMING_SNAKE_CASE__    :		Any			,    SCREAMING_SNAKE_CASE__    :		int			)					->       List[str]:
 for attribute in key.split("""."""			):
  _snake_case      :					Union[str, Any]					=   getattr(SCREAMING_SNAKE_CASE__			,    SCREAMING_SNAKE_CASE__			)

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

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



def  lowercase    (				SCREAMING_SNAKE_CASE__    :		int			,    SCREAMING_SNAKE_CASE__    :		List[Any]			)					->       Optional[Any]:
 _snake_case      :					Dict					=   []
 _snake_case      :					Any					=   fairseq_model.state_dict()

 _snake_case      :					str					=   hf_model.feature_extractor

 for name, value in fairseq_dict.items():
  _snake_case      :					Tuple					=   False
  if "conv_layers" in name:
   load_conv_layer(
       SCREAMING_SNAKE_CASE__			,    SCREAMING_SNAKE_CASE__			,    SCREAMING_SNAKE_CASE__			,    SCREAMING_SNAKE_CASE__			,    hf_model.config.feat_extract_norm == """group"""			,    )
   _snake_case      :					Dict					=   True
  else:
   for key, mapped_key in MAPPING.items():
    _snake_case      :					str					=   mapped_key

    if key in name:
     _snake_case      :					List[str]					=   True
     if "*" in mapped_key:
      _snake_case      :					Any					=   name.split(SCREAMING_SNAKE_CASE__			)[0].split("""."""			)[-2]
      _snake_case      :					Optional[int]					=   mapped_key.replace("""*"""			,    SCREAMING_SNAKE_CASE__			)
     if "weight_g" in name:
      _snake_case      :					Union[str, Any]					=   """weight_g"""
     elif "weight_v" in name:
      _snake_case      :					Any					=   """weight_v"""
     elif "weight" in name:
      _snake_case      :					List[str]					=   """weight"""
     elif "bias" in name:
      _snake_case      :					List[str]					=   """bias"""
     else:
      _snake_case      :					Any					=   None
     set_recursively(SCREAMING_SNAKE_CASE__			,    SCREAMING_SNAKE_CASE__			,    SCREAMING_SNAKE_CASE__			,    SCREAMING_SNAKE_CASE__			,    SCREAMING_SNAKE_CASE__			)
    continue
  if not is_used:
   unused_weights.append(SCREAMING_SNAKE_CASE__			)

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



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



def  lowercase    (				SCREAMING_SNAKE_CASE__    :		Union[str, Any]			)					->       Optional[Any]:
 _snake_case      :					str					=   HubertConfig()
 _snake_case      :					Union[str, Any]					=   model.config

 _snake_case      :					Union[str, Any]					=   fs_config.activation_dropout
 _snake_case      :					int					=   False
 _snake_case      :					Optional[Any]					=   fs_config.attention_dropout
 _snake_case      :					List[Any]					=   False
 _snake_case      :					int					=   eval(fs_config.extractor_conv_feature_layers			)
 _snake_case      :					str					=   [x[0] for x in conv_layers]
 _snake_case      :					Tuple					=   [x[1] for x in conv_layers]
 _snake_case      :					Tuple					=   [x[2] for x in conv_layers]
 _snake_case      :					str					=   """gelu"""
 _snake_case      :					Optional[Any]					=   """layer""" if fs_config.extractor_mode == """layer_norm""" else """group"""
 _snake_case      :					str					=   False
 _snake_case      :					List[Any]					=   0.0
 _snake_case      :					List[Any]					=   0.0
 _snake_case      :					Optional[Any]					=   fs_config.activation_fn
 _snake_case      :					Tuple					=   fs_config.dropout
 _snake_case      :					Optional[int]					=   fs_config.encoder_embed_dim
 _snake_case      :					Union[str, Any]					=   0.0_2
 _snake_case      :					int					=   fs_config.encoder_ffn_embed_dim
 _snake_case      :					Union[str, Any]					=   1e-5
 _snake_case      :					Tuple					=   0.0
 _snake_case      :					Dict					=   fs_config.encoder_attention_heads
 _snake_case      :					str					=   fs_config.conv_pos_groups
 _snake_case      :					Tuple					=   fs_config.conv_pos
 _snake_case      :					List[str]					=   len(SCREAMING_SNAKE_CASE__			)
 _snake_case      :					Optional[Any]					=   fs_config.encoder_layers

 return config



@torch.no_grad()
def  lowercase    (				SCREAMING_SNAKE_CASE__    :		Dict			,    SCREAMING_SNAKE_CASE__    :		Optional[int]=None			)					->       List[str]:
 _snake_case      :					Any					=   distilhubert().model.model

 if config_path is not None:
  _snake_case      :					str					=   HubertConfig.from_pretrained(SCREAMING_SNAKE_CASE__			)
 else:
  _snake_case      :					Tuple					=   convert_config(SCREAMING_SNAKE_CASE__			)
 _snake_case      :					Any					=   model.eval()

 _snake_case      :					Any					=   WavaVecaFeatureExtractor(
     feature_size=1			,    sampling_rate=16_000			,    padding_value=0			,    do_normalize=SCREAMING_SNAKE_CASE__			,    return_attention_mask=SCREAMING_SNAKE_CASE__			,    )
 _snake_case      :					List[str]					=   HubertModel(SCREAMING_SNAKE_CASE__			)

 recursively_load_weights(SCREAMING_SNAKE_CASE__			,    SCREAMING_SNAKE_CASE__			)

 feature_extractor.save_pretrained(SCREAMING_SNAKE_CASE__			)
 hf_model.save_pretrained(SCREAMING_SNAKE_CASE__			)


if __name__ == "__main__":
  a__      =		argparse.ArgumentParser()
  parser.add_argument("""--pytorch_dump_folder_path""", default=None, type=str, help="""Path to the output PyTorch model.""")
  parser.add_argument("""--config_path""", default=None, type=str, help="""Path to hf config.json of model to convert""")
  a__      =		parser.parse_args()
  convert_hubert_checkpoint(args.pytorch_dump_folder_path, args.config_path)





import os
import warnings
from pathlib import Path
from typing import Optional, Union

from .dynamic_module_utils import custom_object_save
from .tokenization_utils_base import PreTrainedTokenizerBase
from .utils import PushToHubMixin, copy_func, direct_transformers_import, logging


a__      =		logging.get_logger(__name__)

# Dynamically import the Transformers module to grab the attribute classes of the processor form their names.
a__      =		direct_transformers_import(Path(__file__).parent)


a__      =		{
    """AutoTokenizer""": """PreTrainedTokenizerBase""",
    """AutoFeatureExtractor""": """FeatureExtractionMixin""",
    """AutoImageProcessor""": """ImageProcessingMixin""",
}







class 						snake_case  (		SCREAMING_SNAKE_CASE_					):



   '''simple docstring'''






   snake_case_   :	List[str]       =						["""feature_extractor""", """tokenizer"""]
   # Names need to be attr_class for attr in attributes
   snake_case_   :	Optional[Any]       =						None
   snake_case_   :	List[str]       =						None
   snake_case_   :	List[str]       =						None

   def __init__(   self							:     int			,						*lowerCAmelCase							:     Any			,						**lowerCAmelCase							:     int)			->			Optional[Any]:

    """simple docstring"""







    for key in kwargs:
     if key not in self.attributes:
      raise TypeError(F'''Unexpected keyword argument {key}.''')
    for arg, attribute_name in zip(lowerCAmelCase			,						self.attributes):
     if attribute_name in kwargs:
      raise TypeError(F'''Got multiple values for argument {attribute_name}.''')
     else:
      _snake_case      :					List[str]					=   arg

    if len(lowerCAmelCase) != len(self.attributes):
     raise ValueError(
         F'''This processor requires {len(self.attributes)} arguments: {', '.join(self.attributes)}. Got '''
         F'''{len(lowerCAmelCase)} arguments instead.''')

    # Check each arg is of the proper class (this will also catch a user initializing in the wrong order)
    for attribute_name, arg in kwargs.items():
     _snake_case      :					Optional[Any]					=   getattr(self			,						F'''{attribute_name}_class''')
     # Nothing is ever going to be an instance of "AutoXxx", in that case we check the base class.
     _snake_case      :					List[str]					=   AUTO_TO_BASE_CLASS_MAPPING.get(lowerCAmelCase			,						lowerCAmelCase)
     if isinstance(lowerCAmelCase			,						lowerCAmelCase):
      _snake_case      :					Optional[Any]					=   tuple(getattr(lowerCAmelCase			,						lowerCAmelCase) for n in class_name if n is not None)
     else:
      _snake_case      :					List[Any]					=   getattr(lowerCAmelCase			,						lowerCAmelCase)

     if not isinstance(lowerCAmelCase			,						lowerCAmelCase):
      raise ValueError(
          F'''Received a {type(lowerCAmelCase).__name__} for argument {attribute_name}, but a {class_name} was expected.''')

     setattr(self			,						lowerCAmelCase			,						lowerCAmelCase)

   def __repr__(   self							:     int)			->			Dict:

    """simple docstring"""







    _snake_case      :					Union[str, Any]					=   [F'''- {name}: {repr(getattr(self			,						lowerCAmelCase))}''' for name in self.attributes]
    _snake_case      :					str					=   """\n""".join(lowerCAmelCase)
    return F'''{self.__class__.__name__}:\n{attributes_repr}'''

   def        UpperCamelCase_					(   self							:     Optional[int]			,						lowerCAmelCase							:     Optional[int]			,						lowerCAmelCase							:     bool = False			,						**lowerCAmelCase							:     Union[str, Any])			->			Dict:

    """simple docstring"""







    os.makedirs(lowerCAmelCase			,						exist_ok=lowerCAmelCase)

    if push_to_hub:
     _snake_case      :					str					=   kwargs.pop("""commit_message"""			,						lowerCAmelCase)
     _snake_case      :					Union[str, Any]					=   kwargs.pop("""repo_id"""			,						save_directory.split(os.path.sep)[-1])
     _snake_case      :					Tuple					=   self._create_repo(lowerCAmelCase			,						**lowerCAmelCase)
     _snake_case      :					Any					=   self._get_files_timestamps(lowerCAmelCase)
    # If we have a custom config, we copy the file defining it in the folder and set the attributes so it can be
    # loaded from the Hub.
    if self._auto_class is not None:
     _snake_case      :					Any					=   [getattr(self			,						lowerCAmelCase) for attribute_name in self.attributes]
     _snake_case      :					Any					=   [(a.init_kwargs if isinstance(lowerCAmelCase			,						lowerCAmelCase) else a) for a in attrs]
     custom_object_save(self			,						lowerCAmelCase			,						config=lowerCAmelCase)

    for attribute_name in self.attributes:
     _snake_case      :					Optional[int]					=   getattr(self			,						lowerCAmelCase)
     # Include the processor class in the attribute config so this processor can then be reloaded with the
     # `AutoProcessor` API.
     if hasattr(lowerCAmelCase			,						"""_set_processor_class"""):
      attribute._set_processor_class(self.__class__.__name__)
     attribute.save_pretrained(lowerCAmelCase)

    if self._auto_class is not None:
     # We added an attribute to the init_kwargs of the tokenizers, which needs to be cleaned up.
     for attribute_name in self.attributes:
      _snake_case      :					List[str]					=   getattr(self			,						lowerCAmelCase)
      if isinstance(lowerCAmelCase			,						lowerCAmelCase):
       del attribute.init_kwargs["auto_map"]

    if push_to_hub:
     self._upload_modified_files(
         lowerCAmelCase			,						lowerCAmelCase			,						lowerCAmelCase			,						commit_message=lowerCAmelCase			,						token=kwargs.get("""use_auth_token""")			,						)

   @classmethod
   def        UpperCamelCase_					(   cls							:     List[Any]			,						lowerCAmelCase							:     Union[str, os.PathLike]			,						lowerCAmelCase							:     Optional[Union[str, os.PathLike]] = None			,						lowerCAmelCase							:     bool = False			,						lowerCAmelCase							:     bool = False			,						lowerCAmelCase							:     Optional[Union[str, bool]] = None			,						lowerCAmelCase							:     str = "main"			,						**lowerCAmelCase							:     Dict			,						)			->			Tuple:

    """simple docstring"""







    _snake_case      :					Dict					=   cache_dir
    _snake_case      :					Optional[Any]					=   force_download
    _snake_case      :					Dict					=   local_files_only
    _snake_case      :					Any					=   revision

    _snake_case      :					Dict					=   kwargs.pop("""use_auth_token"""			,						lowerCAmelCase)
    if use_auth_token is not None:
     warnings.warn(
         """The `use_auth_token` argument is deprecated and will be removed in v5 of Transformers."""			,						lowerCAmelCase)
     if token is not None:
      raise ValueError(
          """`token` and `use_auth_token` are both specified. Please set only the argument `token`.""")
     _snake_case      :					str					=   use_auth_token

    if token is not None:
     # change to `token` in a follow-up PR
     _snake_case      :					Tuple					=   token

    _snake_case      :					Tuple					=   cls._get_arguments_from_pretrained(lowerCAmelCase			,						**lowerCAmelCase)
    return cls(*lowerCAmelCase)

   @classmethod
   def        UpperCamelCase_					(   cls							:     Optional[Any]			,						lowerCAmelCase							:     str="AutoProcessor")			->			int:

    """simple docstring"""







    if not isinstance(lowerCAmelCase			,						lowerCAmelCase):
     _snake_case      :					List[Any]					=   auto_class.__name__

    import transformers.models.auto as auto_module

    if not hasattr(lowerCAmelCase			,						lowerCAmelCase):
     raise ValueError(F'''{auto_class} is not a valid auto class.''')

    _snake_case      :					List[str]					=   auto_class

   @classmethod
   def        UpperCamelCase_					(   cls							:     List[Any]			,						lowerCAmelCase							:     Any			,						**lowerCAmelCase							:     Union[str, Any])			->			Optional[Any]:

    """simple docstring"""







    _snake_case      :					Optional[Any]					=   []
    for attribute_name in cls.attributes:
     _snake_case      :					Optional[Any]					=   getattr(cls			,						F'''{attribute_name}_class''')
     if isinstance(lowerCAmelCase			,						lowerCAmelCase):
      _snake_case      :					Tuple					=   tuple(getattr(lowerCAmelCase			,						lowerCAmelCase) if n is not None else None for n in class_name)
      _snake_case      :					str					=   kwargs.get("""use_fast"""			,						lowerCAmelCase)
      if use_fast and classes[1] is not None:
       _snake_case      :					Any					=   classes[1]
      else:
       _snake_case      :					Union[str, Any]					=   classes[0]
     else:
      _snake_case      :					Optional[Any]					=   getattr(lowerCAmelCase			,						lowerCAmelCase)

     args.append(attribute_class.from_pretrained(lowerCAmelCase			,						**lowerCAmelCase))
    return args


   @property
   def        UpperCamelCase_					(   self							:     Optional[int])			->			str:

    """simple docstring"""







    _snake_case      :					Tuple					=   getattr(self			,						self.attributes[0])
    return getattr(lowerCAmelCase			,						"""model_input_names"""			,						lowerCAmelCase)
a__      =		copy_func(ProcessorMixin.push_to_hub)
if ProcessorMixin.push_to_hub.__doc__ is not None:
  a__      =		ProcessorMixin.push_to_hub.__doc__.format(
      object="""processor""", object_class="""AutoProcessor""", object_files="""processor files"""
  )
