
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        lowerCAmelCase       (				snake_case__   :      Dict				,					snake_case__   :      Optional[int]				,					snake_case__   :      List[Any]				,					snake_case__   :      str=None )->			Tuple:
    A_								=					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.
        A_								=					weight.view(torch.intaa )
        A_								=					"bfloat16"
    A_								=					weight.cpu().numpy()
    A_								=					os.path.join(_SCREAMING_SNAKE_CASE				,					f'{weight_name}.dat' )
    if index is not None:
        if dtype is None:
            A_								=					str(array.dtype )
        A_								=					{"dtype": dtype, "shape": list(array.shape )}
    if array.ndim == 0:
        A_								=					array[None]
    A_								=					np.memmap(_SCREAMING_SNAKE_CASE				,					dtype=array.dtype				,					mode="w+"				,					shape=array.shape )
    A_								=					array[:]
    file_array.flush()
    return index





def        lowerCAmelCase       (				snake_case__   :      int				,					snake_case__   :      Optional[Any] )->			List[Any]:
    A_								=					tuple(weight_info["shape"] )
    if shape == ():
        # NumPy memory-mapped arrays can't have 0 dims so it was saved as 1d tensor
        A_								=					(1,)

    A_								=					weight_info["dtype"]
    if dtype == "bfloat16":
        # NumPy does not support bfloat16 so this was saved as a int16
        A_								=					"int16"

    A_								=					np.memmap(_SCREAMING_SNAKE_CASE				,					dtype=_SCREAMING_SNAKE_CASE				,					shape=_SCREAMING_SNAKE_CASE				,					mode="r" )

    if len(weight_info["shape"] ) == 0:
        A_								=					weight[0]
    A_								=					torch.tensor(_SCREAMING_SNAKE_CASE )
    if weight_info["dtype"] == "bfloat16":
        A_								=					weight.view(torch.bfloataa )

    return weight





def        lowerCAmelCase       (				snake_case__   :      str				,					snake_case__   :      Tuple )->			List[str]:
    if index is None or len(_SCREAMING_SNAKE_CASE ) == 0:
        # Nothing to save
        return

    A_								=					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:
            A_								=					json.load(_SCREAMING_SNAKE_CASE )
    else:
        A_								=					{}
    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        lowerCAmelCase       (				snake_case__   :      Union[str, os.PathLike]				,					snake_case__   :      Dict[str, torch.Tensor] )->			List[Any]:
    os.makedirs(_SCREAMING_SNAKE_CASE				,					exist_ok=_SCREAMING_SNAKE_CASE )
    A_								=					{}
    for name, parameter in state_dict.items():
        A_								=					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 							lowerCamelCase       (		__UpperCAmelCase					):



  """simple docstring"""





  def __init__(					self						,    __UpperCamelCase						,    __UpperCamelCase					):
      A_								=					dataset
      A_								=					prefix

  def __getitem__(					self						,    __UpperCamelCase					):
      return self.dataset[f'{self.prefix}{key}']

  def __iter__(					self					):
      return iter([key for key in self.dataset if key.startswith(self.prefix					)]					)

  def __len__(					self					):
      return len(self.dataset					)



class 							lowerCamelCase       (		__UpperCAmelCase					):



  """simple docstring"""





  def __init__(					self						,    __UpperCamelCase = None						,    __UpperCamelCase = None						,    __UpperCamelCase = None						,    __UpperCamelCase=None						,    ):
      if state_dict is None and save_folder is None:
          raise ValueError("Need either a `state_dict` or a `save_folder` containing offloaded weights."					)

      A_								=					{} if state_dict is None else state_dict
      A_								=					save_folder
      if index is None and save_folder is not None:
          with open(os.path.join(__UpperCamelCase						,    "index.json"					)					) as f:
              A_								=					json.load(__UpperCamelCase					)
      A_								=					{} if index is None else index
      A_								=					list(self.state_dict.keys()					)
      self.all_keys.extend([key for key in self.index if key not in self.all_keys]					)
      A_								=					device

  def __getitem__(					self						,    __UpperCamelCase					):
      # State dict gets priority
      if key in self.state_dict:
          return self.state_dict[key]
      A_								=					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

          A_								=					"cpu" if self.device is None else self.device
          with safe_open(weight_info["safetensors_file"]						,    framework="pt"						,    device=__UpperCamelCase					) as f:
              A_								=					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

      A_								=					os.path.join(self.save_folder						,    f'{key}.dat'					)
      return load_offloaded_weight(__UpperCamelCase						,    __UpperCamelCase					)

  def __iter__(					self					):
      return iter(self.all_keys					)

  def __len__(					self					):
      return len(self.all_keys					)





def        lowerCAmelCase       (				snake_case__   :      Dict[str, torch.Tensor]				,					snake_case__   :      List[str] )->			Union[str, Any]:
    A_								=					{}
    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








'''simple docstring'''







from ..utils import DummyObject, requires_backends


class       lowerCAmelCase			(					metaclass=__UpperCAmelCase      ):
		a				:       Any							=       ["""torch""", """transformers""", """invisible_watermark"""]




		def __init__(					self       ,    *UpperCamelCase       ,    **UpperCamelCase   ):
								requires_backends(self       ,    ["torch", "transformers", "invisible_watermark"]   )




		@classmethod
		def      lowercase							(					cls       ,    *UpperCamelCase       ,    **UpperCamelCase   ):
								requires_backends(cls       ,    ["torch", "transformers", "invisible_watermark"]   )




		@classmethod
		def      lowercase							(					cls       ,    *UpperCamelCase       ,    **UpperCamelCase   ):
								requires_backends(cls       ,    ["torch", "transformers", "invisible_watermark"]   )


class       lowerCAmelCase			(					metaclass=__UpperCAmelCase      ):
		a				:       Optional[Any]							=       ["""torch""", """transformers""", """invisible_watermark"""]




		def __init__(					self       ,    *UpperCamelCase       ,    **UpperCamelCase   ):
								requires_backends(self       ,    ["torch", "transformers", "invisible_watermark"]   )




		@classmethod
		def      lowercase							(					cls       ,    *UpperCamelCase       ,    **UpperCamelCase   ):
								requires_backends(cls       ,    ["torch", "transformers", "invisible_watermark"]   )




		@classmethod
		def      lowercase							(					cls       ,    *UpperCamelCase       ,    **UpperCamelCase   ):
								requires_backends(cls       ,    ["torch", "transformers", "invisible_watermark"]   )