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


__UpperCamelCase									=			logging.get_logger(__name__)




@add_end_docstrings(lowerCAmelCase__   )
class  UpperCamelCase					(				lowerCAmelCase__   ):







       def __init__( self,							**lowerCAmelCase__)      ->      Tuple:
            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__)      ->      Dict:
            return super().__call__(lowerCAmelCase__,							**lowerCAmelCase__)







       def 		a_						( self,							**lowerCAmelCase__)      ->      List[Any]:
            snake_case_					       =					{}
            if "candidate_labels" in kwargs:
                 snake_case_					       =					kwargs['candidate_labels']
            if "hypothesis_template" in kwargs:
                 snake_case_					       =					kwargs['hypothesis_template']

            return preprocess_params, {}, {}







       def 		a_						( self,							lowerCAmelCase__,							lowerCAmelCase__=None,							lowerCAmelCase__="This is a sound of {}.")      ->      int:
            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
                      snake_case_					       =					requests.get(lowerCAmelCase__).content
                 else:
                      with open(lowerCAmelCase__,							'rb') as f:
                           snake_case_					       =					f.read()

            if isinstance(lowerCAmelCase__,							lowerCAmelCase__):
                 snake_case_					       =					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')

            snake_case_					       =					self.feature_extractor(
                [audio],							sampling_rate=self.feature_extractor.sampling_rate,							return_tensors='pt')
            snake_case_					       =					candidate_labels
            snake_case_					       =					[hypothesis_template.format(lowerCAmelCase__) for x in candidate_labels]
            snake_case_					       =					self.tokenizer(lowerCAmelCase__,							return_tensors=self.framework,							padding=lowerCAmelCase__)
            snake_case_					       =					[text_inputs]
            return inputs







       def 		a_						( self,							lowerCAmelCase__)      ->      List[str]:
            snake_case_					       =					model_inputs.pop('candidate_labels')
            snake_case_					       =					model_inputs.pop('text_inputs')
            if isinstance(text_inputs[0],							lowerCAmelCase__):
                 snake_case_					       =					text_inputs[0]
            else:
                 # Batching case.
                 snake_case_					       =					text_inputs[0][0]

            snake_case_					       =					self.model(**lowerCAmelCase__,							**lowerCAmelCase__)

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







       def 		a_						( self,							lowerCAmelCase__)      ->      int:
            snake_case_					       =					model_outputs.pop('candidate_labels')
            snake_case_					       =					model_outputs['logits'][0]

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

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






import warnings
from typing import Dict, List, Optional, Tuple

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


__A   			=logging.get_logger(__name__)

