

import json
import os
from collections.abc import Mapping
from typing import Dict, List, Optional, Union

import numpy as np
import torch

from .imports import is_safetensors_available







def _a				( SCREAMING_SNAKE_CASE	:    Optional[Any]					,      SCREAMING_SNAKE_CASE	:    Optional[Any]					,      SCREAMING_SNAKE_CASE	:    Tuple					,      SCREAMING_SNAKE_CASE	:    Union[str, Any]=None	)					->					Dict:

 """simple docstring"""






 __lowerCAmelCase:     Union[str, Any]     			=   None
 # Check the string instead of the dtype to be compatible with versions of PyTorch that don't have bfloat16.
 if str(weight.dtype	) == "torch.bfloat16":
  # Need to reinterpret the underlined data as int16 since NumPy does not handle bfloat16s.
  __lowerCAmelCase:     Optional[int]     			=   weight.view(torch.intaa	)
  __lowerCAmelCase:     List[str]     			=   'bfloat16'
 __lowerCAmelCase:     Dict     			=   weight.cpu().numpy()
 __lowerCAmelCase:     Dict     			=   os.path.join(SCREAMING_SNAKE_CASE					,      f'''{weight_name}.dat'''	)
 if index is not None:
  if dtype is None:
   __lowerCAmelCase:     Union[str, Any]     			=   str(array.dtype	)
  __lowerCAmelCase:     Optional[Any]     			=   {'dtype': dtype, 'shape': list(array.shape	)}
 if array.ndim == 0:
  __lowerCAmelCase:     Optional[int]     			=   array[None]
 __lowerCAmelCase:     str     			=   np.memmap(SCREAMING_SNAKE_CASE					,      dtype=array.dtype					,      mode='w+'					,      shape=array.shape	)
 __lowerCAmelCase:     List[Any]     			=   array[:]
 file_array.flush()
 return index







def _a				( SCREAMING_SNAKE_CASE	:    List[Any]					,      SCREAMING_SNAKE_CASE	:    List[Any]	)					->					Any:

 """simple docstring"""






 __lowerCAmelCase:     Dict     			=   tuple(weight_info['shape']	)
 if shape == ():
  # NumPy memory-mapped arrays can't have 0 dims so it was saved as 1d tensor
  __lowerCAmelCase:     Tuple     			=   (1,)

 __lowerCAmelCase:     Dict     			=   weight_info['dtype']
 if dtype == "bfloat16":
  # NumPy does not support bfloat16 so this was saved as a int16
  __lowerCAmelCase:     Optional[Any]     			=   'int16'

 __lowerCAmelCase:     Tuple     			=   np.memmap(SCREAMING_SNAKE_CASE					,      dtype=SCREAMING_SNAKE_CASE					,      shape=SCREAMING_SNAKE_CASE					,      mode='r'	)

 if len(weight_info['shape']	) == 0:
  __lowerCAmelCase:     Optional[int]     			=   weight[0]
 __lowerCAmelCase:     Dict     			=   torch.tensor(SCREAMING_SNAKE_CASE	)
 if weight_info["dtype"] == "bfloat16":
  __lowerCAmelCase:     Tuple     			=   weight.view(torch.bfloataa	)

 return weight







def _a				( SCREAMING_SNAKE_CASE	:    Any					,      SCREAMING_SNAKE_CASE	:    int	)					->					List[str]:

 """simple docstring"""






 if index is None or len(SCREAMING_SNAKE_CASE	) == 0:
  # Nothing to save
  return

 __lowerCAmelCase:     List[str]     			=   os.path.join(SCREAMING_SNAKE_CASE					,      'index.json'	)
 if os.path.isfile(SCREAMING_SNAKE_CASE	):
  with open(SCREAMING_SNAKE_CASE					,      'r'					,      encoding='utf-8'	) as f:
   __lowerCAmelCase:     Tuple     			=   json.load(SCREAMING_SNAKE_CASE	)
 else:
  __lowerCAmelCase:     List[Any]     			=   {}
 current_index.update(SCREAMING_SNAKE_CASE	)

 with open(SCREAMING_SNAKE_CASE					,      'w'					,      encoding='utf-8'	) as f:
  json.dump(SCREAMING_SNAKE_CASE					,      SCREAMING_SNAKE_CASE					,      indent=2	)







