'''simple docstring'''


import argparse

import fairseq
import torch

from transformers import UniSpeechSatConfig, UniSpeechSatForCTC, UniSpeechSatForPreTraining, logging


logging.set_verbosity_info()
snake_case__					:      Dict				      =  logging.get_logger(__name__)

snake_case__					:      int				      =  {
    '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',
    'encoder.layer_norm_for_extract': 'layer_norm_for_extract',
    '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',
    'label_embs_concat': 'label_embeddings_concat',
    'mask_emb': 'masked_spec_embed',
    'spk_proj': 'speaker_proj',
}
snake_case__					:      str				      =  [
    'lm_head',
    'quantizer.weight_proj',
    'quantizer.codevectors',
    'project_q',
    'project_hid',
    'label_embeddings_concat',
    'speaker_proj',
    'layer_norm_for_extract',
]


def  _lowerCamelCase   (				lowerCamelCase_ :							str     ,					lowerCamelCase_ :							List[Any]     ,					lowerCamelCase_ :							Optional[Any]     ,					lowerCamelCase_ :							List[str]     ,					lowerCamelCase_ :							List[str]       ):



     """simple docstring"""





     for attribute in key.split('.'       ):
          UpperCAmelCase_							:       Optional[int]       =				getattr(UpperCamelCase__     ,					UpperCamelCase__       )

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

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


def  _lowerCamelCase   (				lowerCamelCase_ :							str     ,					lowerCamelCase_ :							Union[str, Any]       ):



     """simple docstring"""





     UpperCAmelCase_							:       Dict       =				[]
     UpperCAmelCase_							:       List[Any]       =				fairseq_model.state_dict()

     UpperCAmelCase_							:       List[Any]       =				hf_model.unispeech_sat.feature_extractor

     for name, value in fairseq_dict.items():
          UpperCAmelCase_							:       Union[str, Any]       =				False
          if "conv_layers" in name:
               load_conv_layer(
                   UpperCamelCase__     ,					UpperCamelCase__     ,					UpperCamelCase__     ,					UpperCamelCase__     ,					hf_model.config.feat_extract_norm == 'group'     ,					)
               UpperCAmelCase_							:       str       =				True
          else:
               for key, mapped_key in MAPPING.items():
                    UpperCAmelCase_							:       int       =				'unispeech_sat.' + 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]:
                         if "layer_norm_for_extract" in name and (".".join(name.split('.'       )[:-1]       ) != key):
                              # special case since naming is very similar
                              continue
                         UpperCAmelCase_							:       str       =				True
                         if "*" in mapped_key:
                              UpperCAmelCase_							:       int       =				name.split(UpperCamelCase__       )[0].split('.'       )[-2]
                              UpperCAmelCase_							:       List[Any]       =				mapped_key.replace('*'     ,					UpperCamelCase__       )
                         if "weight_g" in name:
                              UpperCAmelCase_							:       Tuple       =				'weight_g'
                         elif "weight_v" in name:
                              UpperCAmelCase_							:       Any       =				'weight_v'
                         elif "bias" in name:
                              UpperCAmelCase_							:       List[str]       =				'bias'
                         elif "weight" in name:
                              # TODO: don't match quantizer.weight_proj
                              UpperCAmelCase_							:       Optional[int]       =				'weight'
                         else:
                              UpperCAmelCase_							:       List[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  _lowerCamelCase   (				lowerCamelCase_ :							Optional[int]     ,					lowerCamelCase_ :							Tuple     ,					lowerCamelCase_ :							Optional[Any]     ,					lowerCamelCase_ :							Tuple     ,					lowerCamelCase_ :							int       ):



     """simple docstring"""





     UpperCAmelCase_							:       int       =				full_name.split('conv_layers.'       )[-1]
     UpperCAmelCase_							:       Dict       =				name.split('.'       )
     UpperCAmelCase_							:       int       =				int(items[0]       )
     UpperCAmelCase_							:       Union[str, Any]       =				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.'''       )
               UpperCAmelCase_							:       int       =				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.'''       )
               UpperCAmelCase_							:       int       =				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[layer_id].layer_norm.bias.data.shape} was found.'''       )
               UpperCAmelCase_							:       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[layer_id].layer_norm.weight.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}.'''       )
     else:
          unused_weights.append(UpperCamelCase__       )


