





'''simple docstring'''


import enum
import warnings

from ..tokenization_utils import TruncationStrategy
from ..utils import add_end_docstrings, is_tf_available, is_torch_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_tf_available():
       import tensorflow as tf

       from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING

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

_SCREAMING_SNAKE_CASE	 =							logging.get_logger(__name__)

class    lowerCAmelCase_		(		enum.Enum		):
 __lowerCamelCase						:							str         =					0
 __lowerCamelCase						:							Union[str, Any]         =					1







@add_end_docstrings(lowercase__		)
class    lowerCAmelCase_		(		lowercase__		):
 __lowerCamelCase						:							int         =					"generated"




 def __init__(	self			,     *_lowerCAmelCase			,     **_lowerCAmelCase		)    ->							Tuple:
     super().__init__(*_lowerCAmelCase			,     **_lowerCAmelCase		)

     self.check_model_type(
         TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING
         if self.framework == "tf"
         else MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING		)




 def 							_snake_case   (	self			,     _lowerCAmelCase=None			,     _lowerCAmelCase=None			,     _lowerCAmelCase=None			,     _lowerCAmelCase=None			,     _lowerCAmelCase=None			,     _lowerCAmelCase=None			,     **_lowerCAmelCase			,     )    ->							Union[str, Any]:
     _lowerCAmelCase				=      {}
     if truncation is not None:
         _lowerCAmelCase				=      truncation

     _lowerCAmelCase				=      generate_kwargs

     _lowerCAmelCase				=      {}
     if return_tensors is not None and return_type is None:
         _lowerCAmelCase				=      ReturnType.TENSORS if return_tensors else ReturnType.TEXT
     if return_type is not None:
         _lowerCAmelCase				=      return_type

     if clean_up_tokenization_spaces is not None:
         _lowerCAmelCase				=      clean_up_tokenization_spaces

     if stop_sequence is not None:
         _lowerCAmelCase				=      self.tokenizer.encode(_lowerCAmelCase			,     add_special_tokens=_lowerCAmelCase		)
         if len(_lowerCAmelCase		) > 1:
             warnings.warn(
                 "Stopping on a multiple token sequence is not yet supported on transformers. The first token of"
                 " the stop sequence will be used as the stop sequence string in the interim."		)
         _lowerCAmelCase				=      stop_sequence_ids[0]

     return preprocess_params, forward_params, postprocess_params




 def 							_snake_case   (	self			,     _lowerCAmelCase			,     _lowerCAmelCase			,     _lowerCAmelCase		)    ->							Any:
     return True




 def 							_snake_case   (	self			,     *_lowerCAmelCase			,     _lowerCAmelCase		)    ->							Any:
     _lowerCAmelCase				=      self.model.config.prefix if self.model.config.prefix is not None else ""
     if isinstance(args[0]			,     _lowerCAmelCase		):
         if self.tokenizer.pad_token_id is None:
             raise ValueError("Please make sure that the tokenizer has a pad_token_id when using a batch input"		)
         _lowerCAmelCase				=      ([prefix + arg for arg in args[0]],)
         _lowerCAmelCase				=      True

     elif isinstance(args[0]			,     _lowerCAmelCase		):
         _lowerCAmelCase				=      (prefix + args[0],)
         _lowerCAmelCase				=      False
     else:
         raise ValueError(
             f''' `args[0]`: {args[0]} have the wrong format. The should be either of type `str` or type `list`'''		)
     _lowerCAmelCase				=      self.tokenizer(*_lowerCAmelCase			,     padding=_lowerCAmelCase			,     truncation=_lowerCAmelCase			,     return_tensors=self.framework		)
     # This is produced by tokenizers but is an invalid generate kwargs
     if "token_type_ids" in inputs:
         del inputs["token_type_ids"]
     return inputs




 def __call__(	self			,     *_lowerCAmelCase			,     **_lowerCAmelCase		)    ->							List[str]:
     _lowerCAmelCase				=      super().__call__(*_lowerCAmelCase			,     **_lowerCAmelCase		)
     if (
         isinstance(args[0]			,     _lowerCAmelCase		)
         and all(isinstance(_lowerCAmelCase			,     _lowerCAmelCase		) for el in args[0]		)
         and all(len(_lowerCAmelCase		) == 1 for res in result		)
     ):
         return [res[0] for res in result]
     return result




 def 							_snake_case   (	self			,     _lowerCAmelCase			,     _lowerCAmelCase=TruncationStrategy.DO_NOT_TRUNCATE			,     **_lowerCAmelCase		)    ->							Any:
     _lowerCAmelCase				=      self._parse_and_tokenize(_lowerCAmelCase			,     truncation=_lowerCAmelCase			,     **_lowerCAmelCase		)
     return inputs




 def 							_snake_case   (	self			,     _lowerCAmelCase			,     **_lowerCAmelCase		)    ->							Optional[int]:
     if self.framework == "pt":
         _lowerCAmelCase				=      model_inputs["input_ids"].shape
     elif self.framework == "tf":
         _lowerCAmelCase				=      tf.shape(model_inputs["input_ids"]		).numpy()

     _lowerCAmelCase				=      generate_kwargs.get("min_length"			,     self.model.config.min_length		)
     _lowerCAmelCase				=      generate_kwargs.get("max_length"			,     self.model.config.max_length		)
     self.check_inputs(_lowerCAmelCase			,     generate_kwargs["min_length"]			,     generate_kwargs["max_length"]		)
     _lowerCAmelCase				=      self.model.generate(**_lowerCAmelCase			,     **_lowerCAmelCase		)
     _lowerCAmelCase				=      output_ids.shape[0]
     if self.framework == "pt":
         _lowerCAmelCase				=      output_ids.reshape(_lowerCAmelCase			,     out_b // in_b			,     *output_ids.shape[1:]		)
     elif self.framework == "tf":
         _lowerCAmelCase				=      tf.reshape(_lowerCAmelCase			,     (in_b, out_b // in_b, *output_ids.shape[1:])		)
     return {"output_ids": output_ids}





 def 							_snake_case   (	self			,     _lowerCAmelCase			,     _lowerCAmelCase=ReturnType.TEXT			,     _lowerCAmelCase=False		)    ->							str:
     _lowerCAmelCase				=      []
     for output_ids in model_outputs["output_ids"][0]:
         if return_type == ReturnType.TENSORS:
             _lowerCAmelCase				=      {f'''{self.return_name}_token_ids''': output_ids}
         elif return_type == ReturnType.TEXT:
             _lowerCAmelCase				=      {
                 f'''{self.return_name}_text''': self.tokenizer.decode(
                     _lowerCAmelCase			,     skip_special_tokens=_lowerCAmelCase			,     clean_up_tokenization_spaces=_lowerCAmelCase			,     )
             }
         records.append(_lowerCAmelCase		)
     return records







@add_end_docstrings(lowercase__		)
class    lowerCAmelCase_		(		lowercase__		):
 __lowerCamelCase						:							Any         =					"summary"




 def __call__(	self			,     *_lowerCAmelCase			,     **_lowerCAmelCase		)    ->							int:
     return super().__call__(*_lowerCAmelCase			,     **_lowerCAmelCase		)





 def 							_snake_case   (	self			,     _lowerCAmelCase			,     _lowerCAmelCase			,     _lowerCAmelCase		)    ->							bool:
     if max_length < min_length:
         logger.warning(f'''Your min_length={min_length} must be inferior than your max_length={max_length}.'''		)

     if input_length < max_length:
         logger.warning(
             f'''Your max_length is set to {max_length}, but your input_length is only {input_length}. Since this is '''
             "a summarization task, where outputs shorter than the input are typically wanted, you might "
             f'''consider decreasing max_length manually, e.g. summarizer(\'...\', max_length={input_length//2})'''		)







@add_end_docstrings(lowercase__		)
class    lowerCAmelCase_		(		lowercase__		):
 __lowerCamelCase						:							Optional[Any]         =					"translation"




 def 							_snake_case   (	self			,     _lowerCAmelCase			,     _lowerCAmelCase			,     _lowerCAmelCase		)    ->							List[Any]:
     if input_length > 0.9 * max_length:
         logger.warning(
             f'''Your input_length: {input_length} is bigger than 0.9 * max_length: {max_length}. You might consider '''
             "increasing your max_length manually, e.g. translator('...', max_length=400)"		)
     return True




 def 							_snake_case   (	self			,     *_lowerCAmelCase			,     _lowerCAmelCase=TruncationStrategy.DO_NOT_TRUNCATE			,     _lowerCAmelCase=None			,     _lowerCAmelCase=None		)    ->							Optional[int]:
     if getattr(self.tokenizer			,     "_build_translation_inputs"			,     _lowerCAmelCase		):
         return self.tokenizer._build_translation_inputs(
             *_lowerCAmelCase			,     return_tensors=self.framework			,     truncation=_lowerCAmelCase			,     src_lang=_lowerCAmelCase			,     tgt_lang=_lowerCAmelCase		)
     else:
         return super()._parse_and_tokenize(*_lowerCAmelCase			,     truncation=_lowerCAmelCase		)




 def 							_snake_case   (	self			,     _lowerCAmelCase=None			,     _lowerCAmelCase=None			,     **_lowerCAmelCase		)    ->							Any:
     _lowerCAmelCase				=      super()._sanitize_parameters(**_lowerCAmelCase		)
     if src_lang is not None:
         _lowerCAmelCase				=      src_lang
     if tgt_lang is not None:
         _lowerCAmelCase				=      tgt_lang
     if src_lang is None and tgt_lang is None:
         # Backward compatibility, direct arguments use is preferred.
         _lowerCAmelCase				=      kwargs.get("task"			,     self.task		)
         _lowerCAmelCase				=      task.split("_"		)
         if task and len(_lowerCAmelCase		) == 4:
             # translation, XX, to YY
             _lowerCAmelCase				=      items[1]
             _lowerCAmelCase				=      items[3]
     return preprocess_params, forward_params, postprocess_params





 def __call__(	self			,     *_lowerCAmelCase			,     **_lowerCAmelCase		)    ->							int:
     return super().__call__(*_lowerCAmelCase			,     **_lowerCAmelCase		)









'''simple docstring'''


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

import sentencepiece as spm

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


if is_torch_available():
       import torch


if TYPE_CHECKING:
       from transformers.pipelines.conversational import Conversation


_SCREAMING_SNAKE_CASE	 =							logging.get_logger(__name__)
_SCREAMING_SNAKE_CASE	 =							{"vocab_file": "spiece.model"}

_SCREAMING_SNAKE_CASE	 =							{
    "vocab_file": {
        "AI-Sweden/gpt-sw3-126m": "https://huggingface.co/AI-Sweden/gpt-sw3-126m/resolve/main/spiece.model",
        "AI-Sweden/gpt-sw3-350m": "https://huggingface.co/AI-Sweden/gpt-sw3-350m/resolve/main/spiece.model",
        "AI-Sweden/gpt-sw3-1.6b": "https://huggingface.co/AI-Sweden/gpt-sw3-1.6b/resolve/main/spiece.model",
        "AI-Sweden/gpt-sw3-6.7b": "https://huggingface.co/AI-Sweden/gpt-sw3-6.7b/resolve/main/spiece.model",
        "AI-Sweden/gpt-sw3-20b": "https://huggingface.co/AI-Sweden/gpt-sw3-20b/resolve/main/spiece.model",
    }
}

_SCREAMING_SNAKE_CASE	 =							{
    "AI-Sweden/gpt-sw3-126m": 20_48,
    "AI-Sweden/gpt-sw3-350m": 20_48,
    "AI-Sweden/gpt-sw3-1.6b": 20_48,
    "AI-Sweden/gpt-sw3-6.7b": 20_48,
    "AI-Sweden/gpt-sw3-20b": 20_48,
}

class    lowerCAmelCase_		(		__magic_name__		):
 __lowerCamelCase						:							List[str]         =					VOCAB_FILES_NAMES
 __lowerCamelCase						:							List[Any]         =					PRETRAINED_VOCAB_FILES_MAP
 __lowerCamelCase						:							Tuple         =					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 __lowerCamelCase						:							List[str]         =					["input_ids", "attention_mask"]




 def __init__(	self			,     _lowerCAmelCase			,     _lowerCAmelCase=False			,     _lowerCAmelCase=False			,     _lowerCAmelCase=False			,     _lowerCAmelCase=None			,     _lowerCAmelCase=None			,     _lowerCAmelCase=None			,     _lowerCAmelCase=None			,     _lowerCAmelCase = None			,     **_lowerCAmelCase			,     )    ->							None:
     _lowerCAmelCase				=      {} if sp_model_kwargs is None else sp_model_kwargs

     _lowerCAmelCase				=      kwargs.get("name_or_path"		)
     if name_or_path is None:
         logger.warning(
             "name_or_path not provided, will work for all GPTSw3 models except gpt-sw3-7b,"
             " you are testing the model, this can safely be ignored"		)
         _lowerCAmelCase				=      "None"

     # Default definitions for our 2 tokenizer versions, with None-checks to enable proper testing
     _lowerCAmelCase				=      "<|endoftext|>" if eos_token is None else eos_token
     _lowerCAmelCase				=      "<unk>" if unk_token is None else unk_token
     if "gpt-sw3-7b" in name_or_path:
         _lowerCAmelCase				=      unk_token if pad_token is None else pad_token
         _lowerCAmelCase				=      eos_token if bos_token is None else bos_token
     else:
         _lowerCAmelCase				=      "<pad>" if pad_token is None else pad_token
         _lowerCAmelCase				=      "<s>" if bos_token is None else bos_token

     super().__init__(
         do_lower_case=_lowerCAmelCase			,     remove_space=_lowerCAmelCase			,     keep_accents=_lowerCAmelCase			,     bos_token=_lowerCAmelCase			,     eos_token=_lowerCAmelCase			,     unk_token=_lowerCAmelCase			,     pad_token=_lowerCAmelCase			,     sp_model_kwargs=self.sp_model_kwargs			,     **_lowerCAmelCase			,     )

     _lowerCAmelCase				=      do_lower_case
     _lowerCAmelCase				=      remove_space
     _lowerCAmelCase				=      keep_accents
     _lowerCAmelCase				=      vocab_file

     _lowerCAmelCase				=      spm.SentencePieceProcessor(**self.sp_model_kwargs		)
     self.sp_model.Load(_lowerCAmelCase		)

     # Used for whitespace normalization in input texts
     # fmt : off
     _lowerCAmelCase				=      {" ", " ", " ", " ", " ", "　", " ", " ", " ", " ", "￼", ""}
     # fmt : on

     # Regular expression to remove non-printing characters (e.g. some unicode control chars) in preprocessing
     _lowerCAmelCase				=      re.compile(
         f'''[{''.join(map(_lowerCAmelCase			,     list(range(0			,     9		)		) + list(range(11			,     32		)		) + list(range(127			,     160		)		) + [160, 173, 8203]		)		)}]'''		)




 def __getstate__(	self		)    ->							Any:
     _lowerCAmelCase				=      self.__dict__.copy()
     _lowerCAmelCase				=      None
     return state




 def __setstate__(	self			,     _lowerCAmelCase		)    ->							Union[str, Any]:
     _lowerCAmelCase				=      d

     # for backward compatibility
     if not hasattr(self			,     "sp_model_kwargs"		):
         _lowerCAmelCase				=      {}

     _lowerCAmelCase				=      spm.SentencePieceProcessor(**self.sp_model_kwargs		)
     self.sp_model.Load(self.vocab_file		)




 @property
 # Copied from transformers.models.albert.tokenization_albert.AlbertTokenizer.vocab_size
 def 							_snake_case   (	self		)    ->							int:
     return len(self.sp_model		)




 def 							_snake_case   (	self			,     _lowerCAmelCase		)    ->							str:
     _lowerCAmelCase				=      self.non_printing_characters_re.sub(""			,     _lowerCAmelCase		)

     # Normalize whitespaces
     _lowerCAmelCase				=      "".join([char if char not in self.whitespaces else " " for char in text]		)

     # NFC Unicode normalization
     _lowerCAmelCase				=      unicodedata.normalize("NFC"			,     _lowerCAmelCase		)
     return text




 def 							_snake_case   (	self			,     _lowerCAmelCase			,     **_lowerCAmelCase		)    ->							List[str]:
     _lowerCAmelCase				=      self.preprocess_text(_lowerCAmelCase		)
     return self.sp_model.encode(_lowerCAmelCase			,     out_type=_lowerCAmelCase		)




 def 							_snake_case   (	self			,     _lowerCAmelCase		)    ->							int:
     return self.sp_model.PieceToId(_lowerCAmelCase		)




 def 							_snake_case   (	self			,     _lowerCAmelCase		)    ->							str:
     return self.sp_model.IdToPiece(_lowerCAmelCase		)




 @staticmethod
 def 							_snake_case   (	_lowerCAmelCase		)    ->							str:
     return out_string




 def 							_snake_case   (	self			,     _lowerCAmelCase		)    ->							str:
     _lowerCAmelCase				=      []
     _lowerCAmelCase				=      ""
     _lowerCAmelCase				=      False
     for token in tokens:
         # make sure that special tokens are not decoded using sentencepiece model
         if token in self.all_special_tokens:
             # TODO: Check if this is needed, as it ensures that decode(encode(doc)) != doc by adding extra whitespace in the decoded document
             if not prev_is_special:
                 out_string += " "

             out_string += self.sp_model.decode(_lowerCAmelCase		) + token
             _lowerCAmelCase				=      True
             _lowerCAmelCase				=      []
         else:
             current_sub_tokens.append(_lowerCAmelCase		)
             _lowerCAmelCase				=      False
     out_string += self.sp_model.decode(_lowerCAmelCase		)

     return out_string




 def 							_snake_case   (	self		)    ->							Dict[str, int]:
     _lowerCAmelCase				=      {self.convert_ids_to_tokens(_lowerCAmelCase		): i for i in range(self.vocab_size		)}
     vocab.update(self.added_tokens_encoder		)
     return vocab




 def 							_snake_case   (	self			,     _lowerCAmelCase			,     _lowerCAmelCase = None		)    ->							Tuple[str]:
     if not os.path.isdir(_lowerCAmelCase		):
         logger.error(f'''Vocabulary path ({save_directory}) should be a directory'''		)
         return
     _lowerCAmelCase				=      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		) and os.path.isfile(self.vocab_file		):
         copyfile(self.vocab_file			,     _lowerCAmelCase		)
     elif not os.path.isfile(self.vocab_file		):
         with open(_lowerCAmelCase			,     "wb"		) as fi:
             _lowerCAmelCase				=      self.sp_model.serialized_model_proto()
             fi.write(_lowerCAmelCase		)

     return (out_vocab_file,)




 def 							_snake_case   (	self			,     _lowerCAmelCase			,     _lowerCAmelCase = False		)    ->							Union[List[int], List[List[int]], "torch.Tensor"]:

     if isinstance(_lowerCAmelCase			,     _lowerCAmelCase		):
         _lowerCAmelCase				=      self.preprocess_text(_lowerCAmelCase		)
         _lowerCAmelCase				=      self.sp_model.encode(_lowerCAmelCase		)
     else:
         _lowerCAmelCase				=      [self.preprocess_text(_lowerCAmelCase		) for t in text]
         _lowerCAmelCase				=      self.sp_model.encode(_lowerCAmelCase		)

     if return_tensors is True or return_tensors == "pt":
         _lowerCAmelCase				=      torch.tensor(_lowerCAmelCase		)

     return token_ids




 def 							_snake_case   (	self			,     _lowerCAmelCase		)    ->							str:
     return self.sp_model.decode(_lowerCAmelCase		)





 def 							_snake_case   (	self			,     _lowerCAmelCase		)    ->							List[int]:
     _lowerCAmelCase				=      [f'''User: {text}''' if is_user else f'''Bot: {text}''' for is_user, text in conversation.iter_texts()]
     _lowerCAmelCase				=      (
         f'''{self.eos_token}{self.bos_token}''' + f'''{self.bos_token}'''.join(_lowerCAmelCase		) + f'''{self.bos_token}Bot:'''
     )
     return self.encode(text=_lowerCAmelCase		)