def _a				( SCREAMING_SNAKE_CASE	:    Union[str, os.PathLike]					,      SCREAMING_SNAKE_CASE	:    Dict[str, torch.Tensor]	)					->					Any:

 """simple docstring"""






 os.makedirs(SCREAMING_SNAKE_CASE					,      exist_ok=SCREAMING_SNAKE_CASE	)
 __lowerCAmelCase:     Dict     			=   {}
 for name, parameter in state_dict.items():
  __lowerCAmelCase:     Tuple     			=   offload_weight(SCREAMING_SNAKE_CASE					,      SCREAMING_SNAKE_CASE					,      SCREAMING_SNAKE_CASE					,      index=SCREAMING_SNAKE_CASE	)

 # Update index
 save_offload_index(SCREAMING_SNAKE_CASE					,      SCREAMING_SNAKE_CASE	)






class 							A_				(	snake_case__      ):


     def __init__(							self    :	Tuple    ,						UpperCAmelCase    :	Mapping    ,						UpperCAmelCase    :	str      )							->			List[str]:
      __lowerCAmelCase:     Tuple     			=   dataset
      __lowerCAmelCase:     str     			=   prefix


     def __getitem__(							self    :	List[str]    ,						UpperCAmelCase    :	int      )							->			List[str]:
      return self.dataset[F'''{self.prefix}{key}''']


     def __iter__(							self    :	Union[str, Any]      )							->			Optional[int]:
      return iter([key for key in self.dataset if key.startswith(self.prefix      )]      )


     def __len__(							self    :	str      )							->			Optional[int]:
      return len(self.dataset      )







class 							A_				(	snake_case__      ):


     def __init__(							self    :	List[Any]    ,						UpperCAmelCase    :	Dict[str, torch.Tensor] = None    ,						UpperCAmelCase    :	Optional[Union[str, os.PathLike]] = None    ,						UpperCAmelCase    :	Mapping = None    ,						UpperCAmelCase    :	Any=None    ,						)							->			Dict:
      if state_dict is None and save_folder is None:
       raise ValueError('Need either a `state_dict` or a `save_folder` containing offloaded weights.'      )

      __lowerCAmelCase:     List[Any]     			=   {} if state_dict is None else state_dict
      __lowerCAmelCase:     List[Any]     			=   save_folder
      if index is None and save_folder is not None:
       with open(os.path.join(UpperCAmelCase    ,						'index.json'      )      ) as f:
        __lowerCAmelCase:     List[str]     			=   json.load(UpperCAmelCase      )
      __lowerCAmelCase:     List[Any]     			=   {} if index is None else index
      __lowerCAmelCase:     str     			=   list(self.state_dict.keys()      )
      self.all_keys.extend([key for key in self.index if key not in self.all_keys]      )
      __lowerCAmelCase:     Optional[Any]     			=   device


     def __getitem__(							self    :	str    ,						UpperCAmelCase    :	str      )							->			int:
      # State dict gets priority
      if key in self.state_dict:
       return self.state_dict[key]
      __lowerCAmelCase:     Optional[Any]     			=   self.index[key]
      if weight_info.get('safetensors_file'      ) is not None:
       if not is_safetensors_available():
        raise ImportError('These offloaded weights require the use of safetensors: `pip install safetensors`.'      )

       from safetensors import safe_open

       __lowerCAmelCase:     Dict     			=   'cpu' if self.device is None else self.device
       with safe_open(weight_info['safetensors_file']    ,						framework='pt'    ,						device=UpperCAmelCase      ) as f:
        __lowerCAmelCase:     str     			=   f.get_tensor(weight_info.get('weight_name'    ,						UpperCAmelCase      )      )

       if "dtype" in weight_info:
        return tensor.to(getattr(UpperCAmelCase    ,						weight_info['dtype']      )      )
       else:
        return tensor

      __lowerCAmelCase:     Optional[Any]     			=   os.path.join(self.save_folder    ,						F'''{key}.dat'''      )
      return load_offloaded_weight(UpperCAmelCase    ,						UpperCAmelCase      )


     def __iter__(							self    :	str      )							->			List[Any]:
      return iter(self.all_keys      )


     def __len__(							self    :	int      )							->			str:
      return len(self.all_keys      )