class 		UpperCAmelCase__			(			__UpperCamelCase   ):




			'''simple docstring'''






			UpperCamelCase									=       ["""input_ids""", """attention_mask"""]




			def __init__(   self							: List[str]       ,       a_							: List[Any]="</s>"       ,       a_							: str="<unk>"       ,       a_							: Tuple="<pad>"       ,       a_							: List[Any]=1_25       ,       a_							: Any=None       ,       **a_							: Tuple       ,       ):
					'''simple docstring'''





					if extra_ids > 0 and additional_special_tokens is None:
							__UpperCAmelCase    :     Optional[Any]													=	[F'<extra_id_{i}>' for i in range(a_ )]
					elif extra_ids > 0 and additional_special_tokens is not None:
							# Check that we have the right number of extra_id special tokens
							__UpperCAmelCase    :     Optional[Any]													=	len(set(filter(lambda a_ : bool('''extra_id''' in str(a_ ) )       ,       a_ ) ) )
							if extra_tokens != extra_ids:
									raise ValueError(
									    F'Both extra_ids ({extra_ids}) and additional_special_tokens ({additional_special_tokens}) are'
									    ''' provided to ByT5Tokenizer. In this case the additional_special_tokens must include the'''
									    ''' extra_ids tokens''' )

					__UpperCAmelCase    :     str													=	AddedToken(a_       ,       lstrip=a_       ,       rstrip=a_ ) if isinstance(a_       ,       a_ ) else pad_token
					__UpperCAmelCase    :     str													=	AddedToken(a_       ,       lstrip=a_       ,       rstrip=a_ ) if isinstance(a_       ,       a_ ) else eos_token
					__UpperCAmelCase    :     int													=	AddedToken(a_       ,       lstrip=a_       ,       rstrip=a_ ) if isinstance(a_       ,       a_ ) else unk_token

					super().__init__(
					    eos_token=a_       ,       unk_token=a_       ,       pad_token=a_       ,       extra_ids=a_       ,       additional_special_tokens=a_       ,       **a_       ,       )

					__UpperCAmelCase    :     Dict													=	extra_ids

					__UpperCAmelCase    :     Any													=	2**8  # utf is 8 bits

					# define special tokens dict
					__UpperCAmelCase    :     Dict[int, str]													=	{
					    self.pad_token: 0,
					    self.eos_token: 1,
					    self.unk_token: 2,
					}
					__UpperCAmelCase    :     Optional[int]													=	len(self.special_tokens_encoder )
					__UpperCAmelCase    :     Union[str, Any]													=	len(a_ )
					for i, token in enumerate(a_ ):
							__UpperCAmelCase    :     Union[str, Any]													=	self.vocab_size + i - n
					__UpperCAmelCase    :     Dict[str, int]													=	{v: k for k, v in self.special_tokens_encoder.items()}




			@property
			def  snake_case__     (   self							: Tuple ):
					'''simple docstring'''





					return self._utf_vocab_size + self._num_special_tokens + self._extra_ids




			def  snake_case__     (   self							: Any       ,       a_							: List[int]       ,       a_							: Optional[List[int]] = None       ,       a_							: bool = False ):
					'''simple docstring'''





					if already_has_special_tokens:
							return super().get_special_tokens_mask(
							    token_ids_a=a_       ,       token_ids_a=a_       ,       already_has_special_tokens=a_ )

					# normal case: some special tokens
					if token_ids_a is None:
							return ([0] * len(a_ )) + [1]
					return ([0] * len(a_ )) + [1] + ([0] * len(a_ )) + [1]




			def  snake_case__     (   self							: Any       ,       a_							: List[int] ):
					'''simple docstring'''





					if len(a_ ) > 0 and token_ids[-1] == self.eos_token_id:
							warnings.warn(
							    F'This sequence already has {self.eos_token}. In future versions this behavior may lead to duplicated'
							    ''' eos tokens being added.''' )
							return token_ids
					else:
							return token_ids + [self.eos_token_id]




			def  snake_case__     (   self							: Union[str, Any]       ,       a_							: List[int]       ,       a_							: Optional[List[int]] = None ):
					'''simple docstring'''





					__UpperCAmelCase    :     str													=	[self.eos_token_id]

					if token_ids_a is None:
							return len(token_ids_a + eos ) * [0]
					return len(token_ids_a + eos + token_ids_a + eos ) * [0]




			def  snake_case__     (   self							: Optional[Any]       ,       a_							: List[int]       ,       a_							: Optional[List[int]] = None ):
					'''simple docstring'''





					__UpperCAmelCase    :     Union[str, Any]													=	self._add_eos_if_not_present(a_ )
					if token_ids_a is None:
							return token_ids_a
					else:
							__UpperCAmelCase    :     List[str]													=	self._add_eos_if_not_present(a_ )
							return token_ids_a + token_ids_a




			def  snake_case__     (   self							: List[Any]       ,       a_							: str ):
					'''simple docstring'''





					__UpperCAmelCase    :     str													=	[chr(a_ ) for i in text.encode('''utf-8''' )]
					return tokens




			def  snake_case__     (   self							: Optional[Any]       ,       a_							: int ):
					'''simple docstring'''





					if token in self.special_tokens_encoder:
							__UpperCAmelCase    :     Union[str, Any]													=	self.special_tokens_encoder[token]
					elif token in self.added_tokens_encoder:
							__UpperCAmelCase    :     Optional[Any]													=	self.added_tokens_encoder[token]
					elif len(a_ ) != 1:
							__UpperCAmelCase    :     int													=	self.unk_token_id
					else:
							__UpperCAmelCase    :     List[str]													=	ord(a_ ) + self._num_special_tokens
					return token_id




			def  snake_case__     (   self							: Union[str, Any]       ,       a_							: Optional[int] ):
					'''simple docstring'''





					if index in self.special_tokens_decoder:
							__UpperCAmelCase    :     Dict													=	self.special_tokens_decoder[index]
					else:
							__UpperCAmelCase    :     List[Any]													=	chr(index - self._num_special_tokens )
					return token




			def  snake_case__     (   self							: int       ,       a_							: Dict ):
					'''simple docstring'''





					__UpperCAmelCase    :     List[Any]													=	b''''''
					for token in tokens:
							if token in self.special_tokens_decoder:
									__UpperCAmelCase    :     Optional[int]													=	self.special_tokens_decoder[token].encode('''utf-8''' )
							elif token in self.added_tokens_decoder:
									__UpperCAmelCase    :     List[str]													=	self.special_tokens_decoder[token].encode('''utf-8''' )
							elif token in self.special_tokens_encoder:
									__UpperCAmelCase    :     Tuple													=	token.encode('''utf-8''' )
							elif token in self.added_tokens_encoder:
									__UpperCAmelCase    :     int													=	token.encode('''utf-8''' )
							else:
									__UpperCAmelCase    :     Any													=	bytes([ord(a_ )] )
							bstring += tok_string
					__UpperCAmelCase    :     Dict													=	bstring.decode('''utf-8'''       ,       errors='''ignore''' )
					return string




			def  snake_case__     (   self							: Tuple       ,       a_							: str       ,       a_							: Optional[str] = None ):
					'''simple docstring'''





					return ()





