



'''simple docstring'''


import glob
import os
from typing import Dict, List, Union

import torch

from diffusers.utils import is_safetensors_available


if is_safetensors_available():
  import safetensors.torch

from huggingface_hub import snapshot_download

from diffusers import DiffusionPipeline, __version__
from diffusers.schedulers.scheduling_utils import SCHEDULER_CONFIG_NAME
from diffusers.utils import CONFIG_NAME, DIFFUSERS_CACHE, ONNX_WEIGHTS_NAME, WEIGHTS_NAME







class 						UpperCAmelCase						(		__a):


 '''simple docstring'''




 def __init__( self)			->   Dict:



      """simple docstring"""







      self.register_to_config()
      super().__init__()




 def    lowercase_	( self				,	lowerCAmelCase_				,	lowerCAmelCase_)			->   List[Any]:



      """simple docstring"""







      if dicta == dicta:
           return True
      else:
           a_		,    a_									=self._remove_meta_keys(lowerCAmelCase_)
           a_		,    a_									=self._remove_meta_keys(lowerCAmelCase_)
           if configa == configa:
                print(f"""Warning !: Mismatch in keys {meta_keysa} and {meta_keysa}.""")
                return True
      return False




 def    lowercase_	( self				,	lowerCAmelCase_)			->   List[Any]:



      """simple docstring"""







      a_									=[]
      a_									=config_dict.copy()
      for key in config_dict.keys():
           if key.startswith("_"):
                temp_dict.pop(lowerCAmelCase_)
                meta_keys.append(lowerCAmelCase_)
      return (temp_dict, meta_keys)




 @torch.no_grad()
 def    lowercase_	( self				,	lowerCAmelCase_				,	**lowerCAmelCase_)			->   Union[str, Any]:



      """simple docstring"""







      a_									=kwargs.pop("cache_dir"				,	lowerCAmelCase_)
      a_									=kwargs.pop("resume_download"				,	lowerCAmelCase_)
      a_									=kwargs.pop("force_download"				,	lowerCAmelCase_)
      a_									=kwargs.pop("proxies"				,	lowerCAmelCase_)
      a_									=kwargs.pop("local_files_only"				,	lowerCAmelCase_)
      a_									=kwargs.pop("use_auth_token"				,	lowerCAmelCase_)
      a_									=kwargs.pop("revision"				,	lowerCAmelCase_)
      a_									=kwargs.pop("torch_dtype"				,	lowerCAmelCase_)
      a_									=kwargs.pop("device_map"				,	lowerCAmelCase_)

      a_									=kwargs.pop("alpha"				,	0.5)
      a_									=kwargs.pop("interp"				,	lowerCAmelCase_)

      print("Received list"				,	lowerCAmelCase_)
      print(f"""Combining with alpha={alpha}, interpolation mode={interp}""")

      a_									=len(lowerCAmelCase_)
      # Ignore result from model_index_json comparision of the two checkpoints
      a_									=kwargs.pop("force"				,	lowerCAmelCase_)

      # If less than 2 checkpoints, nothing to merge. If more than 3, not supported for now.
      if checkpoint_count > 3 or checkpoint_count < 2:
           raise ValueError(
               "Received incorrect number of checkpoints to merge. Ensure that either 2 or 3 checkpoints are being"
               " passed.")

      print("Received the right number of checkpoints")
      # chkpt0, chkpt1 = pretrained_model_name_or_path_list[0:2]
      # chkpt2 = pretrained_model_name_or_path_list[2] if checkpoint_count == 3 else None

      # Validate that the checkpoints can be merged
      # Step 1: Load the model config and compare the checkpoints. We'll compare the model_index.json first while ignoring the keys starting with '_'
      a_									=[]
      for pretrained_model_name_or_path in pretrained_model_name_or_path_list:
           a_									=DiffusionPipeline.load_config(
               lowerCAmelCase_				,	cache_dir=lowerCAmelCase_				,	resume_download=lowerCAmelCase_				,	force_download=lowerCAmelCase_				,	proxies=lowerCAmelCase_				,	local_files_only=lowerCAmelCase_				,	use_auth_token=lowerCAmelCase_				,	revision=lowerCAmelCase_				,	)
           config_dicts.append(lowerCAmelCase_)

      a_									=True
      for idx in range(1				,	len(lowerCAmelCase_)):
           comparison_result &= self._compare_model_configs(config_dicts[idx - 1]				,	config_dicts[idx])
           if not force and comparison_result is False:
                raise ValueError("Incompatible checkpoints. Please check model_index.json for the models.")
                print(config_dicts[0]				,	config_dicts[1])
      print("Compatible model_index.json files found")
      # Step 2: Basic Validation has succeeded. Let's download the models and save them into our local files.
      a_									=[]
      for pretrained_model_name_or_path, config_dict in zip(lowerCAmelCase_				,	lowerCAmelCase_):
           a_									=[k for k in config_dict.keys() if not k.startswith("_")]
           a_									=[os.path.join(lowerCAmelCase_				,	"*") for k in folder_names]
           allow_patterns += [
               WEIGHTS_NAME,
               SCHEDULER_CONFIG_NAME,
               CONFIG_NAME,
               ONNX_WEIGHTS_NAME,
               DiffusionPipeline.config_name,
           ]
           a_									=config_dict.get("_class_name")
           a_									={"diffusers": __version__, "pipeline_class": requested_pipeline_class}

           a_									=(
               pretrained_model_name_or_path
               if os.path.isdir(lowerCAmelCase_)
               else snapshot_download(
                   lowerCAmelCase_				,	cache_dir=lowerCAmelCase_				,	resume_download=lowerCAmelCase_				,	proxies=lowerCAmelCase_				,	local_files_only=lowerCAmelCase_				,	use_auth_token=lowerCAmelCase_				,	revision=lowerCAmelCase_				,	allow_patterns=lowerCAmelCase_				,	user_agent=lowerCAmelCase_				,	)
           )
           print("Cached Folder"				,	lowerCAmelCase_)
           cached_folders.append(lowerCAmelCase_)

      # Step 3:-
      # Load the first checkpoint as a diffusion pipeline and modify its module state_dict in place
      a_									=DiffusionPipeline.from_pretrained(
          cached_folders[0]				,	torch_dtype=lowerCAmelCase_				,	device_map=lowerCAmelCase_)
      final_pipe.to(self.device)

      a_									=None
      if len(lowerCAmelCase_) > 2:
           a_									=os.path.join(cached_folders[2])

      if interp == "sigmoid":
           a_									=CheckpointMergerPipeline.sigmoid
      elif interp == "inv_sigmoid":
           a_									=CheckpointMergerPipeline.inv_sigmoid
      elif interp == "add_diff":
           a_									=CheckpointMergerPipeline.add_difference
      else:
           a_									=CheckpointMergerPipeline.weighted_sum

      # Find each module's state dict.
      for attr in final_pipe.config.keys():
           if not attr.startswith("_"):
                a_									=os.path.join(cached_folders[1]				,	lowerCAmelCase_)
                if os.path.exists(lowerCAmelCase_):
                     a_									=[
                         *glob.glob(os.path.join(lowerCAmelCase_				,	"*.safetensors")),
                         *glob.glob(os.path.join(lowerCAmelCase_				,	"*.bin")),
                     ]
                     a_									=files[0] if len(lowerCAmelCase_) > 0 else None
                if len(lowerCAmelCase_) < 3:
                     a_									=None
                else:
                     a_									=os.path.join(cached_folders[2]				,	lowerCAmelCase_)
                     if os.path.exists(lowerCAmelCase_):
                          a_									=[
                              *glob.glob(os.path.join(lowerCAmelCase_				,	"*.safetensors")),
                              *glob.glob(os.path.join(lowerCAmelCase_				,	"*.bin")),
                          ]
                          a_									=files[0] if len(lowerCAmelCase_) > 0 else None
                # For an attr if both checkpoint_path_1 and 2 are None, ignore.
                # If atleast one is present, deal with it according to interp method, of course only if the state_dict keys match.
                if checkpoint_path_a is None and checkpoint_path_a is None:
                     print(f"""Skipping {attr}: not present in 2nd or 3d model""")
                     continue
                try:
                     a_									=getattr(lowerCAmelCase_				,	lowerCAmelCase_)
                     if isinstance(lowerCAmelCase_				,	lowerCAmelCase_):  # ignore requires_safety_checker boolean
                          continue
                     a_									=getattr(lowerCAmelCase_				,	"state_dict")
                     a_									=theta_a()

                     a_									=getattr(lowerCAmelCase_				,	"load_state_dict")
                     a_									=(
                         safetensors.torch.load_file(lowerCAmelCase_)
                         if (is_safetensors_available() and checkpoint_path_a.endswith(".safetensors"))
                         else torch.load(lowerCAmelCase_				,	map_location="cpu")
                     )
                     a_									=None
                     if checkpoint_path_a:
                          a_									=(
                              safetensors.torch.load_file(lowerCAmelCase_)
                              if (is_safetensors_available() and checkpoint_path_a.endswith(".safetensors"))
                              else torch.load(lowerCAmelCase_				,	map_location="cpu")
                          )

                     if not theta_a.keys() == theta_a.keys():
                          print(f"""Skipping {attr}: key mismatch""")
                          continue
                     if theta_a and not theta_a.keys() == theta_a.keys():
                          print(f"""Skipping {attr}:y mismatch""")
                except Exception as e:
                     print(f"""Skipping {attr} do to an unexpected error: {str(lowerCAmelCase_)}""")
                     continue
                print(f"""MERGING {attr}""")

                for key in theta_a.keys():
                     if theta_a:
                          a_									=theta_func(theta_a[key]				,	theta_a[key]				,	theta_a[key]				,	lowerCAmelCase_)
                     else:
                          a_									=theta_func(theta_a[key]				,	theta_a[key]				,	lowerCAmelCase_				,	lowerCAmelCase_)

                del theta_a
                del theta_a
                update_theta_a(lowerCAmelCase_)

                del theta_a
      return final_pipe




 @staticmethod
 def    lowercase_	( lowerCAmelCase_				,	lowerCAmelCase_				,	lowerCAmelCase_				,	lowerCAmelCase_)			->   str:



      """simple docstring"""







      return ((1 - alpha) * thetaa) + (alpha * thetaa)




 @staticmethod
 def    lowercase_	( lowerCAmelCase_				,	lowerCAmelCase_				,	lowerCAmelCase_				,	lowerCAmelCase_)			->   str:



      """simple docstring"""







      a_									=alpha * alpha * (3 - (2 * alpha))
      return thetaa + ((thetaa - thetaa) * alpha)




 @staticmethod
 def    lowercase_	( lowerCAmelCase_				,	lowerCAmelCase_				,	lowerCAmelCase_				,	lowerCAmelCase_)			->   str:



      """simple docstring"""







      import math

      a_									=0.5 - math.sin(math.asin(1.0 - 2.0 * alpha) / 3.0)
      return thetaa + ((thetaa - thetaa) * alpha)




 @staticmethod
 def    lowercase_	( lowerCAmelCase_				,	lowerCAmelCase_				,	lowerCAmelCase_				,	lowerCAmelCase_)			->   Optional[int]:



      """simple docstring"""







      return thetaa + (thetaa - thetaa) * (1.0 - alpha)







