"""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  lowerCAmelCase__     (      _UpperCamelCase							:					Optional[int]			,      _UpperCamelCase							:					Tuple			,      _UpperCamelCase							:					Union[str, Any]			,      _UpperCamelCase							:					Dict=None       )       ->					str:





						"""simple docstring"""






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





def  lowerCAmelCase__     (      _UpperCamelCase							:					int			,      _UpperCamelCase							:					str       )       ->					Optional[int]:





						"""simple docstring"""






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

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

						snake_case             =      np.memmap(_UpperCamelCase			,      dtype=_UpperCamelCase			,      shape=_UpperCamelCase			,      mode='r'       )

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

						return weight





def  lowerCAmelCase__     (      _UpperCamelCase							:					Union[str, Any]			,      _UpperCamelCase							:					Dict       )       ->					Any:





						"""simple docstring"""






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

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

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





def  lowerCAmelCase__     (      _UpperCamelCase							:					Union[str, os.PathLike]			,      _UpperCamelCase							:					Dict[str, torch.Tensor]       )       ->					List[str]:





						"""simple docstring"""






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

						# Update index
						save_offload_index(_UpperCamelCase			,      _UpperCamelCase       )



class 	lowerCAmelCase_     (      lowerCAmelCase       ):



				"""simple docstring"""

				def __init__(					self		,     lowerCAmelCase		,     lowerCAmelCase     ):






										"""simple docstring"""

										snake_case             =      dataset
										snake_case             =      prefix

				def __getitem__(					self		,     lowerCAmelCase     ):






										"""simple docstring"""

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

				def __iter__(					self     ):






										"""simple docstring"""

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

				def __len__(					self     ):






										"""simple docstring"""

										return len(self.dataset     )





class 	lowerCAmelCase_     (      lowerCAmelCase       ):



				"""simple docstring"""

				def __init__(					self		,     lowerCAmelCase = None		,     lowerCAmelCase = None		,     lowerCAmelCase = None		,     lowerCAmelCase=None		,     ):






										"""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.'     )

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

				def __getitem__(					self		,     lowerCAmelCase     ):






										"""simple docstring"""

										if key in self.state_dict:
																return self.state_dict[key]
										snake_case             =      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

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

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

				def __iter__(					self     ):






										"""simple docstring"""

										return iter(self.all_keys     )

				def __len__(					self     ):






										"""simple docstring"""

										return len(self.all_keys     )





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





						"""simple docstring"""






						snake_case             =      {}
						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  lowerCAmelCase__     (      _UpperCamelCase							:					list       )       ->					list:





						"""simple docstring"""






						if len(_UpperCamelCase       ) <= 1:
												return data
						else:
												return [
												    *quick_sort([e for e in data[1:] if e <= data[0]]       ),
												    data[0],
												    *quick_sort([e for e in data[1:] if e > data[0]]       ),
												]


if __name__ == "__main__":
					import doctest

					doctest.testmod()


