

"""simple docstring"""





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

import sentencepiece as spm

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


_a       =							logging.get_logger(__name__)

_a       =							{'vocab_file': 'spiece.model'}

_a       =							{
    'vocab_file': {
        'xlnet-base-cased': 'https://huggingface.co/xlnet-base-cased/resolve/main/spiece.model',
        'xlnet-large-cased': 'https://huggingface.co/xlnet-large-cased/resolve/main/spiece.model',
    }
}

_a       =							{
    'xlnet-base-cased': None,
    'xlnet-large-cased': None,
}

# Segments (not really needed)
_a       =							0
_a       =							1
_a       =							2
_a       =							3
_a       =							4




class 					_lowerCAmelCase  (	lowercase   ):







 """simple docstring"""







 __UpperCAmelCase	:					Any   		=	VOCAB_FILES_NAMES
 __UpperCAmelCase	:					Union[str, Any]   		=	PRETRAINED_VOCAB_FILES_MAP
 __UpperCAmelCase	:					Dict   		=	PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 __UpperCAmelCase	:					List[Any]   		=	"left"


 def __init__(			self							:			Dict,					UpperCAmelCase__							:			List[str],					UpperCAmelCase__							:			Optional[int]=False,					UpperCAmelCase__							:			Union[str, Any]=True,					UpperCAmelCase__							:			Dict=False,					UpperCAmelCase__							:			Dict="<s>",					UpperCAmelCase__							:			int="</s>",					UpperCAmelCase__							:			List[Any]="<unk>",					UpperCAmelCase__							:			List[Any]="<sep>",					UpperCAmelCase__							:			List[Any]="<pad>",					UpperCAmelCase__							:			List[Any]="<cls>",					UpperCAmelCase__							:			Any="<mask>",					UpperCAmelCase__							:			str=["<eop>", "<eod>"],					UpperCAmelCase__							:			Optional[Dict[str, Any]] = None,					**UpperCAmelCase__							:			int,					):
    # Mask token behave like a normal word, i.e. include the space before it
    __lowercase								=						AddedToken(UpperCAmelCase__,					lstrip=UpperCAmelCase__,					rstrip=UpperCAmelCase__	) if isinstance(UpperCAmelCase__,					UpperCAmelCase__	) else mask_token

    __lowercase								=						{} if sp_model_kwargs is None else sp_model_kwargs

    super().__init__(
        do_lower_case=UpperCAmelCase__,					remove_space=UpperCAmelCase__,					keep_accents=UpperCAmelCase__,					bos_token=UpperCAmelCase__,					eos_token=UpperCAmelCase__,					unk_token=UpperCAmelCase__,					sep_token=UpperCAmelCase__,					pad_token=UpperCAmelCase__,					cls_token=UpperCAmelCase__,					mask_token=UpperCAmelCase__,					additional_special_tokens=UpperCAmelCase__,					sp_model_kwargs=self.sp_model_kwargs,					**UpperCAmelCase__,					)

    __lowercase								=						3

    __lowercase								=						do_lower_case
    __lowercase								=						remove_space
    __lowercase								=						keep_accents
    __lowercase								=						vocab_file

    __lowercase								=						spm.SentencePieceProcessor(**self.sp_model_kwargs	)
    self.sp_model.Load(UpperCAmelCase__	)


 @property
 def 					_lowercase (			self							:			Union[str, Any]	):
    return len(self.sp_model	)


 def 					_lowercase (			self							:			int	):
    __lowercase								=						{self.convert_ids_to_tokens(UpperCAmelCase__	): i for i in range(self.vocab_size	)}
    vocab.update(self.added_tokens_encoder	)
    return vocab


 def __getstate__(			self							:			str	):
    __lowercase								=						self.__dict__.copy()
    __lowercase								=						None
    return state


 def __setstate__(			self							:			Tuple,					UpperCAmelCase__							:			str	):
    __lowercase								=						d

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

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


 def 					_lowercase (			self							:			Optional[Any],					UpperCAmelCase__							:			Optional[int]	):
    if self.remove_space:
       __lowercase								=						" ".join(inputs.strip().split()	)
    else:
       __lowercase								=						inputs
    __lowercase								=						outputs.replace("``",					"\""	).replace("''",					"\""	)

    if not self.keep_accents:
       __lowercase								=						unicodedata.normalize("NFKD",					UpperCAmelCase__	)
       __lowercase								=						"".join([c for c in outputs if not unicodedata.combining(UpperCAmelCase__	)]	)
    if self.do_lower_case:
       __lowercase								=						outputs.lower()

    return outputs


 def 					_lowercase (			self							:			Optional[int],					UpperCAmelCase__							:			str	):
    __lowercase								=						self.preprocess_text(UpperCAmelCase__	)
    __lowercase								=						self.sp_model.encode(UpperCAmelCase__,					out_type=UpperCAmelCase__	)
    __lowercase								=						[]
    for piece in pieces:
       if len(UpperCAmelCase__	) > 1 and piece[-1] == str(","	) and piece[-2].isdigit():
          __lowercase								=						self.sp_model.EncodeAsPieces(piece[:-1].replace(UpperCAmelCase__,					""	)	)
          if piece[0] != SPIECE_UNDERLINE and cur_pieces[0][0] == SPIECE_UNDERLINE:
             if len(cur_pieces[0]	) == 1:
                __lowercase								=						cur_pieces[1:]
             else:
                __lowercase								=						cur_pieces[0][1:]
          cur_pieces.append(piece[-1]	)
          new_pieces.extend(UpperCAmelCase__	)
       else:
          new_pieces.append(UpperCAmelCase__	)

    return new_pieces


 def 					_lowercase (			self							:			Dict,					UpperCAmelCase__							:			Any	):
    return self.sp_model.PieceToId(UpperCAmelCase__	)


 def 					_lowercase (			self							:			List[Any],					UpperCAmelCase__							:			Any	):
    return self.sp_model.IdToPiece(UpperCAmelCase__	)


 def 					_lowercase (			self							:			Optional[int],					UpperCAmelCase__							:			Optional[int]	):
    __lowercase								=						"".join(UpperCAmelCase__	).replace(UpperCAmelCase__,					" "	).strip()
    return out_string


 def 					_lowercase (			self							:			Union[str, Any],					UpperCAmelCase__							:			List[int],					UpperCAmelCase__							:			bool = False,					UpperCAmelCase__							:			bool = None,					UpperCAmelCase__							:			bool = True,					**UpperCAmelCase__							:			Union[str, Any],					):
    __lowercase								=						kwargs.pop("use_source_tokenizer",					UpperCAmelCase__	)

    __lowercase								=						self.convert_ids_to_tokens(UpperCAmelCase__,					skip_special_tokens=UpperCAmelCase__	)

    # To avoid mixing byte-level and unicode for byte-level BPT
    # we need to build string separately for added tokens and byte-level tokens
    # cf. https://github.com/huggingface/transformers/issues/1133
    __lowercase								=						[]
    __lowercase								=						[]
    for token in filtered_tokens:
       if skip_special_tokens and token in self.all_special_ids:
          continue
       if token in self.added_tokens_encoder:
          if current_sub_text:
             sub_texts.append(self.convert_tokens_to_string(UpperCAmelCase__	)	)
             __lowercase								=						[]
          sub_texts.append(UpperCAmelCase__	)
       else:
          current_sub_text.append(UpperCAmelCase__	)
    if current_sub_text:
       sub_texts.append(self.convert_tokens_to_string(UpperCAmelCase__	)	)

    # Mimic the behavior of the Rust tokenizer:
    # By default, there are no spaces between special tokens
    __lowercase								=						"".join(UpperCAmelCase__	)

    __lowercase								=						(
        clean_up_tokenization_spaces
        if clean_up_tokenization_spaces is not None
        else self.clean_up_tokenization_spaces
    )
    if clean_up_tokenization_spaces:
       __lowercase								=						self.clean_up_tokenization(UpperCAmelCase__	)
       return clean_text
    else:
       return text


 def 					_lowercase (			self							:			Any,					UpperCAmelCase__							:			List[int],					UpperCAmelCase__							:			Optional[List[int]] = None	):
    __lowercase								=						[self.sep_token_id]
    __lowercase								=						[self.cls_token_id]
    if token_ids_a is None:
       return token_ids_a + sep + cls
    return token_ids_a + sep + token_ids_a + sep + cls


 def 					_lowercase (			self							:			Dict,					UpperCAmelCase__							:			List[int],					UpperCAmelCase__							:			Optional[List[int]] = None,					UpperCAmelCase__							:			bool = False	):

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

    if token_ids_a is not None:
       return ([0] * len(UpperCAmelCase__	)) + [1] + ([0] * len(UpperCAmelCase__	)) + [1, 1]
    return ([0] * len(UpperCAmelCase__	)) + [1, 1]


 def 					_lowercase (			self							:			Optional[Any],					UpperCAmelCase__							:			List[int],					UpperCAmelCase__							:			Optional[List[int]] = None	):
    __lowercase								=						[self.sep_token_id]
    __lowercase								=						[2]

    if token_ids_a is None:
       return len(token_ids_a + sep	) * [0] + cls_segment_id
    return len(token_ids_a + sep	) * [0] + len(token_ids_a + sep	) * [1] + cls_segment_id



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

    return (out_vocab_file,)







