






'''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()
_A     :      Optional[int]   					=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''',
    '''w2v_model.layer_norm''': '''feature_projection.layer_norm''',
    '''w2v_encoder.proj''': '''lm_head''',
    '''mask_emb''': '''masked_spec_embed''',
}





def 			SCREAMING_SNAKE_CASE_	(UpperCamelCase    ,      UpperCamelCase    ,      UpperCamelCase    ,      UpperCamelCase    ,      UpperCamelCase				)				->	Optional[int]:
							for attribute in key.split("""."""				):
														lowerCamelCase__ :     List[Any]	  =	getattr(UpperCamelCase    ,      UpperCamelCase				)

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

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





def 			SCREAMING_SNAKE_CASE_	(UpperCamelCase    ,      UpperCamelCase    ,      UpperCamelCase				)				->	List[Any]:
							lowerCamelCase__ :     Any	  =	[]
							lowerCamelCase__ :     Union[str, Any]	  =	fairseq_model.state_dict()

							lowerCamelCase__ :     Dict	  =	hf_model.hubert.feature_extractor if is_finetuned else hf_model.feature_extractor

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

							if is_finetuned:
														if dict_path:
																					lowerCamelCase__ :     int	  =	Dictionary.load(UpperCamelCase				)

																					# important change bos & pad token id since CTC symbol is <pad> and
																					# not <s> as in fairseq
																					lowerCamelCase__ :     str	  =	target_dict.pad_index
																					lowerCamelCase__ :     Any	  =	target_dict.bos_index
																					lowerCamelCase__ :     Union[str, Any]	  =	target_dict.eos_index
																					lowerCamelCase__ :     List[Any]	  =	len(target_dict.symbols				)
																					lowerCamelCase__ :     Tuple	  =	os.path.join(UpperCamelCase    ,      """vocab.json"""				)
																					if not os.path.isdir(UpperCamelCase				):
																												logger.error("""--pytorch_dump_folder_path ({}) should be a directory""".format(UpperCamelCase				)				)
																												return
																					os.makedirs(UpperCamelCase    ,      exist_ok=UpperCamelCase				)
																					with open(UpperCamelCase    ,      """w"""    ,      encoding="""utf-8"""				) as vocab_handle:
																												json.dump(target_dict.indices    ,      UpperCamelCase				)
																					lowerCamelCase__ :     List[str]	  =	WavaVecaCTCTokenizer(
																					    UpperCamelCase    ,      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=UpperCamelCase    ,      )
																					lowerCamelCase__ :     Tuple	  =	True if config.feat_extract_norm == """layer""" else False
																					lowerCamelCase__ :     Optional[int]	  =	WavaVecaFeatureExtractor(
																					    feature_size=1    ,      sampling_rate=16000    ,      padding_value=0    ,      do_normalize=UpperCamelCase    ,      return_attention_mask=UpperCamelCase    ,      )
																					lowerCamelCase__ :     List[str]	  =	WavaVecaProcessor(feature_extractor=UpperCamelCase    ,      tokenizer=UpperCamelCase				)
																					processor.save_pretrained(UpperCamelCase				)

														lowerCamelCase__ :     Optional[Any]	  =	HubertForCTC(UpperCamelCase				)
							else:
														lowerCamelCase__ :     Dict	  =	HubertModel(UpperCamelCase				)

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

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

							recursively_load_weights(UpperCamelCase    ,      UpperCamelCase    ,      UpperCamelCase				)

							hf_wavavec.save_pretrained(UpperCamelCase				)


