






import os
from shutil import copyfile
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple

import sentencepiece as spm

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


if TYPE_CHECKING:
	from transformers.pipelines.conversational import Conversation

a_							=      logging.get_logger(__name__)

a_							=      {'''vocab_file''': '''tokenizer.model'''}

a_							=      {
    '''vocab_file''': {
        '''hf-internal-testing/llama-tokenizer''': '''https://huggingface.co/hf-internal-testing/llama-tokenizer/resolve/main/tokenizer.model''',
    },
    '''tokenizer_file''': {
        '''hf-internal-testing/llama-tokenizer''': '''https://huggingface.co/hf-internal-testing/llama-tokenizer/resolve/main/tokenizer_config.json''',
    },
}
a_							=      {
    '''hf-internal-testing/llama-tokenizer''': 2048,
}
a_							=      '''▁'''

a_,  a_							=      '''[INST]''', '''[/INST]'''
a_,  a_							=      '''<<SYS>>\n''', '''\n<</SYS>>\n\n'''

# fmt: off
a_							=      '''You are a helpful, respectful and honest assistant. Always answer as helpfully as possible, while being safe. Your\
answers should not include any harmful, unethical, racist, sexist, toxic, dangerous, or illegal content. Please ensure\
that your responses are socially unbiased and positive in nature.

If a question does not make any sense, or is not factually coherent, explain why instead of answering something not\
correct. If you don\'t know the answer to a question, please don\'t share false information.'''


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

		def __init__(							self						,      __UpperCAmelCase						,      __UpperCAmelCase="<unk>"						,      __UpperCAmelCase="<s>"						,      __UpperCAmelCase="</s>"						,      __UpperCAmelCase=None						,      __UpperCAmelCase = None						,      __UpperCAmelCase=True						,      __UpperCAmelCase=False						,      __UpperCAmelCase=False						,      __UpperCAmelCase=True						,      **__UpperCAmelCase						,      )->   Dict:







								'''simple docstring'''





								lowerCAmelCase__	 =  {} if sp_model_kwargs is None else sp_model_kwargs
								lowerCAmelCase__	 =  AddedToken(__UpperCAmelCase						,      lstrip=__UpperCAmelCase						,      rstrip=__UpperCAmelCase    ) if isinstance(__UpperCAmelCase						,      __UpperCAmelCase    ) else bos_token
								lowerCAmelCase__	 =  AddedToken(__UpperCAmelCase						,      lstrip=__UpperCAmelCase						,      rstrip=__UpperCAmelCase    ) if isinstance(__UpperCAmelCase						,      __UpperCAmelCase    ) else eos_token
								lowerCAmelCase__	 =  AddedToken(__UpperCAmelCase						,      lstrip=__UpperCAmelCase						,      rstrip=__UpperCAmelCase    ) if isinstance(__UpperCAmelCase						,      __UpperCAmelCase    ) else unk_token
								lowerCAmelCase__	 =  AddedToken(__UpperCAmelCase						,      lstrip=__UpperCAmelCase						,      rstrip=__UpperCAmelCase    ) if isinstance(__UpperCAmelCase						,      __UpperCAmelCase    ) else pad_token
								super().__init__(
								    bos_token=__UpperCAmelCase						,      eos_token=__UpperCAmelCase						,      unk_token=__UpperCAmelCase						,      pad_token=__UpperCAmelCase						,      add_bos_token=__UpperCAmelCase						,      add_eos_token=__UpperCAmelCase						,      sp_model_kwargs=self.sp_model_kwargs						,      clean_up_tokenization_spaces=__UpperCAmelCase						,      legacy=__UpperCAmelCase						,      **__UpperCAmelCase						,      )
								if legacy:
														logger.warning_once(
														    F"You are using the legacy behaviour of the {self.__class__}. This means that tokens that come after special tokens will not be properly handled. We recommend you to"
														    " read the related pull request available at https://github.com/huggingface/transformers/pull/24565"    )
								lowerCAmelCase__	 =  legacy
								lowerCAmelCase__	 =  vocab_file
								lowerCAmelCase__	 =  add_bos_token
								lowerCAmelCase__	 =  add_eos_token
								lowerCAmelCase__	 =  spm.SentencePieceProcessor(**self.sp_model_kwargs    )
								self.sp_model.Load(__UpperCAmelCase    )

		def __getstate__(							self    )->   List[Any]:







								'''simple docstring'''





								lowerCAmelCase__	 =  self.__dict__.copy()
								lowerCAmelCase__	 =  None
								lowerCAmelCase__	 =  self.sp_model.serialized_model_proto()
								return state

		def __setstate__(							self						,      __UpperCAmelCase    )->   Tuple:







								'''simple docstring'''





								lowerCAmelCase__	 =  d
								lowerCAmelCase__	 =  spm.SentencePieceProcessor(**self.sp_model_kwargs    )
								self.sp_model.LoadFromSerializedProto(self.sp_model_proto    )

		@property
		def 						UpperCAmelCase					(							self    )->   Any:







								'''simple docstring'''





								return self.sp_model.get_piece_size()

		def 						UpperCAmelCase					(							self    )->   Optional[Any]:







								'''simple docstring'''





								lowerCAmelCase__	 =  {self.convert_ids_to_tokens(__UpperCAmelCase    ): i for i in range(self.vocab_size    )}
								vocab.update(self.added_tokens_encoder    )
								return vocab

		def 						UpperCAmelCase					(							self						,      __UpperCAmelCase						,      **__UpperCAmelCase    )->   List[str]:







								'''simple docstring'''





								if not self.legacy:
														lowerCAmelCase__	 =  SPIECE_UNDERLINE + text.replace(__UpperCAmelCase						,      " "    )
								return super().tokenize(__UpperCAmelCase						,      **__UpperCAmelCase    )

		def 						UpperCAmelCase					(							self						,      __UpperCAmelCase    )->   List[Any]:







								'''simple docstring'''





								if not self.legacy:
														lowerCAmelCase__	 =  text.startswith(__UpperCAmelCase    )
														if is_first:
																				lowerCAmelCase__	 =  text[1:]

								lowerCAmelCase__	 =  self.sp_model.encode(__UpperCAmelCase						,      out_type=__UpperCAmelCase    )

								if not self.legacy and not is_first and not text.startswith(" "    ) and tokens[0].startswith(__UpperCAmelCase    ):
														lowerCAmelCase__	 =  ([tokens[0][1:]] if len(tokens[0]    ) > 1 else []) + tokens[1:]
								return tokens

		def 						UpperCAmelCase					(							self						,      __UpperCAmelCase    )->   Optional[Any]:







								'''simple docstring'''





								return self.sp_model.piece_to_id(__UpperCAmelCase    )

		def 						UpperCAmelCase					(							self						,      __UpperCAmelCase    )->   str:







								'''simple docstring'''





								lowerCAmelCase__	 =  self.sp_model.IdToPiece(__UpperCAmelCase    )
								return token

		def 						UpperCAmelCase					(							self						,      __UpperCAmelCase    )->   Dict:







								'''simple docstring'''





								lowerCAmelCase__	 =  []
								lowerCAmelCase__	 =  ""
								lowerCAmelCase__	 =  False
								for i, token in enumerate(__UpperCAmelCase    ):
														# make sure that special tokens are not decoded using sentencepiece model
														if token in self.all_special_tokens:
																				if not prev_is_special and i != 0:
																										out_string += " "
																				out_string += self.sp_model.decode(__UpperCAmelCase    ) + token
																				lowerCAmelCase__	 =  True
																				lowerCAmelCase__	 =  []
														else:
																				current_sub_tokens.append(__UpperCAmelCase    )
																				lowerCAmelCase__	 =  False
								out_string += self.sp_model.decode(__UpperCAmelCase    )
								return out_string

		def 						UpperCAmelCase					(							self						,      __UpperCAmelCase						,      __UpperCAmelCase = None    )->   Tuple[str]:







								'''simple docstring'''





								if not os.path.isdir(__UpperCAmelCase    ):
														logger.error(F"Vocabulary path ({save_directory}) should be a directory"    )
														return
								lowerCAmelCase__	 =  os.path.join(
								    __UpperCAmelCase						,      (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]    )

								if os.path.abspath(self.vocab_file    ) != os.path.abspath(__UpperCAmelCase    ) and os.path.isfile(self.vocab_file    ):
														copyfile(self.vocab_file						,      __UpperCAmelCase    )
								elif not os.path.isfile(self.vocab_file    ):
														with open(__UpperCAmelCase						,      "wb"    ) as fi:
																				lowerCAmelCase__	 =  self.sp_model.serialized_model_proto()
																				fi.write(__UpperCAmelCase    )

								return (out_vocab_file,)

		def 						UpperCAmelCase					(							self						,      __UpperCAmelCase						,      __UpperCAmelCase=None    )->   Union[str, Any]:







								'''simple docstring'''





								lowerCAmelCase__	 =  [self.bos_token_id] if self.add_bos_token else []
								lowerCAmelCase__	 =  [self.eos_token_id] if self.add_eos_token else []

								lowerCAmelCase__	 =  bos_token_id + token_ids_a + eos_token_id

								if token_ids_a is not None:
														lowerCAmelCase__	 =  output + bos_token_id + token_ids_a + eos_token_id

								return output

		def 						UpperCAmelCase					(							self						,      __UpperCAmelCase						,      __UpperCAmelCase = None						,      __UpperCAmelCase = False    )->   List[int]:







								'''simple docstring'''





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

								lowerCAmelCase__	 =  [1] if self.add_bos_token else []
								lowerCAmelCase__	 =  [1] if self.add_eos_token else []

								if token_ids_a is None:
														return bos_token_id + ([0] * len(__UpperCAmelCase    )) + eos_token_id
								return (
								    bos_token_id
								    + ([0] * len(__UpperCAmelCase    ))
								    + eos_token_id
								    + bos_token_id
								    + ([0] * len(__UpperCAmelCase    ))
								    + eos_token_id
								)

		def 						UpperCAmelCase					(							self						,      __UpperCAmelCase						,      __UpperCAmelCase = None    )->   List[int]:







								'''simple docstring'''





								lowerCAmelCase__	 =  [self.bos_token_id] if self.add_bos_token else []
								lowerCAmelCase__	 =  [self.eos_token_id] if self.add_eos_token else []

								lowerCAmelCase__	 =  [0] * len(bos_token_id + token_ids_a + eos_token_id    )

								if token_ids_a is not None:
														output += [1] * len(bos_token_id + token_ids_a + eos_token_id    )

								return output


		def 						UpperCAmelCase					(							self						,      __UpperCAmelCase    )->   List[int]:







								'''simple docstring'''





								lowerCAmelCase__	 =  list(conversation.iter_texts()    )
								if not all([is_user for is_user, msg in dialogue[::2]]    ) or not all(
								    [not is_user for is_user, msg in dialogue[1::2]]    ):
														raise ValueError(
														    "The model only supports 'user' and 'assistant' roles, starting with user and alternating (u/a/u/a/u...)"    )

								lowerCAmelCase__	 =  []
								if len(conversation.past_user_inputs    ) > 0:
														if not conversation.past_user_inputs[0].startswith(__UpperCAmelCase    ) or E_SYS not in conversation.past_user_inputs[0]:
																				lowerCAmelCase__	 =  (
																				    B_SYS + DEFAULT_SYSTEM_PROMPT + E_SYS + conversation.past_user_inputs[0]
																				)
								elif not dialogue[0][1].startswith(__UpperCAmelCase    ) or E_SYS not in dialogue[0][1]:
														lowerCAmelCase__	 =  (dialogue[0][0], B_SYS + DEFAULT_SYSTEM_PROMPT + E_SYS + dialogue[0][1])

								dialog_tokens += sum(
								    [
								        [self.bos_token_id]
								        + self.encode(
								            F"{B_INST} {(prompt[1]).strip()} {E_INST} {(answer[1]).strip()} "						,      add_special_tokens=__UpperCAmelCase    )
								        + [self.eos_token_id]
								        for prompt, answer in zip(dialogue[::2]						,      dialogue[1::2]    )
								    ]						,      []						,      )
								if not (dialogue[-1][0]):
														raise ValueError(F"Last message must be from user, got {dialogue[-1]['role']}"    )
								dialog_tokens += [self.bos_token_id] + self.encode(
								    F"{B_INST} {(dialogue[-1][1]).strip()} {E_INST}"						,      add_special_tokens=__UpperCAmelCase    )
								return dialog_tokens














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

