




import argparse
import pathlib
from pathlib import Path
from tempfile import TemporaryDirectory

import esm as esm_module
import torch
from esm.esmfold.va.misc import batch_encode_sequences as esmfold_encode_sequences
from esm.esmfold.va.pretrained import esmfold_va

from transformers.models.esm.configuration_esm import EsmConfig, EsmFoldConfig
from transformers.models.esm.modeling_esm import (
    EsmForMaskedLM,
    EsmForSequenceClassification,
    EsmIntermediate,
    EsmLayer,
    EsmOutput,
    EsmSelfAttention,
    EsmSelfOutput,
)
from transformers.models.esm.modeling_esmfold import EsmForProteinFolding
from transformers.models.esm.tokenization_esm import EsmTokenizer
from transformers.utils import logging


logging.set_verbosity_info()
UpperCAmelCase__						:       int							=    logging.get_logger(__name__)

UpperCAmelCase__						:       int							=    [
    (
        'protein1',
        'MNGTEGPNFYVPFSNATGVVRSPFEYPQYYLAEPWQFSMLAAYMFLLIVLGFPINFLTLYVTVQHKKLRTPLNYILLNLAVADLFMVLGGFTSTLYTSLHGYFVFGPTGCNLEGFFATLGGEIALWSLVVLAIERYVVVCKPMSNFRFGENHAIMGVAFTWVMALACAAPPLAGWSRYIPEGLQCSCGIDYYTLKPEVNNESFVIYMFVVHFTIPMIIIFFCYGQLVFTVKEAAAQQQESATTQKAEKEVTRMVIIMVIAFLICWVPYASVAFYIFTHQGSNFGPIFMTIPAFFAKSAAIYNPVIYIMMNKQFRNCMLTTICCGKNPLGDDEASATVSKTETSQVAPA',
    ),
    ('protein2', 'MKTVRQERLKSIVRILERSKEPVSGAQLAEELSVSRQVIVQDIAYLRSLGYNIVATPRGYVLA'),
    ('protein3', 'MKTVRQERLKSI<mask>RILERSKEPVSGAQLAEELS<mask>SRQVIVQDIAYLRSLGYN<mask>VATPRGYVLAGG'),
    ('protein4', 'MKTVRQERLKSI<mask>RILERSKEPVSGAQLAEELS<mask>SRQVIVQDIAYLRSLGYN<mask>VATPRGYVLA'),
]

UpperCAmelCase__						:       Optional[Any]							=    {
    'esm1b_t33_650M_UR50S': esm_module.pretrained.esm1b_t33_650M_UR50S,
    'esm1v_t33_650M_UR90S_1': esm_module.pretrained.esm1v_t33_650M_UR90S_1,
    'esm1v_t33_650M_UR90S_2': esm_module.pretrained.esm1v_t33_650M_UR90S_2,
    'esm1v_t33_650M_UR90S_3': esm_module.pretrained.esm1v_t33_650M_UR90S_3,
    'esm1v_t33_650M_UR90S_4': esm_module.pretrained.esm1v_t33_650M_UR90S_4,
    'esm1v_t33_650M_UR90S_5': esm_module.pretrained.esm1v_t33_650M_UR90S_5,
    'esm2_t48_15B_UR50D': esm_module.pretrained.esm2_t48_15B_UR50D,
    'esm2_t36_3B_UR50D': esm_module.pretrained.esm2_t36_3B_UR50D,
    'esm2_t33_650M_UR50D': esm_module.pretrained.esm2_t33_650M_UR50D,
    'esm2_t30_150M_UR50D': esm_module.pretrained.esm2_t30_150M_UR50D,
    'esm2_t12_35M_UR50D': esm_module.pretrained.esm2_t12_35M_UR50D,
    'esm2_t6_8M_UR50D': esm_module.pretrained.esm2_t6_8M_UR50D,
    'esmfold_v1': esmfold_va,
}

UpperCAmelCase__						:       Any							=    list('ARNDCQEGHILKMFPSTWYV')

UpperCAmelCase__						:       List[Any]							=    restypes + ['X']
UpperCAmelCase__						:       Union[str, Any]							=    restypes_with_x + ['<pad>', '<mask>', '<cls>', '<sep>', '<eos>']


