


"""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()
__snake_case												=							logging.get_logger(__name__)

__snake_case												=							{
    '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',
}
__snake_case												=							[
    'lm_head',
]
def    A_ ( _lowerCAmelCase  :       Optional[Any],  _lowerCAmelCase  :       Dict,  _lowerCAmelCase  :       Tuple,  _lowerCAmelCase  :       str,  _lowerCAmelCase  :       int							):






     """simple docstring"""

     for attribute in key.split('''.'''							):
          _a     						=					getattr(_A,  _A							)

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

     logger.info(f'{key + "." + weight_type if weight_type is not None else ""} was initialized from {full_name}.'							)
def    A_ ( _lowerCAmelCase  :       Dict,  _lowerCAmelCase  :       Dict,  _lowerCAmelCase  :       Optional[Any]							):






     """simple docstring"""

     _a     						=					[]
     _a     						=					fairseq_model.state_dict()

     if not is_headless:
          _a     						=					hf_model.dataavec_audio.feature_extractor
          _a     						=					hf_model.dataavec_audio.encoder.pos_conv_embed

     else:
          _a     						=					hf_model.feature_extractor
          _a     						=					hf_model.encoder.pos_conv_embed

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

     logger.warning(f'Unused weights: {unused_weights}'							)
def    A_ ( _lowerCAmelCase  :       Optional[Any],  _lowerCAmelCase  :       Union[str, Any]							):






     """simple docstring"""

     _a     						=					path.split('''.'''							)
     return reduce(_A,  _A,  _A							)
def    A_ ( _lowerCAmelCase  :       str,  _lowerCAmelCase  :       List[Any],  _lowerCAmelCase  :       str,  _lowerCAmelCase  :       Dict							):






     """simple docstring"""

     _a     						=					access_by_string(_A,  _A							)
     _a     						=					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     						=					fsq_value
     logger.info(f'{full_name} was correctly initialized from {hf_weight_path}.'							)
def    A_ ( _lowerCAmelCase  :       Union[str, Any],  _lowerCAmelCase  :       Any,  _lowerCAmelCase  :       List[str],  _lowerCAmelCase  :       List[Any]							):






     """simple docstring"""

     _a     						=					full_name.split('''conv_layers.'''							)[-1]
     _a     						=					name.split('''.'''							)
     _a     						=					int(items[0]							)
     _a     						=					int(items[1]							)

     _a     						=					name.split('''.'''							)[-1]
     if type_id == 0:
          _a     						=					'''conv'''
     elif type_id == 2:
          _a     						=					'''layer_norm'''
     else:
          unused_weights.append(_A							)
          return

     set_weights(_A,  _A,  _A,  f'conv_layers.{layer_id}.{layer_type}.{weight_type}'							)
def    A_ ( _lowerCAmelCase  :       Union[str, Any],  _lowerCAmelCase  :       Optional[int],  _lowerCAmelCase  :       Optional[int],  _lowerCAmelCase  :       List[Any]							):






     """simple docstring"""

     _a     						=					full_name.split('''pos_conv.'''							)[-1]
     _a     						=					name.split('''.'''							)
     _a     						=					int(items[0]							)
     _a     						=					int(items[1]							)

     _a     						=					name.split('''.'''							)[-1]
     if type_id != 0:
          unused_weights.append(_A							)
          return
     else:
          _a     						=					'''conv'''

     set_weights(_A,  _A,  _A,  f'layers.{layer_id}.{layer_type}.{weight_type}'							)
@torch.no_grad()
def    A_ ( _lowerCAmelCase  :       Optional[Any],  _lowerCAmelCase  :       Optional[int],  _lowerCAmelCase  :       List[Any]=None,  _lowerCAmelCase  :       int=None,  _lowerCAmelCase  :       Union[str, Any]=True							):






     """simple docstring"""

     if config_path is not None:
          _a     						=					DataaVecAudioConfig.from_pretrained(_A							)
     else:
          _a     						=					DataaVecAudioConfig()

     if not is_finetuned:
          # Modify final_proj layer name
          _a     						=					DataaVecAudioModel(_A							)
          _a     						=					os.path.dirname(_A							)

          _a     						=					torch.load(_A							)
          _a     						=					state_dict['''model'''].pop('''final_proj.0.weight'''							)
          _a     						=					state_dict['''model'''].pop('''final_proj.0.bias'''							)
          _a     						=					os.path.join(_A,  '''converted.pt'''							)
          torch.save(_A,  _A							)
     else:
          _a     						=					DataaVecAudioForCTC(_A							)
          _a     						=					checkpoint_path

     def load_dataavec(_lowerCAmelCase  :       str							):
          _a						,   _a						,   _a     						=					fairseq.checkpoint_utils.load_model_ensemble_and_task([path]							)
          return model[0].eval()

     _a     						=					load_dataavec(_A							)

     recursively_load_weights(_A,  _A,  not is_finetuned							)

     _a     						=					WavaVecaProcessor.from_pretrained('''facebook/wav2vec2-large-lv60'''							)

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

     _a     						=					processor(_A,  return_tensors='''pt''',  padding=_A							)

     _a     						=					inputs.input_values
     _a     						=					inputs.attention_mask
     #    input_values = inputs.input_values[:, :-1]
     #    attention_mask = inputs.attention_mask[:, :-1]

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

          _a     						=					torch.argmax(_A,  dim=-1							)
          _a     						=					processor.batch_decode(_A							)

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

     print(our_output.shape,  their_output.shape							)
     _a     						=					torch.max(torch.abs(our_output - their_output							)							).item()
     print(f'max_absolute_diff = {max_absolute_diff}'							)  # ~ 1e-7
     _a     						=					torch.allclose(_A,  _A,  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(_A							)

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


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


from math import factorial




def        __UpperCamelCase     (      _A					:							Union[str, Any]      ,  _A					:							Tuple		)				->Any:



							"""simple docstring"""
							return factorial(_A		) / (factorial(_A		) * factorial(n - r		))




def        __UpperCamelCase     (      )				->Any:



							"""simple docstring"""
							lowerCamelCase_								=0

							for i in range(1      ,  101		):
														for j in range(1      ,  i + 1		):
																					if combinations(_A      ,  _A		) > 1E6:
																												total += 1
							return total


if __name__ == "__main__":
	print(solution())
