





'''simple docstring'''




import argparse
import json
import os

import fairseq
import torch
from fairseq.data import Dictionary

from transformers import (
    UniSpeechConfig,
    UniSpeechForCTC,
    UniSpeechForPreTraining,
    WavaVecaFeatureExtractor,
    WavaVecaPhonemeCTCTokenizer,
    WavaVecaProcessor,
    logging,
)


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

__UpperCAmelCase			    =    {
    """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""",
    """quantizer.weight_proj""": """quantizer.weight_proj""",
    """quantizer.vars""": """quantizer.codevectors""",
    """project_q""": """project_q""",
    """final_proj""": """project_hid""",
    """w2v_encoder.proj""": """ctc_proj""",
    """mask_emb""": """masked_spec_embed""",
}
__UpperCAmelCase			    =    [
    """ctc_proj""",
    """quantizer.weight_proj""",
    """quantizer.codevectors""",
    """project_q""",
    """project_hid""",
]



def 						__A						(   lowerCamelCase_       ,       lowerCamelCase_       ,       lowerCamelCase_       ,       lowerCamelCase_       ,       lowerCamelCase_       ,       lowerCamelCase_     ):


		"""simple docstring"""



		for attribute in key.split("""."""     ):
				if is_finetuned:
						if attribute in ["quantizer", "project_q", "project_hid"]:
								# those layers are only relevant for pretraining and should be dropped
								return

						if attribute == "ctc_proj":
								# we should rename `ctc_proj` to `lm_head` for fine-tuned phoneme models
								SCREAMING_SNAKE_CASE      :							str						=       "lm_head"

				SCREAMING_SNAKE_CASE      :							int						=       getattr(__snake_case       ,       __snake_case     )

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

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



def 						__A						(   lowerCamelCase_       ,       lowerCamelCase_       ,       lowerCamelCase_     ):


		"""simple docstring"""



		SCREAMING_SNAKE_CASE      :							Optional[Any]						=       []
		SCREAMING_SNAKE_CASE      :							int						=       fairseq_model.state_dict()

		SCREAMING_SNAKE_CASE      :							Any						=       hf_model.unispeech.feature_extractor

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

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



def 						__A						(   lowerCamelCase_       ,       lowerCamelCase_       ,       lowerCamelCase_       ,       lowerCamelCase_       ,       lowerCamelCase_     ):


		"""simple docstring"""



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



@torch.no_grad()
def 						__A						(   lowerCamelCase_       ,       lowerCamelCase_       ,       lowerCamelCase_=None       ,       lowerCamelCase_=None       ,       lowerCamelCase_=True     ):


		"""simple docstring"""



		if config_path is not None:
				SCREAMING_SNAKE_CASE      :							Optional[int]						=       UniSpeechConfig.from_pretrained(__snake_case     )
		else:
				SCREAMING_SNAKE_CASE      :							str						=       UniSpeechConfig()

		if is_finetuned:
				if dict_path:
						SCREAMING_SNAKE_CASE      :							Tuple						=       Dictionary.load_from_json(__snake_case     )

						# important change bos & pad token id since CTC symbol is <pad> and
						# not <s> as in fairseq
						SCREAMING_SNAKE_CASE      :							Optional[Any]						=       target_dict.pad_index
						SCREAMING_SNAKE_CASE      :							Union[str, Any]						=       target_dict.bos_index
						SCREAMING_SNAKE_CASE      :							List[str]						=       target_dict.eos_index
						SCREAMING_SNAKE_CASE      :							int						=       len(target_dict.symbols     )
						SCREAMING_SNAKE_CASE      :							int						=       os.path.join(__snake_case       ,       """vocab.json"""     )
						if not os.path.isdir(__snake_case     ):
								logger.error("""--pytorch_dump_folder_path ({}) should be a directory""".format(__snake_case     )     )
								return
						os.makedirs(__snake_case       ,       exist_ok=__snake_case     )
						SCREAMING_SNAKE_CASE      :							Dict						=       target_dict.indices

						# fairseq has the <pad> and <s> switched
						SCREAMING_SNAKE_CASE      :							Union[str, Any]						=       42
						SCREAMING_SNAKE_CASE      :							Optional[Any]						=       43
						with open(__snake_case       ,       """w"""       ,       encoding="""utf-8"""     ) as vocab_handle:
								json.dump(__snake_case       ,       __snake_case     )
						SCREAMING_SNAKE_CASE      :							int						=       WavaVecaPhonemeCTCTokenizer(
						    __snake_case       ,       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=__snake_case       ,       )
						SCREAMING_SNAKE_CASE      :							Dict						=       True if config.feat_extract_norm == "layer" else False
						SCREAMING_SNAKE_CASE      :							Optional[int]						=       WavaVecaFeatureExtractor(
						    feature_size=1       ,       sampling_rate=1_60_00       ,       padding_value=0       ,       do_normalize=__snake_case       ,       return_attention_mask=__snake_case       ,       )
						SCREAMING_SNAKE_CASE      :							int						=       WavaVecaProcessor(feature_extractor=__snake_case       ,       tokenizer=__snake_case     )
						processor.save_pretrained(__snake_case     )

				SCREAMING_SNAKE_CASE      :							str						=       UniSpeechForCTC(__snake_case     )
		else:
				SCREAMING_SNAKE_CASE      :							Dict						=       UniSpeechForPreTraining(__snake_case     )

		if is_finetuned:
				SCREAMING_SNAKE_CASE      :							Union[str, Any]						=       fairseq.checkpoint_utils.load_model_ensemble_and_task(
				    [checkpoint_path]       ,       arg_overrides={"""data""": """/""".join(dict_path.split("""/"""     )[:-1]     ), """w2v_path""": checkpoint_path}     )
		else:
				SCREAMING_SNAKE_CASE      :							Union[str, Any]						=       fairseq.checkpoint_utils.load_model_ensemble_and_task([checkpoint_path]     )

		SCREAMING_SNAKE_CASE      :							Tuple						=       model[0].eval()

		recursively_load_weights(__snake_case       ,       __snake_case       ,       __snake_case     )

		hf_unispeech.save_pretrained(__snake_case     )