@torch.no_grad()
def  _lowerCamelCase   (				lowerCamelCase_ :							Dict     ,					lowerCamelCase_ :							str     ,					lowerCamelCase_ :							Union[str, Any]=None     ,					lowerCamelCase_ :							Tuple=None     ,					lowerCamelCase_ :							int=True       ):



     """simple docstring"""





     if config_path is not None:
          UpperCAmelCase_							:       Tuple       =				UniSpeechSatConfig.from_pretrained(UpperCamelCase__       )
     else:
          UpperCAmelCase_							:       str       =				UniSpeechSatConfig()

     UpperCAmelCase_							:       List[str]       =				''

     if is_finetuned:
          UpperCAmelCase_							:       List[str]       =				UniSpeechSatForCTC(UpperCamelCase__       )
     else:
          UpperCAmelCase_							:       List[str]       =				UniSpeechSatForPreTraining(UpperCamelCase__       )

     UpperCAmelCase_       ,      UpperCAmelCase_       ,      UpperCAmelCase_							:       Tuple       =				fairseq.checkpoint_utils.load_model_ensemble_and_task(
         [checkpoint_path]     ,					arg_overrides={'data': '/'.join(dict_path.split('/'       )[:-1]       )}       )
     UpperCAmelCase_							:       List[Any]       =				model[0].eval()

     recursively_load_weights(UpperCamelCase__     ,					UpperCamelCase__       )

     hf_wavavec.save_pretrained(UpperCamelCase__       )


if __name__ == "__main__":
 snake_case__					:      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'''
 )
 snake_case__					:      Optional[Any]				      =  parser.parse_args()
 convert_unispeech_sat_checkpoint(
     args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.dict_path, not args.not_finetuned
 )







'''simple docstring'''


from __future__ import annotations

from collections import deque
from collections.abc import Iterator
from dataclasses import dataclass






@dataclass
class 					__SCREAMING_SNAKE_CASE						:







  '''simple docstring'''






  lowerCamelCase_		:int
  lowerCamelCase_		:int







class 					__SCREAMING_SNAKE_CASE						:







  '''simple docstring'''






  def __init__(			self				,		snake_case_						):




       '''simple docstring'''

       UpperCAmelCase_							:       list[list[Edge]]       =				[[] for _ in range(snake_case_						)]
       UpperCAmelCase_							:       Any       =				size


  def __getitem__(			self				,		snake_case_						):




       '''simple docstring'''

       return iter(self._graph[vertex]						)


  @property
  def _UpperCamelCase					(			self						):




       '''simple docstring'''

       return self._size


  def _UpperCamelCase					(			self				,		snake_case_				,		snake_case_				,		snake_case_						):




       '''simple docstring'''

       if weight not in (0, 1):
            raise ValueError('Edge weight must be either 0 or 1.'						)

       if to_vertex < 0 or to_vertex >= self.size:
            raise ValueError('Vertex indexes must be in [0; size).'						)

       self._graph[from_vertex].append(Edge(snake_case_				,		snake_case_						)						)






  def _UpperCamelCase					(			self				,		snake_case_				,		snake_case_						):




       '''simple docstring'''

       UpperCAmelCase_							:       Optional[int]       =				deque([start_vertex]						)
       UpperCAmelCase_							:       list[int | None]       =				[None] * self.size
       UpperCAmelCase_							:       str       =				0

       while queue:
            UpperCAmelCase_							:       Optional[int]       =				queue.popleft()
            UpperCAmelCase_							:       List[str]       =				distances[current_vertex]
            if current_distance is None:
                 continue

            for edge in self[current_vertex]:
                 UpperCAmelCase_							:       Union[str, Any]       =				current_distance + edge.weight
                 UpperCAmelCase_							:       Union[str, Any]       =				distances[edge.destination_vertex]
                 if (
                     isinstance(snake_case_				,		snake_case_						)
                     and new_distance >= dest_vertex_distance
                 ):
                      continue
                 UpperCAmelCase_							:       Optional[Any]       =				new_distance
                 if edge.weight == 0:
                      queue.appendleft(edge.destination_vertex						)
                 else:
                      queue.append(edge.destination_vertex						)

       if distances[finish_vertex] is None:
            raise ValueError('No path from start_vertex to finish_vertex.'						)

       return distances[finish_vertex]


if __name__ == "__main__":
 import doctest

 doctest.testmod()






