






"""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 		_a       (						lowerCAmelCase_   ):





			'''simple docstring'''






			def __init__(   self)	-> Tuple:




										'''simple docstring'''




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





			def __lowercase    (   self			,				UpperCAmelCase_			,				UpperCAmelCase_)	-> Optional[int]:




										'''simple docstring'''




										if dicta == dicta:
																	return True
										else:
																	lowercase__:    Union[str, Any]        =      self._remove_meta_keys(_snake_case)
																	lowercase__:    int        =      self._remove_meta_keys(_snake_case)
																	if configa == configa:
																								print(F"""Warning !: Mismatch in keys {meta_keysa} and {meta_keysa}.""")
																								return True
										return False





			def __lowercase    (   self			,				UpperCAmelCase_)	-> Dict:




										'''simple docstring'''




										lowercase__:    int        =      []
										lowercase__:    Dict        =      config_dict.copy()
										for key in config_dict.keys():
																	if key.startswith("_"):
																								temp_dict.pop(_snake_case)
																								meta_keys.append(_snake_case)
										return (temp_dict, meta_keys)





			@torch.no_grad()
			def __lowercase    (   self			,				UpperCAmelCase_			,				**UpperCAmelCase_)	-> Optional[Any]:




										'''simple docstring'''




										lowercase__:    Optional[int]        =      kwargs.pop("cache_dir"			,				_snake_case)
										lowercase__:    List[Any]        =      kwargs.pop("resume_download"			,				_snake_case)
										lowercase__:    Dict        =      kwargs.pop("force_download"			,				_snake_case)
										lowercase__:    List[Any]        =      kwargs.pop("proxies"			,				_snake_case)
										lowercase__:    Optional[int]        =      kwargs.pop("local_files_only"			,				_snake_case)
										lowercase__:    Union[str, Any]        =      kwargs.pop("use_auth_token"			,				_snake_case)
										lowercase__:    Optional[Any]        =      kwargs.pop("revision"			,				_snake_case)
										lowercase__:    int        =      kwargs.pop("torch_dtype"			,				_snake_case)
										lowercase__:    Tuple        =      kwargs.pop("device_map"			,				_snake_case)

										lowercase__:    List[str]        =      kwargs.pop("alpha"			,				0.5)
										lowercase__:    Tuple        =      kwargs.pop("interp"			,				_snake_case)

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

										lowercase__:    List[Any]        =      len(_snake_case)
										# Ignore result from model_index_json comparision of the two checkpoints
										lowercase__:    int        =      kwargs.pop("force"			,				_snake_case)

										# 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 '_'
										lowercase__:    Optional[Any]        =      []
										for pretrained_model_name_or_path in pretrained_model_name_or_path_list:
																	lowercase__:    Optional[int]        =      DiffusionPipeline.load_config(
																	    _snake_case			,				cache_dir=_snake_case			,				resume_download=_snake_case			,				force_download=_snake_case			,				proxies=_snake_case			,				local_files_only=_snake_case			,				use_auth_token=_snake_case			,				revision=_snake_case			,				)
																	config_dicts.append(_snake_case)

										lowercase__:    Dict        =      True
										for idx in range(1			,				len(_snake_case)):
																	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.
										lowercase__:    Union[str, Any]        =      []
										for pretrained_model_name_or_path, config_dict in zip(_snake_case			,				_snake_case):
																	lowercase__:    List[Any]        =      [k for k in config_dict.keys() if not k.startswith("_")]
																	lowercase__:    Optional[int]        =      [os.path.join(_snake_case			,				"*") for k in folder_names]
																	allow_patterns += [
																	    WEIGHTS_NAME,
																	    SCHEDULER_CONFIG_NAME,
																	    CONFIG_NAME,
																	    ONNX_WEIGHTS_NAME,
																	    DiffusionPipeline.config_name,
																	]
																	lowercase__:    List[str]        =      config_dict.get("_class_name")
																	lowercase__:    Optional[int]        =      {'diffusers': __version__, 'pipeline_class': requested_pipeline_class}

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

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

										lowercase__:    Any        =      None
										if len(_snake_case) > 2:
																	lowercase__:    Union[str, Any]        =      os.path.join(cached_folders[2])

										if interp == "sigmoid":
																	lowercase__:    int        =      CheckpointMergerPipeline.sigmoid
										elif interp == "inv_sigmoid":
																	lowercase__:    Tuple        =      CheckpointMergerPipeline.inv_sigmoid
										elif interp == "add_diff":
																	lowercase__:    Any        =      CheckpointMergerPipeline.add_difference
										else:
																	lowercase__:    Optional[int]        =      CheckpointMergerPipeline.weighted_sum

										# Find each module's state dict.
										for attr in final_pipe.config.keys():
																	if not attr.startswith("_"):
																								lowercase__:    List[Any]        =      os.path.join(cached_folders[1]			,				_snake_case)
																								if os.path.exists(_snake_case):
																															lowercase__:    Any        =      [
																															    *glob.glob(os.path.join(_snake_case			,				"*.safetensors")),
																															    *glob.glob(os.path.join(_snake_case			,				"*.bin")),
																															]
																															lowercase__:    Optional[Any]        =      files[0] if len(_snake_case) > 0 else None
																								if len(_snake_case) < 3:
																															lowercase__:    Tuple        =      None
																								else:
																															lowercase__:    Optional[int]        =      os.path.join(cached_folders[2]			,				_snake_case)
																															if os.path.exists(_snake_case):
																																						lowercase__:    Optional[int]        =      [
																																						    *glob.glob(os.path.join(_snake_case			,				"*.safetensors")),
																																						    *glob.glob(os.path.join(_snake_case			,				"*.bin")),
																																						]
																																						lowercase__:    List[Any]        =      files[0] if len(_snake_case) > 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:
																															lowercase__:    Any        =      getattr(_snake_case			,				_snake_case)
																															if isinstance(_snake_case			,				_snake_case):  # ignore requires_safety_checker boolean
																																						continue
																															lowercase__:    str        =      getattr(_snake_case			,				"state_dict")
																															lowercase__:    Union[str, Any]        =      theta_a()

																															lowercase__:    int        =      getattr(_snake_case			,				"load_state_dict")
																															lowercase__:    str        =      (
																															    safetensors.torch.load_file(_snake_case)
																															    if (is_safetensors_available() and checkpoint_path_a.endswith(".safetensors"))
																															    else torch.load(_snake_case			,				map_location="cpu")
																															)
																															lowercase__:    Optional[Any]        =      None
																															if checkpoint_path_a:
																																						lowercase__:    Any        =      (
																																						    safetensors.torch.load_file(_snake_case)
																																						    if (is_safetensors_available() and checkpoint_path_a.endswith(".safetensors"))
																																						    else torch.load(_snake_case			,				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(_snake_case)}""")
																															continue
																								print(F"""MERGING {attr}""")

																								for key in theta_a.keys():
																															if theta_a:
																																						lowercase__:    int        =      theta_func(theta_a[key]			,				theta_a[key]			,				theta_a[key]			,				_snake_case)
																															else:
																																						lowercase__:    str        =      theta_func(theta_a[key]			,				theta_a[key]			,				_snake_case			,				_snake_case)

																								del theta_a
																								del theta_a
																								update_theta_a(_snake_case)

																								del theta_a
										return final_pipe





			@staticmethod
			def __lowercase    (   UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_)	-> Tuple:




										'''simple docstring'''




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





			@staticmethod
			def __lowercase    (   UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_)	-> Union[str, Any]:




										'''simple docstring'''




										lowercase__:    Optional[int]        =      alpha * alpha * (3 - (2 * alpha))
										return thetaa + ((thetaa - thetaa) * alpha)





			@staticmethod
			def __lowercase    (   UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_)	-> List[str]:




										'''simple docstring'''




										import math

										lowercase__:    Optional[Any]        =      0.5 - math.sin(math.asin(1.0 - 2.0 * alpha) / 3.0)
										return thetaa + ((thetaa - thetaa) * alpha)





			@staticmethod
			def __lowercase    (   UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_)	-> Optional[int]:




										'''simple docstring'''




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














