'''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      _UpperCAmelCase    (							_UpperCamelCase    :	str,   _UpperCamelCase    :	List[Any],   _UpperCamelCase    :	Dict,   _UpperCamelCase    :	List[str]=None   )	->      Tuple:
    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(_UpperCamelCase,   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(_UpperCamelCase,   dtype=array.dtype,   mode='''w+''',   shape=array.shape   )
    A_     =   array[:]
    file_array.flush()
    return index





def      _UpperCAmelCase    (							_UpperCamelCase    :	int,   _UpperCamelCase    :	Optional[int]   )	->      Tuple:
    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(_UpperCamelCase,   dtype=_UpperCamelCase,   shape=_UpperCamelCase,   mode='''r'''   )

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

    return weight





def      _UpperCAmelCase    (							_UpperCamelCase    :	Union[str, Any],   _UpperCamelCase    :	str   )	->      Optional[int]:
    if index is None or len(_UpperCamelCase   ) == 0:
        # Nothing to save
        return

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

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





def      _UpperCAmelCase    (							_UpperCamelCase    :	Union[str, os.PathLike],   _UpperCamelCase    :	Dict[str, torch.Tensor]   )	->      str:
    os.makedirs(_UpperCamelCase,   exist_ok=_UpperCamelCase   )
    A_     =   {}
    for name, parameter in state_dict.items():
        A_     =   offload_weight(_UpperCamelCase,   _UpperCamelCase,   _UpperCamelCase,   index=_UpperCamelCase   )

    # Update index
    save_offload_index(_UpperCamelCase,   _UpperCamelCase   )




class 	__UpperCAmelCase				(				_UpperCamelCase     ):





       '''simple docstring'''



       def __init__(      self    ,     _SCREAMING_SNAKE_CASE    ,     _SCREAMING_SNAKE_CASE )		->     Union[str, Any]:
           A_     =   dataset
           A_     =   prefix

       def __getitem__(      self    ,     _SCREAMING_SNAKE_CASE )		->     Tuple:
           return self.dataset[F'''{self.prefix}{key}''']

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

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




class 	__UpperCAmelCase				(				_UpperCamelCase     ):





       '''simple docstring'''



       def __init__(      self    ,     _SCREAMING_SNAKE_CASE = None    ,     _SCREAMING_SNAKE_CASE = None    ,     _SCREAMING_SNAKE_CASE = None    ,     _SCREAMING_SNAKE_CASE=None    ,     )		->     List[str]:
           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(_SCREAMING_SNAKE_CASE    ,     '''index.json''' ) ) as f:
                   A_     =   json.load(_SCREAMING_SNAKE_CASE )
           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    ,     _SCREAMING_SNAKE_CASE )		->     Tuple:
           # 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=_SCREAMING_SNAKE_CASE ) as f:
                   A_     =   f.get_tensor(weight_info.get('''weight_name'''    ,     _SCREAMING_SNAKE_CASE ) )

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

           A_     =   os.path.join(self.save_folder    ,     F'''{key}.dat''' )
           return load_offloaded_weight(_SCREAMING_SNAKE_CASE    ,     _SCREAMING_SNAKE_CASE )

       def __iter__(      self )		->     Any:
           return iter(self.all_keys )

       def __len__(      self )		->     List[Any]:
           return len(self.all_keys )





def      _UpperCAmelCase    (							_UpperCamelCase    :	Dict[str, torch.Tensor],   _UpperCamelCase    :	List[str]   )	->      str:
    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'''




import tempfile
import unittest

from diffusers import (
    DDIMScheduler,
    DDPMScheduler,
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    PNDMScheduler,
    logging,
)
from diffusers.configuration_utils import ConfigMixin, register_to_config
from diffusers.utils.testing_utils import CaptureLogger




class 	__UpperCAmelCase				(				_UpperCamelCase     ):





       '''simple docstring'''



       __lowercase					:					str    =				'config.json'

       @register_to_config
       def __init__(      self    ,     _SCREAMING_SNAKE_CASE=2    ,     _SCREAMING_SNAKE_CASE=5    ,     _SCREAMING_SNAKE_CASE=(2, 5)    ,     _SCREAMING_SNAKE_CASE="for diffusion"    ,     _SCREAMING_SNAKE_CASE=[1, 3]    ,     )		->     Union[str, Any]:
           pass




class 	__UpperCAmelCase				(				_UpperCamelCase     ):





       '''simple docstring'''



       __lowercase					:					Tuple    =				'config.json'

       @register_to_config
       def __init__(      self    ,     _SCREAMING_SNAKE_CASE=2    ,     _SCREAMING_SNAKE_CASE=5    ,     _SCREAMING_SNAKE_CASE=(2, 5)    ,     _SCREAMING_SNAKE_CASE="for diffusion"    ,     _SCREAMING_SNAKE_CASE=[1, 3]    ,     )		->     Union[str, Any]:
           pass




