

"""simple docstring"""




import ast
import os
import re
import shutil
import tempfile
import unittest
from unittest import mock

import torch

from accelerate.test_utils.examples import compare_against_test
from accelerate.test_utils.testing import TempDirTestCase, require_trackers, run_command, slow
from accelerate.utils import write_basic_config


# DataLoaders built from `test_samples/MRPC` for quick testing
# Should mock `{script_name}.get_dataloaders` via:
# @mock.patch("{script_name}.get_dataloaders", mocked_dataloaders)

A_							:			int											= [
    "cross_validation.py",
    "gradient_accumulation.py",
    "local_sgd.py",
    "multi_process_metrics.py",
    "memory.py",
    "automatic_gradient_accumulation.py",
    "fsdp_with_peak_mem_tracking.py",
    "deepspeed_with_config_support.py",
    "megatron_lm_gpt_pretraining.py",
]





class    lowerCamelCase						(unittest.TestCase			):

			def 					SCREAMING_SNAKE_CASE				(					self				:						Any    ,				__UpperCAmelCase				:						str    ,				__UpperCAmelCase				:						bool    ,				__UpperCAmelCase				:						str = None    ,				__UpperCAmelCase				:						list = None       )   ->    Optional[int]:
						SCREAMING_SNAKE_CASE__								=      None
						SCREAMING_SNAKE_CASE__								=      os.path.abspath(os.path.join("""examples"""    ,				"""by_feature"""       )       )
						SCREAMING_SNAKE_CASE__								=      os.path.abspath("""examples"""       )
						for item in os.listdir(__snake_case       ):
									if item not in EXCLUDE_EXAMPLES:
												SCREAMING_SNAKE_CASE__								=      os.path.join(__snake_case    ,				__snake_case       )
												if os.path.isfile(__snake_case       ) and ".py" in item_path:
															with self.subTest(
															    tested_script=__snake_case    ,				feature_script=__snake_case    ,				tested_section="""main()""" if parser_only else """training_function()"""    ,				):
																		SCREAMING_SNAKE_CASE__								=      compare_against_test(
																		    os.path.join(__snake_case    ,				__snake_case       )    ,				__snake_case    ,				__snake_case    ,				__snake_case       )
																		SCREAMING_SNAKE_CASE__								=      '''\n'''.join(__snake_case       )
																		if special_strings is not None:
																					for string in special_strings:
																								SCREAMING_SNAKE_CASE__								=      diff.replace(__snake_case    ,				""""""       )
																		self.assertEqual(__snake_case    ,				""""""       )

			def 					SCREAMING_SNAKE_CASE				(					self				:						Dict       )   ->    List[Any]:
						self.one_complete_example("""complete_nlp_example.py"""    ,				__snake_case       )
						self.one_complete_example("""complete_nlp_example.py"""    ,				__snake_case       )





			def 					SCREAMING_SNAKE_CASE				(					self				:						Optional[Any]       )   ->    Optional[Any]:
						SCREAMING_SNAKE_CASE__								=      os.path.abspath(os.path.join("""examples"""    ,				"""cv_example.py"""       )       )
						SCREAMING_SNAKE_CASE__								=      [
						    ''' ''' * 1_6 + '''{\n\n''',
						    ''' ''' * 2_0 + '''"accuracy": eval_metric["accuracy"],\n\n''',
						    ''' ''' * 2_0 + '''"f1": eval_metric["f1"],\n\n''',
						    ''' ''' * 2_0 + '''"train_loss": total_loss.item() / len(train_dataloader),\n\n''',
						    ''' ''' * 2_0 + '''"epoch": epoch,\n\n''',
						    ''' ''' * 1_6 + '''},\n\n''',
						    ''' ''' * 1_6 + '''step=epoch,\n''',
						    ''' ''' * 1_2,
						    ''' ''' * 8 + '''for step, batch in enumerate(active_dataloader):\n''',
						]
						self.one_complete_example("""complete_cv_example.py"""    ,				__snake_case    ,				__snake_case    ,				__snake_case       )
						self.one_complete_example("""complete_cv_example.py"""    ,				__snake_case    ,				__snake_case    ,				__snake_case       )