"""simple docstring"""





UpperCamelCase			=      {
    """0""": 1,
    """1""": 1,
    """2""": 2,
    """3""": 6,
    """4""": 24,
    """5""": 1_20,
    """6""": 7_20,
    """7""": 50_40,
    """8""": 4_03_20,
    """9""": 36_28_80,
}

UpperCamelCase			=      {1_45: 1_45}

UpperCamelCase			=      {
    1_45: 0,
    1_69: 3,
    3_63_01: 3,
    14_54: 3,
    8_71: 2,
    4_53_61: 2,
    8_72: 2,
}






def     A(  snake_case_		):
       """simple docstring"""





       if n in CACHE_SUM_DIGIT_FACTORIALS:
              return CACHE_SUM_DIGIT_FACTORIALS[n]
       lowercase__:    Any        =      sum(DIGIT_FACTORIALS[let] for let in str(snake_case_		)		)
       lowercase__:    Optional[Any]        =      ret
       return ret






def     A(  snake_case_						,       snake_case_ = None		):
       """simple docstring"""





       lowercase__:    Optional[int]        =      previous or set()
       if n in CHAIN_LENGTH_CACHE:
              return CHAIN_LENGTH_CACHE[n]
       lowercase__:    Tuple        =      sum_digit_factorials(snake_case_		)
       if next_number in previous:
              lowercase__:    List[Any]        =      0
              return 0
       else:
              previous.add(snake_case_		)
              lowercase__:    Optional[int]        =      1 + chain_length(snake_case_						,       snake_case_		)
              lowercase__:    Any        =      ret
              return ret






def     A(  snake_case_ = 60						,       snake_case_ = 1000000		):
       """simple docstring"""





       return sum(1 for i in range(1						,       snake_case_		) if chain_length(snake_case_		) == num_terms		)


if __name__ == "__main__":
  print(F"{solution() = }")