class 	__UpperCAmelCase				(				_UpperCamelCase     ):





       '''simple docstring'''



       __lowercase					:					Optional[Any]    =				'config.json'

       @register_to_config
       def __init__(      self    ,     _SCREAMING_SNAKE_CASE=2    ,     _SCREAMING_SNAKE_CASE=5    ,     _SCREAMING_SNAKE_CASE=(2, 5)    ,     _SCREAMING_SNAKE_CASE="for diffusion"    ,     _SCREAMING_SNAKE_CASE=[1, 3]    ,     _SCREAMING_SNAKE_CASE=[1, 3]    ,     )		->     Dict:
           pass




class 	__UpperCAmelCase				(				_UpperCamelCase     ):





       '''simple docstring'''



       __lowercase					:					Optional[Any]    =				'config.json'

       @register_to_config
       def __init__(      self    ,     _SCREAMING_SNAKE_CASE=2    ,     _SCREAMING_SNAKE_CASE=5    ,     _SCREAMING_SNAKE_CASE=(2, 5)    ,     _SCREAMING_SNAKE_CASE="for diffusion"    ,     _SCREAMING_SNAKE_CASE=[1, 5]    ,     _SCREAMING_SNAKE_CASE=[5, 4]    ,     )		->     List[str]:
           pass




