



'''simple docstring'''
import json
import os
from functools import lru_cache
from typing import TYPE_CHECKING, List, Optional, Tuple

import regex as re

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


if TYPE_CHECKING:
					from transformers.pipelines.conversational import Conversation

snake_case_					:      str    			=    logging.get_logger(__name__)


snake_case_					:      Dict    			=    {
    'vocab_file': 'vocab.json',
    'merges_file': 'merges.txt',
    'tokenizer_config_file': 'tokenizer_config.json',
}

snake_case_					:      Union[str, Any]    			=    {
    'vocab_file': {'facebook/blenderbot-3B': 'https://huggingface.co/facebook/blenderbot-3B/resolve/main/vocab.json'},
    'merges_file': {'facebook/blenderbot-3B': 'https://huggingface.co/facebook/blenderbot-3B/resolve/main/merges.txt'},
    'tokenizer_config_file': {
        'facebook/blenderbot-3B': 'https://huggingface.co/facebook/blenderbot-3B/resolve/main/tokenizer_config.json'
    },
}

snake_case_					:      str    			=    {'facebook/blenderbot-3B': 128}






@lru_cache()
# Copied from transformers.models.roberta.tokenization_roberta.bytes_to_unicode
def A__       (	):
						_UpperCamelCase   :						str												=  (
						    list(range(ord('!'				)    ,    ord('~'				) + 1				)				) + list(range(ord('¡'				)    ,    ord('¬'				) + 1				)				) + list(range(ord('®'				)    ,    ord('ÿ'				) + 1				)				)
						)
						_UpperCamelCase   :						Any												=  bs[:]
						_UpperCamelCase   :						Optional[int]												=  0
						for b in range(2**8				):
												if b not in bs:
																		bs.append(UpperCAmelCase_				)
																		cs.append(2**8 + n				)
																		n += 1
						_UpperCamelCase   :						Any												=  [chr(UpperCAmelCase_				) for n in cs]
						return dict(zip(UpperCAmelCase_    ,    UpperCAmelCase_				)				)






def A__       (	UpperCAmelCase_				):
						_UpperCamelCase   :						str												=  set()
						_UpperCamelCase   :						List[str]												=  word[0]
						for char in word[1:]:
												pairs.add((prev_char, char)				)
												_UpperCamelCase   :						Any												=  char
						return pairs






