




import unittest

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

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


if is_torch_available():
  import torch

  from transformers import (
      RemBertConfig,
      RemBertForCausalLM,
      RemBertForMaskedLM,
      RemBertForMultipleChoice,
      RemBertForQuestionAnswering,
      RemBertForSequenceClassification,
      RemBertForTokenClassification,
      RemBertModel,
  )
  from transformers.models.rembert.modeling_rembert import REMBERT_PRETRAINED_MODEL_ARCHIVE_LIST
class       _UpperCamelCase   :
      def __init__(      self :int			,   lowerCamelCase :Optional[int]			,   lowerCamelCase :int=13			,   lowerCamelCase :Optional[Any]=7			,   lowerCamelCase :str=True			,   lowerCamelCase :int=True			,   lowerCamelCase :List[str]=True			,   lowerCamelCase :Union[str, Any]=True			,   lowerCamelCase :Optional[int]=99			,   lowerCamelCase :Tuple=32			,   lowerCamelCase :Tuple=18			,   lowerCamelCase :Union[str, Any]=43			,   lowerCamelCase :Tuple=5			,   lowerCamelCase :List[str]=4			,   lowerCamelCase :List[Any]=37			,   lowerCamelCase :int="gelu"			,   lowerCamelCase :Optional[Any]=0.1			,   lowerCamelCase :Tuple=0.1			,   lowerCamelCase :List[Any]=512			,   lowerCamelCase :str=16			,   lowerCamelCase :Dict=2			,   lowerCamelCase :Any=0.02			,   lowerCamelCase :Optional[Any]=3			,   lowerCamelCase :Optional[int]=4			,   lowerCamelCase :Optional[int]=None			,   )  ->      Tuple:
       UpperCAmelCase__							=    parent
       UpperCAmelCase__							=    batch_size
       UpperCAmelCase__							=    seq_length
       UpperCAmelCase__							=    is_training
       UpperCAmelCase__							=    use_input_mask
       UpperCAmelCase__							=    use_token_type_ids
       UpperCAmelCase__							=    use_labels
       UpperCAmelCase__							=    vocab_size
       UpperCAmelCase__							=    hidden_size
       UpperCAmelCase__							=    input_embedding_size
       UpperCAmelCase__							=    output_embedding_size
       UpperCAmelCase__							=    num_hidden_layers
       UpperCAmelCase__							=    num_attention_heads
       UpperCAmelCase__							=    intermediate_size
       UpperCAmelCase__							=    hidden_act
       UpperCAmelCase__							=    hidden_dropout_prob
       UpperCAmelCase__							=    attention_probs_dropout_prob
       UpperCAmelCase__							=    max_position_embeddings
       UpperCAmelCase__							=    type_vocab_size
       UpperCAmelCase__							=    type_sequence_label_size
       UpperCAmelCase__							=    initializer_range
       UpperCAmelCase__							=    num_labels
       UpperCAmelCase__							=    num_choices
       UpperCAmelCase__							=    scope
      def 						UpperCAmelCase_    (      self :int   )  ->      List[Any]:
       UpperCAmelCase__							=    ids_tensor([self.batch_size, self.seq_length]			,   self.vocab_size   )

       UpperCAmelCase__							=    None
       if self.use_input_mask:
        UpperCAmelCase__							=    random_attention_mask([self.batch_size, self.seq_length]   )

       UpperCAmelCase__							=    None
       if self.use_token_type_ids:
        UpperCAmelCase__							=    ids_tensor([self.batch_size, self.seq_length]			,   self.type_vocab_size   )

       UpperCAmelCase__							=    None
       UpperCAmelCase__							=    None
       UpperCAmelCase__							=    None
       if self.use_labels:
        UpperCAmelCase__							=    ids_tensor([self.batch_size]			,   self.type_sequence_label_size   )
        UpperCAmelCase__							=    ids_tensor([self.batch_size, self.seq_length]			,   self.num_labels   )
        UpperCAmelCase__							=    ids_tensor([self.batch_size]			,   self.num_choices   )

       UpperCAmelCase__							=    RemBertConfig(
           vocab_size=self.vocab_size			,   hidden_size=self.hidden_size			,   input_embedding_size=self.input_embedding_size			,   output_embedding_size=self.output_embedding_size			,   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			,   is_decoder=lowerCamelCase			,   initializer_range=self.initializer_range			,   )

       return config, input_ids, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels
      def 						UpperCAmelCase_    (      self :str   )  ->      str:
       (
           (
           UpperCAmelCase__
       )			,			(
           UpperCAmelCase__
       )			,			(
           UpperCAmelCase__
       )			,			(
           UpperCAmelCase__
       )			,			(
           UpperCAmelCase__
       )			,			(
           UpperCAmelCase__
       )			,			(
           UpperCAmelCase__
       )			,			
       )							=    self.prepare_config_and_inputs()

       UpperCAmelCase__							=    True
       UpperCAmelCase__							=    floats_tensor([self.batch_size, self.seq_length, self.hidden_size]   )
       UpperCAmelCase__							=    ids_tensor([self.batch_size, self.seq_length]			,   vocab_size=2   )

       return (
           config,
           input_ids,
           token_type_ids,
           input_mask,
           sequence_labels,
           token_labels,
           choice_labels,
           encoder_hidden_states,
           encoder_attention_mask,
       )
      def 						UpperCAmelCase_    (      self :str			,   lowerCamelCase :Any			,   lowerCamelCase :Optional[int]			,   lowerCamelCase :Tuple			,   lowerCamelCase :Dict			,   lowerCamelCase :Union[str, Any]			,   lowerCamelCase :int			,   lowerCamelCase :List[str]   )  ->      Optional[Any]:
       UpperCAmelCase__							=    RemBertModel(config=lowerCamelCase   )
       model.to(lowerCamelCase   )
       model.eval()
       UpperCAmelCase__							=    model(lowerCamelCase			,   attention_mask=lowerCamelCase			,   token_type_ids=lowerCamelCase   )
       UpperCAmelCase__							=    model(lowerCamelCase			,   token_type_ids=lowerCamelCase   )
       UpperCAmelCase__							=    model(lowerCamelCase   )
       self.parent.assertEqual(result.last_hidden_state.shape			,   (self.batch_size, self.seq_length, self.hidden_size)   )
      def 						UpperCAmelCase_    (      self :str			,   lowerCamelCase :Any			,   lowerCamelCase :Union[str, Any]			,   lowerCamelCase :Dict			,   lowerCamelCase :Optional[Any]			,   lowerCamelCase :Optional[Any]			,   lowerCamelCase :Dict			,   lowerCamelCase :Tuple			,   lowerCamelCase :Optional[int]			,   lowerCamelCase :str			,   )  ->      Optional[int]:
       UpperCAmelCase__							=    True
       UpperCAmelCase__							=    RemBertModel(lowerCamelCase   )
       model.to(lowerCamelCase   )
       model.eval()
       UpperCAmelCase__							=    model(
           lowerCamelCase			,   attention_mask=lowerCamelCase			,   token_type_ids=lowerCamelCase			,   encoder_hidden_states=lowerCamelCase			,   encoder_attention_mask=lowerCamelCase			,   )
       UpperCAmelCase__							=    model(
           lowerCamelCase			,   attention_mask=lowerCamelCase			,   token_type_ids=lowerCamelCase			,   encoder_hidden_states=lowerCamelCase			,   )
       UpperCAmelCase__							=    model(lowerCamelCase			,   attention_mask=lowerCamelCase			,   token_type_ids=lowerCamelCase   )
       self.parent.assertEqual(result.last_hidden_state.shape			,   (self.batch_size, self.seq_length, self.hidden_size)   )
      def 						UpperCAmelCase_    (      self :List[Any]			,   lowerCamelCase :Union[str, Any]			,   lowerCamelCase :int			,   lowerCamelCase :Union[str, Any]			,   lowerCamelCase :Optional[Any]			,   lowerCamelCase :Dict			,   lowerCamelCase :int			,   lowerCamelCase :List[str]			,   lowerCamelCase :List[Any]			,   lowerCamelCase :Dict			,   )  ->      Optional[Any]:
       UpperCAmelCase__							=    RemBertForCausalLM(config=lowerCamelCase   )
       model.to(lowerCamelCase   )
       model.eval()
       UpperCAmelCase__							=    model(lowerCamelCase			,   attention_mask=lowerCamelCase			,   token_type_ids=lowerCamelCase			,   labels=lowerCamelCase   )
       self.parent.assertEqual(result.logits.shape			,   (self.batch_size, self.seq_length, self.vocab_size)   )
      def 						UpperCAmelCase_    (      self :Tuple			,   lowerCamelCase :Union[str, Any]			,   lowerCamelCase :List[str]			,   lowerCamelCase :List[Any]			,   lowerCamelCase :Optional[Any]			,   lowerCamelCase :Tuple			,   lowerCamelCase :List[str]			,   lowerCamelCase :List[Any]   )  ->      List[Any]:
       UpperCAmelCase__							=    RemBertForMaskedLM(config=lowerCamelCase   )
       model.to(lowerCamelCase   )
       model.eval()
       UpperCAmelCase__							=    model(lowerCamelCase			,   attention_mask=lowerCamelCase			,   token_type_ids=lowerCamelCase			,   labels=lowerCamelCase   )
       self.parent.assertEqual(result.logits.shape			,   (self.batch_size, self.seq_length, self.vocab_size)   )
      def 						UpperCAmelCase_    (      self :Optional[Any]			,   lowerCamelCase :int			,   lowerCamelCase :List[Any]			,   lowerCamelCase :Union[str, Any]			,   lowerCamelCase :List[str]			,   lowerCamelCase :Dict			,   lowerCamelCase :int			,   lowerCamelCase :Optional[int]			,   lowerCamelCase :int			,   lowerCamelCase :Optional[Any]			,   )  ->      Tuple:
       UpperCAmelCase__							=    True
       UpperCAmelCase__							=    True
       UpperCAmelCase__							=    RemBertForCausalLM(config=lowerCamelCase   )
       model.to(lowerCamelCase   )
       model.eval()

       # first forward pass
       UpperCAmelCase__							=    model(
           lowerCamelCase			,   attention_mask=lowerCamelCase			,   encoder_hidden_states=lowerCamelCase			,   encoder_attention_mask=lowerCamelCase			,   use_cache=lowerCamelCase			,   )
       UpperCAmelCase__							=    outputs.past_key_values

       # create hypothetical multiple next token and extent to next_input_ids
       UpperCAmelCase__							=    ids_tensor((self.batch_size, 3)			,   config.vocab_size   )
       UpperCAmelCase__							=    ids_tensor((self.batch_size, 3)			,   vocab_size=2   )

       # append to next input_ids and
       UpperCAmelCase__							=    torch.cat([input_ids, next_tokens]			,   dim=-1   )
       UpperCAmelCase__							=    torch.cat([input_mask, next_mask]			,   dim=-1   )

       UpperCAmelCase__							=    model(
           lowerCamelCase			,   attention_mask=lowerCamelCase			,   encoder_hidden_states=lowerCamelCase			,   encoder_attention_mask=lowerCamelCase			,   output_hidden_states=lowerCamelCase			,   )["hidden_states"][0]
       UpperCAmelCase__							=    model(
           lowerCamelCase			,   attention_mask=lowerCamelCase			,   encoder_hidden_states=lowerCamelCase			,   encoder_attention_mask=lowerCamelCase			,   past_key_values=lowerCamelCase			,   output_hidden_states=lowerCamelCase			,   )["hidden_states"][0]

       # select random slice
       UpperCAmelCase__							=    ids_tensor((1,)			,   output_from_past.shape[-1]   ).item()
       UpperCAmelCase__							=    output_from_no_past[:, -3:, random_slice_idx].detach()
       UpperCAmelCase__							=    output_from_past[:, :, random_slice_idx].detach()

       self.parent.assertTrue(output_from_past_slice.shape[1] == next_tokens.shape[1]   )

       # test that outputs are equal for slice
       self.parent.assertTrue(torch.allclose(lowerCamelCase			,   lowerCamelCase			,   atol=1e-3   )   )
      def 						UpperCAmelCase_    (      self :Union[str, Any]			,   lowerCamelCase :List[Any]			,   lowerCamelCase :Any			,   lowerCamelCase :int			,   lowerCamelCase :List[str]			,   lowerCamelCase :Tuple			,   lowerCamelCase :Optional[int]			,   lowerCamelCase :Dict   )  ->      int:
       UpperCAmelCase__							=    RemBertForQuestionAnswering(config=lowerCamelCase   )
       model.to(lowerCamelCase   )
       model.eval()
       UpperCAmelCase__							=    model(
           lowerCamelCase			,   attention_mask=lowerCamelCase			,   token_type_ids=lowerCamelCase			,   start_positions=lowerCamelCase			,   end_positions=lowerCamelCase			,   )
       self.parent.assertEqual(result.start_logits.shape			,   (self.batch_size, self.seq_length)   )
       self.parent.assertEqual(result.end_logits.shape			,   (self.batch_size, self.seq_length)   )
      def 						UpperCAmelCase_    (      self :List[str]			,   lowerCamelCase :Union[str, Any]			,   lowerCamelCase :Dict			,   lowerCamelCase :Optional[int]			,   lowerCamelCase :Optional[Any]			,   lowerCamelCase :Optional[int]			,   lowerCamelCase :Tuple			,   lowerCamelCase :Optional[int]   )  ->      Optional[int]:
       UpperCAmelCase__							=    self.num_labels
       UpperCAmelCase__							=    RemBertForSequenceClassification(lowerCamelCase   )
       model.to(lowerCamelCase   )
       model.eval()
       UpperCAmelCase__							=    model(lowerCamelCase			,   attention_mask=lowerCamelCase			,   token_type_ids=lowerCamelCase			,   labels=lowerCamelCase   )
       self.parent.assertEqual(result.logits.shape			,   (self.batch_size, self.num_labels)   )
      def 						UpperCAmelCase_    (      self :Any			,   lowerCamelCase :Optional[int]			,   lowerCamelCase :Dict			,   lowerCamelCase :Dict			,   lowerCamelCase :Dict			,   lowerCamelCase :Optional[Any]			,   lowerCamelCase :List[Any]			,   lowerCamelCase :Dict   )  ->      int:
       UpperCAmelCase__							=    self.num_labels
       UpperCAmelCase__							=    RemBertForTokenClassification(config=lowerCamelCase   )
       model.to(lowerCamelCase   )
       model.eval()
       UpperCAmelCase__							=    model(lowerCamelCase			,   attention_mask=lowerCamelCase			,   token_type_ids=lowerCamelCase			,   labels=lowerCamelCase   )
       self.parent.assertEqual(result.logits.shape			,   (self.batch_size, self.seq_length, self.num_labels)   )
      def 						UpperCAmelCase_    (      self :Union[str, Any]			,   lowerCamelCase :Dict			,   lowerCamelCase :Tuple			,   lowerCamelCase :Optional[int]			,   lowerCamelCase :Optional[int]			,   lowerCamelCase :str			,   lowerCamelCase :int			,   lowerCamelCase :List[Any]   )  ->      Optional[Any]:
       UpperCAmelCase__							=    self.num_choices
       UpperCAmelCase__							=    RemBertForMultipleChoice(config=lowerCamelCase   )
       model.to(lowerCamelCase   )
       model.eval()
       UpperCAmelCase__							=    input_ids.unsqueeze(1   ).expand(-1			,   self.num_choices			,   -1   ).contiguous()
       UpperCAmelCase__							=    token_type_ids.unsqueeze(1   ).expand(-1			,   self.num_choices			,   -1   ).contiguous()
       UpperCAmelCase__							=    input_mask.unsqueeze(1   ).expand(-1			,   self.num_choices			,   -1   ).contiguous()
       UpperCAmelCase__							=    model(
           lowerCamelCase			,   attention_mask=lowerCamelCase			,   token_type_ids=lowerCamelCase			,   labels=lowerCamelCase			,   )
       self.parent.assertEqual(result.logits.shape			,   (self.batch_size, self.num_choices)   )







      def 						UpperCAmelCase_    (      self :Optional[int]   )  ->      Optional[Any]:
       UpperCAmelCase__							=    self.prepare_config_and_inputs()
       (
           (
           UpperCAmelCase__
       )			,			(
           UpperCAmelCase__
       )			,			(
           UpperCAmelCase__
       )			,			(
           UpperCAmelCase__
       )			,			(
           UpperCAmelCase__
       )			,			(
           UpperCAmelCase__
       )			,			(
           UpperCAmelCase__
       )			,			
       )							=    config_and_inputs
       UpperCAmelCase__							=    {"input_ids": input_ids, "token_type_ids": token_type_ids, "attention_mask": input_mask}
       return config, inputs_dict



