




import importlib
import importlib.metadata
import os
import warnings
from distutils.util import strtobool
from functools import lru_cache

import torch
from packaging import version
from packaging.version import parse

from .environment import parse_flag_from_env
from .versions import compare_versions, is_torch_version


try:
   import torch_xla.core.xla_model as xm  # noqa: F401

   snake_case__ :						Tuple     =			True
except ImportError:
   snake_case__ :						int     =			False


# Cache this result has it's a C FFI call which can be pretty time-consuming
snake_case__ :						Any     =			torch.distributed.is_available()


def 						lowercase		(     _lowerCAmelCase      ):
  # Check we're not importing a "pkg_name" directory somewhere but the actual library by trying to grab the version
  UpperCAmelCase__    						=   importlib.util.find_spec(_lowerCAmelCase      ) is not None
  if package_exists:
    try:
      UpperCAmelCase__    						=   importlib.metadata.metadata(_lowerCAmelCase      )
      return True
    except importlib.metadata.PackageNotFoundError:
      return False


def 						lowercase		(     ):
  return _torch_distributed_available


def 						lowercase		(     ):
  try:
    pass
  except ImportError:
    print(
        """Intel(R) oneCCL Bindings for PyTorch* is required to run DDP on Intel(R) GPUs, but it is not"""
        """ detected. If you see \"ValueError: Invalid backend: 'ccl'\" error, please install Intel(R) oneCCL"""
        """ Bindings for PyTorch*."""      )
  return (
      importlib.util.find_spec("""torch_ccl"""      ) is not None
      or importlib.util.find_spec("""oneccl_bindings_for_pytorch"""      ) is not None
  )


def 						lowercase		(     ):
  return importlib.metadata.version("""oneccl_bind_pt"""      )


def 						lowercase		(     ):
  return _is_package_available("""transformer_engine"""      )


@lru_cache
def 						lowercase		(     _lowerCAmelCase=True      ):
  # Due to bugs on the amp series GPUs, we disable torch-xla on them
  if torch.cuda.is_available():
    return False
  if _tpu_available and check_device:
    try:
      # Will raise a RuntimeError if no XLA configuration is found
      UpperCAmelCase__    						=   xm.xla_device()
      return True
    except RuntimeError:
      return False
  return _tpu_available


def 						lowercase		(     ):
  return _is_package_available("""deepspeed"""      )


def 						lowercase		(     _lowerCAmelCase=False      ):
  if is_tpu_available():
    return not ignore_tpu
  if torch.cuda.is_available():
    return torch.cuda.is_bfaa_supported()
  if is_npu_available():
    return False
  return True


def 						lowercase		(     ):
  UpperCAmelCase__    						=   _is_package_available("""bitsandbytes"""      )
  if package_exists:
    UpperCAmelCase__    						=   version.parse(importlib.metadata.version("""bitsandbytes"""      )      )
    return compare_versions(_lowerCAmelCase   ,     """>="""   ,     """0.39.0"""      )
  return False


def 						lowercase		(     ):
  UpperCAmelCase__    						=   _is_package_available("""bitsandbytes"""      )
  if package_exists:
    UpperCAmelCase__    						=   version.parse(importlib.metadata.version("""bitsandbytes"""      )      )
    return compare_versions(_lowerCAmelCase   ,     """>="""   ,     """0.37.2"""      )
  return False


def 						lowercase		(     ):
  return _is_package_available("""bitsandbytes"""      )


def 						lowercase		(     ):
  if strtobool(os.environ.get("""ACCELERATE_USE_MEGATRON_LM"""   ,     """False"""      )      ) == 1:
    UpperCAmelCase__    						=   importlib.util.find_spec("""megatron"""      ) is not None
    if package_exists:
      try:
        UpperCAmelCase__    						=   parse(importlib.metadata.version("""megatron-lm"""      )      )
        return compare_versions(_lowerCAmelCase   ,     """>="""   ,     """2.2.0"""      )
      except Exception as e:
        warnings.warn(F'''Parse Megatron version failed. Exception:{e}'''      )
        return False


def 						lowercase		(     ):
  return _is_package_available("""safetensors"""      )


def 						lowercase		(     ):
  return _is_package_available("""transformers"""      )


def 						lowercase		(     ):
  return _is_package_available("""datasets"""      )


def 						lowercase		(     ):
  return _is_package_available("""aim"""      )


def 						lowercase		(     ):
  return _is_package_available("""tensorboard"""      ) or _is_package_available("""tensorboardX"""      )


