

'''simple docstring'''

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

from tokenizers import pre_tokenizers

from ...tokenization_utils_base import BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging


if TYPE_CHECKING:
    from transformers.pipelines.conversational import Conversation


A_							    =				logging.get_logger(__name__)

A_							    =				{"""tokenizer_file""": """tokenizer.json"""}

A_							    =				{
    """tokenizer_file""": {
        """bigscience/tokenizer""": """https://huggingface.co/bigscience/tokenizer/blob/main/tokenizer.json""",
        """bigscience/bloom-560m""": """https://huggingface.co/bigscience/bloom-560m/blob/main/tokenizer.json""",
        """bigscience/bloom-1b1""": """https://huggingface.co/bigscience/bloom-1b1/blob/main/tokenizer.json""",
        """bigscience/bloom-1b7""": """https://huggingface.co/bigscience/bloom-1b7/blob/main/tokenizer.json""",
        """bigscience/bloom-3b""": """https://huggingface.co/bigscience/bloom-3b/blob/main/tokenizer.json""",
        """bigscience/bloom-7b1""": """https://huggingface.co/bigscience/bloom-7b1/blob/main/tokenizer.json""",
        """bigscience/bloom""": """https://huggingface.co/bigscience/bloom/blob/main/tokenizer.json""",
    },
}





class 				_snake_case			(							_A    ):
   _A							:  Optional[int]      =       VOCAB_FILES_NAMES
   _A							:  List[Any]      =       PRETRAINED_VOCAB_FILES_MAP
   _A							:  int      =       ["""input_ids""", """attention_mask"""]
   _A							:  List[Any]      =       None
   def __init__(			self      :   Dict  ,SCREAMING_SNAKE_CASE__      :   Optional[Any]=None  ,SCREAMING_SNAKE_CASE__      :   str=None  ,SCREAMING_SNAKE_CASE__      :   Any=None  ,SCREAMING_SNAKE_CASE__      :   Optional[int]="<unk>"  ,SCREAMING_SNAKE_CASE__      :   Dict="<s>"  ,SCREAMING_SNAKE_CASE__      :   int="</s>"  ,SCREAMING_SNAKE_CASE__      :   Union[str, Any]="<pad>"  ,SCREAMING_SNAKE_CASE__      :   List[Any]=False  ,SCREAMING_SNAKE_CASE__      :   Tuple=False  ,**SCREAMING_SNAKE_CASE__      :   Optional[int]  ,):
     super().__init__(
         snake_case_  ,snake_case_  ,tokenizer_file=snake_case_  ,unk_token=snake_case_  ,bos_token=snake_case_  ,eos_token=snake_case_  ,pad_token=snake_case_  ,add_prefix_space=snake_case_  ,clean_up_tokenization_spaces=snake_case_  ,**snake_case_  ,)
     SCREAMING_SNAKE_CASE:str       =       json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__()				)
     if pre_tok_state.get("add_prefix_space"  ,snake_case_				) != add_prefix_space:
       SCREAMING_SNAKE_CASE:List[str]       =       getattr(snake_case_  ,pre_tok_state.pop("type"				)				)
       SCREAMING_SNAKE_CASE:Union[str, Any]       =       add_prefix_space
       SCREAMING_SNAKE_CASE:Tuple       =       pre_tok_class(**snake_case_				)

     SCREAMING_SNAKE_CASE:Optional[Any]       =       add_prefix_space
   def      __UpperCamelCase    (			self      :   Any  ,*SCREAMING_SNAKE_CASE__      :   str  ,**SCREAMING_SNAKE_CASE__      :   Optional[Any]				):
     SCREAMING_SNAKE_CASE:List[str]       =       kwargs.get("is_split_into_words"  ,snake_case_				)
     if not (self.add_prefix_space or not is_split_into_words):
       raise Exception(
           F'''You need to instantiate {self.__class__.__name__} with add_prefix_space=True to use it with'''
           " pretokenized inputs."				)

     return super()._batch_encode_plus(*snake_case_  ,**snake_case_				)
   def      __UpperCamelCase    (			self      :   Optional[Any]  ,*SCREAMING_SNAKE_CASE__      :   Tuple  ,**SCREAMING_SNAKE_CASE__      :   List[Any]				):
     SCREAMING_SNAKE_CASE:int       =       kwargs.get("is_split_into_words"  ,snake_case_				)

     if not (self.add_prefix_space or not is_split_into_words):
       raise Exception(
           F'''You need to instantiate {self.__class__.__name__} with add_prefix_space=True to use it with'''
           " pretokenized inputs."				)

     return super()._encode_plus(*snake_case_  ,**snake_case_				)
   def      __UpperCamelCase    (			self      :   int  ,SCREAMING_SNAKE_CASE__      :   str  ,SCREAMING_SNAKE_CASE__      :   Optional[str] = None				):
     SCREAMING_SNAKE_CASE:str       =       self._tokenizer.model.save(snake_case_  ,name=snake_case_				)
     return tuple(snake_case_				)







   def      __UpperCamelCase    (			self      :   List[Any]  ,SCREAMING_SNAKE_CASE__      :   "Conversation"				):
     SCREAMING_SNAKE_CASE:Optional[Any]       =       []
     for is_user, text in conversation.iter_texts():
       input_ids.extend(self.encode(snake_case_  ,add_special_tokens=snake_case_				) + [self.eos_token_id]				)

     if len(snake_case_				) > self.model_max_length:
       SCREAMING_SNAKE_CASE:Tuple       =       input_ids[-self.model_max_length :]
     return input_ids








