


'''simple docstring'''




import argparse

import fairseq
import torch

from transformers import UniSpeechSatConfig, UniSpeechSatForCTC, UniSpeechSatForPreTraining, logging


logging.set_verbosity_info()
_a   :   Optional[Any]												=   logging.get_logger(__name__)

_a   :   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",
}
_a   :   str												=   [
    "lm_head",
    "quantizer.weight_proj",
    "quantizer.codevectors",
    "project_q",
    "project_hid",
    "label_embeddings_concat",
    "speaker_proj",
    "layer_norm_for_extract",
]





def   _a		(lowercase__			:   Tuple							,   lowercase__			:   Optional[int]							,   lowercase__			:   Union[str, Any]							,   lowercase__			:   str							,   lowercase__			:   Dict						)			->			int:






     """simple docstring"""
     for attribute in key.split('.'						):
          __snake_case           =						getattr(lowercase__							,   lowercase__						)

     if weight_type is not None:
          __snake_case           =						getattr(lowercase__							,   lowercase__						).shape
     else:
          __snake_case           =						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":
          __snake_case           =						value
     elif weight_type == "weight_g":
          __snake_case           =						value
     elif weight_type == "weight_v":
          __snake_case           =						value
     elif weight_type == "bias":
          __snake_case           =						value
     else:
          __snake_case           =						value

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





def   _a		(lowercase__			:   Any							,   lowercase__			:   List[Any]						)			->			int:






     """simple docstring"""
     __snake_case           =						[]
     __snake_case           =						fairseq_model.state_dict()

     __snake_case           =						hf_model.unispeech_sat.feature_extractor

     for name, value in fairseq_dict.items():
          __snake_case           =						False
          if "conv_layers" in name:
               load_conv_layer(
                   lowercase__							,   lowercase__							,   lowercase__							,   lowercase__							,   hf_model.config.feat_extract_norm == 'group'							,   )
               __snake_case           =						True
          else:
               for key, mapped_key in MAPPING.items():
                    __snake_case           =						'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
                         __snake_case           =						True
                         if "*" in mapped_key:
                              __snake_case           =						name.split(lowercase__						)[0].split('.'						)[-2]
                              __snake_case           =						mapped_key.replace('*'							,   lowercase__						)
                         if "weight_g" in name:
                              __snake_case           =						'weight_g'
                         elif "weight_v" in name:
                              __snake_case           =						'weight_v'
                         elif "bias" in name:
                              __snake_case           =						'bias'
                         elif "weight" in name:
                              # TODO: don't match quantizer.weight_proj
                              __snake_case           =						'weight'
                         else:
                              __snake_case           =						None
                         set_recursively(lowercase__							,   lowercase__							,   lowercase__							,   lowercase__							,   lowercase__						)
                    continue
          if not is_used:
               unused_weights.append(lowercase__						)

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





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






     """simple docstring"""
     __snake_case           =						full_name.split('conv_layers.'						)[-1]
     __snake_case           =						name.split('.'						)
     __snake_case           =						int(items[0]						)
     __snake_case           =						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.'						)
               __snake_case           =						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.'						)
               __snake_case           =						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.'						)
               __snake_case           =						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.'						)
               __snake_case           =						value
               logger.info(f'Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}.'						)
     else:
          unused_weights.append(lowercase__						)





