



'''simple docstring'''






import unittest

from transformers import EsmConfig, is_torch_available
from transformers.testing_utils import TestCasePlus, require_torch, slow, torch_device

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
     import torch

     from transformers import EsmForMaskedLM, EsmForSequenceClassification, EsmForTokenClassification, EsmModel
     from transformers.models.esm.modeling_esm import (
         ESM_PRETRAINED_MODEL_ARCHIVE_LIST,
         EsmEmbeddings,
         create_position_ids_from_input_ids,
     )

class 							a__:







    def __init__(   self				:						Optional[Any]		,    __snake_case				:						Union[str, Any]		,    __snake_case				:						Any=13		,    __snake_case				:						Optional[Any]=7		,    __snake_case				:						List[Any]=False		,    __snake_case				:						Optional[Any]=True		,    __snake_case				:						Optional[Any]=False		,    __snake_case				:						Union[str, Any]=True		,    __snake_case				:						List[str]=33		,    __snake_case				:						Dict=32		,    __snake_case				:						Any=5		,    __snake_case				:						List[Any]=4		,    __snake_case				:						List[Any]=37		,    __snake_case				:						List[str]="gelu"		,    __snake_case				:						Tuple=0.1		,    __snake_case				:						int=0.1		,    __snake_case				:						Any=5_12		,    __snake_case				:						int=16		,    __snake_case				:						Optional[int]=2		,    __snake_case				:						List[str]=0.02		,    __snake_case				:						Union[str, Any]=3		,    __snake_case				:						Union[str, Any]=4		,    __snake_case				:						List[Any]=None		,    ):
      a		:      Optional[int]       =  parent
      a		:      int       =  batch_size
      a		:      str       =  seq_length
      a		:      Optional[Any]       =  is_training
      a		:      Any       =  use_input_mask
      a		:      Optional[Any]       =  use_token_type_ids
      a		:      Union[str, Any]       =  use_labels
      a		:      Any       =  vocab_size
      a		:      List[Any]       =  hidden_size
      a		:      str       =  num_hidden_layers
      a		:      List[str]       =  num_attention_heads
      a		:      Optional[int]       =  intermediate_size
      a		:      int       =  hidden_act
      a		:      int       =  hidden_dropout_prob
      a		:      Optional[Any]       =  attention_probs_dropout_prob
      a		:      Tuple       =  max_position_embeddings
      a		:      List[Any]       =  type_vocab_size
      a		:      Optional[int]       =  type_sequence_label_size
      a		:      int       =  initializer_range
      a		:      List[Any]       =  num_labels
      a		:      Tuple       =  num_choices
      a		:      int       =  scope







    def     lowercase_ (   self				:						Any			):
      a		:      int       =  ids_tensor([self.batch_size, self.seq_length]		,    self.vocab_size			)

      a		:      Dict       =  None
      if self.use_input_mask:
        a		:      str       =  random_attention_mask([self.batch_size, self.seq_length]			)

      a		:      List[str]       =  None
      a		:      Union[str, Any]       =  None
      a		:      Optional[int]       =  None
      if self.use_labels:
        a		:      Dict       =  ids_tensor([self.batch_size]		,    self.type_sequence_label_size			)
        a		:      Optional[Any]       =  ids_tensor([self.batch_size, self.seq_length]		,    self.num_labels			)
        a		:      Tuple       =  ids_tensor([self.batch_size]		,    self.num_choices			)

      a		:      Optional[int]       =  self.get_config()

      return config, input_ids, input_mask, sequence_labels, token_labels, choice_labels







    def     lowercase_ (   self				:						Union[str, Any]			):
      return EsmConfig(
          vocab_size=self.vocab_size		,    hidden_size=self.hidden_size		,    pad_token_id=1		,    num_hidden_layers=self.num_hidden_layers		,    num_attention_heads=self.num_attention_heads		,    intermediate_size=self.intermediate_size		,    hidden_act=self.hidden_act		,    hidden_dropout_prob=self.hidden_dropout_prob		,    attention_probs_dropout_prob=self.attention_probs_dropout_prob		,    max_position_embeddings=self.max_position_embeddings		,    type_vocab_size=self.type_vocab_size		,    initializer_range=self.initializer_range		,    )







    def     lowercase_ (   self				:						str		,    __snake_case				:						int		,    __snake_case				:						int		,    __snake_case				:						Dict		,    __snake_case				:						Union[str, Any]		,    __snake_case				:						Any		,    __snake_case				:						str			):
      a		:      List[Any]       =  EsmModel(config=__snake_case			)
      model.to(__snake_case			)
      model.eval()
      a		:      List[str]       =  model(__snake_case		,    attention_mask=__snake_case			)
      a		:      Any       =  model(__snake_case			)
      a		:      List[str]       =  model(__snake_case			)

      self.parent.assertEqual(result.last_hidden_state.shape		,    (self.batch_size, self.seq_length, self.hidden_size)			)
      self.parent.assertEqual(result.pooler_output.shape		,    (self.batch_size, self.hidden_size)			)







    def     lowercase_ (   self				:						Dict		,    __snake_case				:						int		,    __snake_case				:						Tuple		,    __snake_case				:						Optional[Any]		,    __snake_case				:						int		,    __snake_case				:						Optional[Any]		,    __snake_case				:						Optional[Any]			):
      a		:      str       =  EsmForMaskedLM(config=__snake_case			)
      model.to(__snake_case			)
      model.eval()
      a		:      List[Any]       =  model(__snake_case		,    attention_mask=__snake_case		,    labels=__snake_case			)
      self.parent.assertEqual(result.logits.shape		,    (self.batch_size, self.seq_length, self.vocab_size)			)







    def     lowercase_ (   self				:						Optional[Any]		,    __snake_case				:						Any		,    __snake_case				:						str		,    __snake_case				:						List[Any]		,    __snake_case				:						List[Any]		,    __snake_case				:						str		,    __snake_case				:						Dict			):
      a		:      str       =  self.num_labels
      a		:      Union[str, Any]       =  EsmForTokenClassification(config=__snake_case			)
      model.to(__snake_case			)
      model.eval()
      a		:      Dict       =  model(__snake_case		,    attention_mask=__snake_case		,    labels=__snake_case			)
      self.parent.assertEqual(result.logits.shape		,    (self.batch_size, self.seq_length, self.num_labels)			)







    def     lowercase_ (   self				:						List[Any]			):
      a		:      Optional[int]       =  self.prepare_config_and_inputs()
      (
          (
          a
      )     ,			(
          a
      )     ,			(
          a
      )     ,			(
          a
      )     ,			(
          a
      )     ,			(
          a
      )     ,			
      )		:      Optional[Any]       =  config_and_inputs
      a		:      List[str]       =  {'input_ids': input_ids, 'attention_mask': input_mask}
      return config, inputs_dict

