
'''simple docstring'''





import os
import sys
import tempfile

import torch

from .state import AcceleratorState
from .utils import PrecisionType, PrepareForLaunch, is_mps_available, patch_environment

def      A (       A_							:					List[str]    ,							A_							:					List[Any]=()    ,							A_							:					List[Any]=None    ,							A_							:					List[str]="no"    ,							A_							:					Union[str, Any]="29500"       ):
						snake_case       :     List[Any] =    False
						snake_case       :     str =    False
						if any(key.startswith('''KAGGLE'''       ) for key in os.environ.keys()       ):
												snake_case       :     Union[str, Any] =    True
						elif "IPython" in sys.modules:
												snake_case       :     Optional[Any] =    '''google.colab''' in str(sys.modules['''IPython'''].get_ipython()       )

						try:
												snake_case       :     Dict =    PrecisionType(mixed_precision.lower()       )
						except ValueError:
												raise ValueError(
												    F"""Unknown mixed_precision mode: {args.mixed_precision.lower()}. Choose between {PrecisionType.list()}."""       )

						if (in_colab or in_kaggle) and (os.environ.get('''TPU_NAME'''    ,							__A       ) is not None):
												# TPU launch
												import torch_xla.distributed.xla_multiprocessing as xmp

												if len(AcceleratorState._shared_state       ) > 0:
																		raise ValueError(
																		    '''To train on TPU in Colab or Kaggle Kernel, the `Accelerator` should only be initialized inside '''
																		    '''your training function. Restart your notebook and make sure no cells initializes an '''
																		    '''`Accelerator`.'''       )
												if num_processes is None:
																		snake_case       :     Dict =    8

												snake_case       :     Optional[int] =    PrepareForLaunch(__A    ,							distributed_type='''TPU'''       )
												print(F"""Launching a training on {num_processes} TPU cores."""       )
												xmp.spawn(__A    ,							args=__A    ,							nprocs=__A    ,							start_method='''fork'''       )
						elif in_colab:
												# No need for a distributed launch otherwise as it's either CPU or one GPU.
												if torch.cuda.is_available():
																		print('''Launching training on one GPU.'''       )
												else:
																		print('''Launching training on one CPU.'''       )
												function(*__A       )
						else:
												if num_processes is None:
																		raise ValueError(
																		    '''You have to specify the number of GPUs you would like to use, add `num_processes=...` to your call.'''       )

												if num_processes > 1:
																		# Multi-GPU launch
																		from torch.multiprocessing import start_processes
																		from torch.multiprocessing.spawn import ProcessRaisedException

																		if len(AcceleratorState._shared_state       ) > 0:
																								raise ValueError(
																								    '''To launch a multi-GPU training from your notebook, the `Accelerator` should only be initialized '''
																								    '''inside your training function. Restart your notebook and make sure no cells initializes an '''
																								    '''`Accelerator`.'''       )

																		if torch.cuda.is_initialized():
																								raise ValueError(
																								    '''To launch a multi-GPU training from your notebook, you need to avoid running any instruction '''
																								    '''using `torch.cuda` in any cell. Restart your notebook and make sure no cells use any CUDA '''
																								    '''function.'''       )

																		# torch.distributed will expect a few environment variable to be here. We set the ones common to each
																		# process here (the other ones will be set be the launcher).
																		with patch_environment(
																		    world_size=__A    ,							master_addr='''127.0.01'''    ,							master_port=__A    ,							mixed_precision=__A       ):
																								snake_case       :     Tuple =    PrepareForLaunch(__A    ,							distributed_type='''MULTI_GPU'''       )
																								print(F"""Launching training on {num_processes} GPUs."""       )
																								try:
																														start_processes(__A    ,							args=__A    ,							nprocs=__A    ,							start_method='''fork'''       )
																								except ProcessRaisedException as e:
																														if "Cannot re-initialize CUDA in forked subprocess" in e.args[0]:
																																				raise RuntimeError(
																																				    '''CUDA has been initialized before the `notebook_launcher` could create a forked subprocess. '''
																																				    '''This likely stems from an outside import causing issues once the `notebook_launcher()` is called. '''
																																				    '''Please review your imports and test them when running the `notebook_launcher()` to identify '''
																																				    '''which one is problematic.'''       ) from e

												else:
																		# No need for a distributed launch otherwise as it's either CPU, GPU or MPS.
																		if is_mps_available():
																								snake_case       :     Tuple =    '''1'''
																								print('''Launching training on MPS.'''       )
																		elif torch.cuda.is_available():
																								print('''Launching training on one GPU.'''       )
																		else:
																								print('''Launching training on CPU.'''       )
																		function(*__A       )