"""simple docstring"""





import json
import os
import subprocess
import unittest
from ast import literal_eval

import pytest
from parameterized import parameterized, parameterized_class

from . import is_sagemaker_available


if is_sagemaker_available():
   from sagemaker import Session, TrainingJobAnalytics
   from sagemaker.huggingface import HuggingFace




@pytest.mark.skipif(
    literal_eval(os.getenv("TEST_SAGEMAKER"     ,"False"   )   ) is not True     ,reason="Skipping test because should only be run when releasing minor transformers version"     ,)
@pytest.mark.usefixtures("sm_env"   )
@parameterized_class(
    [
        {
            "framework": "pytorch",
            "script": "run_glue_model_parallelism.py",
            "model_name_or_path": "roberta-large",
            "instance_type": "ml.p3dn.24xlarge",
            "results": {"train_runtime": 1_6_0_0, "eval_accuracy": 0.3, "eval_loss": 1.2},
        },
        {
            "framework": "pytorch",
            "script": "run_glue.py",
            "model_name_or_path": "roberta-large",
            "instance_type": "ml.p3dn.24xlarge",
            "results": {"train_runtime": 1_6_0_0, "eval_accuracy": 0.3, "eval_loss": 1.2},
        },
    ]   )
class 					_lowerCAmelCase  (	unittest.TestCase   ):







 """simple docstring"""







 def 					_lowercase (			self							:			Optional[int]	):
    if self.framework == "pytorch":
       subprocess.run(
           F"""cp ./examples/pytorch/text-classification/run_glue.py {self.env.test_path}/run_glue.py""".split(),					encoding="utf-8",					check=UpperCAmelCase__,					)
    assert hasattr(self,					"env"	)


 def 					_lowercase (			self							:			str,					UpperCAmelCase__							:			List[Any]	):
    # configuration for running training on smdistributed Model Parallel
    __lowercase								=						{
        "enabled": True,
        "processes_per_host": 8,
    }
    __lowercase								=						{
        "enabled": True,
        "parameters": {
            "microbatches": 4,
            "placement_strategy": "spread",
            "pipeline": "interleaved",
            "optimize": "speed",
            "partitions": 4,
            "ddp": True,
        },
    }

    __lowercase								=						{"smdistributed": {"modelparallel": smp_options}, "mpi": mpi_options}

    __lowercase								=						"trainer" if self.script == "run_glue.py" else "smtrainer"
    # creates estimator
    return HuggingFace(
        entry_point=self.script,					source_dir=self.env.test_path,					role=self.env.role,					image_uri=self.env.image_uri,					base_job_name=F"""{self.env.base_job_name}-{instance_count}-smp-{name_extension}""",					instance_count=UpperCAmelCase__,					instance_type=self.instance_type,					debugger_hook_config=UpperCAmelCase__,					hyperparameters={
            **self.env.hyperparameters,
            "model_name_or_path": self.model_name_or_path,
            "max_steps": 5_0_0,
        },					metric_definitions=self.env.metric_definitions,					distribution=UpperCAmelCase__,					py_version="py36",					)


 def 					_lowercase (			self							:			Tuple,					UpperCAmelCase__							:			int	):
    TrainingJobAnalytics(UpperCAmelCase__	).export_csv(F"""{self.env.test_path}/{job_name}_metrics.csv"""	)



 @parameterized.expand([(1,)]	)
 def 					_lowercase (			self							:			str,					UpperCAmelCase__							:			Union[str, Any]	):
    # create estimator
    __lowercase								=						self.create_estimator(UpperCAmelCase__	)

    # run training
    estimator.fit()

    # result dataframe
    __lowercase								=						TrainingJobAnalytics(estimator.latest_training_job.name	).dataframe()

    # extract kpis
    __lowercase								=						list(result_metrics_df[result_metrics_df.metric_name == "eval_accuracy"]["value"]	)
    __lowercase								=						list(result_metrics_df[result_metrics_df.metric_name == "eval_loss"]["value"]	)
    # get train time from SageMaker job, this includes starting, preprocessing, stopping
    __lowercase								=						(
        Session().describe_training_job(estimator.latest_training_job.name	).get("TrainingTimeInSeconds",					9_9_9_9_9_9	)
    )

    # assert kpis
    assert train_runtime <= self.results["train_runtime"]
    assert all(t >= self.results["eval_accuracy"] for t in eval_accuracy	)
    assert all(t <= self.results["eval_loss"] for t in eval_loss	)

    # dump tests result into json file to share in PR
    with open(F"""{estimator.latest_training_job.name}.json""",					"w"	) as outfile:
       json.dump({"train_time": train_runtime, "eval_accuracy": eval_accuracy, "eval_loss": eval_loss},					UpperCAmelCase__	)




