



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




'''simple docstring'''




import unittest

import numpy as np

from transformers import DistilBertConfig, is_flax_available
from transformers.testing_utils import require_flax, slow

from ...test_modeling_flax_common import FlaxModelTesterMixin, ids_tensor, random_attention_mask


if is_flax_available():
						import jax.numpy as jnp

						from transformers.models.distilbert.modeling_flax_distilbert import (
						    FlaxDistilBertForMaskedLM,
						    FlaxDistilBertForMultipleChoice,
						    FlaxDistilBertForQuestionAnswering,
						    FlaxDistilBertForSequenceClassification,
						    FlaxDistilBertForTokenClassification,
						    FlaxDistilBertModel,
						)







class 	_lowerCAmelCase							(			unittest.TestCase ):


	def __init__(self       ,      lowercase       ,      lowercase=13       ,      lowercase=7       ,      lowercase=True       ,      lowercase=True       ,      lowercase=True       ,      lowercase=True       ,      lowercase=99       ,      lowercase=32       ,      lowercase=5       ,      lowercase=4       ,      lowercase=37       ,      lowercase="gelu"       ,      lowercase=0.1       ,      lowercase=0.1       ,      lowercase=512       ,      lowercase=16       ,      lowercase=2       ,      lowercase=0.02       ,      lowercase=4       ,      ):
				A_	:						List[Any]	=     parent
				A_	:						str	=     batch_size
				A_	:						List[Any]	=     seq_length
				A_	:						Dict	=     is_training
				A_	:						List[Any]	=     use_attention_mask
				A_	:						Any	=     use_token_type_ids
				A_	:						Optional[int]	=     use_labels
				A_	:						Tuple	=     vocab_size
				A_	:						List[str]	=     hidden_size
				A_	:						List[str]	=     num_hidden_layers
				A_	:						Optional[Any]	=     num_attention_heads
				A_	:						int	=     intermediate_size
				A_	:						Optional[Any]	=     hidden_act
				A_	:						List[Any]	=     hidden_dropout_prob
				A_	:						Optional[Any]	=     attention_probs_dropout_prob
				A_	:						Any	=     max_position_embeddings
				A_	:						Union[str, Any]	=     type_vocab_size
				A_	:						int	=     type_sequence_label_size
				A_	:						Any	=     initializer_range
				A_	:						List[str]	=     num_choices


	def     _a	(self							):
				A_	:						List[str]	=     ids_tensor([self.batch_size, self.seq_length]       ,      self.vocab_size							)

				A_	:						Any	=     None
				if self.use_attention_mask:
							A_	:						Any	=     random_attention_mask([self.batch_size, self.seq_length]							)

				A_	:						Union[str, Any]	=     DistilBertConfig(
				    vocab_size=self.vocab_size       ,      dim=self.hidden_size       ,      n_layers=self.num_hidden_layers       ,      n_heads=self.num_attention_heads       ,      hidden_dim=self.intermediate_size       ,      hidden_act=self.hidden_act       ,      dropout=self.hidden_dropout_prob       ,      attention_dropout=self.attention_probs_dropout_prob       ,      max_position_embeddings=self.max_position_embeddings       ,      initializer_range=self.initializer_range       ,      tie_weights_=lowercase       ,      )

				return config, input_ids, attention_mask


	def     _a	(self							):
				A_	:						List[str]	=     self.prepare_config_and_inputs()
				A_,    A_,    A_	:						str	=     config_and_inputs
				A_	:						Any	=     {"""input_ids""": input_ids, """attention_mask""": attention_mask}
				return config, inputs_dict







@require_flax
class 	_lowerCAmelCase							(			__UpperCAmelCase      ,			unittest.TestCase ):
	__SCREAMING_SNAKE_CASE   :		Optional[Any]					=					(
	    (
	        FlaxDistilBertModel,
	        FlaxDistilBertForMaskedLM,
	        FlaxDistilBertForMultipleChoice,
	        FlaxDistilBertForQuestionAnswering,
	        FlaxDistilBertForSequenceClassification,
	        FlaxDistilBertForTokenClassification,
	        FlaxDistilBertForQuestionAnswering,
	    )
	    if is_flax_available()
	    else ()
	)


	def     _a	(self							):
				A_	:						Tuple	=     FlaxDistilBertModelTester(self							)


	@slow
	def     _a	(self							):
				for model_class_name in self.all_model_classes:
							A_	:						Union[str, Any]	=     model_class_name.from_pretrained("""distilbert-base-uncased"""							)
							A_	:						Any	=     model(np.ones((1, 1)							)							)
							self.assertIsNotNone(lowercase							)







@require_flax
class 	_lowerCAmelCase							(			unittest.TestCase ):


	@slow
	def     _a	(self							):
				A_	:						List[str]	=     FlaxDistilBertModel.from_pretrained("""distilbert-base-uncased"""							)
				A_	:						Optional[Any]	=     np.array([[0, 345, 232, 328, 740, 140, 1695, 69, 6078, 1588, 2]]							)
				A_	:						int	=     np.array([[0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]]							)
				A_	:						Optional[int]	=     model(lowercase       ,      attention_mask=lowercase							)[0]
				A_	:						Optional[Any]	=     (1, 11, 768)
				self.assertEqual(output.shape       ,      lowercase							)
				A_	:						Union[str, Any]	=     np.array([[[-0.16_39, 0.32_99, 0.16_48], [-0.17_46, 0.32_89, 0.17_10], [-0.18_84, 0.33_57, 0.18_10]]]							)

				self.assertTrue(jnp.allclose(output[:, 1:4, 1:4]       ,      lowercase       ,      atol=1E-4							)							)