

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    snake_case_(_UpperCamelCase				, _UpperCamelCase				, _UpperCamelCase				, _UpperCamelCase=None  )							->			Optional[int]:



  """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    snake_case_(_UpperCamelCase				, _UpperCamelCase  )							->			Optional[Any]:



  """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    snake_case_(_UpperCamelCase				, _UpperCamelCase  )							->			Optional[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    snake_case_(_UpperCamelCase				, _UpperCamelCase  )							->			Tuple:



  """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 						lowercase_					(						_SCREAMING_SNAKE_CASE  ):



      def __init__(    self   :       Dict    ,      A__   :       Optional[Any]    ,      A__   :       Union[str, Any]	)					->  Optional[Any]:
        _snake_case		=							dataset
        _snake_case		=							prefix



      def __getitem__(    self   :       Optional[int]    ,      A__   :       List[Any]	)					->  Optional[Any]:
        return self.dataset[f"""{self.prefix}{key}"""]



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







      def __len__(    self   :       Any	)					->  Any:
        return len(self.dataset	)



class 						lowercase_					(						_SCREAMING_SNAKE_CASE  ):



      def __init__(    self   :       Optional[int]    ,      A__   :       Optional[int] = None    ,      A__   :       List[Any] = None    ,      A__   :       Union[str, Any] = None    ,      A__   :       Any=None    ,      )					->  Any:
        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(__A    ,      '''index.json'''	)	) as f:
            _snake_case		=							json.load(__A	)
        _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   :       str    ,      A__   :       Optional[Any]	)					->  int:
        # State dict gets priority
        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=__A	) as f:
            _snake_case		=							f.get_tensor(weight_info.get('''weight_name'''    ,      __A	)	)

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

        _snake_case		=							os.path.join(self.save_folder    ,      f"""{key}.dat"""	)
        return load_offloaded_weight(__A    ,      __A	)



      def __iter__(    self   :       Union[str, Any]	)					->  Optional[int]:
        return iter(self.all_keys	)







      def __len__(    self   :       Union[str, Any]	)					->  int:
        return len(self.all_keys	)



def    snake_case_(_UpperCamelCase				, _UpperCamelCase  )							->			int:



  """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




from collections import namedtuple

__A    	=			namedtuple('''from_to''', '''from_ to''')

__A    	=			{
    '''atm''': from_to(1, 1),
    '''pascal''': from_to(0.000_0098, 10_13_25),
    '''bar''': from_to(0.98_6923, 1.0_1325),
    '''kilopascal''': from_to(0.0098_6923, 101.325),
    '''megapascal''': from_to(9.8_6923, 0.10_1325),
    '''psi''': from_to(0.06_8046, 14.6959),
    '''inHg''': from_to(0.033_4211, 29.9213),
    '''torr''': from_to(0.0013_1579, 7_60),
}



def    snake_case_(_UpperCamelCase				, _UpperCamelCase				, _UpperCamelCase  )							->			float:



  """simple docstring"""






  if from_type not in PRESSURE_CONVERSION:
    raise ValueError(
        F"""Invalid 'from_type' value: {from_type!r}  Supported values are:\n"""
        + ''', '''.join(_UpperCamelCase  )  )
  if to_type not in PRESSURE_CONVERSION:
    raise ValueError(
        F"""Invalid 'to_type' value: {to_type!r}.  Supported values are:\n"""
        + ''', '''.join(_UpperCamelCase  )  )
  return (
      value * PRESSURE_CONVERSION[from_type].from_ * PRESSURE_CONVERSION[to_type].to
  )


if __name__ == "__main__":
 import doctest

 doctest.testmod()