'''simple docstring'''


lowercase											=      (1, 153, 370, 371, 1_634, 24_678_051, 115_132_219_018_763_992_565_095_597_973_971_522_401)
lowercase											=      (-153, -1, 0, 1.2, 200, "A", [], {}, None)






def 			UpperCAmelCase_							(		lowercase__							):


     '''simple docstring'''







     if not isinstance(lowercase__							,						lowercase__							) or n < 1:
          return False

     # Initialization of sum and number of digits.
     a_									=0
     a_									=0
     a_									=n
     # Calculation of digits of the number
     while temp > 0:
          number_of_digits += 1
          temp //= 1_0
     # Dividing number into separate digits and find Armstrong number
     a_									=n
     while temp > 0:
          a_									=temp % 1_0
          total += rem**number_of_digits
          temp //= 1_0
     return n == total






def 			UpperCAmelCase_							(		lowercase__							):


     '''simple docstring'''







     if not isinstance(lowercase__							,						lowercase__							) or n < 1:
          return False

     # Init a "histogram" of the digits
     a_									=[0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
     a_									=0
     a_									=0
     a_									=n
     while temp > 0:
          a_		,    a_									=divmod(lowercase__							,						1_0							)
          digit_histogram[rem] += 1
          digit_total += 1

     for cnt, i in zip(lowercase__							,						range(len(lowercase__							)							)							):
          total += cnt * i**digit_total

     return n == total






def 			UpperCAmelCase_							(		lowercase__							):


     '''simple docstring'''







     if not isinstance(lowercase__							,						lowercase__							) or n < 1:
          return False
     a_									=len(str(lowercase__							)							)  # the power that all digits will be raised to
     # check if sum of each digit multiplied expo times is equal to number
     return n == sum(int(lowercase__							) ** expo for i in str(lowercase__							)							)






def 			UpperCAmelCase_							(		):


     '''simple docstring'''







     a_									=int(input("Enter an integer to see if it is an Armstrong number: "							).strip()							)
     print(F"""{num} is {'' if armstrong_number(lowercase__							) else 'not '}an Armstrong number."""							)
     print(F"""{num} is {'' if narcissistic_number(lowercase__							) else 'not '}an Armstrong number."""							)
     print(F"""{num} is {'' if pluperfect_number(lowercase__							) else 'not '}an Armstrong number."""							)


if __name__ == "__main__":
  import doctest

  doctest.testmod()
  main()