def 						lowercase		(     ):
  return _is_package_available("""wandb"""      )


def 						lowercase		(     ):
  return _is_package_available("""comet_ml"""      )


def 						lowercase		(     ):
  return _is_package_available("""boto3"""      )


def 						lowercase		(     ):
  if _is_package_available("""rich"""      ):
    if "ACCELERATE_DISABLE_RICH" in os.environ:
      warnings.warn(
          """`ACCELERATE_DISABLE_RICH` is deprecated and will be removed in v0.22.0 and deactivated by default. Please use `ACCELERATE_ENABLE_RICH` if you wish to use `rich`."""      )
      return not parse_flag_from_env("""ACCELERATE_DISABLE_RICH"""   ,     _lowerCAmelCase      )
    return parse_flag_from_env("""ACCELERATE_ENABLE_RICH"""   ,     _lowerCAmelCase      )
  return False


def 						lowercase		(     ):
  return _is_package_available("""sagemaker"""      )


def 						lowercase		(     ):
  return _is_package_available("""tqdm"""      )


def 						lowercase		(     ):
  return _is_package_available("""mlflow"""      )


def 						lowercase		(     ):
  return is_torch_version(""">="""   ,     """1.12"""      ) and torch.backends.mps.is_available() and torch.backends.mps.is_built()


def 						lowercase		(     ):
  def get_major_and_minor_from_version(_lowerCAmelCase      ):
    return str(version.parse(_lowerCAmelCase      ).major      ) + "." + str(version.parse(_lowerCAmelCase      ).minor      )

  UpperCAmelCase__    						=   importlib.metadata.version("""torch"""      )
  if importlib.util.find_spec("""intel_extension_for_pytorch"""      ) is None:
    return False
  UpperCAmelCase__    						=   """N/A"""
  try:
    UpperCAmelCase__    						=   importlib.metadata.version("""intel_extension_for_pytorch"""      )
  except importlib.metadata.PackageNotFoundError:
    return False
  UpperCAmelCase__    						=   get_major_and_minor_from_version(_torch_version      )
  UpperCAmelCase__    						=   get_major_and_minor_from_version(_ipex_version      )
  if torch_major_and_minor != ipex_major_and_minor:
    warnings.warn(
        F'''Intel Extension for PyTorch {ipex_major_and_minor} needs to work with PyTorch {ipex_major_and_minor}.*,'''
        F''' but PyTorch {_torch_version} is found. Please switch to the matching version and run again.'''      )
    return False
  return True


@lru_cache
def 						lowercase		(     _lowerCAmelCase=False      ):
  if importlib.util.find_spec("""torch"""      ) is None or importlib.util.find_spec("""torch_npu"""      ) is None:
    return False

  import torch
  import torch_npu  # noqa: F401

  if check_device:
    try:
      # Will raise a RuntimeError if no NPU is found
      UpperCAmelCase__    						=   torch.npu.device_count()
      return torch.npu.is_available()
    except RuntimeError:
      return False
  return hasattr(_lowerCAmelCase   ,     """npu"""      ) and torch.npu.is_available()





@lru_cache
def 						lowercase		(     _lowerCAmelCase=False      ):
  if not parse_flag_from_env("""ACCELERATE_USE_XPU"""   ,     default=_lowerCAmelCase      ):
    return False
  if is_ipex_available():
    import torch

    if is_torch_version("""<="""   ,     """1.12"""      ):
      return False
  else:
    return False

  import intel_extension_for_pytorch  # noqa: F401

  if check_device:
    try:
      # Will raise a RuntimeError if no XPU  is found
      UpperCAmelCase__    						=   torch.xpu.device_count()
      return torch.xpu.is_available()
    except RuntimeError:
      return False
  return hasattr(_lowerCAmelCase   ,     """xpu"""      ) and torch.xpu.is_available()










import json
import os
from functools import lru_cache
from typing import TYPE_CHECKING, List, Optional, Tuple, Union

import numpy as np
import regex as re

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 ...tokenization_utils import AddedToken, PreTrainedTokenizer


snake_case__ :						str     =			logging.get_logger(__name__)

snake_case__ :						List[str]     =			{
    '''vocab_file''': '''vocab.json''',
    '''merges_file''': '''merges.txt''',
}

snake_case__ :						Any     =			{
    '''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''',
    },
}

snake_case__ :						Tuple     =			{
    '''Salesforce/codegen-350M-mono''': 2_0_4_8,
}