if __name__ == "__main__":
							__UpperCAmelCase			    =    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"""
							)
							__UpperCAmelCase			    =    parser.parse_args()
							convert_unispeech_checkpoint(
							    args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.dict_path, not args.not_finetuned
							)





"""simple docstring"""




import os
import zipfile

import requests
from get_ci_error_statistics import download_artifact, get_artifacts_links


def   __magic_name__	(      __snake_case				:       Union[str, Any]     , __snake_case				:       List[str]=7  ) ->				str:
       lowercase    :				int       		=							None
       if token is not None:
              lowercase    :				Any       		=							{"Accept": "application/vnd.github+json", "Authorization": f"""Bearer {token}"""}

       # The id of a workflow (not of a workflow run)
       lowercase    :				int       		=							"636036"

       lowercase    :				Dict       		=							f"""https://api.github.com/repos/huggingface/transformers/actions/workflows/{workflow_id}/runs"""
       # On `main` branch + event being `schedule` + not returning PRs + only `num_runs` results
       url += f"""?branch=main&event=schedule&exclude_pull_requests=true&per_page={num_runs}"""

       lowercase    :				int       		=							requests.get(__snake_case     , headers=__snake_case  ).json()

       return result["workflow_runs"]


def   __magic_name__	(      __snake_case				:       Dict  ) ->				Tuple:
       lowercase    :				Tuple       		=							get_daily_ci_runs(__snake_case  )
       lowercase    :				Union[str, Any]       		=							None
       for workflow_run in workflow_runs:
              if workflow_run["status"] == "completed":
                     lowercase    :				List[Any]       		=							workflow_run["id"]
                     break

       return workflow_run_id


def   __magic_name__	(      __snake_case				:       Optional[int]     , __snake_case				:       Optional[Any]     , __snake_case				:       Union[str, Any]  ) ->				int:
       lowercase    :				Dict       		=							get_last_daily_ci_runs(__snake_case  )
       if workflow_run_id is not None:
              lowercase    :				Dict       		=							get_artifacts_links(worflow_run_id=__snake_case     , token=__snake_case  )
              for artifact_name in artifact_names:
                     if artifact_name in artifacts_links:
                            lowercase    :				Optional[int]       		=							artifacts_links[artifact_name]
                            download_artifact(
                                artifact_name=__snake_case     , artifact_url=__snake_case     , output_dir=__snake_case     , token=__snake_case  )


def   __magic_name__	(      __snake_case				:       List[Any]     , __snake_case				:       Optional[int]     , __snake_case				:       Tuple  ) ->				Optional[int]:
       get_last_daily_ci_artifacts(__snake_case     , __snake_case     , __snake_case  )

       lowercase    :				str       		=							{}
       for artifact_name in artifact_names:
              lowercase    :				Optional[Any]       		=							os.path.join(__snake_case     , f"""{artifact_name}.zip"""  )
              if os.path.isfile(__snake_case  ):
                     lowercase    :				List[Any]       		=							{}
                     with zipfile.ZipFile(__snake_case  ) as z:
                            for filename in z.namelist():
                                   if not os.path.isdir(__snake_case  ):
                                          # read the file
                                          with z.open(__snake_case  ) as f:
                                                 lowercase    :				str       		=							f.read().decode("UTF-8"  )

       return results


