





import argparse
import json
import os

import fairseq
import torch
from fairseq.data import Dictionary

from transformers import (
    WavaVecaConformerConfig,
    WavaVecaConformerForCTC,
    WavaVecaConformerForPreTraining,
    WavaVecaCTCTokenizer,
    WavaVecaFeatureExtractor,
    WavaVecaProcessor,
    logging,
)


logging.set_verbosity_info()
__UpperCAmelCase       :   Optional[int]				       =							logging.get_logger(__name__)

__UpperCAmelCase       :   List[Any]				       =							{
    "post_extract_proj": "feature_projection.projection",
    "encoder.pos_conv.0": "encoder.pos_conv_embed.conv",
    "self_attn.linear_k": "encoder.layers.*.self_attn.linear_k",
    "self_attn.linear_v": "encoder.layers.*.self_attn.linear_v",
    "self_attn.linear_q": "encoder.layers.*.self_attn.linear_q",
    "self_attn.pos_bias_u": "encoder.layers.*.self_attn.pos_bias_u",
    "self_attn.pos_bias_v": "encoder.layers.*.self_attn.pos_bias_v",
    "self_attn.linear_out": "encoder.layers.*.self_attn.linear_out",
    "self_attn.linear_pos": "encoder.layers.*.self_attn.linear_pos",
    "self_attn.rotary_emb": "encoder.embed_positions",
    "self_attn_layer_norm": "encoder.layers.*.self_attn_layer_norm",
    "conv_module.pointwise_conv1": "encoder.layers.*.conv_module.pointwise_conv1",
    "conv_module.pointwise_conv2": "encoder.layers.*.conv_module.pointwise_conv2",
    "conv_module.depthwise_conv": "encoder.layers.*.conv_module.depthwise_conv",
    "conv_module.batch_norm": "encoder.layers.*.conv_module.batch_norm",
    "conv_module.layer_norm": "encoder.layers.*.conv_module.layer_norm",
    "ffn1.w_1": "encoder.layers.*.ffn1.intermediate_dense",
    "ffn1.w_2": "encoder.layers.*.ffn1.output_dense",
    "ffn1.layer_norm": "encoder.layers.*.ffn1_layer_norm",
    "ffn2.w_1": "encoder.layers.*.ffn2.intermediate_dense",
    "ffn2.w_2": "encoder.layers.*.ffn2.output_dense",
    "ffn2.layer_norm": "encoder.layers.*.ffn2_layer_norm",
    "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": "lm_head",
    "mask_emb": "masked_spec_embed",
}
__UpperCAmelCase       :   Any				       =							[
    "lm_head",
    "quantizer.weight_proj",
    "quantizer.codevectors",
    "project_q",
    "project_hid",
]







def 			A__							(    SCREAMING_SNAKE_CASE__					,					SCREAMING_SNAKE_CASE__					,					SCREAMING_SNAKE_CASE__					,					SCREAMING_SNAKE_CASE__					,					SCREAMING_SNAKE_CASE__)   ->    Dict:


  for attribute in key.split("""."""):
    __snake_case:						str   =						getattr(__snake_case					,					__snake_case)

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

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