@require_torch
class       _UpperCamelCase   (					lowerCAmelCase						,							lowerCAmelCase						,							unittest.TestCase		):
      UpperCAmelCase_          =		(
          (
              RemBertModel,
              RemBertForMaskedLM,
              RemBertForCausalLM,
              RemBertForMultipleChoice,
              RemBertForQuestionAnswering,
              RemBertForSequenceClassification,
              RemBertForTokenClassification,
          )
          if is_torch_available()
          else ()
      )
      UpperCAmelCase_          =		(RemBertForCausalLM,) if is_torch_available() else ()
      UpperCAmelCase_          =		(
          {
              """feature-extraction""": RemBertModel,
              """fill-mask""": RemBertForMaskedLM,
              """question-answering""": RemBertForQuestionAnswering,
              """text-classification""": RemBertForSequenceClassification,
              """text-generation""": RemBertForCausalLM,
              """token-classification""": RemBertForTokenClassification,
              """zero-shot""": RemBertForSequenceClassification,
          }
          if is_torch_available()
          else {}
      )
      def 						UpperCAmelCase_    (      self :Optional[int]   )  ->      Any:
       UpperCAmelCase__							=    RemBertModelTester(self   )
       UpperCAmelCase__							=    ConfigTester(self			,   config_class=lowerCamelCase			,   hidden_size=37   )
      def 						UpperCAmelCase_    (      self :Union[str, Any]   )  ->      Union[str, Any]:
       self.config_tester.run_common_tests()
      def 						UpperCAmelCase_    (      self :Dict   )  ->      Optional[int]:
       UpperCAmelCase__							=    self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_model(*lowerCamelCase   )
      def 						UpperCAmelCase_    (      self :Optional[Any]   )  ->      str:
       UpperCAmelCase__							=    self.model_tester.prepare_config_and_inputs()
       for type in ["absolute", "relative_key", "relative_key_query"]:
        UpperCAmelCase__							=    type
        self.model_tester.create_and_check_model(*lowerCamelCase   )
      def 						UpperCAmelCase_    (      self :Optional[int]   )  ->      List[str]:
       UpperCAmelCase__							=    self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_for_masked_lm(*lowerCamelCase   )
      def 						UpperCAmelCase_    (      self :Optional[int]   )  ->      List[str]:
       UpperCAmelCase__							=    self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_for_multiple_choice(*lowerCamelCase   )
      def 						UpperCAmelCase_    (      self :Tuple   )  ->      Union[str, Any]:
       UpperCAmelCase__							=    self.model_tester.prepare_config_and_inputs_for_decoder()
       self.model_tester.create_and_check_decoder_model_past_large_inputs(*lowerCamelCase   )
      def 						UpperCAmelCase_    (      self :List[Any]   )  ->      Union[str, Any]:
       UpperCAmelCase__							=    self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_for_question_answering(*lowerCamelCase   )
      def 						UpperCAmelCase_    (      self :List[str]   )  ->      List[str]:
       UpperCAmelCase__							=    self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_for_sequence_classification(*lowerCamelCase   )
      def 						UpperCAmelCase_    (      self :int   )  ->      str:
       UpperCAmelCase__							=    self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_for_token_classification(*lowerCamelCase   )
      def 						UpperCAmelCase_    (      self :Dict   )  ->      Dict:
       UpperCAmelCase__							=    self.model_tester.prepare_config_and_inputs_for_decoder()
       self.model_tester.create_and_check_model_as_decoder(*lowerCamelCase   )
      def 						UpperCAmelCase_    (      self :int   )  ->      int:
       # This regression test was failing with PyTorch < 1.3
       (
           (
           UpperCAmelCase__
       )			,			(
           UpperCAmelCase__
       )			,			(
           UpperCAmelCase__
       )			,			(
           UpperCAmelCase__
       )			,			(
           UpperCAmelCase__
       )			,			(
           UpperCAmelCase__
       )			,			(
           UpperCAmelCase__
       )			,			(
           UpperCAmelCase__
       )			,			(
           UpperCAmelCase__
       )			,			
       )							=    self.model_tester.prepare_config_and_inputs_for_decoder()

       UpperCAmelCase__							=    None

       self.model_tester.create_and_check_model_as_decoder(
           lowerCamelCase			,   lowerCamelCase			,   lowerCamelCase			,   lowerCamelCase			,   lowerCamelCase			,   lowerCamelCase			,   lowerCamelCase			,   lowerCamelCase			,   lowerCamelCase			,   )







      @slow
      def 						UpperCAmelCase_    (      self :int   )  ->      Tuple:
       for model_name in REMBERT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
        UpperCAmelCase__							=    RemBertModel.from_pretrained(lowerCamelCase   )
        self.assertIsNotNone(lowerCamelCase   )



