

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


__A			:					List[str]						=			logging.get_logger(__name__)
__A			:					Union[str, Any]						=			{'vocab_file': 'spiece.model'}

__A			:					int						=			{
    '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',
    }
}

__A			:					Optional[int]						=			{
    'AI-Sweden/gpt-sw3-126m': 2_0_4_8,
    'AI-Sweden/gpt-sw3-350m': 2_0_4_8,
    'AI-Sweden/gpt-sw3-1.6b': 2_0_4_8,
    'AI-Sweden/gpt-sw3-6.7b': 2_0_4_8,
    'AI-Sweden/gpt-sw3-20b': 2_0_4_8,
}





class 			_SCREAMING_SNAKE_CASE (   __snake_case		):







   '''simple docstring'''




   lowerCamelCase__				=      VOCAB_FILES_NAMES
   lowerCamelCase__				=      PRETRAINED_VOCAB_FILES_MAP
   lowerCamelCase__				=      PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
   lowerCamelCase__				=      ["input_ids", "attention_mask"]





   def __init__(       self					:			Tuple     , __lowerCamelCase					:			Any     , __lowerCamelCase					:			str=False     , __lowerCamelCase					:			int=False     , __lowerCamelCase					:			Dict=False     , __lowerCamelCase					:			List[str]=None     , __lowerCamelCase					:			Optional[Any]=None     , __lowerCamelCase					:			Optional[Any]=None     , __lowerCamelCase					:			Optional[int]=None     , __lowerCamelCase					:			Optional[Dict[str, Any]] = None     , **__lowerCamelCase					:			Dict     , ):
       SCREAMING_SNAKE_CASE										=   {} if sp_model_kwargs is None else sp_model_kwargs

       SCREAMING_SNAKE_CASE										=   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"			)
           SCREAMING_SNAKE_CASE										=   "None"

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

       SCREAMING_SNAKE_CASE										=   do_lower_case
       SCREAMING_SNAKE_CASE										=   remove_space
       SCREAMING_SNAKE_CASE										=   keep_accents
       SCREAMING_SNAKE_CASE										=   vocab_file

       SCREAMING_SNAKE_CASE										=   spm.SentencePieceProcessor(**self.sp_model_kwargs			)
       self.sp_model.Load(__lowerCamelCase			)

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

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





   def __getstate__(       self					:			int			):
       SCREAMING_SNAKE_CASE										=   self.__dict__.copy()
       SCREAMING_SNAKE_CASE										=   None
       return state





   def __setstate__(       self					:			Union[str, Any]     , __lowerCamelCase					:			str			):
       SCREAMING_SNAKE_CASE										=   d

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

       SCREAMING_SNAKE_CASE										=   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					:			Tuple			):
       return len(self.sp_model			)





   def      _snake_case		(       self					:			Optional[Any]     , __lowerCamelCase					:			str			):
       SCREAMING_SNAKE_CASE										=   self.non_printing_characters_re.sub(""     , __lowerCamelCase			)

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

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





   def      _snake_case		(       self					:			Dict     , __lowerCamelCase					:			str     , **__lowerCamelCase					:			Tuple			):
       SCREAMING_SNAKE_CASE										=   self.preprocess_text(__lowerCamelCase			)
       return self.sp_model.encode(__lowerCamelCase     , out_type=__lowerCamelCase			)





   def      _snake_case		(       self					:			List[Any]     , __lowerCamelCase					:			str			):
       return self.sp_model.PieceToId(__lowerCamelCase			)





   def      _snake_case		(       self					:			Any     , __lowerCamelCase					:			int			):
       return self.sp_model.IdToPiece(__lowerCamelCase			)





   @staticmethod
   def      _snake_case		(       __lowerCamelCase					:			str			):
       return out_string





   def      _snake_case		(       self					:			List[Any]     , __lowerCamelCase					:			List[str]			):
       SCREAMING_SNAKE_CASE										=   []
       SCREAMING_SNAKE_CASE										=   ""
       SCREAMING_SNAKE_CASE										=   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
               SCREAMING_SNAKE_CASE										=   True
               SCREAMING_SNAKE_CASE										=   []
           else:
               current_sub_tokens.append(__lowerCamelCase			)
               SCREAMING_SNAKE_CASE										=   False
       out_string += self.sp_model.decode(__lowerCamelCase			)

       return out_string





   def      _snake_case		(       self					:			Tuple			):
       SCREAMING_SNAKE_CASE										=   {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					:			List[Any]     , __lowerCamelCase					:			str     , __lowerCamelCase					:			Optional[str] = None			):
       if not os.path.isdir(__lowerCamelCase			):
           logger.error(f"Vocabulary path ({save_directory}) should be a directory"			)
           return
       SCREAMING_SNAKE_CASE										=   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:
               SCREAMING_SNAKE_CASE										=   self.sp_model.serialized_model_proto()
               fi.write(__lowerCamelCase			)

       return (out_vocab_file,)





   def      _snake_case		(       self					:			Any     , __lowerCamelCase					:			Union[str, List[str]]     , __lowerCamelCase					:			Union[str, bool] = False			):

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

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

       return token_ids





   def      _snake_case		(       self					:			Union[str, Any]     , __lowerCamelCase					:			Union[int, List[int]]			):
       return self.sp_model.decode(__lowerCamelCase			)





   def      _snake_case		(       self					:			Dict     , __lowerCamelCase					:			"Conversation"			):
       SCREAMING_SNAKE_CASE										=   [f"User: {text}" if is_user else f"Bot: {text}" for is_user, text in conversation.iter_texts()]
       SCREAMING_SNAKE_CASE										=   (
           f"{self.eos_token}{self.bos_token}" + f"{self.bos_token}".join(__lowerCamelCase			) + f"{self.bos_token}Bot:"
       )
       return self.encode(text=__lowerCamelCase			)