def      A (       A_							:					str    ,							A_							:					Optional[Any]=()    ,							A_							:					Tuple=2       ):
						from torch.multiprocessing import start_processes

						with tempfile.NamedTemporaryFile() as tmp_file:
												# torch.distributed will expect a few environment variable to be here. We set the ones common to each
												# process here (the other ones will be set be the launcher).
												with patch_environment(
												    world_size=__A    ,							master_addr='''127.0.01'''    ,							master_port='''29500'''    ,							accelerate_mixed_precision='''no'''    ,							accelerate_debug_rdv_file=tmp_file.name    ,							accelerate_use_cpu='''yes'''    ,							):
																		snake_case       :     Optional[Any] =    PrepareForLaunch(__A    ,							debug=__A       )
																		start_processes(__A    ,							args=__A    ,							nprocs=__A    ,							start_method='''fork'''       )



'''simple docstring'''





import json
import re
from typing import TYPE_CHECKING, List, Optional, Tuple, Union

import numpy as np

from ...utils import is_tf_available, is_torch_available, logging


if TYPE_CHECKING:
						if is_torch_available():
												import torch
						if is_tf_available():
												import tensorflow as tf

from tokenizers import pre_tokenizers

from ...tokenization_utils_base import BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from .tokenization_codegen import CodeGenTokenizer


UpperCAmelCase    					=   logging.get_logger(__name__)

UpperCAmelCase    					=   {"vocab_file": "vocab.json", "merges_file": "merges.txt", "tokenizer_file": "tokenizer.json"}

UpperCAmelCase    					=   {
    "vocab_file": {
        "Salesforce/codegen-350M-mono": "https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/vocab.json",
    },
    "merges_file": {
        "Salesforce/codegen-350M-mono": "https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/merges.txt",
    },
    "tokenizer_file": {
        "Salesforce/codegen-350M-mono": (
            "https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/tokenizer.json"
        ),
    },
}

UpperCAmelCase    					=   {
    "Salesforce/codegen-350M-mono": 2_048,
}

