
'''simple docstring'''







# Copyright 2021 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from pathlib import Path

import torch

from ...utils import is_npu_available, is_xpu_available
from .config_args import ClusterConfig, default_json_config_file
from .config_utils import SubcommandHelpFormatter


_lowercase :  str    							=       "Create a default config file for Accelerate with only a few flags set."
def 					snake_case_		(				__SCREAMING_SNAKE_CASE					: Any="no"      ,      __SCREAMING_SNAKE_CASE					: str = default_json_config_file      ,      __SCREAMING_SNAKE_CASE					: bool = False							):



						"""simple docstring"""





						lowercase_	:     Union[str, Any]													=       Path(__SCREAMING_SNAKE_CASE							)
						path.parent.mkdir(parents=__SCREAMING_SNAKE_CASE      ,      exist_ok=__SCREAMING_SNAKE_CASE							)
						if path.exists():
												print(
												    F'''Configuration already exists at {save_location}, will not override. Run `accelerate config` manually or pass a different `save_location`.'''							)
												return False
						lowercase_	:     List[str]													=       mixed_precision.lower()
						if mixed_precision not in ["no", "fp16", "bf16", "fp8"]:
												raise ValueError(
												    F'''`mixed_precision` should be one of \'no\', \'fp16\', \'bf16\', or \'fp8\'. Received {mixed_precision}'''							)
						lowercase_	:     Any													=       {
						    '''compute_environment''': '''LOCAL_MACHINE''',
						    '''mixed_precision''': mixed_precision,
						}
						if torch.cuda.is_available():
												lowercase_	:     List[str]													=       torch.cuda.device_count()
												lowercase_	:     str													=       num_gpus
												lowercase_	:     Optional[int]													=       False
												if num_gpus > 1:
																		lowercase_	:     List[str]													=       '''MULTI_GPU'''
												else:
																		lowercase_	:     Dict													=       '''NO'''
						elif is_xpu_available() and use_xpu:
												lowercase_	:     List[str]													=       torch.xpu.device_count()
												lowercase_	:     List[Any]													=       num_xpus
												lowercase_	:     Optional[Any]													=       False
												if num_xpus > 1:
																		lowercase_	:     Dict													=       '''MULTI_XPU'''
												else:
																		lowercase_	:     Optional[int]													=       '''NO'''
						elif is_npu_available():
												lowercase_	:     List[str]													=       torch.npu.device_count()
												lowercase_	:     int													=       num_npus
												lowercase_	:     Tuple													=       False
												if num_npus > 1:
																		lowercase_	:     Union[str, Any]													=       '''MULTI_NPU'''
												else:
																		lowercase_	:     int													=       '''NO'''
						else:
												lowercase_	:     List[Any]													=       0
												lowercase_	:     Union[str, Any]													=       True
												lowercase_	:     Optional[int]													=       1
												lowercase_	:     Tuple													=       '''NO'''
						lowercase_	:     Optional[int]													=       ClusterConfig(**__SCREAMING_SNAKE_CASE							)
						config.to_json_file(__SCREAMING_SNAKE_CASE							)
						return path
def 					snake_case_		(				__SCREAMING_SNAKE_CASE					: Dict      ,      __SCREAMING_SNAKE_CASE					: Optional[int]							):



						"""simple docstring"""





						lowercase_	:     Optional[int]													=       parser.add_parser('''default'''      ,      parents=__SCREAMING_SNAKE_CASE      ,      help=__SCREAMING_SNAKE_CASE      ,      formatter_class=__SCREAMING_SNAKE_CASE							)
						parser.add_argument(
						    '''--config_file'''      ,      default=__SCREAMING_SNAKE_CASE      ,      help=(
						        '''The path to use to store the config file. Will default to a file named default_config.yaml in the cache '''
						        '''location, which is the content of the environment `HF_HOME` suffixed with \'accelerate\', or if you don\'t have '''
						        '''such an environment variable, your cache directory (\'~/.cache\' or the content of `XDG_CACHE_HOME`) suffixed '''
						        '''with \'huggingface\'.'''
						    )      ,      dest='''save_location'''      ,      )

						parser.add_argument(
						    '''--mixed_precision'''      ,      choices=['''no''', '''fp16''', '''bf16''']      ,      type=__SCREAMING_SNAKE_CASE      ,      help='''Whether or not to use mixed precision training. '''
						    '''Choose between FP16 and BF16 (bfloat16) training. '''
						    '''BF16 training is only supported on Nvidia Ampere GPUs and PyTorch 1.10 or later.'''      ,      default='''no'''      ,      )
						parser.set_defaults(func=__SCREAMING_SNAKE_CASE							)
						return parser




def 					snake_case_		(				__SCREAMING_SNAKE_CASE					: Any							):



						"""simple docstring"""





						lowercase_	:     Dict													=       write_basic_config(args.mixed_precision      ,      args.save_location							)
						if config_file:
												print(F'''accelerate configuration saved at {config_file}'''							)




'''simple docstring'''




import os
from shutil import copyfile
from typing import List, Optional, Tuple

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging


if is_sentencepiece_available():
  from .tokenization_pegasus import PegasusTokenizer
else:
  __SCREAMING_SNAKE_CASE       :     Optional[Any]     =  None


__SCREAMING_SNAKE_CASE       :     Dict     =  logging.get_logger(__name__)


__SCREAMING_SNAKE_CASE       :     Union[str, Any]     =  """▁"""

