





"""simple docstring"""



import argparse
import os
from functools import reduce

import fairseq
import torch
from datasets import load_dataset

from transformers import WavaVecaProcessor, logging
from transformers.models.dataavec.configuration_dataavec_audio import DataaVecAudioConfig

# Copied from https://github.com/pytorch/fairseq/blob/main/examples/data2vec/models/data2vec_audio.py
from transformers.models.dataavec.dataavec_audio import DataaVecAudioModel as Dummy  # noqa: F401
from transformers.models.dataavec.modeling_dataavec_audio import DataaVecAudioForCTC, DataaVecAudioModel


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

a_				=    {
    """post_extract_proj""": """feature_projection.projection""",
    """models.0.layer_norm""": """feature_projection.layer_norm""",
    """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""",
}
a_				=    [
    """lm_head""",
]







def 	__lowercase  (     snake_case_						:						Tuple       ,snake_case_						:						Optional[Any]       ,snake_case_						:						int       ,snake_case_						:						int       ,snake_case_						:						Tuple		):



    '''simple docstring'''







    for attribute in key.split('''.'''		):
        __A					:		Tuple						=  getattr(lowerCAmelCase__       ,lowerCAmelCase__		)

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

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







def 	__lowercase  (     snake_case_						:						Dict       ,snake_case_						:						Optional[Any]       ,snake_case_						:						Dict		):



    '''simple docstring'''







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

    if not is_headless:
        __A					:		str						=  hf_model.dataavec_audio.feature_extractor
        __A					:		List[Any]						=  hf_model.dataavec_audio.encoder.pos_conv_embed

    else:
        __A					:		Dict						=  hf_model.feature_extractor
        __A					:		Union[str, Any]						=  hf_model.encoder.pos_conv_embed

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

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







def 	__lowercase  (     snake_case_						:						int       ,snake_case_						:						Optional[int]		):



    '''simple docstring'''







    __A					:		Optional[Any]						=  path.split('''.'''		)
    return reduce(lowerCAmelCase__       ,lowerCAmelCase__       ,lowerCAmelCase__		)







def 	__lowercase  (     snake_case_						:						Tuple       ,snake_case_						:						Optional[int]       ,snake_case_						:						str       ,snake_case_						:						Any		):



    '''simple docstring'''







    __A					:		Any						=  access_by_string(lowerCAmelCase__       ,lowerCAmelCase__		)
    __A					:		List[str]						=  hf_weight.data

    if fsq_value.shape != hf_value.shape:
        raise ValueError(F"""{full_name} has size {fsq_value.shape}, but {hf_value.shape} was found."""		)
    __A					:		str						=  fsq_value
    logger.info(F"""{full_name} was correctly initialized from {hf_weight_path}."""		)







def 	__lowercase  (     snake_case_						:						Tuple       ,snake_case_						:						List[Any]       ,snake_case_						:						List[Any]       ,snake_case_						:						List[Any]		):



    '''simple docstring'''







    __A					:		int						=  full_name.split('''conv_layers.'''		)[-1]
    __A					:		Optional[int]						=  name.split('''.'''		)
    __A					:		str						=  int(items[0]		)
    __A					:		str						=  int(items[1]		)

    __A					:		Dict						=  name.split('''.'''		)[-1]
    if type_id == 0:
        __A					:		Any						=  'conv'
    elif type_id == 2:
        __A					:		Union[str, Any]						=  'layer_norm'
    else:
        unused_weights.append(lowerCAmelCase__		)
        return

    set_weights(lowerCAmelCase__       ,lowerCAmelCase__       ,lowerCAmelCase__       ,F"""conv_layers.{layer_id}.{layer_type}.{weight_type}"""		)







def 	__lowercase  (     snake_case_						:						int       ,snake_case_						:						str       ,snake_case_						:						int       ,snake_case_						:						str		):



    '''simple docstring'''







    __A					:		List[Any]						=  full_name.split('''pos_conv.'''		)[-1]
    __A					:		Tuple						=  name.split('''.'''		)
    __A					:		Tuple						=  int(items[0]		)
    __A					:		Dict						=  int(items[1]		)

    __A					:		int						=  name.split('''.'''		)[-1]
    if type_id != 0:
        unused_weights.append(lowerCAmelCase__		)
        return
    else:
        __A					:		Union[str, Any]						=  'conv'

    set_weights(lowerCAmelCase__       ,lowerCAmelCase__       ,lowerCAmelCase__       ,F"""layers.{layer_id}.{layer_type}.{weight_type}"""		)







