



'''simple docstring'''
import json
import os
import shutil
import tempfile
import unittest

import numpy as np

from transformers import BertTokenizerFast
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES, BertTokenizer
from transformers.testing_utils import require_tokenizers, require_vision
from transformers.utils import IMAGE_PROCESSOR_NAME, is_vision_available


if is_vision_available():
					from PIL import Image

					from transformers import VisionTextDualEncoderProcessor, ViTImageProcessor






@require_tokenizers
@require_vision
class 				lowercase__      (				unittest.TestCase  ):





	def       UpperCamelCase_    (					self   :							int					):







							'''simple docstring'''
							_UpperCamelCase   :						Optional[Any]												=  tempfile.mkdtemp()

							# fmt: off
							_UpperCamelCase   :						Optional[Any]												=  ['[UNK]', '[CLS]', '[SEP]', '[PAD]', '[MASK]', 'want', '##want', '##ed', 'wa', 'un', 'runn', '##ing', ',', 'low', 'lowest']
							# fmt: on
							_UpperCamelCase   :						List[str]												=  os.path.join(self.tmpdirname					,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   :						Any												=  {
							    'do_resize': True,
							    'size': {'height': 18, 'width': 18},
							    'do_normalize': True,
							    'image_mean': [0.5, 0.5, 0.5],
							    'image_std': [0.5, 0.5, 0.5],
							}
							_UpperCamelCase   :						str												=  os.path.join(self.tmpdirname					,lowerCamelCase__					)
							with open(self.image_processor_file					,'w'					,encoding='utf-8'					) as fp:
													json.dump(lowerCamelCase__					,lowerCamelCase__					)





	def       UpperCamelCase_    (					self   :							Optional[Any]					,**lowerCamelCase__   :							List[Any]					):







							'''simple docstring'''
							return BertTokenizer.from_pretrained(self.tmpdirname					,**lowerCamelCase__					)





	def       UpperCamelCase_    (					self   :							Any					,**lowerCamelCase__   :							str					):







							'''simple docstring'''
							return ViTImageProcessor.from_pretrained(self.tmpdirname					,**lowerCamelCase__					)





	def       UpperCamelCase_    (					self   :							Optional[Any]					):







							'''simple docstring'''
							shutil.rmtree(self.tmpdirname					)





	def       UpperCamelCase_    (					self   :							Dict					):







							'''simple docstring'''
							_UpperCamelCase   :						Optional[Any]												=  [np.random.randint(255					,size=(3, 30, 400)					,dtype=np.uinta					)]

							_UpperCamelCase   :						List[str]												=  [Image.fromarray(np.moveaxis(lowerCamelCase__					,0					,-1					)					) for x in image_inputs]

							return image_inputs





	def       UpperCamelCase_    (					self   :							Any					):







							'''simple docstring'''
							_UpperCamelCase   :						List[Any]												=  self.get_tokenizer()
							_UpperCamelCase   :						List[str]												=  self.get_image_processor()

							_UpperCamelCase   :						Dict												=  VisionTextDualEncoderProcessor(tokenizer=lowerCamelCase__					,image_processor=lowerCamelCase__					)

							processor.save_pretrained(self.tmpdirname					)
							_UpperCamelCase   :						Any												=  VisionTextDualEncoderProcessor.from_pretrained(self.tmpdirname					)

							self.assertEqual(processor.tokenizer.get_vocab()					,tokenizer.get_vocab()					)
							self.assertIsInstance(processor.tokenizer					,(BertTokenizer, BertTokenizerFast)					)

							self.assertEqual(processor.image_processor.to_json_string()					,image_processor.to_json_string()					)
							self.assertIsInstance(processor.image_processor					,lowerCamelCase__					)





	def       UpperCamelCase_    (					self   :							List[str]					):







							'''simple docstring'''
							_UpperCamelCase   :						int												=  VisionTextDualEncoderProcessor(
							    tokenizer=self.get_tokenizer()					,image_processor=self.get_image_processor()					)
							processor.save_pretrained(self.tmpdirname					)

							_UpperCamelCase   :						int												=  self.get_tokenizer(bos_token='(BOS)'					,eos_token='(EOS)'					)
							_UpperCamelCase   :						str												=  self.get_image_processor(do_normalize=lowerCamelCase__					,padding_value=1.0					)

							_UpperCamelCase   :						str												=  VisionTextDualEncoderProcessor.from_pretrained(
							    self.tmpdirname					,bos_token='(BOS)'					,eos_token='(EOS)'					,do_normalize=lowerCamelCase__					,padding_value=1.0					)

							self.assertEqual(processor.tokenizer.get_vocab()					,tokenizer_add_kwargs.get_vocab()					)
							self.assertIsInstance(processor.tokenizer					,(BertTokenizer, BertTokenizerFast)					)

							self.assertEqual(processor.image_processor.to_json_string()					,image_processor_add_kwargs.to_json_string()					)
							self.assertIsInstance(processor.image_processor					,lowerCamelCase__					)





	def       UpperCamelCase_    (					self   :							Optional[int]					):







							'''simple docstring'''
							_UpperCamelCase   :						str												=  self.get_image_processor()
							_UpperCamelCase   :						int												=  self.get_tokenizer()

							_UpperCamelCase   :						Tuple												=  VisionTextDualEncoderProcessor(tokenizer=lowerCamelCase__					,image_processor=lowerCamelCase__					)

							_UpperCamelCase   :						int												=  self.prepare_image_inputs()

							_UpperCamelCase   :						List[Any]												=  image_processor(lowerCamelCase__					,return_tensors='np'					)
							_UpperCamelCase   :						Union[str, Any]												=  processor(images=lowerCamelCase__					,return_tensors='np'					)

							for key in input_feat_extract.keys():
													self.assertAlmostEqual(input_feat_extract[key].sum()					,input_processor[key].sum()					,delta=1E-2					)





	def       UpperCamelCase_    (					self   :							Tuple					):







							'''simple docstring'''
							_UpperCamelCase   :						Optional[int]												=  self.get_image_processor()
							_UpperCamelCase   :						Tuple												=  self.get_tokenizer()

							_UpperCamelCase   :						int												=  VisionTextDualEncoderProcessor(tokenizer=lowerCamelCase__					,image_processor=lowerCamelCase__					)

							_UpperCamelCase   :						Any												=  'lower newer'

							_UpperCamelCase   :						Union[str, Any]												=  processor(text=lowerCamelCase__					)

							_UpperCamelCase   :						Dict												=  tokenizer(lowerCamelCase__					)

							for key in encoded_tok.keys():
													self.assertListEqual(encoded_tok[key]					,encoded_processor[key]					)





	def       UpperCamelCase_    (					self   :							List[str]					):







							'''simple docstring'''
							_UpperCamelCase   :						int												=  self.get_image_processor()
							_UpperCamelCase   :						int												=  self.get_tokenizer()

							_UpperCamelCase   :						Dict												=  VisionTextDualEncoderProcessor(tokenizer=lowerCamelCase__					,image_processor=lowerCamelCase__					)

							_UpperCamelCase   :						Optional[int]												=  'lower newer'
							_UpperCamelCase   :						List[str]												=  self.prepare_image_inputs()

							_UpperCamelCase   :						Any												=  processor(text=lowerCamelCase__					,images=lowerCamelCase__					)

							self.assertListEqual(list(inputs.keys()					)					,['input_ids', 'token_type_ids', 'attention_mask', 'pixel_values']					)

							# test if it raises when no input is passed
							with self.assertRaises(lowerCamelCase__					):
													processor()





	def       UpperCamelCase_    (					self   :							int					):







							'''simple docstring'''
							_UpperCamelCase   :						Any												=  self.get_image_processor()
							_UpperCamelCase   :						Optional[int]												=  self.get_tokenizer()

							_UpperCamelCase   :						Optional[Any]												=  VisionTextDualEncoderProcessor(tokenizer=lowerCamelCase__					,image_processor=lowerCamelCase__					)

							_UpperCamelCase   :						Union[str, Any]												=  [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

							_UpperCamelCase   :						str												=  processor.batch_decode(lowerCamelCase__					)
							_UpperCamelCase   :						Any												=  tokenizer.batch_decode(lowerCamelCase__					)

							self.assertListEqual(lowerCamelCase__					,lowerCamelCase__					)





	def       UpperCamelCase_    (					self   :							Optional[Any]					):







							'''simple docstring'''
							_UpperCamelCase   :						Union[str, Any]												=  self.get_image_processor()
							_UpperCamelCase   :						Any												=  self.get_tokenizer()

							_UpperCamelCase   :						str												=  VisionTextDualEncoderProcessor(tokenizer=lowerCamelCase__					,image_processor=lowerCamelCase__					)

							_UpperCamelCase   :						Optional[Any]												=  'lower newer'
							_UpperCamelCase   :						Tuple												=  self.prepare_image_inputs()

							_UpperCamelCase   :						Any												=  processor(text=lowerCamelCase__					,images=lowerCamelCase__					)

							self.assertListEqual(list(inputs.keys()					)					,processor.model_input_names					)







'''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


snake_case_					:      List[Any]    			=    logging.get_logger(__name__)






@add_end_docstrings(lowercase  )
class 				lowercase__      (				lowercase  ):





	def __init__(					self   :							Tuple					,**lowerCamelCase__   :							Union[str, Any]					):







							'''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   :							Dict					,lowerCamelCase__   :							Union[np.ndarray, bytes, str]					,**lowerCamelCase__   :							Any					):







							'''simple docstring'''
							return super().__call__(lowerCamelCase__					,**lowerCamelCase__					)





	def       UpperCamelCase_    (					self   :							int					,**lowerCamelCase__   :							Optional[int]					):







							'''simple docstring'''
							_UpperCamelCase   :						List[Any]												=  {}
							if "candidate_labels" in kwargs:
													_UpperCamelCase   :						Any												=  kwargs['candidate_labels']
							if "hypothesis_template" in kwargs:
													_UpperCamelCase   :						Any												=  kwargs['hypothesis_template']

							return preprocess_params, {}, {}





	def       UpperCamelCase_    (					self   :							Any					,lowerCamelCase__   :							List[str]					,lowerCamelCase__   :							str=None					,lowerCamelCase__   :							List[Any]="This is a sound of {}."					):







							'''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
																			_UpperCamelCase   :						Optional[Any]												=  requests.get(lowerCamelCase__					).content
													else:
																			with open(lowerCamelCase__					,'rb'					) as f:
																									_UpperCamelCase   :						List[str]												=  f.read()

							if isinstance(lowerCamelCase__					,lowerCamelCase__					):
													_UpperCamelCase   :						int												=  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'					)

							_UpperCamelCase   :						Dict												=  self.feature_extractor(
							    [audio]					,sampling_rate=self.feature_extractor.sampling_rate					,return_tensors='pt'					)
							_UpperCamelCase   :						List[Any]												=  candidate_labels
							_UpperCamelCase   :						Tuple												=  [hypothesis_template.format(lowerCamelCase__					) for x in candidate_labels]
							_UpperCamelCase   :						Dict												=  self.tokenizer(lowerCamelCase__					,return_tensors=self.framework					,padding=lowerCamelCase__					)
							_UpperCamelCase   :						Optional[Any]												=  [text_inputs]
							return inputs





	def       UpperCamelCase_    (					self   :							str					,lowerCamelCase__   :							Tuple					):







							'''simple docstring'''
							_UpperCamelCase   :						Any												=  model_inputs.pop('candidate_labels'					)
							_UpperCamelCase   :						str												=  model_inputs.pop('text_inputs'					)
							if isinstance(text_inputs[0]					,lowerCamelCase__					):
													_UpperCamelCase   :						Union[str, Any]												=  text_inputs[0]
							else:
													# Batching case.
													_UpperCamelCase   :						int												=  text_inputs[0][0]

							_UpperCamelCase   :						List[str]												=  self.model(**lowerCamelCase__					,**lowerCamelCase__					)

							_UpperCamelCase   :						Optional[Any]												=  {
							    'candidate_labels': candidate_labels,
							    'logits': outputs.logits_per_audio,
							}
							return model_outputs





	def       UpperCamelCase_    (					self   :							int					,lowerCamelCase__   :							Optional[Any]					):







							'''simple docstring'''
							_UpperCamelCase   :						Optional[Any]												=  model_outputs.pop('candidate_labels'					)
							_UpperCamelCase   :						int												=  model_outputs['logits'][0]

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

							_UpperCamelCase   :						Union[str, Any]												=  [
							    {'score': score, 'label': candidate_label}
							    for score, candidate_label in sorted(zip(lowerCamelCase__					,lowerCamelCase__					)					,key=lambda lowerCamelCase__					: -x[0]					)
							]
							return result


