






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


a_				=     logging.get_logger(__name__)


@add_end_docstrings(A__						)
class   lowercase__  (      A__						):

	def __init__(				self		,     **__UpperCAmelCase							)->       Dict:







							'''simple docstring'''





							super().__init__(**_lowerCamelCase							)

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

	def __call__(				self		,     __UpperCAmelCase		,     **__UpperCAmelCase							)->       Tuple:







							'''simple docstring'''





							return super().__call__(_lowerCamelCase		,     **_lowerCamelCase							)

	def 		UpperCAmelCase	(				self		,     **__UpperCAmelCase							)->       Optional[Any]:







							'''simple docstring'''





							lowerCAmelCase__			   =    {}
							if "candidate_labels" in kwargs:
													lowerCAmelCase__			   =    kwargs['candidate_labels']
							if "hypothesis_template" in kwargs:
													lowerCAmelCase__			   =    kwargs['hypothesis_template']

							return preprocess_params, {}, {}

	def 		UpperCAmelCase	(				self		,     __UpperCAmelCase		,     __UpperCAmelCase=None		,     __UpperCAmelCase="This is a sound of {}."							)->       Optional[int]:







							'''simple docstring'''





							if isinstance(_lowerCamelCase		,     _lowerCamelCase							):
													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
																			lowerCAmelCase__			   =    requests.get(_lowerCamelCase							).content
													else:
																			with open(_lowerCamelCase		,     "rb"							) as f:
																									lowerCAmelCase__			   =    f.read()

							if isinstance(_lowerCamelCase		,     _lowerCamelCase							):
													lowerCAmelCase__			   =    ffmpeg_read(_lowerCamelCase		,     self.feature_extractor.sampling_rate							)

							if not isinstance(_lowerCamelCase		,     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"							)

							lowerCAmelCase__			   =    self.feature_extractor(
							    [audio]		,     sampling_rate=self.feature_extractor.sampling_rate		,     return_tensors="pt"							)
							lowerCAmelCase__			   =    candidate_labels
							lowerCAmelCase__			   =    [hypothesis_template.format(_lowerCamelCase							) for x in candidate_labels]
							lowerCAmelCase__			   =    self.tokenizer(_lowerCamelCase		,     return_tensors=self.framework		,     padding=_lowerCamelCase							)
							lowerCAmelCase__			   =    [text_inputs]
							return inputs

	def 		UpperCAmelCase	(				self		,     __UpperCAmelCase							)->       str:







							'''simple docstring'''





							lowerCAmelCase__			   =    model_inputs.pop("candidate_labels"							)
							lowerCAmelCase__			   =    model_inputs.pop("text_inputs"							)
							if isinstance(text_inputs[0]		,     _lowerCamelCase							):
													lowerCAmelCase__			   =    text_inputs[0]
							else:
													# Batching case.
													lowerCAmelCase__			   =    text_inputs[0][0]

							lowerCAmelCase__			   =    self.model(**_lowerCamelCase		,     **_lowerCamelCase							)

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


	def 		UpperCAmelCase	(				self		,     __UpperCAmelCase							)->       Dict:







							'''simple docstring'''





							lowerCAmelCase__			   =    model_outputs.pop("candidate_labels"							)
							lowerCAmelCase__			   =    model_outputs['logits'][0]

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

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








"""simple docstring"""






import os
import shutil
import tempfile
from unittest import TestCase
from unittest.mock import patch

import numpy as np
from datasets import Dataset

from transformers.models.realm.configuration_realm import RealmConfig
from transformers.models.realm.retrieval_realm import _REALM_BLOCK_RECORDS_FILENAME, RealmRetriever
from transformers.models.realm.tokenization_realm import VOCAB_FILES_NAMES, RealmTokenizer





class 				a__					(						A__       ):



     def  lowerCamelCase_						(							self :List[str]   ):







      '''simple docstring'''



      UpperCamelCase_     :					Optional[Any]   	=tempfile.mkdtemp()
      UpperCamelCase_     :					str   	=5

      # Realm tok
      UpperCamelCase_     :					List[Any]   	=[
          '[UNK]',
          '[CLS]',
          '[SEP]',
          '[PAD]',
          '[MASK]',
          'test',
          'question',
          'this',
          'is',
          'the',
          'first',
          'second',
          'third',
          'fourth',
          'fifth',
          'record',
          'want',
          '##want',
          '##ed',
          'wa',
          'un',
          'runn',
          '##ing',
          ',',
          'low',
          'lowest',
      ]
      UpperCamelCase_     :					List[str]   	=os.path.join(self.tmpdirname      ,  'realm_tokenizer'   )
      os.makedirs(_lowerCamelCase      ,  exist_ok=_lowerCamelCase   )
      UpperCamelCase_     :					int   	=os.path.join(_lowerCamelCase      ,  VOCAB_FILES_NAMES['vocab_file']   )
      with open(self.vocab_file      ,  'w'      ,  encoding='utf-8'   ) as vocab_writer:
       vocab_writer.write(''.join([x + '\n' for x in vocab_tokens]   )   )

      UpperCamelCase_     :					str   	=os.path.join(self.tmpdirname      ,  'realm_block_records'   )
      os.makedirs(_lowerCamelCase      ,  exist_ok=_lowerCamelCase   )



     def  lowerCamelCase_						(							self :Union[str, Any]   ):







      '''simple docstring'''



      return RealmTokenizer.from_pretrained(os.path.join(self.tmpdirname      ,  'realm_tokenizer'   )   )



     def  lowerCamelCase_						(							self :Dict   ):







      '''simple docstring'''



      shutil.rmtree(self.tmpdirname   )



     def  lowerCamelCase_						(							self :Dict   ):







      '''simple docstring'''



      UpperCamelCase_     :					List[str]   	=RealmConfig(num_block_records=self.num_block_records   )
      return config



     def  lowerCamelCase_						(							self :Optional[int]   ):







      '''simple docstring'''



      UpperCamelCase_     :					str   	=Dataset.from_dict(
          {
              'id': ['0', '1'],
              'question': ['foo', 'bar'],
              'answers': [['Foo', 'Bar'], ['Bar']],
          }   )
      return dataset



     def  lowerCamelCase_						(							self :Optional[Any]   ):







      '''simple docstring'''



      UpperCamelCase_     :					Union[str, Any]   	=np.array(
          [
              b'This is the first record',
              b'This is the second record',
              b'This is the third record',
              b'This is the fourth record',
              b'This is the fifth record',
              b'This is a longer longer longer record',
          ]      ,  dtype=_lowerCamelCase      ,  )
      return block_records



     def  lowerCamelCase_						(							self :Any   ):







      '''simple docstring'''



      UpperCamelCase_     :					List[Any]   	=RealmRetriever(
          block_records=self.get_dummy_block_records()      ,  tokenizer=self.get_tokenizer()      ,  )
      return retriever



     def  lowerCamelCase_						(							self :int   ):







      '''simple docstring'''



      UpperCamelCase_     :					Any   	=self.get_config()
      UpperCamelCase_     :					Tuple   	=self.get_dummy_retriever()
      UpperCamelCase_     :					Tuple   	=retriever.tokenizer

      UpperCamelCase_     :					List[str]   	=np.array([0, 3]      ,  dtype='long'   )
      UpperCamelCase_     :					Optional[Any]   	=tokenizer(['Test question']   ).input_ids
      UpperCamelCase_     :					Union[str, Any]   	=tokenizer(
          ['the fourth']      ,  add_special_tokens=_lowerCamelCase      ,  return_token_type_ids=_lowerCamelCase      ,  return_attention_mask=_lowerCamelCase      ,  ).input_ids
      UpperCamelCase_     :					Union[str, Any]   	=config.reader_seq_len

      UpperCamelCase_	,     UpperCamelCase_	,     UpperCamelCase_	,     UpperCamelCase_     :					Any   	=retriever(
          _lowerCamelCase      ,  _lowerCamelCase      ,  answer_ids=_lowerCamelCase      ,  max_length=_lowerCamelCase      ,  return_tensors='np'   )

      self.assertEqual(len(_lowerCamelCase   )      ,  2   )
      self.assertEqual(len(_lowerCamelCase   )      ,  2   )
      self.assertEqual(len(_lowerCamelCase   )      ,  2   )
      self.assertEqual(concat_inputs.input_ids.shape      ,  (2, 10)   )
      self.assertEqual(concat_inputs.attention_mask.shape      ,  (2, 10)   )
      self.assertEqual(concat_inputs.token_type_ids.shape      ,  (2, 10)   )
      self.assertEqual(concat_inputs.special_tokens_mask.shape      ,  (2, 10)   )
      self.assertEqual(
          tokenizer.convert_ids_to_tokens(concat_inputs.input_ids[0]   )      ,  ['[CLS]', 'test', 'question', '[SEP]', 'this', 'is', 'the', 'first', 'record', '[SEP]']      ,  )
      self.assertEqual(
          tokenizer.convert_ids_to_tokens(concat_inputs.input_ids[1]   )      ,  ['[CLS]', 'test', 'question', '[SEP]', 'this', 'is', 'the', 'fourth', 'record', '[SEP]']      ,  )



     def  lowerCamelCase_						(							self :Optional[int]   ):







      '''simple docstring'''



      UpperCamelCase_     :					Optional[int]   	=self.get_config()
      UpperCamelCase_     :					int   	=self.get_dummy_retriever()
      UpperCamelCase_     :					Any   	=retriever.tokenizer

      UpperCamelCase_     :					Optional[int]   	=np.array([0, 3, 5]      ,  dtype='long'   )
      UpperCamelCase_     :					Optional[int]   	=tokenizer(['Test question']   ).input_ids
      UpperCamelCase_     :					int   	=tokenizer(
          ['the fourth', 'longer longer']      ,  add_special_tokens=_lowerCamelCase      ,  return_token_type_ids=_lowerCamelCase      ,  return_attention_mask=_lowerCamelCase      ,  ).input_ids
      UpperCamelCase_     :					Union[str, Any]   	=config.reader_seq_len

      UpperCamelCase_	,     UpperCamelCase_	,     UpperCamelCase_	,     UpperCamelCase_     :					Union[str, Any]   	=retriever(
          _lowerCamelCase      ,  _lowerCamelCase      ,  answer_ids=_lowerCamelCase      ,  max_length=_lowerCamelCase      ,  return_tensors='np'   )

      self.assertEqual([False, True, True]      ,  _lowerCamelCase   )
      self.assertEqual([[-1, -1, -1], [6, -1, -1], [6, 7, 8]]      ,  _lowerCamelCase   )
      self.assertEqual([[-1, -1, -1], [7, -1, -1], [7, 8, 9]]      ,  _lowerCamelCase   )




     def  lowerCamelCase_						(							self :List[Any]   ):







      '''simple docstring'''



      UpperCamelCase_     :					Tuple   	=self.get_dummy_retriever()
      retriever.save_pretrained(os.path.join(self.tmpdirname      ,  'realm_block_records'   )   )

      # Test local path
      UpperCamelCase_     :					int   	=retriever.from_pretrained(os.path.join(self.tmpdirname      ,  'realm_block_records'   )   )
      self.assertEqual(retriever.block_records[0]      ,  b'This is the first record'   )

      # Test mocked remote path
      with patch('transformers.models.realm.retrieval_realm.hf_hub_download'   ) as mock_hf_hub_download:
       UpperCamelCase_     :					Dict   	=os.path.join(
           os.path.join(self.tmpdirname      ,  'realm_block_records'   )      ,  _REALM_BLOCK_RECORDS_FILENAME   )
       UpperCamelCase_     :					int   	=RealmRetriever.from_pretrained('google/realm-cc-news-pretrained-openqa'   )

      self.assertEqual(retriever.block_records[0]      ,  b'This is the first record'   )