@require_torch
class 							a__(      lowerCamelCase__	,				lowerCamelCase__	,				unittest.TestCase			):
    lowercase__				=					False

    lowercase__				=					(
        (
            EsmForMaskedLM,
            EsmModel,
            EsmForSequenceClassification,
            EsmForTokenClassification,
        )
        if is_torch_available()
        else ()
    )
    lowercase__				=					()
    lowercase__				=					(
        {
            """feature-extraction""": EsmModel,
            """fill-mask""": EsmForMaskedLM,
            """text-classification""": EsmForSequenceClassification,
            """token-classification""": EsmForTokenClassification,
            """zero-shot""": EsmForSequenceClassification,
        }
        if is_torch_available()
        else {}
    )
    lowercase__				=					True







    def     lowercase_ (   self				:						Any			):
      a		:      Any       =  EsmModelTester(self			)
      a		:      Optional[Any]       =  ConfigTester(self		,    config_class=__snake_case		,    hidden_size=37			)







    def     lowercase_ (   self				:						Optional[Any]			):
      self.config_tester.run_common_tests()







    def     lowercase_ (   self				:						Any			):
      a		:      List[str]       =  self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_model(*__snake_case			)







    def     lowercase_ (   self				:						Optional[Any]			):
      a		:      Any       =  self.model_tester.prepare_config_and_inputs()
      for type in ["absolute", "relative_key", "relative_key_query"]:
        a		:      List[str]       =  type
        self.model_tester.create_and_check_model(*__snake_case			)







    def     lowercase_ (   self				:						Tuple			):
      a		:      Tuple       =  self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_for_masked_lm(*__snake_case			)







    def     lowercase_ (   self				:						Optional[int]			):
      a		:      Any       =  self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_for_token_classification(*__snake_case			)







    @slow
    def     lowercase_ (   self				:						List[str]			):
      for model_name in ESM_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
        a		:      List[str]       =  EsmModel.from_pretrained(__snake_case			)
        self.assertIsNotNone(__snake_case			)







    def     lowercase_ (   self				:						Any			):
      a		:      List[str]       =  self.model_tester.prepare_config_and_inputs()[0]
      a		:      Optional[int]       =  EsmEmbeddings(config=__snake_case			)

      a		:      Optional[int]       =  torch.as_tensor([[12, 31, 13, model.padding_idx]]			)
      a		:      List[str]       =  torch.as_tensor(
          [
              [
                  0 + model.padding_idx + 1,
                  1 + model.padding_idx + 1,
                  2 + model.padding_idx + 1,
                  model.padding_idx,
              ]
          ]			)
      a		:      List[Any]       =  create_position_ids_from_input_ids(__snake_case		,    model.padding_idx			)
      self.assertEqual(position_ids.shape		,    expected_positions.shape			)
      self.assertTrue(torch.all(torch.eq(__snake_case		,    __snake_case			)			)			)







    def     lowercase_ (   self				:						str			):
      a		:      str       =  self.model_tester.prepare_config_and_inputs()[0]
      a		:      Any       =  EsmEmbeddings(config=__snake_case			)

      a		:      Union[str, Any]       =  torch.empty(2		,    4		,    30			)
      a		:      List[str]       =  [
          0 + embeddings.padding_idx + 1,
          1 + embeddings.padding_idx + 1,
          2 + embeddings.padding_idx + 1,
          3 + embeddings.padding_idx + 1,
      ]
      a		:      List[Any]       =  torch.as_tensor([expected_single_positions, expected_single_positions]			)
      a		:      Dict       =  embeddings.create_position_ids_from_inputs_embeds(__snake_case			)
      self.assertEqual(position_ids.shape		,    expected_positions.shape			)
      self.assertTrue(torch.all(torch.eq(__snake_case		,    __snake_case			)			)			)







    @unittest.skip('Esm does not support embedding resizing'			)
    def     lowercase_ (   self				:						str			):
      pass







    @unittest.skip('Esm does not support embedding resizing'			)
    def     lowercase_ (   self				:						List[Any]			):
      pass







    @unittest.skip('Will be fixed soon by reducing the size of the model used for common tests.'			)
    def     lowercase_ (   self				:						List[str]			):
      pass