@torch.no_grad()
def 	__lowercase  (     snake_case_						:						Union[str, Any]       ,snake_case_						:						List[str]       ,snake_case_						:						str=None       ,snake_case_						:						Optional[int]=None       ,snake_case_						:						List[str]=True		):



    '''simple docstring'''







    if config_path is not None:
        __A					:		Union[str, Any]						=  DataaVecAudioConfig.from_pretrained(lowerCAmelCase__		)
    else:
        __A					:		int						=  DataaVecAudioConfig()

    if not is_finetuned:
        # Modify final_proj layer name
        __A					:		Tuple						=  DataaVecAudioModel(lowerCAmelCase__		)
        __A					:		Optional[Any]						=  os.path.dirname(lowerCAmelCase__		)

        __A					:		List[Any]						=  torch.load(lowerCAmelCase__		)
        __A					:		Dict						=  state_dict['model'].pop('''final_proj.0.weight'''		)
        __A					:		Optional[Any]						=  state_dict['model'].pop('''final_proj.0.bias'''		)
        __A					:		Union[str, Any]						=  os.path.join(lowerCAmelCase__       ,'''converted.pt'''		)
        torch.save(lowerCAmelCase__       ,lowerCAmelCase__		)
    else:
        __A					:		Optional[Any]						=  DataaVecAudioForCTC(lowerCAmelCase__		)
        __A					:		Optional[Any]						=  checkpoint_path

    def load_dataavec(snake_case_						:						Dict		):
        __A					:		Tuple						=  fairseq.checkpoint_utils.load_model_ensemble_and_task([path]		)
        return model[0].eval()

    __A					:		List[str]						=  load_dataavec(lowerCAmelCase__		)

    recursively_load_weights(lowerCAmelCase__       ,lowerCAmelCase__       ,not is_finetuned		)

    __A					:		int						=  WavaVecaProcessor.from_pretrained('''facebook/wav2vec2-large-lv60'''		)

    __A					:		Tuple						=  load_dataset('''patrickvonplaten/librispeech_asr_dummy'''       ,'''clean'''       ,split='''validation'''		)
    __A					:		Any						=  [x['array'] for x in ds[:4]['audio']]

    __A					:		List[str]						=  processor(lowerCAmelCase__       ,return_tensors='''pt'''       ,padding=lowerCAmelCase__		)

    __A					:		Any						=  inputs.input_values
    __A					:		int						=  inputs.attention_mask
    #    input_values = inputs.input_values[:, :-1]
    #    attention_mask = inputs.attention_mask[:, :-1]

    hf_wavavec.eval()
    model.eval()
    if is_finetuned:
        __A					:		Dict						=  model(source=lowerCAmelCase__       ,padding_mask=(1 - attention_mask)       ,mask=lowerCAmelCase__       ,features_only=lowerCAmelCase__		)[
            'encoder_out'
        ].transpose(0       ,1		)
        __A					:		Any						=  hf_wavavec(lowerCAmelCase__       ,attention_mask=lowerCAmelCase__		)['logits']

        __A					:		int						=  torch.argmax(lowerCAmelCase__       ,dim=-1		)
        __A					:		Any						=  processor.batch_decode(lowerCAmelCase__		)

        print(F"""Expected Output: {ds[:4]['text']}, Pred: {output_string}"""		)
    else:
        __A					:		str						=  model(source=lowerCAmelCase__       ,padding_mask=(1 - attention_mask)       ,mask=lowerCAmelCase__       ,features_only=lowerCAmelCase__		)[
            'layer_results'
        ][-1][0].transpose(0       ,1		)
        __A					:		str						=  hf_wavavec(lowerCAmelCase__       ,attention_mask=lowerCAmelCase__		)['last_hidden_state']

    print(our_output.shape       ,their_output.shape		)
    __A					:		List[str]						=  torch.max(torch.abs(our_output - their_output		)		).item()
    print(F"""max_absolute_diff = {max_absolute_diff}"""		)  # ~ 1e-7
    __A					:		Optional[Any]						=  torch.allclose(lowerCAmelCase__       ,lowerCAmelCase__       ,atol=1e-3		)
    print('''Do both models output the same tensors?'''       ,'''🔥''' if success else '''💩'''		)
    if not success:
        raise Exception('''Something went wRoNg'''		)

    hf_wavavec.save_pretrained(lowerCAmelCase__		)

    if is_finetuned:
        processor.save_pretrained(lowerCAmelCase__		)
    else:
        processor.feature_extractor.save_pretrained(lowerCAmelCase__		)


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("""--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_				=    parser.parse_args()
       convert_wavaveca_checkpoint(
           args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.dict_path, not args.not_finetuned
       )











"""simple docstring"""



def 	__lowercase  (     snake_case_						:						list		)     ->list:



    '''simple docstring'''







    for insert_index, insert_value in enumerate(collection[1:]		):
        __A					:		Any						=  insert_index
        while insert_index >= 0 and insert_value < collection[insert_index]:
            __A					:		List[Any]						=  collection[insert_index]
            insert_index -= 1
        if insert_index != temp_index:
            __A					:		str						=  insert_value
    return collection


if __name__ == "__main__":
       from doctest import testmod

       testmod()

       a_				=    input("""Enter numbers separated by a comma:\n""").strip()
       a_				=    [int(item) for item in user_input.split(""",""")]
       print(f'''{insertion_sort(unsorted) = }''')