import os
import sys
import tempfile

import torch

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







def 							A__       (    lowerCamelCase  ,     lowerCamelCase=()  ,     lowerCamelCase=None  ,     lowerCamelCase="no"  ,     lowerCamelCase="29500"    )	->						List[Any]:
				UpperCamelCase_:				Any						=    False
				UpperCamelCase_:				int						=    False
				if any(key.startswith("""KAGGLE"""    ) for key in os.environ.keys()    ):
								UpperCamelCase_:				str						=    True
				elif "IPython" in sys.modules:
								UpperCamelCase_:				Tuple						=    """google.colab""" in str(sys.modules["""IPython"""].get_ipython()    )

				try:
								UpperCamelCase_:				Union[str, Any]						=    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"""  ,     lowerCamelCase    ) 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:
												UpperCamelCase_:				str						=    8

								UpperCamelCase_:				int						=    PrepareForLaunch(lowerCamelCase  ,     distributed_type="""TPU"""    )
								print(F'''Launching a training on {num_processes} TPU cores.'''    )
								xmp.spawn(lowerCamelCase  ,     args=lowerCamelCase  ,     nprocs=lowerCamelCase  ,     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(*lowerCamelCase    )
				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=lowerCamelCase  ,     master_addr="""127.0.01"""  ,     master_port=lowerCamelCase  ,     mixed_precision=lowerCamelCase    ):
																UpperCamelCase_:				Optional[int]						=    PrepareForLaunch(lowerCamelCase  ,     distributed_type="""MULTI_GPU"""    )
																print(F'''Launching training on {num_processes} GPUs.'''    )
																try:
																				start_processes(lowerCamelCase  ,     args=lowerCamelCase  ,     nprocs=lowerCamelCase  ,     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():
																UpperCamelCase_:				List[Any]						=    """1"""
																print("""Launching training on MPS."""    )
												elif torch.cuda.is_available():
																print("""Launching training on one GPU."""    )
												else:
																print("""Launching training on CPU."""    )
												function(*lowerCamelCase    )







def 							A__       (    lowerCamelCase  ,     lowerCamelCase=()  ,     lowerCamelCase=2    )	->						List[str]:
				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=lowerCamelCase  ,     master_addr="""127.0.01"""  ,     master_port="""29500"""  ,     accelerate_mixed_precision="""no"""  ,     accelerate_debug_rdv_file=tmp_file.name  ,     accelerate_use_cpu="""yes"""  ,     ):
												UpperCamelCase_:				str						=    PrepareForLaunch(lowerCamelCase  ,     debug=lowerCamelCase    )
												start_processes(lowerCamelCase  ,     args=lowerCamelCase  ,     nprocs=lowerCamelCase  ,     start_method="""fork"""    )