def _a				( SCREAMING_SNAKE_CASE	:    Dict[str, torch.Tensor]					,      SCREAMING_SNAKE_CASE	:    List[str]	)					->					List[Any]:

 """simple docstring"""






 __lowerCAmelCase:     Optional[int]     			=   {}
 for module_name in submodule_names:
  # We want to catch module_name parameter (module_name.xxx) or potentially module_name, but not any of the
  # submodules that could being like module_name (transformers.h.1 and transformers.h.10 for instance)
  result.update(
      {
          key: param
          for key, param in state_dict.items()
          if key == module_name or key.startswith(module_name + '.'	)
      }	)
 return result








from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_flax_available,
    is_keras_nlp_available,
    is_tensorflow_text_available,
    is_tf_available,
    is_tokenizers_available,
    is_torch_available,
)


_a	 =     {
    '''configuration_gpt2''': ['''GPT2_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''GPT2Config''', '''GPT2OnnxConfig'''],
    '''tokenization_gpt2''': ['''GPT2Tokenizer'''],
}

try:
  if not is_tokenizers_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  _a	 =     ['''GPT2TokenizerFast''']

try:
  if not is_torch_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  _a	 =     [
      '''GPT2_PRETRAINED_MODEL_ARCHIVE_LIST''',
      '''GPT2DoubleHeadsModel''',
      '''GPT2ForQuestionAnswering''',
      '''GPT2ForSequenceClassification''',
      '''GPT2ForTokenClassification''',
      '''GPT2LMHeadModel''',
      '''GPT2Model''',
      '''GPT2PreTrainedModel''',
      '''load_tf_weights_in_gpt2''',
  ]

try:
  if not is_tf_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  _a	 =     [
      '''TF_GPT2_PRETRAINED_MODEL_ARCHIVE_LIST''',
      '''TFGPT2DoubleHeadsModel''',
      '''TFGPT2ForSequenceClassification''',
      '''TFGPT2LMHeadModel''',
      '''TFGPT2MainLayer''',
      '''TFGPT2Model''',
      '''TFGPT2PreTrainedModel''',
  ]

try:
  if not is_keras_nlp_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  _a	 =     ['''TFGPT2Tokenizer''']

try:
  if not is_flax_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  _a	 =     ['''FlaxGPT2LMHeadModel''', '''FlaxGPT2Model''', '''FlaxGPT2PreTrainedModel''']

if TYPE_CHECKING:
  from .configuration_gpta import GPT2_PRETRAINED_CONFIG_ARCHIVE_MAP, GPTaConfig, GPTaOnnxConfig
  from .tokenization_gpta import GPTaTokenizer

  try:
    if not is_tokenizers_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .tokenization_gpta_fast import GPTaTokenizerFast

  try:
    if not is_torch_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .modeling_gpta import (
        GPT2_PRETRAINED_MODEL_ARCHIVE_LIST,
        GPTaDoubleHeadsModel,
        GPTaForQuestionAnswering,
        GPTaForSequenceClassification,
        GPTaForTokenClassification,
        GPTaLMHeadModel,
        GPTaModel,
        GPTaPreTrainedModel,
        load_tf_weights_in_gpta,
    )

  try:
    if not is_tf_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .modeling_tf_gpta import (
        TF_GPT2_PRETRAINED_MODEL_ARCHIVE_LIST,
        TFGPTaDoubleHeadsModel,
        TFGPTaForSequenceClassification,
        TFGPTaLMHeadModel,
        TFGPTaMainLayer,
        TFGPTaModel,
        TFGPTaPreTrainedModel,
    )

  try:
    if not is_keras_nlp_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .tokenization_gpta_tf import TFGPTaTokenizer

  try:
    if not is_flax_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .modeling_flax_gpta import FlaxGPTaLMHeadModel, FlaxGPTaModel, FlaxGPTaPreTrainedModel

else:
  import sys

  _a	 =     _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)





