
"""simple docstring"""






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 __magic_name__  (							_lowerCamelCase:		Optional[Any],				_lowerCamelCase:		Union[str, Any],				_lowerCamelCase:		List[str],				_lowerCamelCase:		List[str]=None )				->	Dict:

	'''simple docstring'''




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




def __magic_name__  (							_lowerCamelCase:		List[str],				_lowerCamelCase:		List[str] )				->	Optional[Any]:

	'''simple docstring'''




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

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

	lowerCAmelCase				       =     np.memmap(_lowerCamelCase,				dtype=_lowerCamelCase,				shape=_lowerCamelCase,				mode='''r''' )

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

	return weight




def __magic_name__  (							_lowerCamelCase:		List[Any],				_lowerCamelCase:		Union[str, Any] )				->	str:

	'''simple docstring'''




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

	lowerCAmelCase				       =     os.path.join(_lowerCamelCase,				'''index.json''' )
	if os.path.isfile(_lowerCamelCase ):
		with open(_lowerCamelCase,				'''r''',				encoding='''utf-8''' ) as f:
			lowerCAmelCase				       =     json.load(_lowerCamelCase )
	else:
		lowerCAmelCase				       =     {}
	current_index.update(_lowerCamelCase )

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




def __magic_name__  (							_lowerCamelCase:		Union[str, os.PathLike],				_lowerCamelCase:		Dict[str, torch.Tensor] )				->	List[Any]:

	'''simple docstring'''




	os.makedirs(_lowerCamelCase,				exist_ok=_lowerCamelCase )
	lowerCAmelCase				       =     {}
	for name, parameter in state_dict.items():
		lowerCAmelCase				       =     offload_weight(_lowerCamelCase,				_lowerCamelCase,				_lowerCamelCase,				index=_lowerCamelCase )

	# Update index
	save_offload_index(_lowerCamelCase,				_lowerCamelCase )



class  lowercase				(   UpperCamelCase__		):






			def __init__(self     :  str			,SCREAMING_SNAKE_CASE_     :  int			,SCREAMING_SNAKE_CASE_     :  int )					->  Optional[Any]:





				"""simple docstring"""


				lowerCAmelCase				       =     dataset
				lowerCAmelCase				       =     prefix






			def __getitem__(self     :  Union[str, Any]			,SCREAMING_SNAKE_CASE_     :  str )					->  str:





				"""simple docstring"""


				return self.dataset[F"""{self.prefix}{key}"""]






			def __iter__(self     :  str )					->  Union[str, Any]:





				"""simple docstring"""


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






			def __len__(self     :  str )					->  Optional[int]:





				"""simple docstring"""


				return len(self.dataset )





class  lowercase				(   UpperCamelCase__		):






			def __init__(self     :  Tuple			,SCREAMING_SNAKE_CASE_     :  Dict = None			,SCREAMING_SNAKE_CASE_     :  Tuple = None			,SCREAMING_SNAKE_CASE_     :  Optional[Any] = None			,SCREAMING_SNAKE_CASE_     :  Optional[int]=None			,)					->  Any:





				"""simple docstring"""


				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				       =     {} if state_dict is None else state_dict
				lowerCAmelCase				       =     save_folder
				if index is None and save_folder is not None:
					with open(os.path.join(lowerCAmelCase__			,'''index.json''' ) ) as f:
						lowerCAmelCase				       =     json.load(lowerCAmelCase__ )
				lowerCAmelCase				       =     {} if index is None else index
				lowerCAmelCase				       =     list(self.state_dict.keys() )
				self.all_keys.extend([key for key in self.index if key not in self.all_keys] )
				lowerCAmelCase				       =     device






			def __getitem__(self     :  Union[str, Any]			,SCREAMING_SNAKE_CASE_     :  Optional[int] )					->  str:





				"""simple docstring"""


				if key in self.state_dict:
					return self.state_dict[key]
				lowerCAmelCase				       =     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				       =     "cpu" if self.device is None else self.device
					with safe_open(weight_info['''safetensors_file''']			,framework='''pt'''			,device=lowerCAmelCase__ ) as f:
						lowerCAmelCase				       =     f.get_tensor(weight_info.get('''weight_name'''			,lowerCAmelCase__ ) )

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

				lowerCAmelCase				       =     os.path.join(self.save_folder			,F"""{key}.dat""" )
				return load_offloaded_weight(lowerCAmelCase__			,lowerCAmelCase__ )






			def __iter__(self     :  Tuple )					->  Optional[Any]:





				"""simple docstring"""


				return iter(self.all_keys )






			def __len__(self     :  Optional[int] )					->  Optional[Any]:





				"""simple docstring"""


				return len(self.all_keys )





def __magic_name__  (							_lowerCamelCase:		Dict[str, torch.Tensor],				_lowerCamelCase:		List[str] )				->	int:

	'''simple docstring'''




	lowerCAmelCase				       =     {}
	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"""






def __magic_name__  (							_lowerCamelCase:		list[int] )				->	int:

	'''simple docstring'''




	lowerCAmelCase				       =     len(_lowerCamelCase )
	if n in (0, 1):
		return 0
	return sum(query * (n - i - 1) for i, query in enumerate(sorted(_lowerCamelCase ) ) )


if __name__ == "__main__":
					import doctest

					doctest.testmod()


