




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







							def __init__(  self       :  str					)			->							List[Any]:


														"""simple docstring"""


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







							def    __lowerCAmelCase			(  self       :  Optional[Any]							,			lowerCAmelCase_       :  Any							,			lowerCAmelCase_       :  Dict					)			->							List[str]:


														"""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    __lowerCAmelCase			(  self       :  List[Any]							,			lowerCAmelCase_       :  Dict					)			->							List[str]:


														"""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    __lowerCAmelCase			(  self       :  Dict							,			lowerCAmelCase_       :  List[Union[str, os.PathLike]]							,			**lowerCAmelCase_       :  Optional[int]					)			->							Optional[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    __lowerCAmelCase			(  lowerCAmelCase_       :  List[Any]							,			lowerCAmelCase_       :  Tuple							,			lowerCAmelCase_       :  Tuple							,			lowerCAmelCase_       :  Any					)			->							List[Any]:


														"""simple docstring"""


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







							@staticmethod
							def    __lowerCAmelCase			(  lowerCAmelCase_       :  str							,			lowerCAmelCase_       :  Dict							,			lowerCAmelCase_       :  Optional[int]							,			lowerCAmelCase_       :  Dict					)			->							List[str]:


														"""simple docstring"""


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







							@staticmethod
							def    __lowerCAmelCase			(  lowerCAmelCase_       :  List[Any]							,			lowerCAmelCase_       :  List[Any]							,			lowerCAmelCase_       :  List[str]							,			lowerCAmelCase_       :  List[Any]					)			->							List[Any]:


														"""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    __lowerCAmelCase			(  lowerCAmelCase_       :  Union[str, Any]							,			lowerCAmelCase_       :  Optional[Any]							,			lowerCAmelCase_       :  List[str]							,			lowerCAmelCase_       :  Union[str, Any]					)			->							str:


														"""simple docstring"""


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







'''simple docstring'''


from __future__ import annotations

from enum import Enum

class 			A    (		_a    ):
							lowercase_							=				24
							lowercase_							=				21
							lowercase_							=				18
							lowercase_							=				15
							lowercase_							=				12
							lowercase_							=				9
							lowercase_							=				6
							lowercase_							=				3
							lowercase_							=				2
							lowercase_							=				1
							lowercase_							=				-1
							lowercase_							=				-2
							lowercase_							=				-3
							lowercase_							=				-6
							lowercase_							=				-9
							lowercase_							=				-12
							lowercase_							=				-15
							lowercase_							=				-18
							lowercase_							=				-21
							lowercase_							=				-24



class 			A    (		_a    ):
							lowercase_							=				8
							lowercase_							=				7
							lowercase_							=				6
							lowercase_							=				5
							lowercase_							=				4
							lowercase_							=				3
							lowercase_							=				2
							lowercase_							=				1



def 		snake_case_    (UpperCamelCase       :  float ,						UpperCamelCase       :  str | SIUnit ,						UpperCamelCase       :  str | SIUnit ,						):





							'''simple docstring'''


							if isinstance(UpperCamelCase ,						UpperCamelCase      ):
														_a	     =				SIUnit[known_prefix.lower()]
							if isinstance(UpperCamelCase ,						UpperCamelCase      ):
														_a	     =				SIUnit[unknown_prefix.lower()]
							_a	     =				known_amount * (
							    10 ** (known_prefix.value - unknown_prefix.value)
							)
							return unknown_amount
def 		snake_case_    (UpperCamelCase       :  float ,						UpperCamelCase       :  str | BinaryUnit ,						UpperCamelCase       :  str | BinaryUnit ,						):





							'''simple docstring'''


							if isinstance(UpperCamelCase ,						UpperCamelCase      ):
														_a	     =				BinaryUnit[known_prefix.lower()]
							if isinstance(UpperCamelCase ,						UpperCamelCase      ):
														_a	     =				BinaryUnit[unknown_prefix.lower()]
							_a	     =				known_amount * (
							    2 ** ((known_prefix.value - unknown_prefix.value) * 10)
							)
							return unknown_amount


if __name__ == "__main__":
				import doctest

				doctest.testmod()