a_							=      logging.get_logger(__name__)


def     _a					(       UpperCamelCase_   :			bytes      ,				UpperCamelCase_   :			int				)	->				np.array:
						"""simple docstring"""







						lowerCAmelCase__	 =  F"{sampling_rate}"
						lowerCAmelCase__	 =  "1"
						lowerCAmelCase__	 =  "f32le"
						lowerCAmelCase__	 =  [
						    "ffmpeg",
						    "-i",
						    "pipe:0",
						    "-ac",
						    ac,
						    "-ar",
						    ar,
						    "-f",
						    format_for_conversion,
						    "-hide_banner",
						    "-loglevel",
						    "quiet",
						    "pipe:1",
						]

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

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





@add_end_docstrings(_UpperCAmelCase		)
class        lowercase__       (     _UpperCAmelCase		):

	def __init__(							self						,      *__UpperCAmelCase						,      **__UpperCAmelCase    )->   Any:







							'''simple docstring'''





							lowerCAmelCase__	 =  5
							super().__init__(*__UpperCAmelCase						,      **__UpperCAmelCase    )

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

							self.check_model_type(__UpperCAmelCase    )

	def __call__(							self						,      __UpperCAmelCase						,      **__UpperCAmelCase						,      )->   List[str]:







							'''simple docstring'''





							return super().__call__(__UpperCAmelCase						,      **__UpperCAmelCase    )

	def 						UpperCAmelCase					(							self						,      __UpperCAmelCase=None						,      **__UpperCAmelCase    )->   Dict:







							'''simple docstring'''





							lowerCAmelCase__	 =  {}
							if top_k is not None:
													if top_k > self.model.config.num_labels:
																			lowerCAmelCase__	 =  self.model.config.num_labels
													lowerCAmelCase__	 =  top_k
							return {}, {}, postprocess_params

	def 						UpperCAmelCase					(							self						,      __UpperCAmelCase    )->   str:







							'''simple docstring'''





							if isinstance(__UpperCAmelCase						,      __UpperCAmelCase    ):
													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
																			lowerCAmelCase__	 =  requests.get(__UpperCAmelCase    ).content
													else:
																			with open(__UpperCAmelCase						,      "rb"    ) as f:
																									lowerCAmelCase__	 =  f.read()

							if isinstance(__UpperCAmelCase						,      __UpperCAmelCase    ):
													lowerCAmelCase__	 =  ffmpeg_read(__UpperCAmelCase						,      self.feature_extractor.sampling_rate    )

							if isinstance(__UpperCAmelCase						,      __UpperCAmelCase    ):
													# 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"    )

													lowerCAmelCase__	 =  inputs.pop("raw"						,      __UpperCAmelCase    )
													if _inputs is None:
																			# Remove path which will not be used from `datasets`.
																			inputs.pop("path"						,      __UpperCAmelCase    )
																			lowerCAmelCase__	 =  inputs.pop("array"						,      __UpperCAmelCase    )
													lowerCAmelCase__	 =  inputs.pop("sampling_rate"    )
													lowerCAmelCase__	 =  _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`."    )

																			lowerCAmelCase__	 =  F.resample(
																			    torch.from_numpy(__UpperCAmelCase    )						,      __UpperCAmelCase						,      self.feature_extractor.sampling_rate    ).numpy()

							if not isinstance(__UpperCAmelCase						,      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"    )

							lowerCAmelCase__	 =  self.feature_extractor(
							    __UpperCAmelCase						,      sampling_rate=self.feature_extractor.sampling_rate						,      return_tensors="pt"    )
							return processed

	def 						UpperCAmelCase					(							self						,      __UpperCAmelCase    )->   Tuple:







							'''simple docstring'''





							lowerCAmelCase__	 =  self.model(**__UpperCAmelCase    )
							return model_outputs


	def 						UpperCAmelCase					(							self						,      __UpperCAmelCase						,      __UpperCAmelCase=5    )->   int:







							'''simple docstring'''





							lowerCAmelCase__	 =  model_outputs.logits[0].softmax(-1    )
							lowerCAmelCase__ ,							lowerCAmelCase__	 =  probs.topk(__UpperCAmelCase    )

							lowerCAmelCase__	 =  scores.tolist()
							lowerCAmelCase__	 =  ids.tolist()

							lowerCAmelCase__	 =  [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(__UpperCAmelCase						,      __UpperCAmelCase    )]

							return labels






