



'''simple docstring'''




import json
import os
import shutil
import tempfile
import unittest

import numpy as np
import pytest

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


if is_vision_available():
						from PIL import Image

						from transformers import ChineseCLIPImageProcessor, ChineseCLIPProcessor







@require_vision
class 	_lowerCAmelCase							(			unittest.TestCase ):


	def     _a	(self							):
				A_	:						List[Any]	=     tempfile.mkdtemp()

				A_	:						Dict	=     [
				    """[UNK]""",
				    """[CLS]""",
				    """[SEP]""",
				    """[PAD]""",
				    """[MASK]""",
				    """的""",
				    """价""",
				    """格""",
				    """是""",
				    """15""",
				    """便""",
				    """alex""",
				    """##andra""",
				    """，""",
				    """。""",
				    """-""",
				    """t""",
				    """shirt""",
				]
				A_	:						Any	=     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]							)							)

				A_	:						Any	=     {
				    """do_resize""": True,
				    """size""": {"""height""": 224, """width""": 224},
				    """do_center_crop""": True,
				    """crop_size""": {"""height""": 18, """width""": 18},
				    """do_normalize""": True,
				    """image_mean""": [0.48_14_54_66, 0.4_57_82_75, 0.40_82_10_73],
				    """image_std""": [0.26_86_29_54, 0.26_13_02_58, 0.27_57_77_11],
				    """do_convert_rgb""": True,
				}
				A_	:						Tuple	=     os.path.join(self.tmpdirname       ,      lowercase							)
				with open(self.image_processor_file       ,      """w"""       ,      encoding="""utf-8"""							) as fp:
							json.dump(lowercase       ,      lowercase							)


	def     _a	(self       ,      **lowercase							):
				return BertTokenizer.from_pretrained(self.tmpdirname       ,      **lowercase							)


	def     _a	(self       ,      **lowercase							):
				return BertTokenizerFast.from_pretrained(self.tmpdirname       ,      **lowercase							)


	def     _a	(self       ,      **lowercase							):
				return ChineseCLIPImageProcessor.from_pretrained(self.tmpdirname       ,      **lowercase							)


	def     _a	(self							):
				shutil.rmtree(self.tmpdirname							)


	def     _a	(self							):
				A_	:						int	=     [np.random.randint(255       ,      size=(3, 30, 400)       ,      dtype=np.uinta							)]

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

				return image_inputs


	def     _a	(self							):
				A_	:						List[str]	=     self.get_tokenizer()
				A_	:						Union[str, Any]	=     self.get_rust_tokenizer()
				A_	:						int	=     self.get_image_processor()

				A_	:						Tuple	=     ChineseCLIPProcessor(tokenizer=lowercase       ,      image_processor=lowercase							)
				processor_slow.save_pretrained(self.tmpdirname							)
				A_	:						Optional[int]	=     ChineseCLIPProcessor.from_pretrained(self.tmpdirname       ,      use_fast=lowercase							)

				A_	:						List[str]	=     ChineseCLIPProcessor(tokenizer=lowercase       ,      image_processor=lowercase							)
				processor_fast.save_pretrained(self.tmpdirname							)
				A_	:						List[Any]	=     ChineseCLIPProcessor.from_pretrained(self.tmpdirname							)

				self.assertEqual(processor_slow.tokenizer.get_vocab()       ,      tokenizer_slow.get_vocab()							)
				self.assertEqual(processor_fast.tokenizer.get_vocab()       ,      tokenizer_fast.get_vocab()							)
				self.assertEqual(tokenizer_slow.get_vocab()       ,      tokenizer_fast.get_vocab()							)
				self.assertIsInstance(processor_slow.tokenizer       ,      lowercase							)
				self.assertIsInstance(processor_fast.tokenizer       ,      lowercase							)

				self.assertEqual(processor_slow.image_processor.to_json_string()       ,      image_processor.to_json_string()							)
				self.assertEqual(processor_fast.image_processor.to_json_string()       ,      image_processor.to_json_string()							)
				self.assertIsInstance(processor_slow.image_processor       ,      lowercase							)
				self.assertIsInstance(processor_fast.image_processor       ,      lowercase							)


	def     _a	(self							):
				A_	:						str	=     ChineseCLIPProcessor(tokenizer=self.get_tokenizer()       ,      image_processor=self.get_image_processor()							)
				processor.save_pretrained(self.tmpdirname							)

				A_	:						str	=     self.get_tokenizer(cls_token="""(CLS)"""       ,      sep_token="""(SEP)"""							)
				A_	:						Tuple	=     self.get_image_processor(do_normalize=lowercase							)

				A_	:						Union[str, Any]	=     ChineseCLIPProcessor.from_pretrained(
				    self.tmpdirname       ,      cls_token="""(CLS)"""       ,      sep_token="""(SEP)"""       ,      do_normalize=lowercase							)

				self.assertEqual(processor.tokenizer.get_vocab()       ,      tokenizer_add_kwargs.get_vocab()							)
				self.assertIsInstance(processor.tokenizer       ,      lowercase							)

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


	def     _a	(self							):
				A_	:						Dict	=     self.get_image_processor()
				A_	:						Tuple	=     self.get_tokenizer()

				A_	:						Optional[int]	=     ChineseCLIPProcessor(tokenizer=lowercase       ,      image_processor=lowercase							)

				A_	:						Dict	=     self.prepare_image_inputs()

				A_	:						Tuple	=     image_processor(lowercase       ,      return_tensors="""np"""							)
				A_	:						int	=     processor(images=lowercase       ,      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     _a	(self							):
				A_	:						Dict	=     self.get_image_processor()
				A_	:						Union[str, Any]	=     self.get_tokenizer()

				A_	:						Dict	=     ChineseCLIPProcessor(tokenizer=lowercase       ,      image_processor=lowercase							)

				A_	:						Union[str, Any]	=     """Alexandra，T-shirt的价格是15便士。"""

				A_	:						Dict	=     processor(text=lowercase							)

				A_	:						Dict	=     tokenizer(lowercase							)

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


	def     _a	(self							):
				A_	:						str	=     self.get_image_processor()
				A_	:						Tuple	=     self.get_tokenizer()

				A_	:						List[Any]	=     ChineseCLIPProcessor(tokenizer=lowercase       ,      image_processor=lowercase							)

				A_	:						Optional[int]	=     """Alexandra，T-shirt的价格是15便士。"""
				A_	:						List[Any]	=     self.prepare_image_inputs()

				A_	:						Optional[int]	=     processor(text=lowercase       ,      images=lowercase							)

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

				# test if it raises when no input is passed
				with pytest.raises(lowercase							):
							processor()


	def     _a	(self							):
				A_	:						Any	=     self.get_image_processor()
				A_	:						int	=     self.get_tokenizer()

				A_	:						Optional[Any]	=     ChineseCLIPProcessor(tokenizer=lowercase       ,      image_processor=lowercase							)

				A_	:						List[Any]	=     [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

				A_	:						Union[str, Any]	=     processor.batch_decode(lowercase							)
				A_	:						List[Any]	=     tokenizer.batch_decode(lowercase							)

				self.assertListEqual(lowercase       ,      lowercase							)


	def     _a	(self							):
				A_	:						Optional[int]	=     self.get_image_processor()
				A_	:						int	=     self.get_tokenizer()

				A_	:						Tuple	=     ChineseCLIPProcessor(tokenizer=lowercase       ,      image_processor=lowercase							)

				A_	:						List[Any]	=     """Alexandra，T-shirt的价格是15便士。"""
				A_	:						Any	=     self.prepare_image_inputs()

				A_	:						str	=     processor(text=lowercase       ,      images=lowercase							)

				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


lowerCamelCase							:Union[str, Any]				    =					logging.get_logger(__name__)







@add_end_docstrings(__UpperCAmelCase )
class 	_lowerCAmelCase							(			__UpperCAmelCase ):


	def __init__(self       ,      **lowercase							):
				super().__init__(**lowercase							)

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


	def __call__(self       ,      lowercase       ,      **lowercase							):
				return super().__call__(lowercase       ,      **lowercase							)


	def     _a	(self       ,      **lowercase							):
				A_	:						Tuple	=     {}
				if "candidate_labels" in kwargs:
							A_	:						Dict	=     kwargs["""candidate_labels"""]
				if "hypothesis_template" in kwargs:
							A_	:						Optional[Any]	=     kwargs["""hypothesis_template"""]

				return preprocess_params, {}, {}


	def     _a	(self       ,      lowercase       ,      lowercase=None       ,      lowercase="This is a sound of {}."							):
				if isinstance(lowercase       ,      lowercase							):
							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_	:						Dict	=     requests.get(lowercase							).content
							else:
										with open(lowercase       ,      """rb"""							) as f:
													A_	:						List[str]	=     f.read()

				if isinstance(lowercase       ,      lowercase							):
							A_	:						List[Any]	=     ffmpeg_read(lowercase       ,      self.feature_extractor.sampling_rate							)

				if not isinstance(lowercase       ,      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_	:						int	=     self.feature_extractor(
				    [audio]       ,      sampling_rate=self.feature_extractor.sampling_rate       ,      return_tensors="""pt"""							)
				A_	:						List[Any]	=     candidate_labels
				A_	:						str	=     [hypothesis_template.format(lowercase							) for x in candidate_labels]
				A_	:						Optional[Any]	=     self.tokenizer(lowercase       ,      return_tensors=self.framework       ,      padding=lowercase							)
				A_	:						Optional[Any]	=     [text_inputs]
				return inputs


	def     _a	(self       ,      lowercase							):
				A_	:						Union[str, Any]	=     model_inputs.pop("""candidate_labels"""							)
				A_	:						List[Any]	=     model_inputs.pop("""text_inputs"""							)
				if isinstance(text_inputs[0]       ,      lowercase							):
							A_	:						Union[str, Any]	=     text_inputs[0]
				else:
							# Batching case.
							A_	:						Optional[int]	=     text_inputs[0][0]

				A_	:						str	=     self.model(**lowercase       ,      **lowercase							)

				A_	:						Union[str, Any]	=     {
				    """candidate_labels""": candidate_labels,
				    """logits""": outputs.logits_per_audio,
				}
				return model_outputs


	def     _a	(self       ,      lowercase							):
				A_	:						Union[str, Any]	=     model_outputs.pop("""candidate_labels"""							)
				A_	:						List[Any]	=     model_outputs["""logits"""][0]

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

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