



"""simple docstring"""


import json
from typing import Iterator, List, Union

from tokenizers import AddedToken, Regex, Tokenizer, decoders, normalizers, pre_tokenizers, trainers
from tokenizers.implementations.base_tokenizer import BaseTokenizer
from tokenizers.models import Unigram
from tokenizers.processors import TemplateProcessing
class 			_a (					lowerCAmelCase):
			"""simple docstring"""



			def __init__(		self     :    Optional[Any]				,   __UpperCamelCase     :    str = "▁"				,   __UpperCamelCase     :    bool = True				,   __UpperCamelCase     :    Union[str, AddedToken] = "<unk>"				,   __UpperCamelCase     :    Union[str, AddedToken] = "</s>"				,   __UpperCamelCase     :    Union[str, AddedToken] = "<pad>"				,   )->Any:
							_UpperCAmelCase					=    {
							    '''pad''': {'''id''': 0, '''token''': pad_token},
							    '''eos''': {'''id''': 1, '''token''': eos_token},
							    '''unk''': {'''id''': 2, '''token''': unk_token},
							}

							_UpperCAmelCase					=    [None] * len(self.special_tokens   )
							for token_dict in self.special_tokens.values():
											_UpperCAmelCase					=    token_dict['''token''']

							_UpperCAmelCase					=    Tokenizer(Unigram()   )

							_UpperCAmelCase					=    normalizers.Sequence(
							    [
							        normalizers.Nmt(),
							        normalizers.NFKC(),
							        normalizers.Replace(Regex(''' {2,}'''   )				,   ''' '''   ),
							        normalizers.Lowercase(),
							    ]   )
							_UpperCAmelCase					=    pre_tokenizers.Sequence(
							    [
							        pre_tokenizers.Metaspace(replacement=__UpperCamelCase				,   add_prefix_space=__UpperCamelCase   ),
							        pre_tokenizers.Digits(individual_digits=__UpperCamelCase   ),
							        pre_tokenizers.Punctuation(),
							    ]   )
							_UpperCAmelCase					=    decoders.Metaspace(replacement=__UpperCamelCase				,   add_prefix_space=__UpperCamelCase   )

							_UpperCAmelCase					=    TemplateProcessing(
							    single=F'$A {self.special_tokens["eos"]["token"]}'				,   special_tokens=[(self.special_tokens['''eos''']['''token'''], self.special_tokens['''eos''']['''id'''])]				,   )

							_UpperCAmelCase					=    {
							    '''model''': '''SentencePieceUnigram''',
							    '''replacement''': replacement,
							    '''add_prefix_space''': add_prefix_space,
							}

							super().__init__(__UpperCamelCase				,   __UpperCamelCase   )



			def  lowercase__						(		self     :    Union[str, Any]				,   __UpperCamelCase     :    Union[str, List[str]]				,   __UpperCamelCase     :    int = 8_0_0_0				,   __UpperCamelCase     :    bool = True				,   )->Any:
							_UpperCAmelCase					=    trainers.UnigramTrainer(
							    vocab_size=__UpperCamelCase				,   special_tokens=self.special_tokens_list				,   show_progress=__UpperCamelCase				,   )

							if isinstance(__UpperCamelCase				,   __UpperCamelCase   ):
											_UpperCAmelCase					=    [files]
							self._tokenizer.train(__UpperCamelCase				,   trainer=__UpperCamelCase   )

							self.add_unk_id()



			def  lowercase__						(		self     :    List[Any]				,   __UpperCamelCase     :    Union[Iterator[str], Iterator[Iterator[str]]]				,   __UpperCamelCase     :    int = 8_0_0_0				,   __UpperCamelCase     :    bool = True				,   )->Union[str, Any]:
							_UpperCAmelCase					=    trainers.UnigramTrainer(
							    vocab_size=__UpperCamelCase				,   special_tokens=self.special_tokens_list				,   show_progress=__UpperCamelCase				,   )

							self._tokenizer.train_from_iterator(__UpperCamelCase				,   trainer=__UpperCamelCase   )

							self.add_unk_id()



			def  lowercase__						(		self     :    Optional[Any]   )->Optional[Any]:
							_UpperCAmelCase					=    json.loads(self._tokenizer.to_str()   )

							_UpperCAmelCase					=    self.special_tokens['''unk''']['''id''']

							_UpperCAmelCase					=    Tokenizer.from_str(json.dumps(__UpperCamelCase   )   )