@torch.no_grad()
def   _a		(lowercase__			:   List[Any]							,   lowercase__			:   Union[str, Any]							,   lowercase__			:   Tuple=None							,   lowercase__			:   Any=None							,   lowercase__			:   List[str]=True						)			->			Dict:






     """simple docstring"""
     if config_path is not None:
          __snake_case           =						UniSpeechSatConfig.from_pretrained(lowercase__						)
     else:
          __snake_case           =						UniSpeechSatConfig()

     __snake_case           =						''

     if is_finetuned:
          __snake_case           =						UniSpeechSatForCTC(lowercase__						)
     else:
          __snake_case           =						UniSpeechSatForPreTraining(lowercase__						)

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

     recursively_load_weights(lowercase__							,   lowercase__						)

     hf_wavavec.save_pretrained(lowercase__						)


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   :   Union[str, 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"""




import json
import os
from typing import Optional, Tuple

import regex as re

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


lowerCAmelCase				:     Union[str, Any]					=				logging.get_logger(__name__)

lowerCAmelCase				:     Optional[int]					=				{
    """vocab_file""": """vocab.json""",
    """merges_file""": """merges.txt""",
}

lowerCAmelCase				:     Optional[Any]					=				{
    """vocab_file""": {"""ctrl""": """https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-vocab.json"""},
    """merges_file""": {"""ctrl""": """https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-merges.txt"""},
}

lowerCAmelCase				:     Optional[Any]					=				{
    """ctrl""": 256,
}

lowerCAmelCase				:     int					=				{
    """Pregnancy""": 16_8629,
    """Christianity""": 7675,
    """Explain""": 10_6423,
    """Fitness""": 6_3440,
    """Saving""": 6_3163,
    """Ask""": 2_7171,
    """Ass""": 9_5985,
    """Joke""": 16_3509,
    """Questions""": 4_5622,
    """Thoughts""": 4_9605,
    """Retail""": 5_2342,
    """Feminism""": 16_4338,
    """Writing""": 1_1992,
    """Atheism""": 19_2263,
    """Netflix""": 4_8616,
    """Computing""": 3_9639,
    """Opinion""": 4_3213,
    """Alone""": 4_4967,
    """Funny""": 5_8917,
    """Gaming""": 4_0358,
    """Human""": 4088,
    """India""": 1331,
    """Joker""": 7_7138,
    """Diet""": 3_6206,
    """Legal""": 1_1859,
    """Norman""": 4939,
    """Tip""": 7_2689,
    """Weight""": 5_2343,
    """Movies""": 4_6273,
    """Running""": 2_3425,
    """Science""": 2090,
    """Horror""": 3_7793,
    """Confession""": 6_0572,
    """Finance""": 1_2250,
    """Politics""": 1_6360,
    """Scary""": 19_1985,
    """Support""": 1_2654,
    """Technologies""": 3_2516,
    """Teenage""": 6_6160,
    """Event""": 3_2769,
    """Learned""": 6_7460,
    """Notion""": 18_2770,
    """Wikipedia""": 3_7583,
    """Books""": 6665,
    """Extract""": 7_6050,
    """Confessions""": 10_2701,
    """Conspiracy""": 7_5932,
    """Links""": 6_3674,
    """Narcissus""": 15_0425,
    """Relationship""": 5_4766,
    """Relationships""": 13_4796,
    """Reviews""": 4_1671,
    """News""": 4256,
    """Translation""": 2_6820,
    """multilingual""": 12_8406,
}



def        a__					(							snake_case__							)		->   Optional[Any]:
			lowerCamelCase      =   set()
			lowerCamelCase      =   word[0]
			for char in word[1:]:
						pairs.add((prev_char, char)							)
						lowerCamelCase      =   char

			lowerCamelCase      =   set(snake_case__							)
			return pairs






class        __magic_name__				(			UpperCAmelCase__							):






				'''simple docstring'''
				__UpperCamelCase         =		VOCAB_FILES_NAMES
				__UpperCamelCase         =		PRETRAINED_VOCAB_FILES_MAP
				__UpperCamelCase         =		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				__UpperCamelCase         =		CONTROL_CODES





				def __init__(			self						,					_a						,					_a						,					_a="<unk>"						,					**_a					):





							"""simple docstring"""
							super().__init__(unk_token=_a						,					**_a					)

							with open(_a						,					encoding="""utf-8"""					) as vocab_handle:
										lowerCamelCase      =   json.load(_a					)
							lowerCamelCase      =   {v: k for k, v in self.encoder.items()}
							with open(_a						,					encoding="""utf-8"""					) as merges_handle:
										lowerCamelCase      =   merges_handle.read().split("""\n"""					)[1:-1]
							lowerCamelCase      =   [tuple(merge.split()					) for merge in merges]
							lowerCamelCase      =   dict(zip(_a						,					range(len(_a					)					)					)					)
							lowerCamelCase      =   {}





				@property
				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							return len(self.encoder					)





				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							return dict(self.encoder						,					**self.added_tokens_encoder					)





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							if token in self.cache:
										return self.cache[token]
							lowerCamelCase      =   tuple(_a					)
							lowerCamelCase      =   tuple(list(word[:-1]					) + [word[-1] + """</w>"""]					)
							lowerCamelCase      =   get_pairs(_a					)

							if not pairs:
										return token

							while True:
										lowerCamelCase      =   min(_a						,					key=lambda _a					: self.bpe_ranks.get(_a						,					float("""inf"""					)					)					)
										if bigram not in self.bpe_ranks:
													break
										lowerCamelCase ,  lowerCamelCase      =   bigram
										lowerCamelCase      =   []
										lowerCamelCase      =   0
										while i < len(_a					):
													try:
																lowerCamelCase      =   word.index(_a						,					_a					)
													except ValueError:
																new_word.extend(word[i:]					)
																break
													else:
																new_word.extend(word[i:j]					)
																lowerCamelCase      =   j

													if word[i] == first and i < len(_a					) - 1 and word[i + 1] == second:
																new_word.append(first + second					)
																i += 2
													else:
																new_word.append(word[i]					)
																i += 1
										lowerCamelCase      =   tuple(_a					)
										lowerCamelCase      =   new_word
										if len(_a					) == 1:
													break
										else:
													lowerCamelCase      =   get_pairs(_a					)
							lowerCamelCase      =   """@@ """.join(_a					)
							lowerCamelCase      =   word[:-4]
							lowerCamelCase      =   word
							return word





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							lowerCamelCase      =   []

							lowerCamelCase      =   re.findall(r"""\S+\n?"""						,					_a					)

							for token in words:
										split_tokens.extend(list(self.bpe(_a					).split(""" """					)					)					)
							return split_tokens





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							return self.encoder.get(_a						,					self.encoder.get(self.unk_token					)					)





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							return self.decoder.get(_a						,					self.unk_token					)





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							lowerCamelCase      =   """ """.join(_a					).replace("""@@ """						,					""""""					).strip()
							return out_string





				def   _lowerCAmelCase							(			self						,					_a						,					_a = None					):





							"""simple docstring"""
							if not os.path.isdir(_a					):
										logger.error(f'Vocabulary path ({save_directory}) should be a directory'					)
										return
							lowerCamelCase      =   os.path.join(
							    _a						,					(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]					)
							lowerCamelCase      =   os.path.join(
							    _a						,					(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""merges_file"""]					)

							with open(_a						,					"""w"""						,					encoding="""utf-8"""					) as f:
										f.write(json.dumps(self.encoder						,					indent=2						,					sort_keys=_a						,					ensure_ascii=_a					) + """\n"""					)

							lowerCamelCase      =   0
							with open(_a						,					"""w"""						,					encoding="""utf-8"""					) as writer:
										writer.write("""#version: 0.2\n"""					)
										for bpe_tokens, token_index in sorted(self.bpe_ranks.items()						,					key=lambda _a					: kv[1]					):
													if index != token_index:
																logger.warning(
																    f'Saving vocabulary to {merge_file}: BPE merge indices are not consecutive.'
																    """ Please check that the tokenizer is not corrupted!"""					)
																lowerCamelCase      =   token_index
													writer.write(""" """.join(_a					) + """\n"""					)
													index += 1

							return vocab_file, merge_file

				# def decode(self, token_ids, skip_special_tokens=False, clean_up_tokenization_spaces=True):
				#     filtered_tokens = ' '.join(self.convert_ids_to_tokens(token_ids, skip_special_tokens=skip_special_tokens))
				#     tokens_generated_so_far = re.sub('(@@ )', '', string=filtered_tokens)
				#     tokens_generated_so_far = re.sub('(@@ ?$)', '', string=tokens_generated_so_far)
				#     return ''.join(tokens_generated_so_far)



