
import os
from shutil import copyfile
from typing import TYPE_CHECKING, Optional, Tuple

from tokenizers import processors

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging
from ...utils.versions import require_version


if TYPE_CHECKING:
    from transformers.pipelines.conversational import Conversation

require_version("""tokenizers>=0.13.3""")

if is_sentencepiece_available():
    from .tokenization_llama import LlamaTokenizer
else:
    snake_case__  							=							None

snake_case__  							=							logging.get_logger(__name__)
snake_case__  							=							{"vocab_file": "tokenizer.model", "tokenizer_file": "tokenizer.json"}

snake_case__  							=							"[INST]", "[/INST]"
snake_case__  							=							"<<SYS>>\n", "\n<</SYS>>\n\n"

# fmt: off
snake_case__  							=							"You are a helpful, respectful and honest assistant. Always answer as helpfully as possible, while being safe. Your\\nanswers should not include any harmful, unethical, racist, sexist, toxic, dangerous, or illegal content. Please ensure\\nthat your responses are socially unbiased and positive in nature.\n\nIf a question does not make any sense, or is not factually coherent, explain why instead of answering something not\\ncorrect. If you don't know the answer to a question, please don't share false information."







class 							UpperCAmelCase	(   SCREAMING_SNAKE_CASE__					):
     a__: int			  =				VOCAB_FILES_NAMES
     a__: str			  =				LlamaTokenizer
     a__: List[Any]			  =				"""left"""
     a__: Optional[Any]			  =				["""input_ids""", """attention_mask"""]

     def __init__(			self      :					List[Any] ,      lowerCAmelCase      :					str=None ,      lowerCAmelCase      :					Tuple=None ,      lowerCAmelCase      :					Union[str, Any]=False ,      lowerCAmelCase      :					str="<unk>" ,      lowerCAmelCase      :					Dict="<s>" ,      lowerCAmelCase      :					Dict="</s>" ,      lowerCAmelCase      :					List[Any]=True ,      lowerCAmelCase      :					int=False ,      **lowerCAmelCase      :					Tuple ,      ):
           super().__init__(
               vocab_file=lowerCAmelCase ,      tokenizer_file=lowerCAmelCase ,      clean_up_tokenization_spaces=lowerCAmelCase ,      unk_token=lowerCAmelCase ,      bos_token=lowerCAmelCase ,      eos_token=lowerCAmelCase ,      **lowerCAmelCase ,      )
           lowercase       :    List[str]											=  add_bos_token
           lowercase       :    Tuple											=  add_eos_token
           self.update_post_processor()

           lowercase       :    str											=  vocab_file
           lowercase       :    List[str]											=  False if not self.vocab_file else True

     def        _lowerCAmelCase	(			self      :					Dict					):
           lowercase       :    Optional[int]											=  self.bos_token
           lowercase       :    str											=  self.bos_token_id

           lowercase       :    Union[str, Any]											=  self.eos_token
           lowercase       :    Optional[Any]											=  self.eos_token_id

           lowercase       :    List[Any]											=  f'''{(bos+":0 ") * self.add_bos_token}$A:0{(" "+eos+":0") * self.add_eos_token}'''
           lowercase       :    Tuple											=  f'''{single}{(" "+bos+":1") * self.add_bos_token} $B:1{(" "+eos+":1") * self.add_eos_token}'''

           lowercase       :    Dict											=  []
           if self.add_bos_token:
                 special_tokens.append((bos, bos_token_id)					)
           if self.add_eos_token:
                 special_tokens.append((eos, eos_token_id)					)
           lowercase       :    Optional[Any]											=  processors.TemplateProcessing(
               single=lowerCAmelCase ,      pair=lowerCAmelCase ,      special_tokens=lowerCAmelCase					)

     @property
     def        _lowerCAmelCase	(			self      :					Optional[Any]					):
           return self._add_eos_token

     @property
     def        _lowerCAmelCase	(			self      :					List[Any]					):
           return self._add_bos_token

     @add_eos_token.setter
     def        _lowerCAmelCase	(			self      :					str ,      lowerCAmelCase      :					List[str]					):
           lowercase       :    Any											=  value
           self.update_post_processor()

     @add_bos_token.setter
     def        _lowerCAmelCase	(			self      :					List[Any] ,      lowerCAmelCase      :					List[Any]					):
           lowercase       :    Optional[int]											=  value
           self.update_post_processor()

     def        _lowerCAmelCase	(			self      :					str ,      lowerCAmelCase      :					str ,      lowerCAmelCase      :					Optional[Any] = None					):



           if not self.can_save_slow_tokenizer:
                 raise ValueError(
                     '''Your fast tokenizer does not have the necessary information to save the vocabulary for a slow '''
                     '''tokenizer.'''					)

           if not os.path.isdir(lowerCAmelCase					):
                 logger.error(f'''Vocabulary path ({save_directory}) should be a directory'''					)
                 return
           lowercase       :    Union[str, Any]											=  os.path.join(
               lowerCAmelCase ,      (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']					)

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

           return (out_vocab_file,)




     def        _lowerCAmelCase	(			self      :					Union[str, Any] ,      lowerCAmelCase      :					Tuple					):
           lowercase       :    Optional[int]											=  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...)'''					)

           lowercase       :    List[Any]											=  []
           if len(conversation.past_user_inputs					) > 0:
                 if not conversation.past_user_inputs[0].startswith(lowerCAmelCase					) or E_SYS not in conversation.past_user_inputs[0]:
                       lowercase       :    Dict											=  (
                           B_SYS + DEFAULT_SYSTEM_PROMPT + E_SYS + conversation.past_user_inputs[0]
                       )
           elif not dialogue[0][1].startswith(lowerCAmelCase					) or E_SYS not in dialogue[0][1]:
                 lowercase       :    Union[str, Any]											=  (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=lowerCAmelCase					)
                   + [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=lowerCAmelCase					)
           return dialog_tokens




from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


snake_case__  							=							logging.get_logger(__name__)







class 							UpperCAmelCase	(   __lowerCamelCase					):
					a__: str			  =				["""input_values""", """attention_mask"""]

					def __init__(			self      :					Union[str, Any] ,      lowerCAmelCase      :					Tuple=1 ,      lowerCAmelCase      :					Dict=1_6000 ,      lowerCAmelCase      :					Optional[Any]=128 ,      lowerCAmelCase      :					Dict=1024 ,      lowerCAmelCase      :					List[Any]=0.0 ,      lowerCAmelCase      :					Any=True ,      lowerCAmelCase      :					Tuple=-4.2_67_73_93 ,      lowerCAmelCase      :					Dict=4.5_68_99_74 ,      lowerCAmelCase      :					int=False ,      **lowerCAmelCase      :					Optional[Any] ,      ):
											super().__init__(feature_size=lowerCAmelCase ,      sampling_rate=lowerCAmelCase ,      padding_value=lowerCAmelCase ,      **lowerCAmelCase					)
											lowercase       :    Optional[int]											=  num_mel_bins
											lowercase       :    List[Any]											=  max_length
											lowercase       :    Any											=  do_normalize
											lowercase       :    Any											=  mean
											lowercase       :    Union[str, Any]											=  std
											lowercase       :    int											=  return_attention_mask

					def        _lowerCAmelCase	(			self      :					int ,      lowerCAmelCase      :					np.ndarray ,      lowerCAmelCase      :					int ,      ):
											lowercase       :    List[Any]											=  torch.from_numpy(lowerCAmelCase					).unsqueeze(0					)
											lowercase       :    List[str]											=  ta_kaldi.fbank(
											    lowerCAmelCase ,      htk_compat=lowerCAmelCase ,      sample_frequency=self.sampling_rate ,      use_energy=lowerCAmelCase ,      window_type='''hanning''' ,      num_mel_bins=self.num_mel_bins ,      dither=0.0 ,      frame_shift=10 ,      )

											lowercase       :    Tuple											=  fbank.shape[0]
											lowercase       :    Optional[Any]											=  max_length - n_frames

											# pad or truncate, depending on difference
											if difference > 0:
																	lowercase       :    int											=  torch.nn.ZeroPadad((0, 0, 0, difference)					)
																	lowercase       :    str											=  pad_module(lowerCAmelCase					)
											elif difference < 0:
																	lowercase       :    Optional[int]											=  fbank[0:max_length, :]

											lowercase       :    List[str]											=  fbank.numpy()

											return fbank

					def        _lowerCAmelCase	(			self      :					Tuple ,      lowerCAmelCase      :					np.ndarray					):
											return (input_values - (self.mean)) / (self.std * 2)




					def __call__(			self      :					Any ,      lowerCAmelCase      :					Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]] ,      lowerCAmelCase      :					Optional[int] = None ,      lowerCAmelCase      :					Optional[Union[str, TensorType]] = None ,      **lowerCAmelCase      :					Tuple ,      ):

											if sampling_rate is not None:
																	if sampling_rate != self.sampling_rate:
																							raise ValueError(
																							    f'''The model corresponding to this feature extractor: {self} was trained using a sampling rate of'''
																							    f''' {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with'''
																							    f''' {self.sampling_rate} and not {sampling_rate}.'''					)
											else:
																	logger.warning(
																	    '''It is strongly recommended to pass the `sampling_rate` argument to this function. '''
																	    '''Failing to do so can result in silent errors that might be hard to debug.'''					)

											lowercase       :    Optional[Any]											=  isinstance(lowerCAmelCase ,      np.ndarray					) and len(raw_speech.shape					) > 1
											if is_batched_numpy and len(raw_speech.shape					) > 2:
																	raise ValueError(f'''Only mono-channel audio is supported for input to {self}'''					)
											lowercase       :    Tuple											=  is_batched_numpy or (
											    isinstance(lowerCAmelCase ,      (list, tuple)					) and (isinstance(raw_speech[0] ,      (np.ndarray, tuple, list)					))
											)

											if is_batched:
																	lowercase       :    Optional[int]											=  [np.asarray(lowerCAmelCase ,      dtype=np.floataa					) for speech in raw_speech]
											elif not is_batched and not isinstance(lowerCAmelCase ,      np.ndarray					):
																	lowercase       :    Dict											=  np.asarray(lowerCAmelCase ,      dtype=np.floataa					)
											elif isinstance(lowerCAmelCase ,      np.ndarray					) and raw_speech.dtype is np.dtype(np.floataa					):
																	lowercase       :    int											=  raw_speech.astype(np.floataa					)

											# always return batch
											if not is_batched:
																	lowercase       :    Dict											=  [raw_speech]

											# extract fbank features and pad/truncate to max_length
											lowercase       :    Dict											=  [self._extract_fbank_features(lowerCAmelCase ,      max_length=self.max_length					) for waveform in raw_speech]

											# convert into BatchFeature
											lowercase       :    Any											=  BatchFeature({'''input_values''': features}					)

											# make sure list is in array format
											lowercase       :    str											=  padded_inputs.get('''input_values'''					)
											if isinstance(input_values[0] ,      lowerCAmelCase					):
																	lowercase       :    Any											=  [np.asarray(lowerCAmelCase ,      dtype=np.floataa					) for feature in input_values]

											# normalization
											if self.do_normalize:
																	lowercase       :    List[str]											=  [self.normalize(lowerCAmelCase					) for feature in input_values]

											if return_tensors is not None:
																	lowercase       :    str											=  padded_inputs.convert_to_tensors(lowerCAmelCase					)

											return padded_inputs