@require_torch
class       _UpperCamelCase   (					unittest.TestCase		):
      @slow
      def 						UpperCAmelCase_    (      self :str   )  ->      Optional[Any]:
       # Test exact values at the last hidden layer
       UpperCAmelCase__							=    RemBertModel.from_pretrained("google/rembert"   )
       UpperCAmelCase__							=    torch.tensor([[312, 5_6498, 313, 2125, 313]]   )
       UpperCAmelCase__							=    torch.tensor([[0, 0, 0, 1, 1]]   )
       with torch.no_grad():
        UpperCAmelCase__							=    model(lowerCamelCase			,   token_type_ids=lowerCamelCase			,   output_hidden_states=lowerCamelCase   )

       UpperCAmelCase__							=    1152

       UpperCAmelCase__							=    torch.Size((1, 5, hidden_size)   )
       self.assertEqual(output["last_hidden_state"].shape			,   lowerCamelCase   )

       UpperCAmelCase__							=    torch.tensor(
           [
               [
                   [0.07_54, -0.20_22, 0.19_04],
                   [-0.33_54, -0.36_92, -0.47_91],
                   [-0.23_14, -0.67_29, -0.07_49],
                   [-0.03_96, -0.31_05, -0.42_34],
                   [-0.15_71, -0.05_25, 0.53_53],
               ]
           ]   )

       # Running on the original tf implementation gives slightly different results here.
       # Not clear why this variations is present
       # TODO: Find reason for discrepancy
       # expected_original_implementation = [[
       #     [0.07630594074726105, -0.20146065950393677, 0.19107051193714142],
       #     [-0.3405614495277405, -0.36971670389175415, -0.4808273911476135],
       #     [-0.22587086260318756, -0.6656315922737122, -0.07844287157058716],
       #     [-0.04145475849509239, -0.3077218234539032, -0.42316967248916626],
       #     [-0.15887849032878876, -0.054529931396245956, 0.5356100797653198]
       # ]]

       self.assertTrue(torch.allclose(output["last_hidden_state"][:, :, :3]			,   lowerCamelCase			,   atol=1e-4   )   )