class 						a   (				__magic_name__    ):
				_snake_case		    = VOCAB_FILES_NAMES
				_snake_case		    = PRETRAINED_VOCAB_FILES_MAP
				_snake_case		    = PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				_snake_case		    = ['''input_ids''', '''attention_mask''']
				_snake_case		    = CodeGenTokenizer




				def __init__(							self	:				Any,    SCREAMING_SNAKE_CASE_	:				Dict=None,    SCREAMING_SNAKE_CASE_	:				Optional[Any]=None,    SCREAMING_SNAKE_CASE_	:				Optional[Any]=None,    SCREAMING_SNAKE_CASE_	:				Optional[int]="<|endoftext|>",    SCREAMING_SNAKE_CASE_	:				List[Any]="<|endoftext|>",    SCREAMING_SNAKE_CASE_	:				List[str]="<|endoftext|>",    SCREAMING_SNAKE_CASE_	:				int=False,    **SCREAMING_SNAKE_CASE_	:				Optional[int],    ):
										super().__init__(
										    SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_,    tokenizer_file=SCREAMING_SNAKE_CASE_,    unk_token=SCREAMING_SNAKE_CASE_,    bos_token=SCREAMING_SNAKE_CASE_,    eos_token=SCREAMING_SNAKE_CASE_,    add_prefix_space=SCREAMING_SNAKE_CASE_,    **SCREAMING_SNAKE_CASE_,    )

										if kwargs.pop('''add_bos_token''',    SCREAMING_SNAKE_CASE_      ):
																snake_case       :     Dict =    kwargs.pop('''name_or_path''',    ''''''      )
																raise ValueError(
																    '''Currenty GPT2\'s fast tokenizer does NOT support adding a BOS token.'''
																    '''Instead you should use GPT2\'s slow tokenizer class `CodeGenTokenizer` as follows: \n'''
																    F"""`CodeGenTokenizer.from_pretrained('{model_id}')`\nor\n"""
																    F"""`AutoTokenizer.from_pretrained('{model_id}', use_fast=False)`\n"""
																    '''This issue will be fixed soon, see: https://github.com/huggingface/tokenizers/pull/1005.'''
																    ''' so that the fast tokenizer works correctly.'''      )

										snake_case       :     int =    json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__()      )
										if pre_tok_state.get('''add_prefix_space''',    SCREAMING_SNAKE_CASE_      ) != add_prefix_space:
																snake_case       :     Optional[Any] =    getattr(SCREAMING_SNAKE_CASE_,    pre_tok_state.pop('''type'''      )      )
																snake_case       :     Union[str, Any] =    add_prefix_space
																snake_case       :     Optional[Any] =    pre_tok_class(**SCREAMING_SNAKE_CASE_      )

										snake_case       :     int =    add_prefix_space




				def 	__snake_case							(							self	:				List[Any],    *SCREAMING_SNAKE_CASE_	:				Tuple,    **SCREAMING_SNAKE_CASE_	:				int      ):
										snake_case       :     Dict =    kwargs.get('''is_split_into_words''',    SCREAMING_SNAKE_CASE_      )
										assert self.add_prefix_space or not is_split_into_words, (
										    F"""You need to instantiate {self.__class__.__name__} with add_prefix_space=True """
										    "to use it with pretokenized inputs."
										)

										return super()._batch_encode_plus(*SCREAMING_SNAKE_CASE_,    **SCREAMING_SNAKE_CASE_      )




				def 	__snake_case							(							self	:				List[str],    *SCREAMING_SNAKE_CASE_	:				str,    **SCREAMING_SNAKE_CASE_	:				Optional[int]      ):
										snake_case       :     Dict =    kwargs.get('''is_split_into_words''',    SCREAMING_SNAKE_CASE_      )

										assert self.add_prefix_space or not is_split_into_words, (
										    F"""You need to instantiate {self.__class__.__name__} with add_prefix_space=True """
										    "to use it with pretokenized inputs."
										)

										return super()._encode_plus(*SCREAMING_SNAKE_CASE_,    **SCREAMING_SNAKE_CASE_      )




				def 	__snake_case							(							self	:				Tuple,    SCREAMING_SNAKE_CASE_	:				str,    SCREAMING_SNAKE_CASE_	:				Optional[str] = None      ):
										snake_case       :     Optional[int] =    self._tokenizer.model.save(SCREAMING_SNAKE_CASE_,    name=SCREAMING_SNAKE_CASE_      )
										return tuple(SCREAMING_SNAKE_CASE_      )




				def 	__snake_case							(							self	:				Dict,    SCREAMING_SNAKE_CASE_	:				Union[int, List[int], "np.ndarray", "torch.Tensor", "tf.Tensor"],    SCREAMING_SNAKE_CASE_	:				bool = False,    SCREAMING_SNAKE_CASE_	:				bool = None,    SCREAMING_SNAKE_CASE_	:				Optional[List[str]] = None,    **SCREAMING_SNAKE_CASE_	:				Union[str, Any],    ):
										snake_case       :     Dict =    super().decode(
										    token_ids=SCREAMING_SNAKE_CASE_,    skip_special_tokens=SCREAMING_SNAKE_CASE_,    clean_up_tokenization_spaces=SCREAMING_SNAKE_CASE_,    **SCREAMING_SNAKE_CASE_,    )

										if truncate_before_pattern is not None and len(SCREAMING_SNAKE_CASE_      ) > 0:
																snake_case       :     Optional[int] =    self.truncate(SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_      )

										return decoded_text




				def 	__snake_case							(							self	:				int,    SCREAMING_SNAKE_CASE_	:				Optional[Any],    SCREAMING_SNAKE_CASE_	:				Tuple      ):
										def find_re(SCREAMING_SNAKE_CASE_	:				Optional[int],    SCREAMING_SNAKE_CASE_	:				Dict,    SCREAMING_SNAKE_CASE_	:				Optional[Any]      ):
																snake_case       :     Optional[Any] =    pattern.search(SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_      )
																return m.start() if m else -1

										snake_case       :     Union[str, Any] =    [re.compile(SCREAMING_SNAKE_CASE_,    re.MULTILINE      ) for pattern in truncate_before_pattern]

										snake_case       :     Union[str, Any] =    list(re.finditer('''^print''',    SCREAMING_SNAKE_CASE_,    re.MULTILINE      )      )

										if len(SCREAMING_SNAKE_CASE_      ) > 1:
																snake_case       :     Tuple =    completion[: prints[1].start()]

										snake_case       :     List[str] =    list(re.finditer('''^def''',    SCREAMING_SNAKE_CASE_,    re.MULTILINE      )      )

										if len(SCREAMING_SNAKE_CASE_      ) > 1:
																snake_case       :     Optional[Any] =    completion[: defs[1].start()]

										snake_case       :     Tuple =    0

										snake_case       :     List[Any] =    [
										    pos for pos in [find_re(SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_      ) for terminal in terminals] if pos != -1
										]

										if len(SCREAMING_SNAKE_CASE_      ) > 0:
																return completion[: min(SCREAMING_SNAKE_CASE_      )]
										else:
																return completion