def 			A__							(    SCREAMING_SNAKE_CASE__					,					SCREAMING_SNAKE_CASE__					,					SCREAMING_SNAKE_CASE__)   ->    List[str]:
  __snake_case:						Dict   =						[]
  __snake_case:						str   =						fairseq_model.state_dict()

  __snake_case:						int   =						hf_model.wavaveca_conformer.feature_extractor

  for name, value in fairseq_dict.items():
    __snake_case:						str   =						False
    if "conv_layers" in name:
      load_conv_layer(
          __snake_case					,					__snake_case					,					__snake_case					,					__snake_case					,					hf_model.config.feat_extract_norm == """group"""					,					)
      __snake_case:						Tuple   =						True
    else:
      for key, mapped_key in MAPPING.items():
        __snake_case:						Optional[Any]   =						"""wav2vec2_conformer.""" + 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]:
          __snake_case:						List[str]   =						True
          if "*" in mapped_key:
            __snake_case:						Dict   =						name.split(__snake_case)[0].split(""".""")[-2]
            __snake_case:						Optional[Any]   =						mapped_key.replace("""*"""					,					__snake_case)
          if "pos_bias_u" in name:
            __snake_case:						Any   =						None
          elif "pos_bias_v" in name:
            __snake_case:						int   =						None
          elif "weight_g" in name:
            __snake_case:						str   =						"""weight_g"""
          elif "weight_v" in name:
            __snake_case:						int   =						"""weight_v"""
          elif "bias" in name:
            __snake_case:						List[Any]   =						"""bias"""
          elif "weight" in name:
            # TODO: don't match quantizer.weight_proj
            __snake_case:						int   =						"""weight"""
          elif "running_mean" in name:
            __snake_case:						List[str]   =						"""running_mean"""
          elif "inv_freq" in name:
            __snake_case:						List[Any]   =						"""inv_freq"""
          elif "running_var" in name:
            __snake_case:						Any   =						"""running_var"""
          elif "num_batches_tracked" in name:
            __snake_case:						List[Any]   =						"""num_batches_tracked"""
          else:
            __snake_case:						Optional[Any]   =						None
          set_recursively(__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__							(    SCREAMING_SNAKE_CASE__					,					SCREAMING_SNAKE_CASE__					,					SCREAMING_SNAKE_CASE__					,					SCREAMING_SNAKE_CASE__					,					SCREAMING_SNAKE_CASE__)   ->    str:
  __snake_case:						Optional[Any]   =						full_name.split("""conv_layers.""")[-1]
  __snake_case:						Optional[Any]   =						name.split(""".""")
  __snake_case:						Tuple   =						int(items[0])
  __snake_case:						Union[str, Any]   =						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:						Dict   =						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:						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:
      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.conv_layers[layer_id].layer_norm.bias.data.shape} was found.''')
      __snake_case:						Union[str, Any]   =						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.conv_layers[layer_id].layer_norm.weight.data.shape} was found.''')
      __snake_case:						Optional[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__							(    SCREAMING_SNAKE_CASE__					,					SCREAMING_SNAKE_CASE__					,					SCREAMING_SNAKE_CASE__=None					,					SCREAMING_SNAKE_CASE__=None					,					SCREAMING_SNAKE_CASE__=True)   ->    str:


  if config_path is not None:
    __snake_case:						Optional[int]   =						WavaVecaConformerConfig.from_pretrained(__snake_case					,					hidden_act="""swish""")
  else:
    __snake_case:						List[Any]   =						WavaVecaConformerConfig()

  if "rope" in checkpoint_path:
    __snake_case:						int   =						"""rotary"""

  if is_finetuned:
    if dict_path:
      __snake_case:						List[str]   =						Dictionary.load(__snake_case)

      # important change bos & pad token id since CTC symbol is <pad> and
      # not <s> as in fairseq
      __snake_case:						int   =						target_dict.pad_index
      __snake_case:						Optional[Any]   =						target_dict.bos_index
      __snake_case:						Dict   =						target_dict.eos_index
      __snake_case:						Optional[Any]   =						len(target_dict.symbols)
      __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)
      __snake_case:						int   =						target_dict.indices

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

    __snake_case:						List[str]   =						WavaVecaConformerForCTC(__snake_case)
  else:
    __snake_case:						Union[str, Any]   =						WavaVecaConformerForPreTraining(__snake_case)

  if is_finetuned:
    __snake_case  ,   __snake_case  ,   __snake_case:						Any   =						fairseq.checkpoint_utils.load_model_ensemble_and_task(
        [checkpoint_path]					,					arg_overrides={"""data""": """/""".join(dict_path.split("""/""")[:-1])})
  else:
    __snake_case:						Optional[Any]   =						argparse.Namespace(task="""audio_pretraining""")
    __snake_case:						Union[str, Any]   =						fairseq.tasks.setup_task(__snake_case)

    __snake_case  ,   __snake_case  ,   __snake_case:						Union[str, Any]   =						fairseq.checkpoint_utils.load_model_ensemble_and_task([checkpoint_path]					,					task=__snake_case)

  __snake_case:						str   =						model[0].eval()

  recursively_load_weights(__snake_case					,					__snake_case					,					not is_finetuned)

  hf_wavavec.save_pretrained(__snake_case)


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










UpperCAmelCase__   	=							8.31_44_62  # Unit - J mol-1 K-1
def 				UpperCAmelCase_					(							__snake_case     ,						__snake_case     ,						__snake_case						)     ->			float:







 """simple docstring"""


 if moles < 0 or kelvin < 0 or volume < 0:
  raise ValueError('''Invalid inputs. Enter positive value.'''						)
 return moles * kelvin * UNIVERSAL_GAS_CONSTANT / volume
def 				UpperCAmelCase_					(							__snake_case     ,						__snake_case     ,						__snake_case						)     ->			float:







 """simple docstring"""


 if moles < 0 or kelvin < 0 or pressure < 0:
  raise ValueError('''Invalid inputs. Enter positive value.'''						)
 return moles * kelvin * UNIVERSAL_GAS_CONSTANT / pressure


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

  testmod()