class 	__UpperCAmelCase				(				unittest.TestCase     ):





       '''simple docstring'''



       def 					__A       (      self )		->     Optional[int]:
           with self.assertRaises(_SCREAMING_SNAKE_CASE ):
               ConfigMixin.load_config('''dummy_path''' )

       def 					__A       (      self )		->     Union[str, Any]:
           A_     =   SampleObject()
           A_     =   obj.config
           assert config["a"] == 2
           assert config["b"] == 5
           assert config["c"] == (2, 5)
           assert config["d"] == "for diffusion"
           assert config["e"] == [1, 3]

           # init ignore private arguments
           A_     =   SampleObject(_name_or_path='''lalala''' )
           A_     =   obj.config
           assert config["a"] == 2
           assert config["b"] == 5
           assert config["c"] == (2, 5)
           assert config["d"] == "for diffusion"
           assert config["e"] == [1, 3]

           # can override default
           A_     =   SampleObject(c=6 )
           A_     =   obj.config
           assert config["a"] == 2
           assert config["b"] == 5
           assert config["c"] == 6
           assert config["d"] == "for diffusion"
           assert config["e"] == [1, 3]

           # can use positional arguments.
           A_     =   SampleObject(1    ,     c=6 )
           A_     =   obj.config
           assert config["a"] == 1
           assert config["b"] == 5
           assert config["c"] == 6
           assert config["d"] == "for diffusion"
           assert config["e"] == [1, 3]

       def 					__A       (      self )		->     int:
           A_     =   SampleObject()
           A_     =   obj.config

           assert config["a"] == 2
           assert config["b"] == 5
           assert config["c"] == (2, 5)
           assert config["d"] == "for diffusion"
           assert config["e"] == [1, 3]

           with tempfile.TemporaryDirectory() as tmpdirname:
               obj.save_config(_SCREAMING_SNAKE_CASE )
               A_     =   SampleObject.from_config(SampleObject.load_config(_SCREAMING_SNAKE_CASE ) )
               A_     =   new_obj.config

           # unfreeze configs
           A_     =   dict(_SCREAMING_SNAKE_CASE )
           A_     =   dict(_SCREAMING_SNAKE_CASE )

           assert config.pop('''c''' ) == (2, 5)  # instantiated as tuple
           assert new_config.pop('''c''' ) == [2, 5]  # saved & loaded as list because of json
           config.pop('''_use_default_values''' )
           assert config == new_config

       def 					__A       (      self )		->     List[Any]:
           A_     =   logging.get_logger('''diffusers.configuration_utils''' )
           # 30 for warning
           logger.setLevel(30 )

           with CaptureLogger(_SCREAMING_SNAKE_CASE ) as cap_logger:
               A_     =   DDIMScheduler.from_pretrained(
                   '''hf-internal-testing/tiny-stable-diffusion-torch'''    ,     subfolder='''scheduler''' )

           assert ddim.__class__ == DDIMScheduler
           # no warning should be thrown
           assert cap_logger.out == ""

       def 					__A       (      self )		->     List[str]:
           A_     =   logging.get_logger('''diffusers.configuration_utils''' )
           # 30 for warning
           logger.setLevel(30 )

           with CaptureLogger(_SCREAMING_SNAKE_CASE ) as cap_logger:
               A_     =   EulerDiscreteScheduler.from_pretrained(
                   '''hf-internal-testing/tiny-stable-diffusion-torch'''    ,     subfolder='''scheduler''' )

           assert euler.__class__ == EulerDiscreteScheduler
           # no warning should be thrown
           assert cap_logger.out == ""

       def 					__A       (      self )		->     Optional[Any]:
           A_     =   logging.get_logger('''diffusers.configuration_utils''' )
           # 30 for warning
           logger.setLevel(30 )

           with CaptureLogger(_SCREAMING_SNAKE_CASE ) as cap_logger:
               A_     =   EulerAncestralDiscreteScheduler.from_pretrained(
                   '''hf-internal-testing/tiny-stable-diffusion-torch'''    ,     subfolder='''scheduler''' )

           assert euler.__class__ == EulerAncestralDiscreteScheduler
           # no warning should be thrown
           assert cap_logger.out == ""

       def 					__A       (      self )		->     Optional[int]:
           A_     =   logging.get_logger('''diffusers.configuration_utils''' )
           # 30 for warning
           logger.setLevel(30 )

           with CaptureLogger(_SCREAMING_SNAKE_CASE ) as cap_logger:
               A_     =   PNDMScheduler.from_pretrained(
                   '''hf-internal-testing/tiny-stable-diffusion-torch'''    ,     subfolder='''scheduler''' )

           assert pndm.__class__ == PNDMScheduler
           # no warning should be thrown
           assert cap_logger.out == ""

       def 					__A       (      self )		->     Tuple:
           A_     =   logging.get_logger('''diffusers.configuration_utils''' )
           # 30 for warning
           logger.setLevel(30 )

           with CaptureLogger(_SCREAMING_SNAKE_CASE ) as cap_logger:
               A_     =   DDPMScheduler.from_pretrained(
                   '''hf-internal-testing/tiny-stable-diffusion-torch'''    ,     subfolder='''scheduler'''    ,     prediction_type='''sample'''    ,     beta_end=8    ,     )

           with CaptureLogger(_SCREAMING_SNAKE_CASE ) as cap_logger_a:
               A_     =   DDPMScheduler.from_pretrained('''google/ddpm-celebahq-256'''    ,     beta_start=88 )

           assert ddpm.__class__ == DDPMScheduler
           assert ddpm.config.prediction_type == "sample"
           assert ddpm.config.beta_end == 8
           assert ddpm_a.config.beta_start == 88

           # no warning should be thrown
           assert cap_logger.out == ""
           assert cap_logger_a.out == ""

       def 					__A       (      self )		->     List[Any]:
           A_     =   logging.get_logger('''diffusers.configuration_utils''' )
           # 30 for warning
           logger.setLevel(30 )

           with CaptureLogger(_SCREAMING_SNAKE_CASE ) as cap_logger:
               A_     =   DPMSolverMultistepScheduler.from_pretrained(
                   '''hf-internal-testing/tiny-stable-diffusion-torch'''    ,     subfolder='''scheduler''' )

           assert dpm.__class__ == DPMSolverMultistepScheduler
           # no warning should be thrown
           assert cap_logger.out == ""

       def 					__A       (      self )		->     Optional[int]:
           # let's first save a config that should be in the form
           #    a=2,
           #    b=5,
           #    c=(2, 5),
           #    d="for diffusion",
           #    e=[1, 3],

           A_     =   SampleObject()

           A_     =   {k: v for k, v in config.config.items() if not k.startswith('''_''' )}

           # make sure that default config has all keys in `_use_default_values`
           assert set(config_dict.keys() ) == set(config.config._use_default_values )

           with tempfile.TemporaryDirectory() as tmpdirname:
               config.save_config(_SCREAMING_SNAKE_CASE )

               # now loading it with SampleObject2 should put f into `_use_default_values`
               A_     =   SampleObjecta.from_config(_SCREAMING_SNAKE_CASE )

               assert "f" in config._use_default_values
               assert config.f == [1, 3]

           # now loading the config, should **NOT** use [1, 3] for `f`, but the default [1, 4] value
           # **BECAUSE** it is part of `config._use_default_values`
           A_     =   SampleObjecta.from_config(config.config )
           assert new_config.f == [5, 4]

           config.config._use_default_values.pop()
           A_     =   SampleObjecta.from_config(config.config )
           assert new_config_a.f == [1, 3]

           # Nevertheless "e" should still be correctly loaded to [1, 3] from SampleObject2 instead of defaulting to [1, 5]
           assert new_config_a.e == [1, 3]