import subprocess
from typing import Union

import numpy as np
import requests

from ..utils import add_end_docstrings, is_torch_available, is_torchaudio_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_torch_available():
  from ..models.auto.modeling_auto import MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING

_lowerCAmelCase    : List[str]		= logging.get_logger(__name__)



def     lowerCAmelCase		(		_lowerCAmelCase				: bytes    ,				_lowerCAmelCase				: int						):




 """simple docstring"""







 UpperCAmelCase__							=    F'''{sampling_rate}'''
 UpperCAmelCase__							=    "1"
 UpperCAmelCase__							=    "f32le"
 UpperCAmelCase__							=    [
     "ffmpeg",
     "-i",
     "pipe:0",
     "-ac",
     ac,
     "-ar",
     ar,
     "-f",
     format_for_conversion,
     "-hide_banner",
     "-loglevel",
     "quiet",
     "pipe:1",
 ]

 try:
  UpperCAmelCase__							=    subprocess.Popen(_lowerCAmelCase    ,				stdin=subprocess.PIPE    ,				stdout=subprocess.PIPE						)
 except FileNotFoundError:
  raise ValueError("ffmpeg was not found but is required to load audio files from filename"						)
 UpperCAmelCase__							=    ffmpeg_process.communicate(_lowerCAmelCase						)
 UpperCAmelCase__							=    output_stream[0]

 UpperCAmelCase__							=    np.frombuffer(_lowerCAmelCase    ,				np.floataa						)
 if audio.shape[0] == 0:
  raise ValueError("Malformed soundfile"						)
 return audio