"""simple docstring"""
import logging
import os

import quant_trainer
import torch
from torch.utils.data import DataLoader

from transformers import Trainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput


_A							=  logging.getLogger(__name__)

if is_torch_tpu_available(check_device=False):
    import torch_xla.core.xla_model as xm
    import torch_xla.debug.metrics as met

class       lowerCamelCase	(       lowerCAmelCase__   ):


  '''simple docstring'''







  def __init__(self							,       *_lowerCamelCase							,       _lowerCamelCase=None							,       _lowerCamelCase=None							,       _lowerCamelCase=None							,       **_lowerCamelCase	):


        """simple docstring"""

        super().__init__(*_lowerCamelCase							,       **_lowerCamelCase	)
        UpperCAmelCase__			:							Any							=    eval_examples
        UpperCAmelCase__			:							Union[str, Any]							=    post_process_function
        UpperCAmelCase__			:							int							=    quant_trainer_args
        UpperCAmelCase__			:							Dict							=    128  # default number of calibration samples



  def  _a      (self							,       _lowerCamelCase=None	):


        """simple docstring"""

        if calib_dataset is None and self.calib_dataset is None:
              raise ValueError("""Trainer: calibration requires an calib_dataset."""	)
        UpperCAmelCase__			:							List[str]							=    calib_dataset if calib_dataset is not None else self.calib_dataset

        UpperCAmelCase__			:							Optional[int]							=    self._remove_unused_columns(_lowerCamelCase							,       description="""Calibration"""	)

        return DataLoader(
            _lowerCamelCase							,       batch_size=self.args.eval_batch_size							,       collate_fn=self.data_collator							,       drop_last=self.args.dataloader_drop_last							,       num_workers=self.args.dataloader_num_workers							,       pin_memory=self.args.dataloader_pin_memory							,       shuffle=_lowerCamelCase							,       )



  def  _a      (self							,       _lowerCamelCase=None	):


        """simple docstring"""

        UpperCAmelCase__			:							int							=    self.train_dataset if calib_dataset is None else calib_dataset
        UpperCAmelCase__			:							List[str]							=    self.get_calib_dataloader(_lowerCamelCase	)

        UpperCAmelCase__			:							List[str]							=    self.model
        quant_trainer.configure_model(_lowerCamelCase							,       self.quant_trainer_args							,       calib=_lowerCamelCase	)
        model.eval()
        quant_trainer.enable_calibration(_lowerCamelCase	)

        logger.info("""***** Running calibration *****"""	)
        logger.info(F"""  Num examples = {self.calib_num}"""	)
        logger.info(F"""  Batch size = {calib_dataloader.batch_size}"""	)

        for step, inputs in enumerate(_lowerCamelCase	):
              # Prediction step
              UpperCAmelCase__					,		UpperCAmelCase__					,		UpperCAmelCase__			:							List[str]							=    self.prediction_step(_lowerCamelCase							,       _lowerCamelCase							,       prediction_loss_only=_lowerCamelCase	)
              if (step + 1) * calib_dataloader.batch_size >= self.calib_num:
                    break

        quant_trainer.finish_calibration(_lowerCamelCase							,       self.quant_trainer_args	)
        UpperCAmelCase__			:							Tuple							=    model



  def  _a      (self							,       _lowerCamelCase=None							,       _lowerCamelCase=None							,       _lowerCamelCase=None							,       _lowerCamelCase = "eval"	):


        """simple docstring"""

        UpperCAmelCase__			:							Tuple							=    self.eval_dataset if eval_dataset is None else eval_dataset
        UpperCAmelCase__			:							List[Any]							=    self.get_eval_dataloader(_lowerCamelCase	)
        UpperCAmelCase__			:							Tuple							=    self.eval_examples if eval_examples is None else eval_examples

        # Temporarily disable metric computation, we will do it in the loop here.
        UpperCAmelCase__			:							Optional[Any]							=    self.compute_metrics
        UpperCAmelCase__			:							int							=    None
        UpperCAmelCase__			:							int							=    self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
        try:
              UpperCAmelCase__			:							str							=    eval_loop(
                  _lowerCamelCase							,       description="""Evaluation"""							,       prediction_loss_only=True if compute_metrics is None else None							,       ignore_keys=_lowerCamelCase							,       )
        finally:
              UpperCAmelCase__			:							List[str]							=    compute_metrics

        if self.post_process_function is not None and self.compute_metrics is not None:
              UpperCAmelCase__			:							int							=    self.post_process_function(_lowerCamelCase							,       _lowerCamelCase							,       output.predictions	)
              UpperCAmelCase__			:							List[Any]							=    self.compute_metrics(_lowerCamelCase	)

              # Prefix all keys with metric_key_prefix + '_'
              for key in list(metrics.keys()	):
                    if not key.startswith(F"""{metric_key_prefix}_"""	):
                          UpperCAmelCase__			:							Optional[int]							=    metrics.pop(_lowerCamelCase	)

              self.log(_lowerCamelCase	)
        else:
              UpperCAmelCase__			:							List[Any]							=    {}

        if self.args.tpu_metrics_debug or self.args.debug:
              # tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
              xm.master_print(met.metrics_report()	)

        UpperCAmelCase__			:							Optional[int]							=    self.callback_handler.on_evaluate(self.args							,       self.state							,       self.control							,       _lowerCamelCase	)
        return metrics



  def  _a      (self							,       _lowerCamelCase							,       _lowerCamelCase							,       _lowerCamelCase=None							,       _lowerCamelCase = "test"	):


        """simple docstring"""

        UpperCAmelCase__			:							Union[str, Any]							=    self.get_test_dataloader(_lowerCamelCase	)

        # Temporarily disable metric computation, we will do it in the loop here.
        UpperCAmelCase__			:							Union[str, Any]							=    self.compute_metrics
        UpperCAmelCase__			:							Optional[Any]							=    None
        UpperCAmelCase__			:							Any							=    self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
        try:
              UpperCAmelCase__			:							Optional[Any]							=    eval_loop(
                  _lowerCamelCase							,       description="""Prediction"""							,       prediction_loss_only=True if compute_metrics is None else None							,       ignore_keys=_lowerCamelCase							,       )
        finally:
              UpperCAmelCase__			:							Dict							=    compute_metrics

        if self.post_process_function is None or self.compute_metrics is None:
              return output

        UpperCAmelCase__			:							Dict							=    self.post_process_function(_lowerCamelCase							,       _lowerCamelCase							,       output.predictions							,       """predict"""	)
        UpperCAmelCase__			:							Any							=    self.compute_metrics(_lowerCamelCase	)

        # Prefix all keys with metric_key_prefix + '_'
        for key in list(metrics.keys()	):
              if not key.startswith(F"""{metric_key_prefix}_"""	):
                    UpperCAmelCase__			:							Any							=    metrics.pop(_lowerCamelCase	)

        return PredictionOutput(predictions=predictions.predictions							,       label_ids=predictions.label_ids							,       metrics=_lowerCamelCase	)




  def  _a      (self							,       _lowerCamelCase="./"	):


        """simple docstring"""

        UpperCAmelCase__			:							Tuple							=    self.eval_dataset
        UpperCAmelCase__			:							int							=    self.get_eval_dataloader(_lowerCamelCase	)

        UpperCAmelCase__			:							Union[str, Any]							=    next(iter(_lowerCamelCase	)	)

        # saving device - to make it consistent
        UpperCAmelCase__			:							int							=    torch.device("""cuda""" if torch.cuda.is_available() else """cpu"""	)

        # convert to tuple
        UpperCAmelCase__			:							List[Any]							=    tuple(v.to(_lowerCamelCase	) for k, v in batch.items()	)

        logger.info("""Converting model to be onnx compatible"""	)
        from pytorch_quantization.nn import TensorQuantizer

        UpperCAmelCase__			:							Optional[Any]							=    True

        UpperCAmelCase__			:							Tuple							=    self.model.to(_lowerCamelCase	)

        model.eval()
        model.float()

        UpperCAmelCase__			:							str							=    model.module if hasattr(_lowerCamelCase							,       """module"""	) else model
        quant_trainer.configure_model(_lowerCamelCase							,       self.quant_trainer_args	)

        UpperCAmelCase__			:							List[str]							=    os.path.join(_lowerCamelCase							,       """model.onnx"""	)
        logger.info(F"""exporting model to {output_model_file}"""	)

        UpperCAmelCase__			:							Optional[Any]							=    {0: """batch_size""", 1: """seq_len"""}

        torch.onnx.export(
            _lowerCamelCase							,       _lowerCamelCase							,       _lowerCamelCase							,       export_params=_lowerCamelCase							,       opset_version=13							,       do_constant_folding=_lowerCamelCase							,       input_names=["""input_ids""", """attention_mask""", """token_type_ids"""]							,       output_names=["""output_start_logits""", """output_end_logits"""]							,       dynamic_axes={
                """input_ids""": axes,
                """attention_mask""": axes,
                """token_type_ids""": axes,
                """output_start_logits""": axes,
                """output_end_logits""": axes,
            }							,       verbose=_lowerCamelCase							,       )
        logger.info("""onnx export finished"""	)