if __name__ == "__main__":
			_A     :      Union[str, 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     :      Optional[int]   					=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'''







from collections import UserDict
from typing import Union

import numpy as np
import requests

from ..utils import (
    add_end_docstrings,
    logging,
)
from .audio_classification import ffmpeg_read
from .base import PIPELINE_INIT_ARGS, Pipeline


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

@add_end_docstrings(_lowercase		)
class _lowercase     (    _lowercase		):




							def __init__(							self:    Union[str, Any]      ,							**UpperCamelCase__:    str					):
														super().__init__(**UpperCamelCase__					)

														if self.framework != "pt":
																					raise ValueError(F'''The {self.__class__} is only available in PyTorch.'''					)
														# No specific FOR_XXX available yet




							def __call__(							self:    Optional[Any]      ,							UpperCamelCase__:    Union[np.ndarray, bytes, str]      ,							**UpperCamelCase__:    int					):
														return super().__call__(UpperCamelCase__      ,							**UpperCamelCase__					)




							def 		lowerCamelCase_    (							self:    Optional[Any]      ,							**UpperCamelCase__:    int					):
														lowerCamelCase__ :     Optional[Any]	  =	{}
														if "candidate_labels" in kwargs:
																					lowerCamelCase__ :     str	  =	kwargs["""candidate_labels"""]
														if "hypothesis_template" in kwargs:
																					lowerCamelCase__ :     int	  =	kwargs["""hypothesis_template"""]

														return preprocess_params, {}, {}




							def 		lowerCamelCase_    (							self:    Optional[Any]      ,							UpperCamelCase__:    int      ,							UpperCamelCase__:    List[str]=None      ,							UpperCamelCase__:    Dict="This is a sound of {}."					):
														if isinstance(UpperCamelCase__      ,							UpperCamelCase__					):
																					if audio.startswith("""http://"""					) or audio.startswith("""https://"""					):
																												# We need to actually check for a real protocol, otherwise it's impossible to use a local file
																												# like http_huggingface_co.png
																												lowerCamelCase__ :     int	  =	requests.get(UpperCamelCase__					).content
																					else:
																												with open(UpperCamelCase__      ,							"""rb"""					) as f:
																																			lowerCamelCase__ :     Dict	  =	f.read()

														if isinstance(UpperCamelCase__      ,							UpperCamelCase__					):
																					lowerCamelCase__ :     str	  =	ffmpeg_read(UpperCamelCase__      ,							self.feature_extractor.sampling_rate					)

														if not isinstance(UpperCamelCase__      ,							np.ndarray					):
																					raise ValueError("""We expect a numpy ndarray as input"""					)
														if len(audio.shape					) != 1:
																					raise ValueError("""We expect a single channel audio input for ZeroShotAudioClassificationPipeline"""					)

														lowerCamelCase__ :     Optional[Any]	  =	self.feature_extractor(
														    [audio]      ,							sampling_rate=self.feature_extractor.sampling_rate      ,							return_tensors="""pt"""					)
														lowerCamelCase__ :     Any	  =	candidate_labels
														lowerCamelCase__ :     Any	  =	[hypothesis_template.format(UpperCamelCase__					) for x in candidate_labels]
														lowerCamelCase__ :     int	  =	self.tokenizer(UpperCamelCase__      ,							return_tensors=self.framework      ,							padding=UpperCamelCase__					)
														lowerCamelCase__ :     Tuple	  =	[text_inputs]
														return inputs




							def 		lowerCamelCase_    (							self:    Tuple      ,							UpperCamelCase__:    Union[str, Any]					):
														lowerCamelCase__ :     Any	  =	model_inputs.pop("""candidate_labels"""					)
														lowerCamelCase__ :     Union[str, Any]	  =	model_inputs.pop("""text_inputs"""					)
														if isinstance(text_inputs[0]      ,							UpperCamelCase__					):
																					lowerCamelCase__ :     Optional[int]	  =	text_inputs[0]
														else:
																					# Batching case.
																					lowerCamelCase__ :     Tuple	  =	text_inputs[0][0]

														lowerCamelCase__ :     Tuple	  =	self.model(**UpperCamelCase__      ,							**UpperCamelCase__					)

														lowerCamelCase__ :     str	  =	{
														    """candidate_labels""": candidate_labels,
														    """logits""": outputs.logits_per_audio,
														}
														return model_outputs




							def 		lowerCamelCase_    (							self:    List[str]      ,							UpperCamelCase__:    Union[str, Any]					):
														lowerCamelCase__ :     List[str]	  =	model_outputs.pop("""candidate_labels"""					)
														lowerCamelCase__ :     int	  =	model_outputs["""logits"""][0]

														if self.framework == "pt":
																					lowerCamelCase__ :     Optional[int]	  =	logits.softmax(dim=0					)
																					lowerCamelCase__ :     Any	  =	probs.tolist()
														else:
																					raise ValueError("""`tf` framework not supported."""					)

														lowerCamelCase__ :     Tuple	  =	[
														    {"""score""": score, """label""": candidate_label}
														    for score, candidate_label in sorted(zip(UpperCamelCase__      ,							UpperCamelCase__					)      ,							key=lambda UpperCamelCase__					: -x[0]					)
														]
														return result