@lru_cache()
def 						lowercase		(     ):
  UpperCAmelCase__    						=   (
      list(range(ord("""!"""      )   ,     ord("""~"""      ) + 1      )      ) + list(range(ord("""¡"""      )   ,     ord("""¬"""      ) + 1      )      ) + list(range(ord("""®"""      )   ,     ord("""ÿ"""      ) + 1      )      )
  )
  UpperCAmelCase__    						=   bs[:]
  UpperCAmelCase__    						=   0
  for b in range(2**8      ):
    if b not in bs:
      bs.append(_lowerCAmelCase      )
      cs.append(2**8 + n      )
      n += 1
  UpperCAmelCase__    						=   [chr(_lowerCAmelCase      ) for n in cs]
  return dict(zip(_lowerCAmelCase   ,     _lowerCAmelCase      )      )


def 						lowercase		(     _lowerCAmelCase      ):
  UpperCAmelCase__    						=   set()
  UpperCAmelCase__    						=   word[0]
  for char in word[1:]:
    pairs.add((prev_char, char)      )
    UpperCAmelCase__    						=   char
  return pairs





class     snake_case  (	_snake_case							):


   '''simple docstring'''







   UpperCamelCase__	:       List[Any]     =      VOCAB_FILES_NAMES
   UpperCamelCase__	:       Union[str, Any]     =      PRETRAINED_VOCAB_FILES_MAP
   UpperCamelCase__	:       int     =      PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
   UpperCamelCase__	:       int     =      ["input_ids", "attention_mask"]




   def __init__(					self     :	Union[str, Any]						,    lowerCamelCase_     :	Tuple						,    lowerCamelCase_     :	Any						,    lowerCamelCase_     :	Union[str, Any]="replace"						,    lowerCamelCase_     :	List[str]="<|endoftext|>"						,    lowerCamelCase_     :	Dict="<|endoftext|>"						,    lowerCamelCase_     :	List[str]="<|endoftext|>"						,    lowerCamelCase_     :	str=None						,    lowerCamelCase_     :	Any=False						,    lowerCamelCase_     :	Optional[int]=False						,    **lowerCamelCase_     :	str						,    )					->Dict:
     '''simple docstring'''







     UpperCAmelCase__    						=   AddedToken(lowerCamelCase_						,    lstrip=lowerCamelCase_						,    rstrip=lowerCamelCase_			) if isinstance(lowerCamelCase_						,    lowerCamelCase_			) else bos_token
     UpperCAmelCase__    						=   AddedToken(lowerCamelCase_						,    lstrip=lowerCamelCase_						,    rstrip=lowerCamelCase_			) if isinstance(lowerCamelCase_						,    lowerCamelCase_			) else eos_token
     UpperCAmelCase__    						=   AddedToken(lowerCamelCase_						,    lstrip=lowerCamelCase_						,    rstrip=lowerCamelCase_			) if isinstance(lowerCamelCase_						,    lowerCamelCase_			) else unk_token
     UpperCAmelCase__    						=   AddedToken(lowerCamelCase_						,    lstrip=lowerCamelCase_						,    rstrip=lowerCamelCase_			) if isinstance(lowerCamelCase_						,    lowerCamelCase_			) else pad_token
     super().__init__(
         errors=lowerCamelCase_						,    unk_token=lowerCamelCase_						,    bos_token=lowerCamelCase_						,    eos_token=lowerCamelCase_						,    pad_token=lowerCamelCase_						,    add_prefix_space=lowerCamelCase_						,    add_bos_token=lowerCamelCase_						,    **lowerCamelCase_						,    )
     UpperCAmelCase__    						=   add_bos_token

     with open(lowerCamelCase_						,    encoding="""utf-8"""			) as vocab_handle:
       UpperCAmelCase__    						=   json.load(lowerCamelCase_			)
     UpperCAmelCase__    						=   {v: k for k, v in self.encoder.items()}
     UpperCAmelCase__    						=   errors  # how to handle errors in decoding
     UpperCAmelCase__    						=   bytes_to_unicode()
     UpperCAmelCase__    						=   {v: k for k, v in self.byte_encoder.items()}
     with open(lowerCamelCase_						,    encoding="""utf-8"""			) as merges_handle:
       UpperCAmelCase__    						=   merges_handle.read().split("""\n"""			)[1:-1]
     UpperCAmelCase__    						=   [tuple(merge.split()			) for merge in bpe_merges]
     UpperCAmelCase__    						=   dict(zip(lowerCamelCase_						,    range(len(lowerCamelCase_			)			)			)			)
     UpperCAmelCase__    						=   {}
     UpperCAmelCase__    						=   add_prefix_space

     # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
     UpperCAmelCase__    						=   re.compile(R"""'s|'t|'re|'ve|'m|'ll|'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+"""			)




   @property
   def 	UpperCAmelCase       (					self     :	Dict			)					->Any:
     '''simple docstring'''







     return len(self.encoder			)




   def 	UpperCAmelCase       (					self     :	Dict			)					->Optional[int]:
     '''simple docstring'''







     return dict(self.encoder						,    **self.added_tokens_encoder			)




   def 	UpperCAmelCase       (					self     :	Tuple						,    lowerCamelCase_     :	List[str]			)					->Optional[Any]:
     '''simple docstring'''







     if token in self.cache:
       return self.cache[token]
     UpperCAmelCase__    						=   tuple(lowerCamelCase_			)
     UpperCAmelCase__    						=   get_pairs(lowerCamelCase_			)

     if not pairs:
       return token

     while True:
       UpperCAmelCase__    						=   min(lowerCamelCase_						,    key=lambda lowerCamelCase_			: self.bpe_ranks.get(lowerCamelCase_						,    float("""inf"""			)			)			)
       if bigram not in self.bpe_ranks:
         break
       UpperCAmelCase__						,					UpperCAmelCase__    						=   bigram
       UpperCAmelCase__    						=   []
       UpperCAmelCase__    						=   0
       while i < len(lowerCamelCase_			):
         try:
           UpperCAmelCase__    						=   word.index(lowerCamelCase_						,    lowerCamelCase_			)
         except ValueError:
           new_word.extend(word[i:]			)
           break
         else:
           new_word.extend(word[i:j]			)
           UpperCAmelCase__    						=   j

         if word[i] == first and i < len(lowerCamelCase_			) - 1 and word[i + 1] == second:
           new_word.append(first + second			)
           i += 2
         else:
           new_word.append(word[i]			)
           i += 1
       UpperCAmelCase__    						=   tuple(lowerCamelCase_			)
       UpperCAmelCase__    						=   new_word
       if len(lowerCamelCase_			) == 1:
         break
       else:
         UpperCAmelCase__    						=   get_pairs(lowerCamelCase_			)
     UpperCAmelCase__    						=   """ """.join(lowerCamelCase_			)
     UpperCAmelCase__    						=   word
     return word




   def 	UpperCAmelCase       (					self     :	int						,    lowerCamelCase_     :	Optional[int]						,    lowerCamelCase_     :	Optional[Any]=None			)					->str:
     '''simple docstring'''







     if self.add_bos_token:
       UpperCAmelCase__    						=   [self.bos_token_id]
     else:
       UpperCAmelCase__    						=   []

     UpperCAmelCase__    						=   bos_token_ids + token_ids_a

     if token_ids_a is None:
       return output

     return output + bos_token_ids + token_ids_a




   def 	UpperCAmelCase       (					self     :	Any						,    lowerCamelCase_     :	Optional[Any]			)					->Union[str, Any]:
     '''simple docstring'''







     UpperCAmelCase__    						=   []
     for token in re.findall(self.pat						,    lowerCamelCase_			):
       UpperCAmelCase__    						=   """""".join(
           self.byte_encoder[b] for b in token.encode("""utf-8"""			)			)  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
       bpe_tokens.extend(bpe_token for bpe_token in self.bpe(lowerCamelCase_			).split(""" """			)			)
     return bpe_tokens




   def 	UpperCAmelCase       (					self     :	Dict						,    lowerCamelCase_     :	Optional[Any]			)					->int:
     '''simple docstring'''







     return self.encoder.get(lowerCamelCase_						,    self.encoder.get(self.unk_token			)			)




   def 	UpperCAmelCase       (					self     :	Union[str, Any]						,    lowerCamelCase_     :	Dict			)					->Union[str, Any]:
     '''simple docstring'''







     return self.decoder.get(lowerCamelCase_			)




   def 	UpperCAmelCase       (					self     :	int						,    lowerCamelCase_     :	Union[str, Any]			)					->Union[str, Any]:
     '''simple docstring'''







     UpperCAmelCase__    						=   """""".join(lowerCamelCase_			)
     UpperCAmelCase__    						=   bytearray([self.byte_decoder[c] for c in text]			).decode("""utf-8"""						,    errors=self.errors			)
     return text




   def 	UpperCAmelCase       (					self     :	Dict						,    lowerCamelCase_     :	str						,    lowerCamelCase_     :	Optional[str] = None			)					->Tuple[str]:
     '''simple docstring'''







     if not os.path.isdir(lowerCamelCase_			):
       logger.error(f'''Vocabulary path ({save_directory}) should be a directory'''			)
       return
     UpperCAmelCase__    						=   os.path.join(
         lowerCamelCase_						,    (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]			)
     UpperCAmelCase__    						=   os.path.join(
         lowerCamelCase_						,    (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""merges_file"""]			)

     with open(lowerCamelCase_						,    """w"""						,    encoding="""utf-8"""			) as f:
       f.write(json.dumps(self.encoder						,    indent=2						,    sort_keys=lowerCamelCase_						,    ensure_ascii=lowerCamelCase_			) + """\n"""			)

     UpperCAmelCase__    						=   0
     with open(lowerCamelCase_						,    """w"""						,    encoding="""utf-8"""			) as writer:
       writer.write("""#version: 0.2\n"""			)
       for bpe_tokens, token_index in sorted(self.bpe_ranks.items()						,    key=lambda lowerCamelCase_			: kv[1]			):
         if index != token_index:
           logger.warning(
               f'''Saving vocabulary to {merge_file}: BPE merge indices are not consecutive.'''
               """ Please check that the tokenizer is not corrupted!"""			)
           UpperCAmelCase__    						=   token_index
         writer.write(""" """.join(lowerCamelCase_			) + """\n"""			)
         index += 1

     return vocab_file, merge_file




   def 	UpperCAmelCase       (					self     :	List[Any]						,    lowerCamelCase_     :	int						,    lowerCamelCase_     :	Tuple=False						,    **lowerCamelCase_     :	Any			)					->Any:
     '''simple docstring'''







     UpperCAmelCase__    						=   kwargs.pop("""add_prefix_space"""						,    self.add_prefix_space			)
     if is_split_into_words or add_prefix_space:
       UpperCAmelCase__    						=   """ """ + text
     return (text, kwargs)




   def 	UpperCAmelCase       (					self     :	List[Any]						,    lowerCamelCase_     :	Union[int, List[int], "np.ndarray", "torch.Tensor", "tf.Tensor"]						,    lowerCamelCase_     :	bool = False						,    lowerCamelCase_     :	bool = None						,    lowerCamelCase_     :	Optional[List[str]] = None						,    **lowerCamelCase_     :	Any						,    )					->str:
     '''simple docstring'''







     UpperCAmelCase__    						=   super()._decode(
         token_ids=lowerCamelCase_						,    skip_special_tokens=lowerCamelCase_						,    clean_up_tokenization_spaces=lowerCamelCase_						,    **lowerCamelCase_						,    )

     if truncate_before_pattern is not None and len(lowerCamelCase_			) > 0:
       UpperCAmelCase__    						=   self.truncate(lowerCamelCase_						,    lowerCamelCase_			)

     return decoded_text







   def 	UpperCAmelCase       (					self     :	Tuple						,    lowerCamelCase_     :	Any						,    lowerCamelCase_     :	str			)					->Optional[int]:
     '''simple docstring'''







     def find_re(lowerCamelCase_     :	List[str]						,    lowerCamelCase_     :	Union[str, Any]						,    lowerCamelCase_     :	int			):
       UpperCAmelCase__    						=   pattern.search(lowerCamelCase_						,    lowerCamelCase_			)
       return m.start() if m else -1

     UpperCAmelCase__    						=   [re.compile(lowerCamelCase_						,    re.MULTILINE			) for pattern in truncate_before_pattern]

     UpperCAmelCase__    						=   list(re.finditer("""^print"""						,    lowerCamelCase_						,    re.MULTILINE			)			)

     if len(lowerCamelCase_			) > 1:
       UpperCAmelCase__    						=   completion[: prints[1].start()]

     UpperCAmelCase__    						=   list(re.finditer("""^def"""						,    lowerCamelCase_						,    re.MULTILINE			)			)

     if len(lowerCamelCase_			) > 1:
       UpperCAmelCase__    						=   completion[: defs[1].start()]

     UpperCAmelCase__    						=   0

     UpperCAmelCase__    						=   [
         pos for pos in [find_re(lowerCamelCase_						,    lowerCamelCase_						,    lowerCamelCase_			) for terminal in terminals] if pos != -1
     ]

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