@mock.patch.dict(os.environ			,{'TESTING_MOCKED_DATALOADERS': '1'}			)
class    lowerCamelCase						(__SCREAMING_SNAKE_CASE			):
			lowerCamelCase__   :		Union[str, Any]							=							False

			@classmethod
			def 					SCREAMING_SNAKE_CASE				(					cls				:						Optional[Any]       )   ->    Dict:
						super().setUpClass()
						SCREAMING_SNAKE_CASE__								=      tempfile.mkdtemp()
						SCREAMING_SNAKE_CASE__								=      os.path.join(cls._tmpdir    ,				"""default_config.yml"""       )

						write_basic_config(save_location=cls.configPath       )
						SCREAMING_SNAKE_CASE__								=      ['''accelerate''', '''launch''', '''--config_file''', cls.configPath]

			@classmethod
			def 					SCREAMING_SNAKE_CASE				(					cls				:						Dict       )   ->    List[Any]:
						super().tearDownClass()
						shutil.rmtree(cls._tmpdir       )

			def 					SCREAMING_SNAKE_CASE				(					self				:						int       )   ->    Dict:
						SCREAMING_SNAKE_CASE__								=      F"""\n        examples/by_feature/checkpointing.py\n        --checkpointing_steps epoch\n        --output_dir {self.tmpdir}\n        """.split()
						run_command(self._launch_args + testargs       )
						self.assertTrue(os.path.exists(os.path.join(self.tmpdir    ,				"""epoch_0"""       )       )       )

			def 					SCREAMING_SNAKE_CASE				(					self				:						List[Any]       )   ->    Optional[int]:
						SCREAMING_SNAKE_CASE__								=      F"""\n        examples/by_feature/checkpointing.py\n        --checkpointing_steps 1\n        --output_dir {self.tmpdir}\n        """.split()
						SCREAMING_SNAKE_CASE__								=      run_command(self._launch_args + testargs       )
						self.assertTrue(os.path.exists(os.path.join(self.tmpdir    ,				"""step_2"""       )       )       )

			def 					SCREAMING_SNAKE_CASE				(					self				:						List[str]       )   ->    Union[str, Any]:
						SCREAMING_SNAKE_CASE__								=      F"""\n        examples/by_feature/checkpointing.py\n        --resume_from_checkpoint {os.path.join(self.tmpdir    ,				"epoch_0"       )}\n        """.split()
						SCREAMING_SNAKE_CASE__								=      run_command(self._launch_args + testargs    ,				return_stdout=__snake_case       )
						self.assertNotIn("""epoch 0:"""    ,				__snake_case       )
						self.assertIn("""epoch 1:"""    ,				__snake_case       )

			def 					SCREAMING_SNAKE_CASE				(					self				:						str       )   ->    Optional[Any]:
						SCREAMING_SNAKE_CASE__								=      F"""\n        examples/by_feature/checkpointing.py\n        --resume_from_checkpoint {os.path.join(self.tmpdir    ,				"step_2"       )}\n        """.split()
						SCREAMING_SNAKE_CASE__								=      run_command(self._launch_args + testargs    ,				return_stdout=__snake_case       )
						if torch.cuda.is_available():
									SCREAMING_SNAKE_CASE__								=      torch.cuda.device_count()
						else:
									SCREAMING_SNAKE_CASE__								=      1
						if num_processes > 1:
									self.assertNotIn("""epoch 0:"""    ,				__snake_case       )
									self.assertIn("""epoch 1:"""    ,				__snake_case       )
						else:
									self.assertIn("""epoch 0:"""    ,				__snake_case       )
									self.assertIn("""epoch 1:"""    ,				__snake_case       )

			@slow
			def 					SCREAMING_SNAKE_CASE				(					self				:						Any       )   ->    int:
						SCREAMING_SNAKE_CASE__								=      '''
        examples/by_feature/cross_validation.py
        --num_folds 2
        '''.split()
						with mock.patch.dict(os.environ    ,				{"""TESTING_MOCKED_DATALOADERS""": """0"""}       ):
									SCREAMING_SNAKE_CASE__								=      run_command(self._launch_args + testargs    ,				return_stdout=__snake_case       )
									SCREAMING_SNAKE_CASE__								=      re.findall("""({.+})"""    ,				__snake_case       )
									SCREAMING_SNAKE_CASE__								=      [r for r in results if '''accuracy''' in r][-1]
									SCREAMING_SNAKE_CASE__								=      ast.literal_eval(__snake_case       )
									self.assertGreaterEqual(results["""accuracy"""]    ,				0.75       )

			def 					SCREAMING_SNAKE_CASE				(					self				:						int       )   ->    Any:
						SCREAMING_SNAKE_CASE__								=      ['''examples/by_feature/multi_process_metrics.py''']
						run_command(self._launch_args + testargs       )

			@require_trackers
			@mock.patch.dict(os.environ    ,				{"""WANDB_MODE""": """offline"""}       )
			def 					SCREAMING_SNAKE_CASE				(					self				:						Any       )   ->    str:
						with tempfile.TemporaryDirectory() as tmpdir:
									SCREAMING_SNAKE_CASE__								=      F"""\n            examples/by_feature/tracking.py\n            --with_tracking\n            --project_dir {tmpdir}\n            """.split()
									run_command(self._launch_args + testargs       )
									self.assertTrue(os.path.exists(os.path.join(__snake_case    ,				"""tracking"""       )       )       )

			def 					SCREAMING_SNAKE_CASE				(					self				:						Tuple       )   ->    Tuple:
						SCREAMING_SNAKE_CASE__								=      ['''examples/by_feature/gradient_accumulation.py''']
						run_command(self._launch_args + testargs       )





			def 					SCREAMING_SNAKE_CASE				(					self				:						Optional[Any]       )   ->    Any:
						SCREAMING_SNAKE_CASE__								=      ['''examples/by_feature/local_sgd.py''']
						run_command(self._launch_args + testargs       )