def 		_lowerCamelCase				(					)			->		Any:





  """simple docstring"""





  with TemporaryDirectory() as tempdir:
    _lowercase: Optional[int]						     =    '\n'.join(UpperCamelCase							)
    _lowercase: Tuple						     =    Path(UpperCamelCase							) / 'vocab.txt'
    vocab_file.write_text(UpperCamelCase							)
    _lowercase: int						     =    EsmTokenizer(vocab_file=str(UpperCamelCase							)							)
  _lowercase: str						     =    0  # Overlaps with 'A' but that seems to be what they want
  return hf_tokenizer


def 		_lowerCamelCase				(					UpperCamelCase					,     UpperCamelCase							)			->		Any:





  """simple docstring"""





  _lowercase: Tuple						     =    our_module.load_state_dict(original_module.state_dict()							)
  if status.missing_keys:
    raise ValueError(F'Missing keys: {status.missing_keys}'							)
  if status.unexpected_keys:
    raise ValueError(F'Unexpected keys: {status.unexpected_keys}'							)


def 		_lowerCamelCase				(					UpperCamelCase					,     UpperCamelCase					,     UpperCamelCase					,     UpperCamelCase					,     UpperCamelCase							)			->		int:





  """simple docstring"""





  if model.startswith('esmfold'							):
    _lowercase: Optional[Any]						     =    MODEL_MAPPING[model]()
  else:
    _lowercase: Any						     =    MODEL_MAPPING[model]()
  esm.eval()  # disable dropout

  if model.startswith('esmfold'							):
    _lowercase: Optional[Any]						     =    esm.esm.embed_dim
    _lowercase: Any						     =    esm.esm.num_layers
    _lowercase: Dict						     =    esm.esm.attention_heads
    _lowercase: Tuple						     =    4 * embed_dim
    _lowercase: Any						     =    esm.esm.token_dropout
    _lowercase: int						     =    False  # This code path does not exist in ESM-2
    _lowercase: Union[str, Any]						     =    'rotary'
    _lowercase: Tuple						     =    True
    _lowercase: Dict						     =    EsmFoldConfig()
    for key, val in esm.cfg.items():
      if hasattr(UpperCamelCase					,     UpperCamelCase							) and key != "trunk":
        setattr(UpperCamelCase					,     UpperCamelCase					,     UpperCamelCase							)
    for key, val in esm.cfg.trunk.items():
      if hasattr(esmfold_config.trunk					,     UpperCamelCase							) and key != "structure_module":
        setattr(esmfold_config.trunk					,     UpperCamelCase					,     UpperCamelCase							)
    for key, val in esm.cfg.trunk.structure_module.items():
      if hasattr(esmfold_config.trunk.structure_module					,     UpperCamelCase							):
        setattr(esmfold_config.trunk.structure_module					,     UpperCamelCase					,     UpperCamelCase							)
  elif hasattr(UpperCamelCase					,     'args'							):
    # Indicates an ESM-1b or ESM-1v model
    _lowercase: List[Any]						     =    esm.args.embed_dim
    _lowercase: Union[str, Any]						     =    esm.args.layers
    _lowercase: str						     =    esm.args.attention_heads
    _lowercase: Union[str, Any]						     =    esm.args.ffn_embed_dim
    _lowercase: Optional[Any]						     =    esm.args.token_dropout
    _lowercase: Optional[Any]						     =    True if esm.emb_layer_norm_before else False
    _lowercase: Any						     =    'absolute'
    _lowercase: List[Any]						     =    False
    _lowercase: List[str]						     =    None
  else:
    # Indicates an ESM-2 model
    _lowercase: Optional[Any]						     =    esm.embed_dim
    _lowercase: str						     =    esm.num_layers
    _lowercase: Dict						     =    esm.attention_heads
    _lowercase: int						     =    4 * embed_dim  # This is hardcoded in ESM-2
    _lowercase: str						     =    esm.token_dropout
    _lowercase: Dict						     =    False  # This code path does not exist in ESM-2
    _lowercase: List[str]						     =    'rotary'
    _lowercase: int						     =    False
    _lowercase: Tuple						     =    None

  if is_folding_model:
    _lowercase: List[Any]						     =    esm.esm.alphabet
  _lowercase: int						     =    tuple(alphabet.all_toks							)
  _lowercase: Union[str, Any]						     =    alphabet.mask_idx
  _lowercase: Optional[Any]						     =    alphabet.padding_idx

  if is_folding_model:
    _lowercase: int						     =    esm.esm
  else:
    _lowercase: List[str]						     =    esm

  _lowercase: Union[str, Any]						     =    EsmConfig(
      vocab_size=original_esm_model.embed_tokens.num_embeddings					,     mask_token_id=UpperCamelCase					,     hidden_size=UpperCamelCase					,     num_hidden_layers=UpperCamelCase					,     num_attention_heads=UpperCamelCase					,     intermediate_size=UpperCamelCase					,     max_position_embeddings=1_026					,     layer_norm_eps=1E-5					,     attention_probs_dropout_prob=0.0					,     hidden_dropout_prob=0.0					,     pad_token_id=UpperCamelCase					,     emb_layer_norm_before=UpperCamelCase					,     token_dropout=UpperCamelCase					,     position_embedding_type=UpperCamelCase					,     is_folding_model=UpperCamelCase					,     esmfold_config=UpperCamelCase					,     vocab_list=UpperCamelCase					,     )
  if classification_head:
    _lowercase: Optional[Any]						     =    esm.classification_heads['mnli'].out_proj.weight.shape[0]
  print('Our ESM config:'					,     UpperCamelCase							)

  if model.startswith('esmfold'							):
    _lowercase: List[Any]						     =    EsmForProteinFolding
  elif classification_head:
    _lowercase: str						     =    EsmForSequenceClassification
  else:
    _lowercase: Optional[int]						     =    EsmForMaskedLM
  _lowercase: Optional[Any]						     =    model_class(UpperCamelCase							)
  model.eval()

  # Now let's copy all the weights.
  # Embeddings
  _lowercase: Dict						     =    original_esm_model.embed_tokens.weight
  if position_embedding_type == "absolute":
    _lowercase: Dict						     =    original_esm_model.embed_positions.weight

  if config.emb_layer_norm_before:
    _lowercase: List[Any]						     =    original_esm_model.emb_layer_norm_before.weight
    _lowercase: Optional[Any]						     =    original_esm_model.emb_layer_norm_before.bias

  _lowercase: Tuple						     =    original_esm_model.emb_layer_norm_after.weight
  _lowercase: Any						     =    original_esm_model.emb_layer_norm_after.bias

  for i in range(config.num_hidden_layers							):
    # Encoder: start of layer
    _lowercase: EsmLayer						     =    model.esm.encoder.layer[i]
    # esm_layer: TransformerSentenceEncoderLayer = original_esm_model.layers[i]
    _lowercase: Dict						     =    original_esm_model.layers[i]

    # self attention
    _lowercase: EsmSelfAttention						     =    layer.attention.self
    assert (
        esm_layer.self_attn.k_proj.weight.data.shape
        == esm_layer.self_attn.q_proj.weight.data.shape
        == esm_layer.self_attn.v_proj.weight.data.shape
        == torch.Size((config.hidden_size, config.hidden_size)							)
    )

    _lowercase: Dict						     =    esm_layer.self_attn.q_proj.weight
    _lowercase: List[str]						     =    esm_layer.self_attn.q_proj.bias
    _lowercase: Union[str, Any]						     =    esm_layer.self_attn.k_proj.weight
    _lowercase: Union[str, Any]						     =    esm_layer.self_attn.k_proj.bias
    _lowercase: List[Any]						     =    esm_layer.self_attn.v_proj.weight
    _lowercase: Optional[Any]						     =    esm_layer.self_attn.v_proj.bias

    if getattr(esm_layer.self_attn					,     'rot_emb'					,     UpperCamelCase							) is not None:
      # Matt: Although inv_freq is not a trainable weight, it is computed at model init and cached.
      # During the training of ESM-2 the model was converted to float16 precision, which also converts
      # the inv_freq tensor, and the loss of precision remains even if the model is loaded later as float32.
      # If we recompute inv_freq without this loss of precision then we will get subtly different rotary
      # embeddings, which are enough to cause significant discrepancies in model outputs. To avoid this,
      # we make sure the new model copies the data from the old inv_freq.
      _lowercase: Optional[int]						     =    esm_layer.self_attn.rot_emb.inv_freq

    # LayerNorm changes for pre-activation
    _lowercase: Tuple						     =    esm_layer.self_attn_layer_norm.weight
    _lowercase: Optional[int]						     =    esm_layer.self_attn_layer_norm.bias
    _lowercase: Optional[int]						     =    esm_layer.final_layer_norm.weight
    _lowercase: Dict						     =    esm_layer.final_layer_norm.bias

    # self-attention output
    _lowercase: EsmSelfOutput						     =    layer.attention.output
    assert self_output.dense.weight.shape == esm_layer.self_attn.out_proj.weight.shape
    _lowercase: Optional[Any]						     =    esm_layer.self_attn.out_proj.weight
    _lowercase: List[Any]						     =    esm_layer.self_attn.out_proj.bias

    # intermediate
    _lowercase: EsmIntermediate						     =    layer.intermediate
    assert intermediate.dense.weight.shape == esm_layer.fca.weight.shape
    _lowercase: Tuple						     =    esm_layer.fca.weight
    _lowercase: Tuple						     =    esm_layer.fca.bias

    # output
    _lowercase: EsmOutput						     =    layer.output
    assert bert_output.dense.weight.shape == esm_layer.fca.weight.shape
    _lowercase: Optional[int]						     =    esm_layer.fca.weight
    _lowercase: List[str]						     =    esm_layer.fca.bias
    # end of layer

  if is_folding_model:
    _lowercase: Any						     =    esm.esm_s_combine.data
    _lowercase: Union[str, Any]						     =    esm.afa_to_esm.data
    transfer_and_check_weights(esm.embedding					,     model.embedding							)
    transfer_and_check_weights(esm.esm_s_mlp					,     model.esm_s_mlp							)
    transfer_and_check_weights(esm.trunk					,     model.trunk							)
    transfer_and_check_weights(esm.distogram_head					,     model.distogram_head							)
    transfer_and_check_weights(esm.ptm_head					,     model.ptm_head							)
    transfer_and_check_weights(esm.lm_head					,     model.lm_head							)
    transfer_and_check_weights(esm.lddt_head					,     model.lddt_head							)

  elif classification_head:
    _lowercase: Optional[Any]						     =    esm.esm.classification_heads['mnli'].dense.weight
    _lowercase: Any						     =    esm.classification_heads['mnli'].dense.bias
    _lowercase: List[str]						     =    esm.classification_heads['mnli'].out_proj.weight
    _lowercase: int						     =    esm.classification_heads['mnli'].out_proj.bias
  else:
    # LM Head
    _lowercase: Dict						     =    esm.lm_head.dense.weight
    _lowercase: Optional[Any]						     =    esm.lm_head.dense.bias
    _lowercase: Tuple						     =    esm.lm_head.layer_norm.weight
    _lowercase: List[Any]						     =    esm.lm_head.layer_norm.bias
    _lowercase: List[Any]						     =    esm.lm_head.weight
    _lowercase: List[Any]						     =    esm.lm_head.bias

  # Contact prediction head
  transfer_and_check_weights(esm.contact_head					,     model.esm.contact_head							)

  # Prepare data (first 2 sequences from ESMStructuralSplitDataset superfamily / 4)
  if is_folding_model:
    # Folding models aren't trained on masked inputs and don't like mask tokens.
    _lowercase: int						     =    SAMPLE_DATA[:2]
  else:
    _lowercase: Optional[int]						     =    SAMPLE_DATA

  if is_folding_model:
    _lowercase: Union[str, Any]						     =    get_esmfold_tokenizer()
    _lowercase: Optional[int]						     =    hf_tokenizer(
        [row[1] for row in sample_data]					,     return_tensors='pt'					,     padding=UpperCamelCase					,     add_special_tokens=UpperCamelCase							)
    _lowercase: Tuple						     =    esmfold_encode_sequences([row[1] for row in sample_data]							)
    _lowercase: str						     =    torch.all(hf_tokens['input_ids'] == esmfold_aas							) and torch.all(
        hf_tokens['attention_mask'] == esmfold_mask							)
  else:
    # Let's check that we get the same results.
    _lowercase: List[Any]						     =    alphabet.get_batch_converter()
    _lowercase: int						     =    batch_converter(UpperCamelCase							)
    # Prepare tokenizer and make sure it matches
    with TemporaryDirectory() as tempdir:
      _lowercase: List[Any]						     =    '\n'.join(alphabet.all_toks							)
      _lowercase: List[Any]						     =    Path(UpperCamelCase							) / 'vocab.txt'
      vocab_file.write_text(UpperCamelCase							)
      _lowercase: Dict						     =    EsmTokenizer(vocab_file=str(UpperCamelCase							)							)

    _lowercase: Optional[int]						     =    hf_tokenizer([row[1] for row in sample_data]					,     return_tensors='pt'					,     padding=UpperCamelCase							)
    _lowercase: Any						     =    torch.all(hf_tokens['input_ids'] == batch_tokens							)

  print('Do both models tokenizers output the same tokens?'					,     '🔥' if success else '💩'							)
  if not success:
    raise Exception('Tokenization does not match!'							)

  with torch.no_grad():
    if is_folding_model:
      # Let's test the model in parts
      # ESMFold always converts the ESM stem to float16, which requires float16 ops
      # that don't exist on CPU. Therefore, to test it we need to run it on GPU. However,
      # ESMFold is what we in the community call a "big boy" and so we desperately avoid putting both the
      # original and the converted model on the GPU at the same time.
      _lowercase: List[Any]						     =    esm.cuda().infer([row[1] for row in sample_data]							)
      _lowercase: List[Any]						     =    model.cuda()(
          input_ids=hf_tokens['input_ids'].cuda()					,     attention_mask=hf_tokens['attention_mask'].cuda()							)
    else:
      _lowercase: Dict						     =    model(**UpperCamelCase					,     output_hidden_states=UpperCamelCase							)
      _lowercase: List[str]						     =    our_output['logits']
      if classification_head:
        _lowercase: int						     =    esm.model.classification_heads['mnli'](esm.extract_features(UpperCamelCase							)							)
      else:
        _lowercase: Optional[int]						     =    esm(hf_tokens['input_ids']					,     repr_layers=list(range(999							)							)							)
        _lowercase: Dict						     =    their_output['logits']

    if is_folding_model:
      _lowercase: Dict						     =    torch.max(torch.abs(our_output['positions'] - their_output['positions']							)							).item()
      _lowercase: int						     =    torch.allclose(our_output['positions']					,     their_output['positions']					,     atol=1E-5							)
    else:
      _lowercase: Optional[int]						     =    torch.max(torch.abs(our_output - their_output							)							).item()
      _lowercase: List[str]						     =    torch.allclose(UpperCamelCase					,     UpperCamelCase					,     atol=1E-5							)

    print(F'max_absolute_diff = {max_absolute_diff}'							)  # ~ 1e-5
    print('Do both models output the same tensors?'					,     '🔥' if success else '💩'							)

    if not success:
      raise Exception('Something went wRoNg'							)

    if not is_folding_model:
      # Let's check contact prediction too
      _lowercase: Optional[int]						     =    model.predict_contacts(hf_tokens['input_ids']					,     hf_tokens['attention_mask']							)
      _lowercase: str						     =    esm.predict_contacts(hf_tokens['input_ids']							)
      _lowercase: Dict						     =    torch.max(torch.abs(our_output - their_output							)							).item()
      _lowercase: str						     =    torch.allclose(UpperCamelCase					,     UpperCamelCase					,     atol=1E-5							)

      print('Contact prediction testing:'							)
      print(F'max_absolute_diff = {max_absolute_diff}'							)  # ~ 1e-5
      print('Do both models output the same tensors?'					,     '🔥' if success else '💩'							)

      if not success:
        raise Exception('Something went wRoNg'							)

    pathlib.Path(UpperCamelCase							).mkdir(parents=UpperCamelCase					,     exist_ok=UpperCamelCase							)
    print(F'Saving model to {pytorch_dump_folder_path}'							)
    model.save_pretrained(UpperCamelCase							)

    del esm  # Free up some memory before continuing

  print(F'Saving tokenizer to {pytorch_dump_folder_path}'							)
  hf_tokenizer.save_pretrained(UpperCamelCase							)

  if push_to_repo:
    model.push_to_hub(repo_id=UpperCamelCase					,     use_auth_token=UpperCamelCase							)
    hf_tokenizer.push_to_hub(repo_id=UpperCamelCase					,     use_auth_token=UpperCamelCase							)


