
import os
import sys
import unittest


lowercase  :					Optional[Any]						    =							os.path.abspath(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
sys.path.append(os.path.join(git_repo_path, """utils"""))

import get_test_info  # noqa: E402
from get_test_info import (  # noqa: E402
    get_model_to_test_mapping,
    get_model_to_tester_mapping,
    get_test_to_tester_mapping,
)


lowercase  :					Optional[int]						    =							os.path.join("""tests""", """models""", """bert""", """test_modeling_bert.py""")
lowercase  :					List[str]						    =							os.path.join("""tests""", """models""", """blip""", """test_modeling_blip.py""")

class __snake_case		(      unittest.TestCase     ):
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           lowercase							:     List[str]       	=      get_test_to_tester_mapping(snake_case      )
           lowercase							:     Dict       	=      get_test_to_tester_mapping(snake_case      )

           lowercase							:     str       	=      {"""BertModelTest""": """BertModelTester"""}

           lowercase							:     Dict       	=      {
               """BlipModelTest""": """BlipModelTester""",
               """BlipTextImageModelTest""": """BlipTextImageModelsModelTester""",
               """BlipTextModelTest""": """BlipTextModelTester""",
               """BlipTextRetrievalModelTest""": """BlipTextRetrievalModelTester""",
               """BlipVQAModelTest""": """BlipVQAModelTester""",
               """BlipVisionModelTest""": """BlipVisionModelTester""",
           }

           self.assertEqual(get_test_info.to_json(snake_case      )		,snake_case      )
           self.assertEqual(get_test_info.to_json(snake_case      )		,snake_case      )
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           lowercase							:     Tuple       	=      get_model_to_test_mapping(snake_case      )
           lowercase							:     str       	=      get_model_to_test_mapping(snake_case      )

           lowercase							:     List[Any]       	=      {
               """BertForMaskedLM""": ["""BertModelTest"""],
               """BertForMultipleChoice""": ["""BertModelTest"""],
               """BertForNextSentencePrediction""": ["""BertModelTest"""],
               """BertForPreTraining""": ["""BertModelTest"""],
               """BertForQuestionAnswering""": ["""BertModelTest"""],
               """BertForSequenceClassification""": ["""BertModelTest"""],
               """BertForTokenClassification""": ["""BertModelTest"""],
               """BertLMHeadModel""": ["""BertModelTest"""],
               """BertModel""": ["""BertModelTest"""],
           }

           lowercase							:     List[str]       	=      {
               """BlipForConditionalGeneration""": ["""BlipTextImageModelTest"""],
               """BlipForImageTextRetrieval""": ["""BlipTextRetrievalModelTest"""],
               """BlipForQuestionAnswering""": ["""BlipVQAModelTest"""],
               """BlipModel""": ["""BlipModelTest"""],
               """BlipTextModel""": ["""BlipTextModelTest"""],
               """BlipVisionModel""": ["""BlipVisionModelTest"""],
           }

           self.assertEqual(get_test_info.to_json(snake_case      )		,snake_case      )
           self.assertEqual(get_test_info.to_json(snake_case      )		,snake_case      )
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           lowercase							:     List[Any]       	=      get_model_to_tester_mapping(snake_case      )
           lowercase							:     Any       	=      get_model_to_tester_mapping(snake_case      )

           lowercase							:     List[str]       	=      {
               """BertForMaskedLM""": ["""BertModelTester"""],
               """BertForMultipleChoice""": ["""BertModelTester"""],
               """BertForNextSentencePrediction""": ["""BertModelTester"""],
               """BertForPreTraining""": ["""BertModelTester"""],
               """BertForQuestionAnswering""": ["""BertModelTester"""],
               """BertForSequenceClassification""": ["""BertModelTester"""],
               """BertForTokenClassification""": ["""BertModelTester"""],
               """BertLMHeadModel""": ["""BertModelTester"""],
               """BertModel""": ["""BertModelTester"""],
           }

           lowercase							:     List[Any]       	=      {
               """BlipForConditionalGeneration""": ["""BlipTextImageModelsModelTester"""],
               """BlipForImageTextRetrieval""": ["""BlipTextRetrievalModelTester"""],
               """BlipForQuestionAnswering""": ["""BlipVQAModelTester"""],
               """BlipModel""": ["""BlipModelTester"""],
               """BlipTextModel""": ["""BlipTextModelTester"""],
               """BlipVisionModel""": ["""BlipVisionModelTester"""],
           }

           self.assertEqual(get_test_info.to_json(snake_case      )		,snake_case      )
           self.assertEqual(get_test_info.to_json(snake_case      )		,snake_case      )


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


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

@add_end_docstrings(lowerCAmelCase     )
class __snake_case		(      lowerCAmelCase     ):
    def __init__(  self		,**snake_case      ):



           '''simple docstring'''







           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		,snake_case		,**snake_case      ):



           '''simple docstring'''







           return super().__call__(snake_case		,**snake_case      )
    def       _SCREAMING_SNAKE_CASE       (  self		,**snake_case      ):



           '''simple docstring'''







           lowercase							:     Union[str, Any]       	=      {}
           if "candidate_labels" in kwargs:
                  lowercase							:     List[str]       	=      kwargs["""candidate_labels"""]
           if "hypothesis_template" in kwargs:
                  lowercase							:     Dict       	=      kwargs["""hypothesis_template"""]

           return preprocess_params, {}, {}
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case=None		,snake_case="This is a sound of {}."      ):



           '''simple docstring'''







           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
                         lowercase							:     Optional[Any]       	=      requests.get(snake_case      ).content
                  else:
                         with open(snake_case		,"""rb"""      ) as f:
                                lowercase							:     Union[str, Any]       	=      f.read()

           if isinstance(snake_case		,snake_case      ):
                  lowercase							:     int       	=      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"""      )

           lowercase							:     Dict       	=      self.feature_extractor(
               [audio]		,sampling_rate=self.feature_extractor.sampling_rate		,return_tensors="""pt"""      )
           lowercase							:     Tuple       	=      candidate_labels
           lowercase							:     Tuple       	=      [hypothesis_template.format(snake_case      ) for x in candidate_labels]
           lowercase							:     Optional[Any]       	=      self.tokenizer(snake_case		,return_tensors=self.framework		,padding=snake_case      )
           lowercase							:     Optional[Any]       	=      [text_inputs]
           return inputs
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case      ):



           '''simple docstring'''







           lowercase							:     List[str]       	=      model_inputs.pop("""candidate_labels"""      )
           lowercase							:     Dict       	=      model_inputs.pop("""text_inputs"""      )
           if isinstance(text_inputs[0]		,snake_case      ):
                  lowercase							:     List[Any]       	=      text_inputs[0]
           else:
                  # Batching case.
                  lowercase							:     Dict       	=      text_inputs[0][0]

           lowercase							:     Optional[Any]       	=      self.model(**snake_case		,**snake_case      )

           lowercase							:     Any       	=      {
               """candidate_labels""": candidate_labels,
               """logits""": outputs.logits_per_audio,
           }
           return model_outputs
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case      ):



           '''simple docstring'''







           lowercase							:     List[Any]       	=      model_outputs.pop("""candidate_labels"""      )
           lowercase							:     Any       	=      model_outputs["""logits"""][0]

           if self.framework == "pt":
                  lowercase							:     Any       	=      logits.softmax(dim=0      )
                  lowercase							:     Tuple       	=      probs.tolist()
           else:
                  raise ValueError("""`tf` framework not supported."""      )

           lowercase							:     Tuple       	=      [
               {"""score""": score, """label""": candidate_label}
               for score, candidate_label in sorted(zip(snake_case		,snake_case      )		,key=lambda snake_case      : -x[0]      )
           ]
           return result
