



import json
import os
import re
import unicodedata
from typing import Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer, _is_control, _is_punctuation, _is_whitespace
from ...utils import logging


__UpperCAmelCase						  =     logging.get_logger(__name__)

__UpperCAmelCase						  =     {
    '''vocab_file''': '''vocab.json''',
    '''merges_file''': '''merges.txt''',
}

__UpperCAmelCase						  =     {
    '''vocab_file''': {'''openai-gpt''': '''https://huggingface.co/openai-gpt/resolve/main/vocab.json'''},
    '''merges_file''': {'''openai-gpt''': '''https://huggingface.co/openai-gpt/resolve/main/merges.txt'''},
}

__UpperCAmelCase						  =     {
    '''openai-gpt''': 5_12,
}





def 						__lowerCamelCase       (      __magic_name__	:  Union[str, Any]						):
							a__:       int											=text.strip()
							if not text:
														return []
							a__:       Optional[int]											=text.split()
							return tokens


class 	lowerCamelCase__ (		_a   ):


						def __init__( self						:  Union[str, Any]      ,       _a						:  str=True      ,       _a						:  Any=None      ,       _a						:  Union[str, Any]=True      ,       _a						:  Any=None      ,       _a						:  Union[str, Any]=True      ,       ):
													if never_split is None:
																				a__:       Optional[int]											=[]
													a__:       Union[str, Any]											=do_lower_case
													a__:       List[Any]											=set(_a			)
													a__:       Dict											=tokenize_chinese_chars
													a__:       Optional[Any]											=strip_accents
													a__:       Tuple											=do_split_on_punc


						def   _lowerCamelCase			( self						:  Any      ,       _a						:  Dict      ,       _a						:  Dict=None			):
													a__:       List[Any]											=self.never_split.union(set(_a			)			) if never_split else self.never_split
													a__:       Union[str, Any]											=self._clean_text(_a			)

													# This was added on November 1st, 2018 for the multilingual and Chinese
													# models. This is also applied to the English models now, but it doesn't
													# matter since the English models were not trained on any Chinese data
													# and generally don't have any Chinese data in them (there are Chinese
													# characters in the vocabulary because Wikipedia does have some Chinese
													# words in the English Wikipedia.).
													if self.tokenize_chinese_chars:
																				a__:       List[Any]											=self._tokenize_chinese_chars(_a			)
													# prevents treating the same character with different unicode codepoints as different characters
													a__:       List[Any]											=unicodedata.normalize("NFC"      ,       _a			)
													a__:       Union[str, Any]											=whitespace_tokenize(_a			)
													a__:       int											=[]
													for token in orig_tokens:
																				if token not in never_split:
																											if self.do_lower_case:
																																		a__:       Optional[Any]											=token.lower()
																																		if self.strip_accents is not False:
																																									a__:       Union[str, Any]											=self._run_strip_accents(_a			)
																											elif self.strip_accents:
																																		a__:       Union[str, Any]											=self._run_strip_accents(_a			)
																				split_tokens.extend(self._run_split_on_punc(_a      ,       _a			)			)

													a__:       Dict											=whitespace_tokenize(" ".join(_a			)			)
													return output_tokens


						def   _lowerCamelCase			( self						:  Union[str, Any]      ,       _a						:  List[str]			):
													a__:       Tuple											=unicodedata.normalize("NFD"      ,       _a			)
													a__:       Optional[int]											=[]
													for char in text:
																				a__:       Union[str, Any]											=unicodedata.category(_a			)
																				if cat == "Mn":
																											continue
																				output.append(_a			)
													return "".join(_a			)


						def   _lowerCamelCase			( self						:  Optional[Any]      ,       _a						:  Optional[int]      ,       _a						:  Tuple=None			):
													if not self.do_split_on_punc or (never_split is not None and text in never_split):
																				return [text]
													a__:       List[Any]											=list(_a			)
													a__:       List[Any]											=0
													a__:       List[str]											=True
													a__:       Union[str, Any]											=[]
													while i < len(_a			):
																				a__:       Optional[int]											=chars[i]
																				if _is_punctuation(_a			):
																											output.append([char]			)
																											a__:       int											=True
																				else:
																											if start_new_word:
																																		output.append([]			)
																											a__:       Any											=False
																											output[-1].append(_a			)
																				i += 1

													return ["".join(_a			) for x in output]


						def   _lowerCamelCase			( self						:  int      ,       _a						:  Dict			):
													a__:       Optional[int]											=[]
													for char in text:
																				a__:       int											=ord(_a			)
																				if self._is_chinese_char(_a			):
																											output.append(" "			)
																											output.append(_a			)
																											output.append(" "			)
																				else:
																											output.append(_a			)
													return "".join(_a			)


						def   _lowerCamelCase			( self						:  Dict      ,       _a						:  str			):
													# This defines a "chinese character" as anything in the CJK Unicode block:
													#   https://en.wikipedia.org/wiki/CJK_Unified_Ideographs_(Unicode_block)
													#
													# Note that the CJK Unicode block is NOT all Japanese and Korean characters,
													# despite its name. The modern Korean Hangul alphabet is a different block,
													# as is Japanese Hiragana and Katakana. Those alphabets are used to write
													# space-separated words, so they are not treated specially and handled
													# like the all of the other languages.
													if (
													    (cp >= 0x4_e00 and cp <= 0x9_fff)
													    or (cp >= 0x3_400 and cp <= 0x4_dbf)  #
													    or (cp >= 0x20_000 and cp <= 0x2a_6df)  #
													    or (cp >= 0x2a_700 and cp <= 0x2b_73f)  #
													    or (cp >= 0x2b_740 and cp <= 0x2b_81f)  #
													    or (cp >= 0x2b_820 and cp <= 0x2c_eaf)  #
													    or (cp >= 0xf_900 and cp <= 0xf_aff)
													    or (cp >= 0x2f_800 and cp <= 0x2f_a1f)  #
													):  #
																				return True

													return False







						def   _lowerCamelCase			( self						:  Optional[Any]      ,       _a						:  Union[str, Any]			):
													a__:       Any											=[]
													for char in text:
																				a__:       str											=ord(_a			)
																				if cp == 0 or cp == 0xf_ffd or _is_control(_a			):
																											continue
																				if _is_whitespace(_a			):
																											output.append(" "			)
																				else:
																											output.append(_a			)
													return "".join(_a			)





