


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




     '''simple docstring'''



     _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(lowerCAmelCase__					,    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(lowerCAmelCase__					,    dtype=array.dtype					,    mode='w+'					,    shape=array.shape		)
     _a         =				array[:]
     file_array.flush()
     return index






def 						_A					(lowerCAmelCase__	:int					,    lowerCAmelCase__	:Dict		) ->	Optional[int]:




     '''simple docstring'''



     _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(lowerCAmelCase__					,    dtype=lowerCAmelCase__					,    shape=lowerCAmelCase__					,    mode='r'		)

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

     return weight






def 						_A					(lowerCAmelCase__	:Any					,    lowerCAmelCase__	:List[str]		) ->	List[str]:




     '''simple docstring'''



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

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

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






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




     '''simple docstring'''



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

     # Update index
     save_offload_index(lowerCAmelCase__					,    lowerCAmelCase__		)




class 			a  (  _SCREAMING_SNAKE_CASE							):
 def __init__(			self   ,					__magic_name__   ,					__magic_name__    )       ->				Dict:
      _a         =				dataset
      _a         =				prefix
 def __getitem__(			self   ,					__magic_name__    )       ->				Any:
      return self.dataset[f'{self.prefix}{key}']
 def __iter__(			self    )       ->				str:
      return iter([key for key in self.dataset if key.startswith(self.prefix    )]    )




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





class 			a  (  _SCREAMING_SNAKE_CASE							):
 def __init__(			self   ,					__magic_name__ = None   ,					__magic_name__ = None   ,					__magic_name__ = None   ,					__magic_name__=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.'    )

      _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(__magic_name__   ,					'index.json'    )    ) as f:
                _a         =				json.load(__magic_name__    )
      _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   ,					__magic_name__    )       ->				List[str]:
      # 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=__magic_name__    ) as f:
                _a         =				f.get_tensor(weight_info.get('weight_name'   ,					__magic_name__    )    )

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

      _a         =				os.path.join(self.save_folder   ,					f'{key}.dat'    )
      return load_offloaded_weight(__magic_name__   ,					__magic_name__    )
 def __iter__(			self    )       ->				Optional[int]:
      return iter(self.all_keys    )




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






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




     '''simple docstring'''



     _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 dataclasses import dataclass
from enum import Enum
from typing import List, Optional, Union

import numpy as np
import PIL
from PIL import Image

from ...utils import BaseOutput, is_torch_available, is_transformers_available




@dataclass
class 			a  (  _SCREAMING_SNAKE_CASE							):
 _lowerCAmelCase			       =						{
     """sld_warmup_steps""": 1_5,
     """sld_guidance_scale""": 2_0,
     """sld_threshold""": 0.0,
     """sld_momentum_scale""": 0.0,
     """sld_mom_beta""": 0.0,
 }
 _lowerCAmelCase			       =						{
     """sld_warmup_steps""": 1_0,
     """sld_guidance_scale""": 1_0_0_0,
     """sld_threshold""": 0.01,
     """sld_momentum_scale""": 0.3,
     """sld_mom_beta""": 0.4,
 }
 _lowerCAmelCase			       =						{
     """sld_warmup_steps""": 7,
     """sld_guidance_scale""": 2_0_0_0,
     """sld_threshold""": 0.025,
     """sld_momentum_scale""": 0.5,
     """sld_mom_beta""": 0.7,
 }
 _lowerCAmelCase			       =						{
     """sld_warmup_steps""": 0,
     """sld_guidance_scale""": 5_0_0_0,
     """sld_threshold""": 1.0,
     """sld_momentum_scale""": 0.5,
     """sld_mom_beta""": 0.7,
 }





@dataclass
class 			a  (  _SCREAMING_SNAKE_CASE							):
 _lowerCAmelCase			       =						42
 _lowerCAmelCase			       =						42
 _lowerCAmelCase			       =						42
 _lowerCAmelCase			       =						42


if is_transformers_available() and is_torch_available():
 from .pipeline_stable_diffusion_safe import StableDiffusionPipelineSafe
 from .safety_checker import SafeStableDiffusionSafetyChecker