@require_torch
class 							a__(      lowerCamelCase__			):







    @slow
    def     lowercase_ (   self				:						Any			):
      with torch.no_grad():
        a		:      str       =  EsmForMaskedLM.from_pretrained('facebook/esm2_t6_8M_UR50D'			)
        model.eval()
        a		:      int       =  torch.tensor([[0, 1, 2, 3, 4, 5]]			)
        a		:      Optional[Any]       =  model(__snake_case			)[0]

        a		:      Optional[Any]       =  33

        a		:      Any       =  torch.Size((1, 6, vocab_size)			)
        self.assertEqual(output.shape		,    __snake_case			)

        a		:      Any       =  torch.tensor(
            [[[8.9215, -10.5898, -6.4671], [-6.3967, -13.9114, -1.1212], [-7.7812, -13.9516, -3.7406]]]			)
        self.assertTrue(torch.allclose(output[:, :3, :3]		,    __snake_case		,    atol=1e-4			)			)







    @slow
    def     lowercase_ (   self				:						Tuple			):
      with torch.no_grad():
        a		:      Optional[int]       =  EsmModel.from_pretrained('facebook/esm2_t6_8M_UR50D'			)
        model.eval()

        a		:      Dict       =  torch.tensor([[0, 6, 4, 13, 5, 4, 16, 12, 11, 7, 2]]			)
        a		:      int       =  model(__snake_case			)[0]
        # compare the actual values for a slice.
        a		:      Optional[int]       =  torch.tensor(
            [[[0.1444, 0.5413, 0.3248], [0.3034, 0.0053, 0.3108], [0.3228, -0.2499, 0.3415]]]			)
        self.assertTrue(torch.allclose(output[:, :3, :3]		,    __snake_case		,    atol=1e-4			)			)