class 				lowercase__      (				lowercase  ):
	lowercase__    							=     VOCAB_FILES_NAMES
	lowercase__    							=     PRETRAINED_VOCAB_FILES_MAP
	lowercase__    							=     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	lowercase__    							=     ["""input_ids""", """attention_mask"""]





	def __init__(					self   :							List[Any]					,lowerCamelCase__   :							Any					,lowerCamelCase__   :							Tuple					,lowerCamelCase__   :							str="replace"					,lowerCamelCase__   :							Optional[Any]="<s>"					,lowerCamelCase__   :							int="</s>"					,lowerCamelCase__   :							List[Any]="</s>"					,lowerCamelCase__   :							str="<s>"					,lowerCamelCase__   :							List[Any]="<unk>"					,lowerCamelCase__   :							Union[str, Any]="<pad>"					,lowerCamelCase__   :							Tuple="<mask>"					,lowerCamelCase__   :							Tuple=False					,**lowerCamelCase__   :							str					,):







							'''simple docstring'''
							_UpperCamelCase   :						Dict												=  AddedToken(lowerCamelCase__					,lstrip=lowerCamelCase__					,rstrip=lowerCamelCase__					) if isinstance(lowerCamelCase__					,lowerCamelCase__					) else bos_token
							_UpperCamelCase   :						List[Any]												=  AddedToken(lowerCamelCase__					,lstrip=lowerCamelCase__					,rstrip=lowerCamelCase__					) if isinstance(lowerCamelCase__					,lowerCamelCase__					) else eos_token
							_UpperCamelCase   :						Tuple												=  AddedToken(lowerCamelCase__					,lstrip=lowerCamelCase__					,rstrip=lowerCamelCase__					) if isinstance(lowerCamelCase__					,lowerCamelCase__					) else sep_token
							_UpperCamelCase   :						str												=  AddedToken(lowerCamelCase__					,lstrip=lowerCamelCase__					,rstrip=lowerCamelCase__					) if isinstance(lowerCamelCase__					,lowerCamelCase__					) else cls_token
							_UpperCamelCase   :						str												=  AddedToken(lowerCamelCase__					,lstrip=lowerCamelCase__					,rstrip=lowerCamelCase__					) if isinstance(lowerCamelCase__					,lowerCamelCase__					) else unk_token
							_UpperCamelCase   :						int												=  AddedToken(lowerCamelCase__					,lstrip=lowerCamelCase__					,rstrip=lowerCamelCase__					) if isinstance(lowerCamelCase__					,lowerCamelCase__					) else pad_token

							# Mask token behave like a normal word, i.e. include the space before it
							_UpperCamelCase   :						List[Any]												=  AddedToken(lowerCamelCase__					,lstrip=lowerCamelCase__					,rstrip=lowerCamelCase__					) if isinstance(lowerCamelCase__					,lowerCamelCase__					) else mask_token

							super().__init__(
							    errors=lowerCamelCase__					,bos_token=lowerCamelCase__					,eos_token=lowerCamelCase__					,unk_token=lowerCamelCase__					,sep_token=lowerCamelCase__					,cls_token=lowerCamelCase__					,pad_token=lowerCamelCase__					,mask_token=lowerCamelCase__					,add_prefix_space=lowerCamelCase__					,**lowerCamelCase__					,)

							with open(lowerCamelCase__					,encoding='utf-8'					) as vocab_handle:
													_UpperCamelCase   :						Tuple												=  json.load(lowerCamelCase__					)
							_UpperCamelCase   :						Optional[int]												=  {v: k for k, v in self.encoder.items()}
							_UpperCamelCase   :						Tuple												=  errors  # how to handle errors in decoding
							_UpperCamelCase   :						Any												=  bytes_to_unicode()
							_UpperCamelCase   :						Any												=  {v: k for k, v in self.byte_encoder.items()}
							with open(lowerCamelCase__					,encoding='utf-8'					) as merges_handle:
													_UpperCamelCase   :						List[Any]												=  merges_handle.read().split('\n'					)[1:-1]
							_UpperCamelCase   :						List[Any]												=  [tuple(merge.split()					) for merge in bpe_merges]
							_UpperCamelCase   :						Any												=  dict(zip(lowerCamelCase__					,range(len(lowerCamelCase__					)					)					)					)
							_UpperCamelCase   :						str												=  {}
							_UpperCamelCase   :						Tuple												=  add_prefix_space

							# Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
							_UpperCamelCase   :						List[Any]												=  re.compile(R'\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+'					)





	@property
	# Copied from transformers.models.roberta.tokenization_roberta.RobertaTokenizer.vocab_size with Roberta->Blenderbot, RoBERTa->Blenderbot
	def       UpperCamelCase_    (					self   :							Optional[Any]					):







							'''simple docstring'''
							return len(self.encoder					)





	def       UpperCamelCase_    (					self   :							Union[str, Any]					):







							'''simple docstring'''
							return dict(self.encoder					,**self.added_tokens_encoder					)





	def       UpperCamelCase_    (					self   :							int					,lowerCamelCase__   :							Optional[int]					):







							'''simple docstring'''
							if token in self.cache:
													return self.cache[token]
							_UpperCamelCase   :						Optional[int]												=  tuple(lowerCamelCase__					)
							_UpperCamelCase   :						int												=  get_pairs(lowerCamelCase__					)

							if not pairs:
													return token

							while True:
													_UpperCamelCase   :						Optional[int]												=  min(lowerCamelCase__					,key=lambda lowerCamelCase__					: self.bpe_ranks.get(lowerCamelCase__					,float('inf'					)					)					)
													if bigram not in self.bpe_ranks:
																			break
													_UpperCamelCase							,     _UpperCamelCase   :						Any												=  bigram
													_UpperCamelCase   :						Optional[int]												=  []
													_UpperCamelCase   :						Dict												=  0
													while i < len(lowerCamelCase__					):
																			try:
																									_UpperCamelCase   :						Optional[int]												=  word.index(lowerCamelCase__					,lowerCamelCase__					)
																			except ValueError:
																									new_word.extend(word[i:]					)
																									break
																			else:
																									new_word.extend(word[i:j]					)
																									_UpperCamelCase   :						Tuple												=  j

																			if word[i] == first and i < len(lowerCamelCase__					) - 1 and word[i + 1] == second:
																									new_word.append(first + second					)
																									i += 2
																			else:
																									new_word.append(word[i]					)
																									i += 1
													_UpperCamelCase   :						int												=  tuple(lowerCamelCase__					)
													_UpperCamelCase   :						Any												=  new_word
													if len(lowerCamelCase__					) == 1:
																			break
													else:
																			_UpperCamelCase   :						str												=  get_pairs(lowerCamelCase__					)
							_UpperCamelCase   :						Optional[int]												=  ' '.join(lowerCamelCase__					)
							_UpperCamelCase   :						str												=  word
							return word





	def       UpperCamelCase_    (					self   :							Optional[int]					,lowerCamelCase__   :							Optional[Any]					):







							'''simple docstring'''
							_UpperCamelCase   :						str												=  []
							for token in re.findall(self.pat					,lowerCamelCase__					):
													_UpperCamelCase   :						List[Any]												=  ''.join(
													    self.byte_encoder[b] for b in token.encode('utf-8'					)					)  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
													bpe_tokens.extend(bpe_token for bpe_token in self.bpe(lowerCamelCase__					).split(' '					)					)
							return bpe_tokens





	def       UpperCamelCase_    (					self   :							Optional[Any]					,lowerCamelCase__   :							List[str]					):







							'''simple docstring'''
							return self.encoder.get(lowerCamelCase__					,self.encoder.get(self.unk_token					)					)





	def       UpperCamelCase_    (					self   :							List[Any]					,lowerCamelCase__   :							Optional[Any]					):







							'''simple docstring'''
							return self.decoder.get(lowerCamelCase__					)





	def       UpperCamelCase_    (					self   :							Union[str, Any]					,lowerCamelCase__   :							int					):







							'''simple docstring'''
							_UpperCamelCase   :						Optional[int]												=  ''.join(lowerCamelCase__					)
							_UpperCamelCase   :						List[str]												=  bytearray([self.byte_decoder[c] for c in text]					).decode('utf-8'					,errors=self.errors					)
							return text





	def       UpperCamelCase_    (					self   :							str					,lowerCamelCase__   :							str					,lowerCamelCase__   :							Optional[str] = None					):







							'''simple docstring'''
							if not os.path.isdir(lowerCamelCase__					):
													logger.error(F'Vocabulary path ({save_directory}) should be a directory'					)
													return
							_UpperCamelCase   :						Union[str, Any]												=  os.path.join(
							    lowerCamelCase__					,(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']					)
							_UpperCamelCase   :						str												=  os.path.join(
							    lowerCamelCase__					,(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['merges_file']					)

							with open(lowerCamelCase__					,'w'					,encoding='utf-8'					) as f:
													f.write(json.dumps(self.encoder					,indent=2					,sort_keys=lowerCamelCase__					,ensure_ascii=lowerCamelCase__					) + '\n'					)

							_UpperCamelCase   :						Dict												=  0
							with open(lowerCamelCase__					,'w'					,encoding='utf-8'					) as writer:
													writer.write('#version: 0.2\n'					)
													for bpe_tokens, token_index in sorted(self.bpe_ranks.items()					,key=lambda lowerCamelCase__					: kv[1]					):
																			if index != token_index:
																									logger.warning(
																									    F'Saving vocabulary to {merge_file}: BPE merge indices are not consecutive.'
																									    ' Please check that the tokenizer is not corrupted!'					)
																									_UpperCamelCase   :						Optional[int]												=  token_index
																			writer.write(' '.join(lowerCamelCase__					) + '\n'					)
																			index += 1

							return vocab_file, merge_file





	def       UpperCamelCase_    (					self   :							List[Any]					,lowerCamelCase__   :							List[int]					,lowerCamelCase__   :							Optional[List[int]] = None					,lowerCamelCase__   :							bool = False					):







							'''simple docstring'''
							if already_has_special_tokens:
													return super().get_special_tokens_mask(
													    token_ids_a=lowerCamelCase__					,token_ids_a=lowerCamelCase__					,already_has_special_tokens=lowerCamelCase__					)

							if token_ids_a is None:
													return [1] + ([0] * len(lowerCamelCase__					)) + [1]
							return [1] + ([0] * len(lowerCamelCase__					)) + [1, 1] + ([0] * len(lowerCamelCase__					)) + [1]





	def       UpperCamelCase_    (					self   :							Dict					,lowerCamelCase__   :							List[int]					,lowerCamelCase__   :							Optional[List[int]] = None					):







							'''simple docstring'''
							_UpperCamelCase   :						Optional[int]												=  [self.sep_token_id]
							_UpperCamelCase   :						List[Any]												=  [self.cls_token_id]

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





	def       UpperCamelCase_    (					self   :							str					,lowerCamelCase__   :							Dict					,lowerCamelCase__   :							Dict=False					,**lowerCamelCase__   :							int					):







							'''simple docstring'''
							_UpperCamelCase   :						str												=  kwargs.pop('add_prefix_space'					,self.add_prefix_space					)
							if (is_split_into_words or add_prefix_space) and (len(lowerCamelCase__					) > 0 and not text[0].isspace()):
													_UpperCamelCase   :						Union[str, Any]												=  ' ' + text
							return (text, kwargs)





	def       UpperCamelCase_    (					self   :							Optional[Any]					,lowerCamelCase__   :							List[int]					,lowerCamelCase__   :							Optional[List[int]] = None					):







							'''simple docstring'''
							return token_ids_a + [self.eos_token_id]





	def       UpperCamelCase_    (					self   :							int					,lowerCamelCase__   :							"Conversation"					):







							'''simple docstring'''
							_UpperCamelCase   :						List[str]												=  []
							for is_user, text in conversation.iter_texts():
													if is_user:
																			# We need to space prefix as it's being done within blenderbot
																			inputs.append(' ' + text					)
													else:
																			# Generated responses should contain them already.
																			inputs.append(lowerCamelCase__					)

							_UpperCamelCase   :						Any												=  '  '.join(lowerCamelCase__					)
							_UpperCamelCase   :						Union[str, Any]												=  self.encode(lowerCamelCase__					)
							if len(lowerCamelCase__					) > self.model_max_length:
													_UpperCamelCase   :						Union[str, Any]												=  input_ids[-self.model_max_length :]
													logger.warning(F'Trimmed input from conversation as it was longer than {self.model_max_length} tokens.'					)
							return input_ids







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


snake_case_					:      List[Any]    			=    logging.get_logger(__name__)






@add_end_docstrings(lowercase  )
class 				lowercase__      (				lowercase  ):





	def __init__(					self   :							Tuple					,**lowerCamelCase__   :							Union[str, Any]					):







							'''simple docstring'''
							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   :							Dict					,lowerCamelCase__   :							Union[np.ndarray, bytes, str]					,**lowerCamelCase__   :							Any					):







							'''simple docstring'''
							return super().__call__(lowerCamelCase__					,**lowerCamelCase__					)





	def       UpperCamelCase_    (					self   :							int					,**lowerCamelCase__   :							Optional[int]					):







							'''simple docstring'''
							_UpperCamelCase   :						List[Any]												=  {}
							if "candidate_labels" in kwargs:
													_UpperCamelCase   :						Any												=  kwargs['candidate_labels']
							if "hypothesis_template" in kwargs:
													_UpperCamelCase   :						Any												=  kwargs['hypothesis_template']

							return preprocess_params, {}, {}





	def       UpperCamelCase_    (					self   :							Any					,lowerCamelCase__   :							List[str]					,lowerCamelCase__   :							str=None					,lowerCamelCase__   :							List[Any]="This is a sound of {}."					):







							'''simple docstring'''
							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   :						Optional[Any]												=  requests.get(lowerCamelCase__					).content
													else:
																			with open(lowerCamelCase__					,'rb'					) as f:
																									_UpperCamelCase   :						List[str]												=  f.read()

							if isinstance(lowerCamelCase__					,lowerCamelCase__					):
													_UpperCamelCase   :						int												=  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   :						Dict												=  self.feature_extractor(
							    [audio]					,sampling_rate=self.feature_extractor.sampling_rate					,return_tensors='pt'					)
							_UpperCamelCase   :						List[Any]												=  candidate_labels
							_UpperCamelCase   :						Tuple												=  [hypothesis_template.format(lowerCamelCase__					) for x in candidate_labels]
							_UpperCamelCase   :						Dict												=  self.tokenizer(lowerCamelCase__					,return_tensors=self.framework					,padding=lowerCamelCase__					)
							_UpperCamelCase   :						Optional[Any]												=  [text_inputs]
							return inputs





	def       UpperCamelCase_    (					self   :							str					,lowerCamelCase__   :							Tuple					):







							'''simple docstring'''
							_UpperCamelCase   :						Any												=  model_inputs.pop('candidate_labels'					)
							_UpperCamelCase   :						str												=  model_inputs.pop('text_inputs'					)
							if isinstance(text_inputs[0]					,lowerCamelCase__					):
													_UpperCamelCase   :						Union[str, Any]												=  text_inputs[0]
							else:
													# Batching case.
													_UpperCamelCase   :						int												=  text_inputs[0][0]

							_UpperCamelCase   :						List[str]												=  self.model(**lowerCamelCase__					,**lowerCamelCase__					)

							_UpperCamelCase   :						Optional[Any]												=  {
							    'candidate_labels': candidate_labels,
							    'logits': outputs.logits_per_audio,
							}
							return model_outputs





	def       UpperCamelCase_    (					self   :							int					,lowerCamelCase__   :							Optional[Any]					):







							'''simple docstring'''
							_UpperCamelCase   :						Optional[Any]												=  model_outputs.pop('candidate_labels'					)
							_UpperCamelCase   :						int												=  model_outputs['logits'][0]

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

							_UpperCamelCase   :						Union[str, Any]												=  [
							    {'score': score, 'label': candidate_label}
							    for score, candidate_label in sorted(zip(lowerCamelCase__					,lowerCamelCase__					)					,key=lambda lowerCamelCase__					: -x[0]					)
							]
							return result