"""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   :							List[str]										=   logging.get_logger(__name__)
@add_end_docstrings(lowerCAmelCase)
class 			_a (					lowerCAmelCase):
			"""simple docstring"""



			def __init__(		self     :    Union[str, Any]				,   **__UpperCamelCase     :    Optional[Any]   )->List[Any]:
							super().__init__(**__UpperCamelCase   )

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



			def __call__(		self     :    Optional[int]				,   __UpperCamelCase     :    Union[np.ndarray, bytes, str]				,   **__UpperCamelCase     :    Tuple   )->List[str]:
							return super().__call__(__UpperCamelCase				,   **__UpperCamelCase   )



			def  lowercase__						(		self     :    Any				,   **__UpperCamelCase     :    Any   )->Union[str, Any]:
							_UpperCAmelCase					=    {}
							if "candidate_labels" in kwargs:
											_UpperCAmelCase					=    kwargs['''candidate_labels''']
							if "hypothesis_template" in kwargs:
											_UpperCAmelCase					=    kwargs['''hypothesis_template''']

							return preprocess_params, {}, {}



			def  lowercase__						(		self     :    Optional[int]				,   __UpperCamelCase     :    Union[str, Any]				,   __UpperCamelCase     :    List[Any]=None				,   __UpperCamelCase     :    Tuple="This is a sound of {}."   )->int:
							if isinstance(__UpperCamelCase				,   __UpperCamelCase   ):
											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					=    requests.get(__UpperCamelCase   ).content
											else:
															with open(__UpperCamelCase				,   '''rb'''   ) as f:
																			_UpperCAmelCase					=    f.read()

							if isinstance(__UpperCamelCase				,   __UpperCamelCase   ):
											_UpperCAmelCase					=    ffmpeg_read(__UpperCamelCase				,   self.feature_extractor.sampling_rate   )

							if not isinstance(__UpperCamelCase				,   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					=    self.feature_extractor(
							    [audio]				,   sampling_rate=self.feature_extractor.sampling_rate				,   return_tensors='''pt'''   )
							_UpperCAmelCase					=    candidate_labels
							_UpperCAmelCase					=    [hypothesis_template.format(__UpperCamelCase   ) for x in candidate_labels]
							_UpperCAmelCase					=    self.tokenizer(__UpperCamelCase				,   return_tensors=self.framework				,   padding=__UpperCamelCase   )
							_UpperCAmelCase					=    [text_inputs]
							return inputs



			def  lowercase__						(		self     :    Union[str, Any]				,   __UpperCamelCase     :    Optional[int]   )->Any:
							_UpperCAmelCase					=    model_inputs.pop('''candidate_labels'''   )
							_UpperCAmelCase					=    model_inputs.pop('''text_inputs'''   )
							if isinstance(text_inputs[0]				,   __UpperCamelCase   ):
											_UpperCAmelCase					=    text_inputs[0]
							else:
											# Batching case.
											_UpperCAmelCase					=    text_inputs[0][0]

							_UpperCAmelCase					=    self.model(**__UpperCamelCase				,   **__UpperCamelCase   )

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



			def  lowercase__						(		self     :    List[str]				,   __UpperCamelCase     :    Any   )->List[Any]:
							_UpperCAmelCase					=    model_outputs.pop('''candidate_labels'''   )
							_UpperCAmelCase					=    model_outputs['''logits'''][0]

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

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