'''simple docstring'''






from collections import UserDict
from typing import Union

import numpy as np
import requests

from ..utils import (
    add_end_docstrings,
    logging,
)
from .audio_classification import ffmpeg_read
from .base import PIPELINE_INIT_ARGS, Pipeline


lowerCAmelCase:      Optional[Any]		 =		logging.get_logger(__name__)

@add_end_docstrings(lowerCamelCase__			)
class 							a__(      lowerCamelCase__			):







    def __init__(   self				:						Union[str, Any]		,    **__snake_case				:						List[str]			):
      super().__init__(**__snake_case			)

      if self.framework != "pt":
        raise ValueError(F"""The {self.__class__} is only available in PyTorch."""			)
      # No specific FOR_XXX available yet







    def __call__(   self				:						Any		,    __snake_case				:						Union[np.ndarray, bytes, str]		,    **__snake_case				:						Optional[Any]			):
      return super().__call__(__snake_case		,    **__snake_case			)







    def     lowercase_ (   self				:						Dict		,    **__snake_case				:						Optional[int]			):
      a		:      str       =  {}
      if "candidate_labels" in kwargs:
        a		:      Optional[int]       =  kwargs['candidate_labels']
      if "hypothesis_template" in kwargs:
        a		:      Any       =  kwargs['hypothesis_template']

      return preprocess_params, {}, {}







    def     lowercase_ (   self				:						int		,    __snake_case				:						Optional[Any]		,    __snake_case				:						List[Any]=None		,    __snake_case				:						List[str]="This is a sound of {}."			):
      if isinstance(__snake_case		,    __snake_case			):
        if audio.startswith('http://'			) or audio.startswith('https://'			):
          # We need to actually check for a real protocol, otherwise it's impossible to use a local file
          # like http_huggingface_co.png
          a		:      List[Any]       =  requests.get(__snake_case			).content
        else:
          with open(__snake_case		,    'rb'			) as f:
            a		:      List[str]       =  f.read()

      if isinstance(__snake_case		,    __snake_case			):
        a		:      Optional[Any]       =  ffmpeg_read(__snake_case		,    self.feature_extractor.sampling_rate			)

      if not isinstance(__snake_case		,    np.ndarray			):
        raise ValueError('We expect a numpy ndarray as input'			)
      if len(audio.shape			) != 1:
        raise ValueError('We expect a single channel audio input for ZeroShotAudioClassificationPipeline'			)

      a		:      str       =  self.feature_extractor(
          [audio]		,    sampling_rate=self.feature_extractor.sampling_rate		,    return_tensors='pt'			)
      a		:      Tuple       =  candidate_labels
      a		:      Union[str, Any]       =  [hypothesis_template.format(__snake_case			) for x in candidate_labels]
      a		:      int       =  self.tokenizer(__snake_case		,    return_tensors=self.framework		,    padding=__snake_case			)
      a		:      Any       =  [text_inputs]
      return inputs







    def     lowercase_ (   self				:						Optional[Any]		,    __snake_case				:						str			):
      a		:      Any       =  model_inputs.pop('candidate_labels'			)
      a		:      List[Any]       =  model_inputs.pop('text_inputs'			)
      if isinstance(text_inputs[0]		,    __snake_case			):
        a		:      str       =  text_inputs[0]
      else:
        # Batching case.
        a		:      Dict       =  text_inputs[0][0]

      a		:      List[Any]       =  self.model(**__snake_case		,    **__snake_case			)

      a		:      Any       =  {
          'candidate_labels': candidate_labels,
          'logits': outputs.logits_per_audio,
      }
      return model_outputs







    def     lowercase_ (   self				:						Union[str, Any]		,    __snake_case				:						str			):
      a		:      Optional[int]       =  model_outputs.pop('candidate_labels'			)
      a		:      List[str]       =  model_outputs['logits'][0]

      if self.framework == "pt":
        a		:      str       =  logits.softmax(dim=0			)
        a		:      List[Any]       =  probs.tolist()
      else:
        raise ValueError('`tf` framework not supported.'			)

      a		:      Dict       =  [
          {'score': score, 'label': candidate_label}
          for score, candidate_label in sorted(zip(__snake_case		,    __snake_case			)		,    key=lambda __snake_case			: -x[0]			)
      ]
      return result