@add_end_docstrings(lowerCAmelCase		)
class       _UpperCamelCase   (					lowerCAmelCase		):
      def __init__(      self :Any			,   *lowerCamelCase :Any			,   **lowerCamelCase :Optional[Any]   )  ->      Optional[int]:
       # Default, might be overriden by the model.config.
       UpperCAmelCase__							=    5
       super().__init__(*lowerCamelCase			,   **lowerCamelCase   )

       if self.framework != "pt":
        raise ValueError(f'''The {self.__class__} is only available in PyTorch.'''   )

       self.check_model_type(lowerCamelCase   )
      def __call__(      self :Optional[Any]			,   lowerCamelCase :Union[np.ndarray, bytes, str]			,   **lowerCamelCase :List[Any]			,   )  ->      Tuple:
       return super().__call__(lowerCamelCase			,   **lowerCamelCase   )
      def 						UpperCAmelCase_    (      self :str			,   lowerCamelCase :Dict=None			,   **lowerCamelCase :str   )  ->      int:
       # No parameters on this pipeline right now
       UpperCAmelCase__							=    {}
       if top_k is not None:
        if top_k > self.model.config.num_labels:
         UpperCAmelCase__							=    self.model.config.num_labels
        UpperCAmelCase__							=    top_k
       return {}, {}, postprocess_params
      def 						UpperCAmelCase_    (      self :Optional[Any]			,   lowerCamelCase :Dict   )  ->      Optional[int]:
       if isinstance(lowerCamelCase			,   lowerCamelCase   ):
        if inputs.startswith("http://"   ) or inputs.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
         UpperCAmelCase__							=    requests.get(lowerCamelCase   ).content
        else:
         with open(lowerCamelCase			,   "rb"   ) as f:
          UpperCAmelCase__							=    f.read()

       if isinstance(lowerCamelCase			,   lowerCamelCase   ):
        UpperCAmelCase__							=    ffmpeg_read(lowerCamelCase			,   self.feature_extractor.sampling_rate   )

       if isinstance(lowerCamelCase			,   lowerCamelCase   ):
        # Accepting `"array"` which is the key defined in `datasets` for
        # better integration
        if not ("sampling_rate" in inputs and ("raw" in inputs or "array" in inputs)):
         raise ValueError(
             "When passing a dictionary to AudioClassificationPipeline, the dict needs to contain a "
             "\"raw\" key containing the numpy array representing the audio and a \"sampling_rate\" key, "
             "containing the sampling_rate associated with that array"   )

        UpperCAmelCase__							=    inputs.pop("raw"			,   lowerCamelCase   )
        if _inputs is None:
         # Remove path which will not be used from `datasets`.
         inputs.pop("path"			,   lowerCamelCase   )
         UpperCAmelCase__							=    inputs.pop("array"			,   lowerCamelCase   )
        UpperCAmelCase__							=    inputs.pop("sampling_rate"   )
        UpperCAmelCase__							=    _inputs
        if in_sampling_rate != self.feature_extractor.sampling_rate:
         import torch

         if is_torchaudio_available():
          from torchaudio import functional as F
         else:
          raise ImportError(
              "torchaudio is required to resample audio samples in AudioClassificationPipeline. "
              "The torchaudio package can be installed through: `pip install torchaudio`."   )

         UpperCAmelCase__							=    F.resample(
             torch.from_numpy(lowerCamelCase   )			,   lowerCamelCase			,   self.feature_extractor.sampling_rate   ).numpy()

       if not isinstance(lowerCamelCase			,   np.ndarray   ):
        raise ValueError("We expect a numpy ndarray as input"   )
       if len(inputs.shape   ) != 1:
        raise ValueError("We expect a single channel audio input for AudioClassificationPipeline"   )

       UpperCAmelCase__							=    self.feature_extractor(
           lowerCamelCase			,   sampling_rate=self.feature_extractor.sampling_rate			,   return_tensors="pt"   )
       return processed
      def 						UpperCAmelCase_    (      self :Tuple			,   lowerCamelCase :Union[str, Any]   )  ->      List[Any]:
       UpperCAmelCase__							=    self.model(**lowerCamelCase   )
       return model_outputs







      def 						UpperCAmelCase_    (      self :List[str]			,   lowerCamelCase :Optional[Any]			,   lowerCamelCase :List[Any]=5   )  ->      Tuple:
       UpperCAmelCase__							=    model_outputs.logits[0].softmax(-1   )
       UpperCAmelCase__			,			UpperCAmelCase__							=    probs.topk(lowerCamelCase   )

       UpperCAmelCase__							=    scores.tolist()
       UpperCAmelCase__							=    ids.tolist()

       UpperCAmelCase__							=    [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(lowerCamelCase			,   lowerCamelCase   )]

       return labels