"""simple docstring"""




import os
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 logging


A_							:			Optional[int]											= logging.get_logger(__name__)

A_							:			Union[str, Any]											= {"vocab_file": "sentencepiece.bpe.model"}

A_							:			str											= {
    "vocab_file": {
        "camembert-base": "https://huggingface.co/camembert-base/resolve/main/sentencepiece.bpe.model",
    }
}

A_							:			Tuple											= {
    "camembert-base": 512,
}

A_							:			int											= "▁"





class    lowerCamelCase						(A__			):
			lowerCamelCase__   :		int							=							VOCAB_FILES_NAMES
			lowerCamelCase__   :		List[Any]							=							PRETRAINED_VOCAB_FILES_MAP
			lowerCamelCase__   :		List[str]							=							PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
			lowerCamelCase__   :		Optional[int]							=							['input_ids', 'attention_mask']

			def __init__(					self				:						Tuple    ,				__UpperCAmelCase				:						Optional[Any]    ,				__UpperCAmelCase				:						int="<s>"    ,				__UpperCAmelCase				:						Union[str, Any]="</s>"    ,				__UpperCAmelCase				:						List[str]="</s>"    ,				__UpperCAmelCase				:						List[Any]="<s>"    ,				__UpperCAmelCase				:						int="<unk>"    ,				__UpperCAmelCase				:						Dict="<pad>"    ,				__UpperCAmelCase				:						int="<mask>"    ,				__UpperCAmelCase				:						List[Any]=["<s>NOTUSED", "</s>NOTUSED"]    ,				__UpperCAmelCase				:						Optional[Dict[str, Any]] = None    ,				**__UpperCAmelCase				:						Optional[Any]    ,				)   ->    None:
						# Mask token behave like a normal word, i.e. include the space before it
						SCREAMING_SNAKE_CASE__								=      AddedToken(__UpperCAmelCase    ,				lstrip=__UpperCAmelCase    ,				rstrip=__UpperCAmelCase       ) if isinstance(__UpperCAmelCase    ,				__UpperCAmelCase       ) else mask_token

						SCREAMING_SNAKE_CASE__								=      {} if sp_model_kwargs is None else sp_model_kwargs

						super().__init__(
						    bos_token=__UpperCAmelCase    ,				eos_token=__UpperCAmelCase    ,				unk_token=__UpperCAmelCase    ,				sep_token=__UpperCAmelCase    ,				cls_token=__UpperCAmelCase    ,				pad_token=__UpperCAmelCase    ,				mask_token=__UpperCAmelCase    ,				additional_special_tokens=__UpperCAmelCase    ,				sp_model_kwargs=self.sp_model_kwargs    ,				**__UpperCAmelCase    ,				)
						SCREAMING_SNAKE_CASE__								=      spm.SentencePieceProcessor(**self.sp_model_kwargs       )
						self.sp_model.Load(str(__UpperCAmelCase       )       )
						SCREAMING_SNAKE_CASE__								=      vocab_file
						# HACK: These tokens were added by fairseq but don't seem to be actually used when duplicated in the actual
						# sentencepiece vocabulary (this is the case for <s> and </s>
						SCREAMING_SNAKE_CASE__								=      {"""<s>NOTUSED""": 0, """<pad>""": 1, """</s>NOTUSED""": 2, """<unk>""": 3}
						SCREAMING_SNAKE_CASE__								=      len(self.fairseq_tokens_to_ids       )
						SCREAMING_SNAKE_CASE__								=      len(self.sp_model       ) + len(self.fairseq_tokens_to_ids       )
						SCREAMING_SNAKE_CASE__								=      {v: k for k, v in self.fairseq_tokens_to_ids.items()}

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

						if token_ids_a is None:
									return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
						SCREAMING_SNAKE_CASE__								=      [self.cls_token_id]
						SCREAMING_SNAKE_CASE__								=      [self.sep_token_id]
						return cls + token_ids_a + sep + sep + token_ids_a + sep

			def 					SCREAMING_SNAKE_CASE				(					self				:						Optional[int]    ,				__UpperCAmelCase				:						List[int]    ,				__UpperCAmelCase				:						Optional[List[int]] = None    ,				__UpperCAmelCase				:						bool = False       )   ->    List[int]:
						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 None:
									return [1] + ([0] * len(__UpperCAmelCase       )) + [1]
						return [1] + ([0] * len(__UpperCAmelCase       )) + [1, 1] + ([0] * len(__UpperCAmelCase       )) + [1]

			def 					SCREAMING_SNAKE_CASE				(					self				:						Tuple    ,				__UpperCAmelCase				:						List[int]    ,				__UpperCAmelCase				:						Optional[List[int]] = None       )   ->    List[int]:
						SCREAMING_SNAKE_CASE__								=      [self.sep_token_id]
						SCREAMING_SNAKE_CASE__								=      [self.cls_token_id]

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

			@property
			def 					SCREAMING_SNAKE_CASE				(					self				:						Any       )   ->    Any:
						return len(self.fairseq_tokens_to_ids       ) + len(self.sp_model       )

			def 					SCREAMING_SNAKE_CASE				(					self				:						int       )   ->    Optional[Any]:
						SCREAMING_SNAKE_CASE__								=      {self.convert_ids_to_tokens(__UpperCAmelCase       ): i for i in range(self.vocab_size       )}
						vocab.update(self.added_tokens_encoder       )
						return vocab

			def 					SCREAMING_SNAKE_CASE				(					self				:						Optional[int]    ,				__UpperCAmelCase				:						str       )   ->    List[str]:
						return self.sp_model.encode(__UpperCAmelCase    ,				out_type=__UpperCAmelCase       )

			def 					SCREAMING_SNAKE_CASE				(					self				:						Tuple    ,				__UpperCAmelCase				:						Tuple       )   ->    Optional[Any]:
						if token in self.fairseq_tokens_to_ids:
									return self.fairseq_tokens_to_ids[token]
						elif self.sp_model.PieceToId(__UpperCAmelCase       ) == 0:
									# Convert sentence piece unk token to fairseq unk token index
									return self.unk_token_id
						return self.fairseq_offset + self.sp_model.PieceToId(__UpperCAmelCase       )

			def 					SCREAMING_SNAKE_CASE				(					self				:						Optional[Any]    ,				__UpperCAmelCase				:						int       )   ->    Dict:
						if index in self.fairseq_ids_to_tokens:
									return self.fairseq_ids_to_tokens[index]
						return self.sp_model.IdToPiece(index - self.fairseq_offset       )

			def 					SCREAMING_SNAKE_CASE				(					self				:						Optional[int]    ,				__UpperCAmelCase				:						List[str]       )   ->    Union[str, Any]:
						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:
												if not prev_is_special:
															out_string += " "
												out_string += self.sp_model.decode(__UpperCAmelCase       ) + token
												SCREAMING_SNAKE_CASE__								=      True
												SCREAMING_SNAKE_CASE__								=      []
									else:
												current_sub_tokens.append(__UpperCAmelCase       )
												SCREAMING_SNAKE_CASE__								=      False
						out_string += self.sp_model.decode(__UpperCAmelCase       )
						return out_string.strip()

			def __getstate__(					self				:						Optional[Any]       )   ->    Any:
						SCREAMING_SNAKE_CASE__								=      self.__dict__.copy()
						SCREAMING_SNAKE_CASE__								=      None
						return state

			def __setstate__(					self				:						int    ,				__UpperCAmelCase				:						Optional[int]       )   ->    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       )





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

						return (out_vocab_file,)


