


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


__A	=logging.get_logger(__name__)


@add_end_docstrings(lowercase__	)
class 		_snake_case					(     lowercase__	):




					def __init__(      self							,							**_lowerCamelCase):
								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							,							_lowerCamelCase							,							**_lowerCamelCase):
								return super().__call__(__lowerCamelCase							,							**__lowerCamelCase)




					def 				snake_case__						(      self							,							**_lowerCamelCase):
								UpperCAmelCase__  :					Optional[Any]        =     {}
								if "candidate_labels" in kwargs:
											UpperCAmelCase__  :					Union[str, Any]        =     kwargs["candidate_labels"]
								if "hypothesis_template" in kwargs:
											UpperCAmelCase__  :					List[Any]        =     kwargs["hypothesis_template"]

								return preprocess_params, {}, {}




					def 				snake_case__						(      self							,							_lowerCamelCase							,							_lowerCamelCase=None							,							_lowerCamelCase="This is a sound of {}."):



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

								if isinstance(__lowerCamelCase							,							__lowerCamelCase):
											UpperCAmelCase__  :					Tuple        =     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__  :					List[str]        =     self.feature_extractor(
								    [audio]							,							sampling_rate=self.feature_extractor.sampling_rate							,							return_tensors="""pt""")
								UpperCAmelCase__  :					Dict        =     candidate_labels
								UpperCAmelCase__  :					List[Any]        =     [hypothesis_template.format(__lowerCamelCase) for x in candidate_labels]
								UpperCAmelCase__  :					List[str]        =     self.tokenizer(__lowerCamelCase							,							return_tensors=self.framework							,							padding=__lowerCamelCase)
								UpperCAmelCase__  :					Union[str, Any]        =     [text_inputs]
								return inputs




					def 				snake_case__						(      self							,							_lowerCamelCase):
								UpperCAmelCase__  :					Union[str, Any]        =     model_inputs.pop("""candidate_labels""")
								UpperCAmelCase__  :					str        =     model_inputs.pop("""text_inputs""")
								if isinstance(text_inputs[0]							,							__lowerCamelCase):
											UpperCAmelCase__  :					int        =     text_inputs[0]
								else:
											# Batching case.
											UpperCAmelCase__  :					List[str]        =     text_inputs[0][0]

								UpperCAmelCase__  :					List[Any]        =     self.model(**__lowerCamelCase							,							**__lowerCamelCase)

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




					def 				snake_case__						(      self							,							_lowerCamelCase):
								UpperCAmelCase__  :					List[Any]        =     model_outputs.pop("""candidate_labels""")
								UpperCAmelCase__  :					int        =     model_outputs["logits"][0]

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

								UpperCAmelCase__  :					Optional[Any]        =     [
								    {"score": score, "label": candidate_label}
								    for score, candidate_label in sorted(zip(__lowerCamelCase							,							__lowerCamelCase)							,							key=lambda _lowerCamelCase: -x[0])
								]
								return result



'''simple docstring'''



import unittest

from transformers import AutoTokenizer, is_flax_available
from transformers.testing_utils import require_flax, require_sentencepiece, require_tokenizers, slow


if is_flax_available():
		import jax.numpy as jnp

		from transformers import FlaxXLMRobertaModel


@require_sentencepiece
@require_tokenizers
@require_flax
class 		_snake_case					(     unittest.TestCase	):




					@slow
					def 				snake_case__						(      self):
								UpperCAmelCase__  :					Union[str, Any]        =     FlaxXLMRobertaModel.from_pretrained("""xlm-roberta-base""")
								UpperCAmelCase__  :					List[str]        =     AutoTokenizer.from_pretrained("""xlm-roberta-base""")
								UpperCAmelCase__  :					Dict        =     """The dog is cute and lives in the garden house"""
								UpperCAmelCase__  :					Dict        =     jnp.array([tokenizer.encode(_lowerCamelCase)])

								UpperCAmelCase__  :					int        =     (1, 12, 768)  # batch_size, sequence_length, embedding_vector_dim
								UpperCAmelCase__  :					Any        =     jnp.array(
								    [[-0.0101, 0.1218, -0.0803, 0.0801, 0.1327, 0.0776, -0.1215, 0.2383, 0.3338, 0.3106, 0.0300, 0.0252]])

								UpperCAmelCase__  :					Any        =     model(_lowerCamelCase)["""last_hidden_state"""]
								self.assertEqual(output.shape							,							_lowerCamelCase)
								# compare the actual values for a slice of last dim
								self.assertTrue(jnp.allclose(output[:, :, -1]							,							_lowerCamelCase							,							atol=1e-3))