




'''simple docstring'''


import argparse
import json
import os

import fairseq
import torch
from torch import nn

from transformers import (
    SpeechaTextaConfig,
    SpeechaTextaForCausalLM,
    SpeechaTextaTokenizer,
    SpeechEncoderDecoderConfig,
    SpeechEncoderDecoderModel,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
    WavaVecaModel,
    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": "lm_head",
    "mask_emb": "masked_spec_embed",
}
UpperCAmelCase_					=   [
    "lm_head",
    "quantizer.weight_proj",
    "quantizer.codevectors",
    "project_q",
    "project_hid",
]


def     SCREAMING_SNAKE_CASE							(   a_						:				Any					,       a_						:				str					,       a_						:				Optional[Any]					,       a_						:				Tuple					,       a_						:				List[str]      ):
  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

  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":
    __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     SCREAMING_SNAKE_CASE							(   a_						:				Optional[Any]					,       a_						:				Dict      ):
  __a								=       []
  __a								=       fairseq_model.state_dict()

  __a								=       hf_model.feature_extractor

  # if encoder has different dim to decoder -> use proj_weight
  __a								=       None

  for name, value in fairseq_dict.items():
    __a								=       False
    if "conv_layers" in name:
      load_conv_layer(
          a_					,       a_					,       a_					,       a_					,       hf_model.config.feat_extract_norm == 'group'					,       )
      __a								=       True
    elif name.split('.'      )[0] == "proj":
      __a								=       fairseq_model.proj
      __a								=       True
    else:
      for key, mapped_key in MAPPING.items():
        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:
            __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}"      )

  return proj_weight


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


def     SCREAMING_SNAKE_CASE							(   a_						:				Union[str, Any]      ):
  __a						,							__a								=       emb.weight.shape
  __a								=       nn.Linear(a_					,       a_					,       bias=a_      )
  __a								=       emb.weight.data
  return lin_layer


def     SCREAMING_SNAKE_CASE							(   a_						:				Union[str, Any]      ):
  with open(a_					,       'r'					,       encoding='utf-8'      ) as f:
    __a								=       f.readlines()
    __a								=       [line.split(' '      )[0] for line in lines]

  __a								=       len(a_      )

  __a								=       {
      '<s>': 0,
      '<pad>': 1,
      '</s>': 2,
      '<unk>': 3,
  }

  vocab_dict.update(dict(zip(a_					,       range(4					,       num_words + 4      )      )      )      )
  return vocab_dict


@torch.no_grad()
def     SCREAMING_SNAKE_CASE							(   a_						:				Optional[Any]					,       a_						:				Optional[int]					,       a_						:				int					,       a_						:				Dict					,       a_						:				List[str]					,       a_						:				Union[str, Any]					,       a_						:				Union[str, Any]					,       ):
  __a								=       WavaVecaConfig.from_pretrained(a_      )
  __a								=       SpeechaTextaConfig.from_pretrained(
      a_					,       vocab_size=a_					,       decoder_layers=a_					,       do_stable_layer_norm=a_      )

  __a								=       WavaVecaFeatureExtractor(
      feature_size=1					,       sampling_rate=1_6000					,       padding_value=0					,       do_normalize=a_					,       return_attention_mask=a_					,       )

  __a						,							__a						,							__a								=       fairseq.checkpoint_utils.load_model_ensemble_and_task(
      [checkpoint_path]					,       arg_overrides={'data': '/'.join(dict_path.split('/'      )[:-1]      )}      )
  __a								=       model[0].eval()

  # set weights for wav2vec2 encoder
  __a								=       WavaVecaModel(a_      )
  __a								=       recursively_load_weights_wavaveca(model.encoder					,       a_      )

  __a								=       SpeechaTextaForCausalLM(a_      )
  __a						,							__a								=       hf_decoder.model.decoder.load_state_dict(model.decoder.state_dict()					,       strict=a_      )

  # set output linear layer
  unexpected_keys.remove('embed_out'      )
  __a								=       nn.Parameter(model.decoder.embed_out.detach()      )

  # layer norm is init to identity matrix so leaving it is fine
  logger.warning(f"The following keys are missing when loading the decoder weights: {missing_keys}"      )
  logger.warning(f"The following keys are unexpected when loading the decoder weights: {unexpected_keys}"      )

  __a								=       SpeechEncoderDecoderModel(encoder=a_					,       decoder=a_      )
  __a								=       False

  # add projection layer
  __a								=       nn.Parameter(projection_layer.weight      )
  __a								=       nn.Parameter(projection_layer.bias      )

  __a								=       create_vocab_dict(a_      )

  with open(os.path.join(a_					,       'vocab.json'      )					,       'w'      ) as fp:
    json.dump(a_					,       a_      )

  __a								=       SpeechaTextaTokenizer(os.path.join(a_					,       'vocab.json'      )      )
  tokenizer.save_pretrained(a_      )

  __a								=       hf_wavavec.config.to_dict()
  __a								=       tokenizer.pad_token_id
  __a								=       tokenizer.bos_token_id
  __a								=       tokenizer.eos_token_id
  __a								=       'speech_to_text_2'
  __a								=       'wav2vec2'

  __a								=       SpeechEncoderDecoderConfig.from_dict(a_      )

  hf_wavavec.save_pretrained(a_      )
  feature_extractor.save_pretrained(a_      )


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(
      "--encoder_config_path",
      default="facebook/wav2vec2-large-lv60",
      type=str,
      help="Path to hf encoder wav2vec2 checkpoint config",
  )
  parser.add_argument(
      "--decoder_config_path",
      default="facebook/s2t-small-mustc-en-fr-st",
      type=str,
      help="Path to hf decoder s2t checkpoint config",
  )
  parser.add_argument("--vocab_size", default=1_02_24, type=int, help="Vocab size of decoder")
  parser.add_argument("--num_decoder_layers", default=7, type=int, help="Number of decoder layers")

  UpperCAmelCase_					=   parser.parse_args()
  convert_wavaveca_checkpoint(
      args.checkpoint_path,
      args.pytorch_dump_folder_path,
      args.dict_path,
      encoder_config_path=args.encoder_config_path,
      decoder_config_path=args.decoder_config_path,
      vocab_size=args.vocab_size,
      num_decoder_layers=args.num_decoder_layers,
  )









'''simple docstring'''


import math


def     SCREAMING_SNAKE_CASE							(   a_						:				float					,       a_						:				float      ):

  if initial_intensity < 0:
    raise ValueError('The value of intensity cannot be negative'      )
    # handling of negative values of initial intensity
  if angle < 0 or angle > 360:
    raise ValueError('In Malus Law, the angle is in the range 0-360 degrees'      )
    # handling of values out of allowed range
  return initial_intensity * (math.cos(math.radians(a_      )      ) ** 2)


if __name__ == "__main__":
  import doctest

  doctest.testmod(name="malus_law")