if __name__ == "__main__":
   UpperCAmelCase__						:       Optional[Any]							=    argparse.ArgumentParser()
   # Required parameters
   parser.add_argument(
       '--pytorch_dump_folder_path', type=str, required=True, help='Path to the output PyTorch model.'
   )
   parser.add_argument(
       '--classification_head', action='store_true', help='Whether to convert a final classification head.'
   )
   parser.add_argument('--model', default=None, type=str, required=True, help='Name of model to convert.')
   parser.add_argument('--push_to_repo', type=str, help='Repo to upload to (including username!).')
   parser.add_argument('--auth_token', type=str, help='HuggingFace auth token.')
   UpperCAmelCase__						:       int							=    parser.parse_args()
   convert_esm_checkpoint_to_pytorch(
       args.model, args.pytorch_dump_folder_path, args.classification_head, args.push_to_repo, args.auth_token
   )






import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin

class      UpperCAmelCase						(			A   ):
   __SCREAMING_SNAKE_CASE		:							Optional[Any]							=					"AutoFeatureExtractor"
   __SCREAMING_SNAKE_CASE		:							Optional[Any]							=					"Speech2Text2Tokenizer"






   def __init__(	self   :       Optional[Any]						,					UpperCamelCase_   :       Any						,					UpperCamelCase_   :       List[Any]     )->			Dict:
     super().__init__(UpperCamelCase_						,					UpperCamelCase_     )
     _lowercase: Optional[int]						     =    self.feature_extractor
     _lowercase: int						     =    False






   def __call__(	self   :       List[Any]						,					*UpperCamelCase_   :       Dict						,					**UpperCamelCase_   :       Optional[int]     )->			str:
     # For backward compatibility
     if self._in_target_context_manager:
       return self.current_processor(*UpperCamelCase_						,					**UpperCamelCase_     )

     if "raw_speech" in kwargs:
       warnings.warn('Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead.'     )
       _lowercase: Tuple						     =    kwargs.pop('raw_speech'     )
     else:
       _lowercase: Optional[Any]						     =    kwargs.pop('audio'						,					UpperCamelCase_     )
     _lowercase: int						     =    kwargs.pop('sampling_rate'						,					UpperCamelCase_     )
     _lowercase: Dict						     =    kwargs.pop('text'						,					UpperCamelCase_     )
     if len(UpperCamelCase_     ) > 0:
       _lowercase: List[Any]						     =    args[0]
       _lowercase: List[Any]						     =    args[1:]

     if audio is None and text is None:
       raise ValueError('You need to specify either an `audio` or `text` input to process.'     )

     if audio is not None:
       _lowercase: Optional[int]						     =    self.feature_extractor(UpperCamelCase_						,					*UpperCamelCase_						,					sampling_rate=UpperCamelCase_						,					**UpperCamelCase_     )
     if text is not None:
       _lowercase: str						     =    self.tokenizer(UpperCamelCase_						,					**UpperCamelCase_     )

     if text is None:
       return inputs
     elif audio is None:
       return encodings
     else:
       _lowercase: int						     =    encodings['input_ids']
       return inputs






   def  _a						(	self   :       Optional[int]						,					*UpperCamelCase_   :       Tuple						,					**UpperCamelCase_   :       Dict     )->			List[str]:
     return self.tokenizer.batch_decode(*UpperCamelCase_						,					**UpperCamelCase_     )






   def  _a						(	self   :       Tuple						,					*UpperCamelCase_   :       List[Any]						,					**UpperCamelCase_   :       Union[str, Any]     )->			Optional[int]:
     return self.tokenizer.decode(*UpperCamelCase_						,					**UpperCamelCase_     )






   @contextmanager
   def  _a						(	self   :       Tuple     )->			Union[str, Any]:
     warnings.warn(
         '`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your '
         'labels by using the argument `text` of the regular `__call__` method (either in the same call as '
         'your audio inputs, or in a separate call.'     )
     _lowercase: Any						     =    True
     _lowercase: Optional[Any]						     =    self.tokenizer
     yield
     _lowercase: Union[str, Any]						     =    self.feature_extractor
     _lowercase: str						     =    False