def 						__lowerCamelCase       (      __magic_name__	:  int						):
							a__:       List[Any]											=set()
							a__:       Tuple											=word[0]
							for char in word[1:]:
														pairs.add((prev_char, char)						)
														a__:       List[Any]											=char
							return pairs





def 						__lowerCamelCase       (      __magic_name__	:  Tuple						):
							a__:       Optional[Any]											=text.replace("—"     , "-"						)
							a__:       Any											=text.replace("–"     , "-"						)
							a__:       Any											=text.replace("―"     , "-"						)
							a__:       List[str]											=text.replace("…"     , "..."						)
							a__:       Optional[int]											=text.replace("´"     , "'"						)
							a__:       List[Any]											=re.sub(R"(-+|~+|!+|\"+|;+|\?+|\++|,+|\)+|\(+|\\+|\/+|\*+|\[+|\]+|}+|{+|\|+|_+)"     , R" \1 "     , __magic_name__						)
							a__:       Any											=re.sub(R"\s*\n\s*"     , " \n "     , __magic_name__						)
							a__:       Optional[int]											=re.sub(R"[^\S\n]+"     , " "     , __magic_name__						)
							return text.strip()







class 	lowerCamelCase__ (		_a   ):
						_lowerCAmelCase								= VOCAB_FILES_NAMES
						_lowerCAmelCase								= PRETRAINED_VOCAB_FILES_MAP
						_lowerCAmelCase								= PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
						_lowerCAmelCase								= ['''input_ids''', '''attention_mask''']


						def __init__( self						:  Dict      ,       _a						:  Optional[Any]      ,       _a						:  List[str]      ,       _a						:  Optional[int]="<unk>"      ,       **_a						:  str			):
													super().__init__(unk_token=_a      ,       **_a			)

													try:
																				import ftfy
																				from spacy.lang.en import English

																				a__:       str											=English()
																				a__:       List[str]											=_nlp.tokenizer
																				a__:       Tuple											=ftfy.fix_text
													except ImportError:
																				logger.warning("ftfy or spacy is not installed using BERT BasicTokenizer instead of SpaCy & ftfy."			)
																				a__:       Tuple											=BasicTokenizer(do_lower_case=_a			)
																				a__:       int											=None

													with open(_a      ,       encoding="utf-8"			) as vocab_handle:
																				a__:       List[str]											=json.load(_a			)
													a__:       int											={v: k for k, v in self.encoder.items()}
													with open(_a      ,       encoding="utf-8"			) as merges_handle:
																				a__:       Tuple											=merges_handle.read().split("\n"			)[1:-1]
													a__:       Tuple											=[tuple(merge.split()			) for merge in merges]
													a__:       Optional[int]											=dict(zip(_a      ,       range(len(_a			)			)			)			)
													a__:       int											={}


						@property
						def   _lowerCamelCase			( self						:  Dict			):
													return True


						@property
						def   _lowerCamelCase			( self						:  int			):
													return len(self.encoder			)


						def   _lowerCamelCase			( self						:  Union[str, Any]			):
													return dict(self.encoder      ,       **self.added_tokens_encoder			)


						def   _lowerCamelCase			( self						:  Tuple      ,       _a						:  List[str]			):
													a__:       Optional[Any]											=tuple(token[:-1]			) + (token[-1] + "</w>",)
													if token in self.cache:
																				return self.cache[token]
													a__:       str											=get_pairs(_a			)

													if not pairs:
																				return token + "</w>"

													while True:
																				a__:       str											=min(_a      ,       key=lambda _a			: self.bpe_ranks.get(_a      ,       float("inf"			)			)			)
																				if bigram not in self.bpe_ranks:
																											break
																				a__							,			a__:       Tuple											=bigram
																				a__:       Dict											=[]
																				a__:       Tuple											=0
																				while i < len(_a			):
																											try:
																																		a__:       Dict											=word.index(_a      ,       _a			)
																											except ValueError:
																																		new_word.extend(word[i:]			)
																																		break
																											else:
																																		new_word.extend(word[i:j]			)
																																		a__:       str											=j

																											if word[i] == first and i < len(_a			) - 1 and word[i + 1] == second:
																																		new_word.append(first + second			)
																																		i += 2
																											else:
																																		new_word.append(word[i]			)
																																		i += 1
																				a__:       Tuple											=tuple(_a			)
																				a__:       Dict											=new_word
																				if len(_a			) == 1:
																											break
																				else:
																											a__:       Tuple											=get_pairs(_a			)
													a__:       Union[str, Any]											=" ".join(_a			)
													if word == "\n  </w>":
																				a__:       Tuple											="\n</w>"
													a__:       Optional[Any]											=word
													return word


						def   _lowerCamelCase			( self						:  List[Any]      ,       _a						:  Any			):
													a__:       str											=[]
													if self.fix_text is None:
																				# Using BERT's BasicTokenizer
																				a__:       List[Any]											=self.nlp.tokenize(_a			)
																				for token in text:
																											split_tokens.extend(list(self.bpe(_a			).split(" "			)			)			)
													else:
																				# Using SpaCy & ftfy (original tokenization process of OpenAI GPT)
																				a__:       str											=self.nlp(text_standardize(self.fix_text(_a			)			)			)
																				for token in text:
																											split_tokens.extend(list(self.bpe(token.text.lower()			).split(" "			)			)			)
													return split_tokens


						def   _lowerCamelCase			( self						:  Union[str, Any]      ,       _a						:  List[Any]			):
													return self.encoder.get(_a      ,       self.encoder.get(self.unk_token			)			)


						def   _lowerCamelCase			( self						:  Tuple      ,       _a						:  Tuple			):
													return self.decoder.get(_a      ,       self.unk_token			)


						def   _lowerCamelCase			( self						:  Dict      ,       _a						:  List[Any]			):
													a__:       Union[str, Any]											="".join(_a			).replace("</w>"      ,       " "			).strip()
													return out_string







						def   _lowerCamelCase			( self						:  Optional[Any]      ,       _a						:  str      ,       _a						:  Optional[str] = None			):
													if not os.path.isdir(_a			):
																				logger.error(F"Vocabulary path ({save_directory}) should be a directory"			)
																				return
													a__:       List[Any]											=os.path.join(
													    _a      ,       (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]			)
													a__:       Union[str, Any]											=os.path.join(
													    _a      ,       (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["merges_file"]			)

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

													a__:       Dict											=0
													with open(_a      ,       "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 _a			: 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!"			)
																																		a__:       Tuple											=token_index
																											writer.write(" ".join(_a			) + "\n"			)
																											index += 1

													return vocab_file, merge_file










import subprocess
from typing import Union

import numpy as np
import requests

from ..utils import add_end_docstrings, is_torch_available, is_torchaudio_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_torch_available():
		from ..models.auto.modeling_auto import MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING

__UpperCAmelCase						  =     logging.get_logger(__name__)





def 						__lowerCamelCase       (      __magic_name__	:  bytes     , __magic_name__	:  int						):
							a__:       Optional[int]											=F"{sampling_rate}"
							a__:       Union[str, Any]											="1"
							a__:       List[str]											="f32le"
							a__:       Dict											=[
							    "ffmpeg",
							    "-i",
							    "pipe:0",
							    "-ac",
							    ac,
							    "-ar",
							    ar,
							    "-f",
							    format_for_conversion,
							    "-hide_banner",
							    "-loglevel",
							    "quiet",
							    "pipe:1",
							]

							try:
														a__:       List[Any]											=subprocess.Popen(__magic_name__     , stdin=subprocess.PIPE     , stdout=subprocess.PIPE						)
							except FileNotFoundError:
														raise ValueError("ffmpeg was not found but is required to load audio files from filename"						)
							a__:       int											=ffmpeg_process.communicate(__magic_name__						)
							a__:       List[str]											=output_stream[0]

							a__:       str											=np.frombuffer(__magic_name__     , np.floataa						)
							if audio.shape[0] == 0:
														raise ValueError("Malformed soundfile"						)
							return audio







@add_end_docstrings(_a   )
class 	lowerCamelCase__ (		_a   ):


						def __init__( self						:  Union[str, Any]      ,       *_a						:  Dict      ,       **_a						:  Tuple			):
													# Default, might be overriden by the model.config.
													a__:       List[Any]											=5
													super().__init__(*_a      ,       **_a			)

													if self.framework != "pt":
																				raise ValueError(F"The {self.__class__} is only available in PyTorch."			)

													self.check_model_type(_a			)


						def __call__( self						:  Tuple      ,       _a						:  Union[np.ndarray, bytes, str]      ,       **_a						:  Union[str, Any]      ,       ):
													return super().__call__(_a      ,       **_a			)


						def   _lowerCamelCase			( self						:  int      ,       _a						:  Tuple=None      ,       **_a						:  int			):
													# No parameters on this pipeline right now
													a__:       Optional[int]											={}
													if top_k is not None:
																				if top_k > self.model.config.num_labels:
																											a__:       Union[str, Any]											=self.model.config.num_labels
																				a__:       Union[str, Any]											=top_k
													return {}, {}, postprocess_params


						def   _lowerCamelCase			( self						:  Any      ,       _a						:  Union[str, Any]			):
													if isinstance(_a      ,       _a			):
																				if inputs.startswith("http://"			) or inputs.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
																											a__:       List[str]											=requests.get(_a			).content
																				else:
																											with open(_a      ,       "rb"			) as f:
																																		a__:       Dict											=f.read()

													if isinstance(_a      ,       _a			):
																				a__:       Tuple											=ffmpeg_read(_a      ,       self.feature_extractor.sampling_rate			)

													if isinstance(_a      ,       _a			):
																				# Accepting `"array"` which is the key defined in `datasets` for
																				# better integration
																				if not ("sampling_rate" in inputs and ("raw" in inputs or "array" in inputs)):
																											raise ValueError(
																											    "When passing a dictionary to AudioClassificationPipeline, the dict needs to contain a "
																											    "\"raw\" key containing the numpy array representing the audio and a \"sampling_rate\" key, "
																											    "containing the sampling_rate associated with that array"			)

																				a__:       Tuple											=inputs.pop("raw"      ,       _a			)
																				if _inputs is None:
																											# Remove path which will not be used from `datasets`.
																											inputs.pop("path"      ,       _a			)
																											a__:       int											=inputs.pop("array"      ,       _a			)
																				a__:       Optional[Any]											=inputs.pop("sampling_rate"			)
																				a__:       Optional[int]											=_inputs
																				if in_sampling_rate != self.feature_extractor.sampling_rate:
																											import torch

																											if is_torchaudio_available():
																																		from torchaudio import functional as F
																											else:
																																		raise ImportError(
																																		    "torchaudio is required to resample audio samples in AudioClassificationPipeline. "
																																		    "The torchaudio package can be installed through: `pip install torchaudio`."			)

																											a__:       Union[str, Any]											=F.resample(
																											    torch.from_numpy(_a			)      ,       _a      ,       self.feature_extractor.sampling_rate			).numpy()

													if not isinstance(_a      ,       np.ndarray			):
																				raise ValueError("We expect a numpy ndarray as input"			)
													if len(inputs.shape			) != 1:
																				raise ValueError("We expect a single channel audio input for AudioClassificationPipeline"			)

													a__:       Union[str, Any]											=self.feature_extractor(
													    _a      ,       sampling_rate=self.feature_extractor.sampling_rate      ,       return_tensors="pt"			)
													return processed


						def   _lowerCamelCase			( self						:  Dict      ,       _a						:  Tuple			):
													a__:       str											=self.model(**_a			)
													return model_outputs







						def   _lowerCamelCase			( self						:  Optional[Any]      ,       _a						:  Tuple      ,       _a						:  Tuple=5			):
													a__:       Tuple											=model_outputs.logits[0].softmax(-1			)
													a__							,			a__:       Any											=probs.topk(_a			)

													a__:       Union[str, Any]											=scores.tolist()
													a__:       Dict											=ids.tolist()

													a__:       Optional[int]											=[{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(_a      ,       _a			)]

													return labels





