



'''simple docstring'''
import os
import re
from shutil import copyfile
from typing import List, Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


snake_case_					:      int    			=    logging.get_logger(__name__)

snake_case_					:      Tuple    			=    {
    'vocab_file': 'vocab.txt',
    'merges_file': 'bpe.codes',
}

snake_case_					:      str    			=    {
    'vocab_file': {
        'vinai/phobert-base': 'https://huggingface.co/vinai/phobert-base/resolve/main/vocab.txt',
        'vinai/phobert-large': 'https://huggingface.co/vinai/phobert-large/resolve/main/vocab.txt',
    },
    'merges_file': {
        'vinai/phobert-base': 'https://huggingface.co/vinai/phobert-base/resolve/main/bpe.codes',
        'vinai/phobert-large': 'https://huggingface.co/vinai/phobert-large/resolve/main/bpe.codes',
    },
}

snake_case_					:      List[Any]    			=    {
    'vinai/phobert-base': 256,
    'vinai/phobert-large': 256,
}






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

						_UpperCamelCase   :						Optional[int]												=  set(UpperCAmelCase_				)
						return pairs






class 				lowercase__      (				lowercase  ):
	lowercase__    							=     VOCAB_FILES_NAMES
	lowercase__    							=     PRETRAINED_VOCAB_FILES_MAP
	lowercase__    							=     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES





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







							'''simple docstring'''
							super().__init__(
							    bos_token=lowerCamelCase__					,eos_token=lowerCamelCase__					,unk_token=lowerCamelCase__					,sep_token=lowerCamelCase__					,cls_token=lowerCamelCase__					,pad_token=lowerCamelCase__					,mask_token=lowerCamelCase__					,**lowerCamelCase__					,)

							_UpperCamelCase   :						Tuple												=  vocab_file
							_UpperCamelCase   :						str												=  merges_file

							_UpperCamelCase   :						Dict												=  {}
							_UpperCamelCase   :						Union[str, Any]												=  0
							_UpperCamelCase   :						List[Any]												=  1
							_UpperCamelCase   :						int												=  2
							_UpperCamelCase   :						Dict												=  3

							self.add_from_file(lowerCamelCase__					)

							_UpperCamelCase   :						List[str]												=  {v: k for k, v in self.encoder.items()}

							with open(lowerCamelCase__					,encoding='utf-8'					) as merges_handle:
													_UpperCamelCase   :						Optional[Any]												=  merges_handle.read().split('\n'					)[:-1]
							_UpperCamelCase   :						Any												=  [tuple(merge.split()[:-1]					) for merge in merges]
							_UpperCamelCase   :						Any												=  dict(zip(lowerCamelCase__					,range(len(lowerCamelCase__					)					)					)					)
							_UpperCamelCase   :						int												=  {}





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







							'''simple docstring'''

							if token_ids_a is None:
													return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
							_UpperCamelCase   :						int												=  [self.cls_token_id]
							_UpperCamelCase   :						Optional[int]												=  [self.sep_token_id]
							return cls + token_ids_a + sep + sep + token_ids_a + sep





	def       UpperCamelCase_    (					self   :							str					,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   :							Optional[int]					,lowerCamelCase__   :							List[int]					,lowerCamelCase__   :							Optional[List[int]] = None					):







							'''simple docstring'''
							_UpperCamelCase   :						List[Any]												=  [self.sep_token_id]
							_UpperCamelCase   :						List[str]												=  [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]





	@property
	def       UpperCamelCase_    (					self   :							List[str]					):







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





	def       UpperCamelCase_    (					self   :							Optional[int]					):







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





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







							'''simple docstring'''
							if token in self.cache:
													return self.cache[token]
							_UpperCamelCase   :						int												=  tuple(lowerCamelCase__					)
							_UpperCamelCase   :						Optional[int]												=  tuple(list(word[:-1]					) + [word[-1] + '</w>']					)
							_UpperCamelCase   :						List[Any]												=  get_pairs(lowerCamelCase__					)

							if not pairs:
													return token

							while True:
													_UpperCamelCase   :						Dict												=  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   :						Any												=  []
													_UpperCamelCase   :						int												=  0
													while i < len(lowerCamelCase__					):
																			try:
																									_UpperCamelCase   :						List[Any]												=  word.index(lowerCamelCase__					,lowerCamelCase__					)
																			except ValueError:
																									new_word.extend(word[i:]					)
																									break
																			else:
																									new_word.extend(word[i:j]					)
																									_UpperCamelCase   :						Optional[int]												=  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   :						Tuple												=  tuple(lowerCamelCase__					)
													_UpperCamelCase   :						List[Any]												=  new_word
													if len(lowerCamelCase__					) == 1:
																			break
													else:
																			_UpperCamelCase   :						List[str]												=  get_pairs(lowerCamelCase__					)
							_UpperCamelCase   :						Tuple												=  '@@ '.join(lowerCamelCase__					)
							_UpperCamelCase   :						Tuple												=  word[:-4]
							_UpperCamelCase   :						List[str]												=  word
							return word





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







							'''simple docstring'''
							_UpperCamelCase   :						List[Any]												=  []

							_UpperCamelCase   :						Dict												=  re.findall(R'\S+\n?'					,lowerCamelCase__					)

							for token in words:
													split_tokens.extend(list(self.bpe(lowerCamelCase__					).split(' '					)					)					)
							return split_tokens





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







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





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







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





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







							'''simple docstring'''
							_UpperCamelCase   :						int												=  ' '.join(lowerCamelCase__					).replace('@@ '					,''					).strip()
							return out_string





	def       UpperCamelCase_    (					self   :							Optional[int]					,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   :						List[Any]												=  os.path.join(
							    lowerCamelCase__					,(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']					)
							_UpperCamelCase   :						Dict												=  os.path.join(
							    lowerCamelCase__					,(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['merges_file']					)

							if os.path.abspath(self.vocab_file					) != os.path.abspath(lowerCamelCase__					):
													copyfile(self.vocab_file					,lowerCamelCase__					)

							if os.path.abspath(self.merges_file					) != os.path.abspath(lowerCamelCase__					):
													copyfile(self.merges_file					,lowerCamelCase__					)

							return out_vocab_file, out_merge_file





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







							'''simple docstring'''
							if isinstance(lowerCamelCase__					,lowerCamelCase__					):
													try:
																			with open(lowerCamelCase__					,'r'					,encoding='utf-8'					) as fd:
																									self.add_from_file(lowerCamelCase__					)
													except FileNotFoundError as fnfe:
																			raise fnfe
													except UnicodeError:
																			raise Exception(F'Incorrect encoding detected in {f}, please rebuild the dataset'					)
													return

							_UpperCamelCase   :						Dict												=  f.readlines()
							for lineTmp in lines:
													_UpperCamelCase   :						List[Any]												=  lineTmp.strip()
													_UpperCamelCase   :						List[Any]												=  line.rfind(' '					)
													if idx == -1:
																			raise ValueError('Incorrect dictionary format, expected \'<token> <cnt>\''					)
													_UpperCamelCase   :						int												=  line[:idx]
													_UpperCamelCase   :						List[str]												=  len(self.encoder					)







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