__SCREAMING_SNAKE_CASE       :     str     =  {"""vocab_file""": """spiece.model""", """tokenizer_file""": """tokenizer.json"""}

__SCREAMING_SNAKE_CASE       :     int     =  {
    """vocab_file""": {"""google/pegasus-xsum""": """https://huggingface.co/google/pegasus-xsum/resolve/main/spiece.model"""},
    """tokenizer_file""": {
        """google/pegasus-xsum""": """https://huggingface.co/google/pegasus-xsum/resolve/main/tokenizer.json"""
    },
}

__SCREAMING_SNAKE_CASE       :     str     =  {
    """google/pegasus-xsum""": 512,
}

class 	lowerCamelCase_	(snake_case__    ):
      '''simple docstring'''

      __UpperCamelCase:  Optional[int]						=						VOCAB_FILES_NAMES
      __UpperCamelCase:  Dict						=						PRETRAINED_VOCAB_FILES_MAP
      __UpperCamelCase:  List[Any]						=						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      __UpperCamelCase:  Optional[int]						=						PegasusTokenizer
      __UpperCamelCase:  Optional[Any]						=						["input_ids", "attention_mask"]




      def __init__(       self						:				Dict						,					A						:				List[str]=None						,					A						:				Union[str, Any]=None						,					A						:				Optional[int]="<pad>"						,					A						:				Tuple="</s>"						,					A						:				Union[str, Any]="<unk>"						,					A						:				Union[str, Any]="<mask_2>"						,					A						:				Dict="<mask_1>"						,					A						:				Union[str, Any]=None						,					A						:				int=103						,					**A						:				Optional[Any]						,					):
             _UpperCAmelCase     :  Dict							      =    offset

             if additional_special_tokens is not None:
                    if not isinstance(A						,					A       ):
                           raise TypeError(
                               F"""additional_special_tokens should be of type {type(A       )}, but is"""
                               F""" {type(A       )}"""       )

                    _UpperCAmelCase     :  Optional[int]							      =    (
                        ([mask_token_sent] + additional_special_tokens)
                        if mask_token_sent not in additional_special_tokens and mask_token_sent is not None
                        else additional_special_tokens
                    )
                    # fill additional tokens with ..., <unk_token_102> in case not all additional tokens are already taken
                    additional_special_tokens_extended += [
                        F"""<unk_{i}>""" for i in range(len(A       )						,					self.offset - 1       )
                    ]

                    if len(set(A       )       ) != len(A       ):
                           raise ValueError(
                               "Please make sure that the provided additional_special_tokens do not contain an incorrectly"
                               F""" shifted list of <unk_x> tokens. Found {additional_special_tokens_extended}."""       )
                    _UpperCAmelCase     :  Any							      =    additional_special_tokens_extended
             else:
                    _UpperCAmelCase     :  Dict							      =    [mask_token_sent] if mask_token_sent is not None else []
                    additional_special_tokens += [F"""<unk_{i}>""" for i in range(2						,					self.offset       )]

             super().__init__(
                 A						,					tokenizer_file=A						,					pad_token=A						,					eos_token=A						,					unk_token=A						,					mask_token=A						,					mask_token_sent=A						,					offset=A						,					additional_special_tokens=A						,					**A						,					)
             _UpperCAmelCase     :  Optional[Any]							      =    vocab_file
             _UpperCAmelCase     :  Optional[Any]							      =    False if not self.vocab_file else True




      def      _A	(       self						:				List[str]						,					A						:				Optional[Any]       ):
             _UpperCAmelCase     :  Any							      =    set(self.all_special_ids       )  # call it once instead of inside list comp
             all_special_ids.remove(self.unk_token_id       )  # <unk> is only sometimes special

             if all_special_ids != set(range(len(self.additional_special_tokens       ) + 3       )       ):
                    raise ValueError(
                        "There should be 3 special tokens: mask_token, pad_token, and eos_token +"
                        F""" {len(self.additional_special_tokens       )} additional_special_tokens, but got {all_special_ids}"""       )

             return [1 if x in all_special_ids else 0 for x in seq]




      def      _A	(       self						:				str						,					A						:				List						,					A						:				Optional[List] = None						,					A						:				bool = False       ):
             if already_has_special_tokens:
                    return self._special_token_mask(A       )
             elif token_ids_a is None:
                    return self._special_token_mask(A       ) + [1]
             else:
                    return self._special_token_mask(token_ids_a + token_ids_a       ) + [1]




      def      _A	(       self						:				Optional[int]						,					A						:				Union[str, Any]						,					A						:				int=None       ):
             if token_ids_a is None:
                    return token_ids_a + [self.eos_token_id]
             # We don't expect to process pairs, but leave the pair logic for API consistency
             return token_ids_a + token_ids_a + [self.eos_token_id]





      def      _A	(       self						:				Union[str, Any]						,					A						:				str						,					A						:				Optional[str] = None       ):
             if not self.can_save_slow_tokenizer:
                    raise ValueError(
                        "Your fast tokenizer does not have the necessary information to save the vocabulary for a slow "
                        "tokenizer."       )

             if not os.path.isdir(A       ):
                    logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""       )
                    return
             _UpperCAmelCase     :  List[Any]							      =    os.path.join(
                 A						,					(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]       )

             if os.path.abspath(self.vocab_file       ) != os.path.abspath(A       ):
                    copyfile(self.vocab_file						,					A       )

             return (out_vocab_file,)

