






'''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 							_SCREAMING_SNAKE_CASE	(					A	:							int	,		A	:							Dict	,		A	:							List[str]	,		A	:							List[str]=None	)     ->			Optional[Any]:







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




def 							_SCREAMING_SNAKE_CASE	(					A	:							str	,		A	:							Union[str, Any]	)     ->			List[Any]:







   """simple docstring"""
   __snake_case					:   int  				=			tuple(weight_info['shape']	)
   if shape == ():
      # NumPy memory-mapped arrays can't have 0 dims so it was saved as 1d tensor
      __snake_case					:   Union[str, Any]  				=			(1,)

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

   __snake_case					:   str  				=			np.memmap(A	,		dtype=A	,		shape=A	,		mode='r'	)

   if len(weight_info['shape']	) == 0:
      __snake_case					:   Optional[int]  				=			weight[0]
   __snake_case					:   Optional[int]  				=			torch.tensor(A	)
   if weight_info["dtype"] == "bfloat16":
      __snake_case					:   Tuple  				=			weight.view(torch.bfloataa	)

   return weight




def 							_SCREAMING_SNAKE_CASE	(					A	:							List[Any]	,		A	:							Any	)     ->			Union[str, Any]:







   """simple docstring"""
   if index is None or len(A	) == 0:
      # Nothing to save
      return

   __snake_case					:   List[Any]  				=			os.path.join(A	,		'index.json'	)
   if os.path.isfile(A	):
      with open(A	,		'r'	,		encoding='utf-8'	) as f:
         __snake_case					:   Dict  				=			json.load(A	)
   else:
      __snake_case					:   Tuple  				=			{}
   current_index.update(A	)

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




def 							_SCREAMING_SNAKE_CASE	(					A	:							int	,		A	:							Optional[int]	)     ->			int:







   """simple docstring"""
   os.makedirs(A	,		exist_ok=A	)
   __snake_case					:   List[Any]  				=			{}
   for name, parameter in state_dict.items():
      __snake_case					:   Optional[int]  				=			offload_weight(A	,		A	,		A	,		index=A	)

   # Update index
   save_offload_index(A	,		A	)




class a_     (     UpperCamelCase_				):







   def __init__(self       , __a       , __a)      ->   List[Any]:
      """simple docstring"""






      __snake_case					:   str  				=			dataset
      __snake_case					:   str  				=			prefix







   def __getitem__(self       , __a)      ->   Tuple:
      """simple docstring"""






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







   def __iter__(self)      ->   Optional[Any]:
      """simple docstring"""






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







   def __len__(self)      ->   Any:
      """simple docstring"""






      return len(self.dataset)




class a_     (     UpperCamelCase_				):







   def __init__(self       , __a = None       , __a = None       , __a = None       , __a=None       , )      ->   str:
      """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					:   str  				=			{} if state_dict is None else state_dict
      __snake_case					:   Optional[int]  				=			save_folder
      if index is None and save_folder is not None:
         with open(os.path.join(_lowerCAmelCase       , 'index.json')) as f:
            __snake_case					:   int  				=			json.load(_lowerCAmelCase)
      __snake_case					:   Tuple  				=			{} if index is None else index
      __snake_case					:   int  				=			list(self.state_dict.keys())
      self.all_keys.extend([key for key in self.index if key not in self.all_keys])
      __snake_case					:   int  				=			device







   def __getitem__(self       , __a)      ->   List[Any]:
      """simple docstring"""






      if key in self.state_dict:
         return self.state_dict[key]
      __snake_case					:   Dict  				=			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					:   List[str]  				=			"cpu" if self.device is None else self.device
         with safe_open(weight_info['safetensors_file']       , framework='pt'       , device=_lowerCAmelCase) as f:
            __snake_case					:   List[Any]  				=			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					:   List[str]  				=			os.path.join(self.save_folder       , F"""{key}.dat""")
      return load_offloaded_weight(_lowerCAmelCase       , _lowerCAmelCase)







   def __iter__(self)      ->   Optional[Any]:
      """simple docstring"""






      return iter(self.all_keys)







   def __len__(self)      ->   List[str]:
      """simple docstring"""






      return len(self.all_keys)




def 							_SCREAMING_SNAKE_CASE	(					A	:							Tuple	,		A	:							str	)     ->			str:







   """simple docstring"""
   __snake_case					:   Any  				=			{}
   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





import tempfile
import unittest

import torch

from diffusers import UNetaDConditionModel
from diffusers.training_utils import EMAModel
from diffusers.utils.testing_utils import enable_full_determinism, skip_mps, torch_device


enable_full_determinism()




class  __lowerCAmelCase							(				unittest.TestCase):
       _a			       =    '''hf-internal-testing/tiny-stable-diffusion-pipe'''
       _a			       =    1
       _a			       =    77
       _a			       =    32
       _a			       =    4
       _a			       =    _a			       =    64
       _a			       =    torch.manual_seed(0)






       def        SCREAMING_SNAKE_CASE     ( self:       Union[str, Any]							,       _lowerCAmelCase:       int=0.99_99					):
         lowercase     :Optional[int]   						=    UNetaDConditionModel.from_pretrained(self.model_id							,       subfolder="unet"					)
         lowercase     :Union[str, Any]   						=    unet.to(_lowerCAmelCase					)
         lowercase     :Any   						=    EMAModel(unet.parameters()							,       decay=_lowerCAmelCase							,       model_cls=_lowerCAmelCase							,       model_config=unet.config					)
         return unet, ema_unet






       def        SCREAMING_SNAKE_CASE     ( self:       List[str]					):
         lowercase     :Optional[Any]   						=    torch.randn(
             self.batch_size							,       self.num_in_channels							,       self.latent_height							,       self.latent_width							,       generator=self.generator					).to(_lowerCAmelCase					)
         lowercase     :Dict   						=    torch.randint(0							,       10_00							,       size=(self.batch_size,)							,       generator=self.generator					).to(_lowerCAmelCase					)
         lowercase     :List[Any]   						=    torch.randn(
             self.batch_size							,       self.prompt_length							,       self.text_encoder_hidden_dim							,       generator=self.generator					).to(_lowerCAmelCase					)
         return noisy_latents, timesteps, encoder_hidden_states






       def        SCREAMING_SNAKE_CASE     ( self:       Optional[int]							,       _lowerCAmelCase:       Optional[Any]					):
         lowercase     :Any   						=    {}
         for k, param in unet.state_dict().items():
           lowercase     :Optional[int]   						=    torch.randn_like(_lowerCAmelCase					) + (param * torch.randn_like(_lowerCAmelCase					))
           updated_state_dict.update({k: updated_param}					)
         unet.load_state_dict(_lowerCAmelCase					)
         return unet






       def        SCREAMING_SNAKE_CASE     ( self:       int					):
         lowercase						,     lowercase     :List[str]   						=    self.get_models()
         # Take the first (hypothetical) EMA step.
         ema_unet.step(unet.parameters()					)
         assert ema_unet.optimization_step == 1

         # Take two more.
         for _ in range(2					):
           ema_unet.step(unet.parameters()					)
         assert ema_unet.optimization_step == 3






       def        SCREAMING_SNAKE_CASE     ( self:       Any					):
         lowercase						,     lowercase     :str   						=    self.get_models()
         # Since the `unet` is not being updated (i.e., backprop'd)
         # there won't be any difference between the `params` of `unet`
         # and `ema_unet` even if we call `ema_unet.step(unet.parameters())`.
         ema_unet.step(unet.parameters()					)
         lowercase     :Tuple   						=    list(unet.parameters()					)
         for s_param, param in zip(ema_unet.shadow_params							,       _lowerCAmelCase					):
           assert torch.allclose(_lowerCAmelCase							,       _lowerCAmelCase					)

         # The above holds true even if we call `ema.step()` multiple times since
         # `unet` params are still not being updated.
         for _ in range(4					):
           ema_unet.step(unet.parameters()					)
         for s_param, param in zip(ema_unet.shadow_params							,       _lowerCAmelCase					):
           assert torch.allclose(_lowerCAmelCase							,       _lowerCAmelCase					)






       def        SCREAMING_SNAKE_CASE     ( self:       Dict					):
         lowercase						,     lowercase     :Optional[int]   						=    self.get_models()
         # Here we simulate the parameter updates for `unet`. Since there might
         # be some parameters which are initialized to zero we take extra care to
         # initialize their values to something non-zero before the multiplication.
         lowercase     :List[str]   						=    self.simulate_backprop(_lowerCAmelCase					)

         # Take the EMA step.
         ema_unet.step(unet_pseudo_updated_step_one.parameters()					)

         # Now the EMA'd parameters won't be equal to the original model parameters.
         lowercase     :str   						=    list(unet_pseudo_updated_step_one.parameters()					)
         for s_param, param in zip(ema_unet.shadow_params							,       _lowerCAmelCase					):
           assert ~torch.allclose(_lowerCAmelCase							,       _lowerCAmelCase					)

         # Ensure this is the case when we take multiple EMA steps.
         for _ in range(4					):
           ema_unet.step(unet.parameters()					)
         for s_param, param in zip(ema_unet.shadow_params							,       _lowerCAmelCase					):
           assert ~torch.allclose(_lowerCAmelCase							,       _lowerCAmelCase					)






       def        SCREAMING_SNAKE_CASE     ( self:       Optional[int]					):
         # If we call EMA step after a backpropagation consecutively for two times,
         # the shadow params from those two steps should be different.
         lowercase						,     lowercase     :Optional[Any]   						=    self.get_models()

         # First backprop + EMA
         lowercase     :Union[str, Any]   						=    self.simulate_backprop(_lowerCAmelCase					)
         ema_unet.step(unet_step_one.parameters()					)
         lowercase     :int   						=    ema_unet.shadow_params

         # Second backprop + EMA
         lowercase     :Optional[int]   						=    self.simulate_backprop(_lowerCAmelCase					)
         ema_unet.step(unet_step_two.parameters()					)
         lowercase     :Optional[int]   						=    ema_unet.shadow_params

         for step_one, step_two in zip(_lowerCAmelCase							,       _lowerCAmelCase					):
           assert ~torch.allclose(_lowerCAmelCase							,       _lowerCAmelCase					)






       def        SCREAMING_SNAKE_CASE     ( self:       Union[str, Any]					):
         # If there's no decay even if there are backprops, EMA steps
         # won't take any effect i.e., the shadow params would remain the
         # same.
         lowercase						,     lowercase     :Any   						=    self.get_models(decay=0.0					)
         lowercase     :List[Any]   						=    self.simulate_backprop(_lowerCAmelCase					)
         ema_unet.step(unet_step_one.parameters()					)
         lowercase     :List[str]   						=    ema_unet.shadow_params

         lowercase     :str   						=    self.simulate_backprop(_lowerCAmelCase					)
         ema_unet.step(unet_step_two.parameters()					)
         lowercase     :List[str]   						=    ema_unet.shadow_params

         for step_one, step_two in zip(_lowerCAmelCase							,       _lowerCAmelCase					):
           assert torch.allclose(_lowerCAmelCase							,       _lowerCAmelCase					)






       @skip_mps
       def        SCREAMING_SNAKE_CASE     ( self:       List[Any]					):
         lowercase						,     lowercase     :Optional[Any]   						=    self.get_models()
         lowercase						,     lowercase						,     lowercase     :str   						=    self.get_dummy_inputs()

         with tempfile.TemporaryDirectory() as tmpdir:
           ema_unet.save_pretrained(_lowerCAmelCase					)
           lowercase     :Optional[int]   						=    UNetaDConditionModel.from_pretrained(_lowerCAmelCase							,       model_cls=_lowerCAmelCase					)
           lowercase     :Tuple   						=    loaded_unet.to(unet.device					)

         # Since no EMA step has been performed the outputs should match.
         lowercase     :Dict   						=    unet(_lowerCAmelCase							,       _lowerCAmelCase							,       _lowerCAmelCase					).sample
         lowercase     :Any   						=    loaded_unet(_lowerCAmelCase							,       _lowerCAmelCase							,       _lowerCAmelCase					).sample

         assert torch.allclose(_lowerCAmelCase							,       _lowerCAmelCase							,       atol=1e-4					)



