






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  snake_case__      (							_lowerCAmelCase   ):
	def __init__(       self				)							->   List[Any]:
							self.register_to_config()
							super().__init__()
	def     __magic_name__       (       self    ,     lowerCAmelCase__    ,     lowerCAmelCase__				)							->   List[Any]:
							if dicta == dicta:
													return True
							else:
													__magic_name__		,__magic_name__    :     List[str]     		=      self._remove_meta_keys(lowerCAmelCase__				)
													__magic_name__		,__magic_name__    :     Tuple     		=      self._remove_meta_keys(lowerCAmelCase__				)
													if configa == configa:
																			print(F'Warning !: Mismatch in keys {meta_keysa} and {meta_keysa}.'				)
																			return True
							return False
	def     __magic_name__       (       self    ,     lowerCAmelCase__				)							->   Optional[int]:
							__magic_name__    :     Optional[Any]     		=      []
							__magic_name__    :     List[Any]     		=      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     __magic_name__       (       self    ,     lowerCAmelCase__    ,     **lowerCAmelCase__				)							->   Dict:
							__magic_name__    :     List[Any]     		=      kwargs.pop("""cache_dir"""    ,     lowerCAmelCase__				)
							__magic_name__    :     Tuple     		=      kwargs.pop("""resume_download"""    ,     lowerCAmelCase__				)
							__magic_name__    :     Any     		=      kwargs.pop("""force_download"""    ,     lowerCAmelCase__				)
							__magic_name__    :     List[str]     		=      kwargs.pop("""proxies"""    ,     lowerCAmelCase__				)
							__magic_name__    :     Optional[int]     		=      kwargs.pop("""local_files_only"""    ,     lowerCAmelCase__				)
							__magic_name__    :     Optional[Any]     		=      kwargs.pop("""use_auth_token"""    ,     lowerCAmelCase__				)
							__magic_name__    :     Dict     		=      kwargs.pop("""revision"""    ,     lowerCAmelCase__				)
							__magic_name__    :     List[str]     		=      kwargs.pop("""torch_dtype"""    ,     lowerCAmelCase__				)
							__magic_name__    :     int     		=      kwargs.pop("""device_map"""    ,     lowerCAmelCase__				)

							__magic_name__    :     Union[str, Any]     		=      kwargs.pop("""alpha"""    ,     0.5				)
							__magic_name__    :     List[str]     		=      kwargs.pop("""interp"""    ,     lowerCAmelCase__				)

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

							__magic_name__    :     Dict     		=      len(lowerCAmelCase__				)
							# Ignore result from model_index_json comparision of the two checkpoints
							__magic_name__    :     Optional[int]     		=      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 '_'
							__magic_name__    :     Dict     		=      []
							for pretrained_model_name_or_path in pretrained_model_name_or_path_list:
													__magic_name__    :     str     		=      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__				)

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

													__magic_name__    :     Tuple     		=      (
													    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
							__magic_name__    :     Optional[Any]     		=      DiffusionPipeline.from_pretrained(
							    cached_folders[0]    ,     torch_dtype=lowerCAmelCase__    ,     device_map=lowerCAmelCase__				)
							final_pipe.to(self.device				)

							__magic_name__    :     int     		=      None
							if len(lowerCAmelCase__				) > 2:
													__magic_name__    :     Optional[int]     		=      os.path.join(cached_folders[2]				)

							if interp == "sigmoid":
													__magic_name__    :     Union[str, Any]     		=      CheckpointMergerPipeline.sigmoid
							elif interp == "inv_sigmoid":
													__magic_name__    :     List[Any]     		=      CheckpointMergerPipeline.inv_sigmoid
							elif interp == "add_diff":
													__magic_name__    :     Any     		=      CheckpointMergerPipeline.add_difference
							else:
													__magic_name__    :     List[str]     		=      CheckpointMergerPipeline.weighted_sum

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

																									__magic_name__    :     Any     		=      getattr(lowerCAmelCase__    ,     """load_state_dict"""				)
																									__magic_name__    :     Union[str, Any]     		=      (
																									    safetensors.torch.load_file(lowerCAmelCase__				)
																									    if (is_safetensors_available() and checkpoint_path_a.endswith(""".safetensors"""				))
																									    else torch.load(lowerCAmelCase__    ,     map_location="""cpu"""				)
																									)
																									__magic_name__    :     Optional[Any]     		=      None
																									if checkpoint_path_a:
																															__magic_name__    :     Optional[int]     		=      (
																															    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:
																															__magic_name__    :     Union[str, Any]     		=      theta_func(theta_a[key]    ,     theta_a[key]    ,     theta_a[key]    ,     lowerCAmelCase__				)
																									else:
																															__magic_name__    :     Optional[Any]     		=      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     __magic_name__       (       lowerCAmelCase__    ,     lowerCAmelCase__    ,     lowerCAmelCase__    ,     lowerCAmelCase__				)							->   Dict:
							return ((1 - alpha) * thetaa) + (alpha * thetaa)
	@staticmethod
	def     __magic_name__       (       lowerCAmelCase__    ,     lowerCAmelCase__    ,     lowerCAmelCase__    ,     lowerCAmelCase__				)							->   Optional[int]:
							__magic_name__    :     Dict     		=      alpha * alpha * (3 - (2 * alpha))
							return thetaa + ((thetaa - thetaa) * alpha)
	@staticmethod
	def     __magic_name__       (       lowerCAmelCase__    ,     lowerCAmelCase__    ,     lowerCAmelCase__    ,     lowerCAmelCase__				)							->   Optional[Any]:
							import math

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







	@staticmethod
	def     __magic_name__       (       lowerCAmelCase__    ,     lowerCAmelCase__    ,     lowerCAmelCase__    ,     lowerCAmelCase__				)							->   int:
							return thetaa + (thetaa - thetaa) * (1.0 - alpha)








class  snake_case__      :
	def __init__(       self				)							->   None:
							# numbermap keys are the number and its values are lists of indexes sorted
							# in ascending order
							__magic_name__    :     dict[int, list[int]]     		=      {}
							# indexmap keys are an index and it's values are the number at that index
							__magic_name__    :     dict[int, int]     		=      {}
	def     __magic_name__       (       self    ,     lowerCAmelCase__    ,     lowerCAmelCase__				)							->   list[int]:
							if isinstance(lowerCAmelCase__    ,     (range, str)				):
													__magic_name__    :     int     		=      list(lowerCAmelCase__				)
							elif not isinstance(lowerCAmelCase__    ,     lowerCAmelCase__				):
													raise TypeError(
													    """binary_search_delete() only accepts either a list, range or str"""				)

							__magic_name__    :     str     		=      0
							__magic_name__    :     int     		=      len(lowerCAmelCase__				) - 1

							while low <= high:
													__magic_name__    :     List[str]     		=      (low + high) // 2
													if array[mid] == item:
																			array.pop(lowerCAmelCase__				)
																			return array
													elif array[mid] < item:
																			__magic_name__    :     Tuple     		=      mid + 1
													else:
																			__magic_name__    :     Dict     		=      mid - 1
							raise ValueError(
							    """Either the item is not in the array or the array was unsorted"""				)
	def     __magic_name__       (       self    ,     lowerCAmelCase__    ,     lowerCAmelCase__				)							->   list[int]:
							if isinstance(lowerCAmelCase__    ,     (range, str)				):
													__magic_name__    :     Tuple     		=      list(lowerCAmelCase__				)
							elif not isinstance(lowerCAmelCase__    ,     lowerCAmelCase__				):
													raise TypeError(
													    """binary_search_insert() only accepts either a list, range or str"""				)

							__magic_name__    :     List[str]     		=      0
							__magic_name__    :     Tuple     		=      len(lowerCAmelCase__				) - 1

							while low <= high:
													__magic_name__    :     Dict     		=      (low + high) // 2
													if array[mid] == index:
																			# If the item already exists in the array,
																			# insert it after the existing item
																			array.insert(mid + 1    ,     lowerCAmelCase__				)
																			return array
													elif array[mid] < index:
																			__magic_name__    :     int     		=      mid + 1
													else:
																			__magic_name__    :     Tuple     		=      mid - 1

        # If the item doesn't exist in the array, insert it at the appropriate position
							array.insert(lowerCAmelCase__    ,     lowerCAmelCase__				)
							return array
	def     __magic_name__       (       self    ,     lowerCAmelCase__    ,     lowerCAmelCase__				)							->   None:
							# Remove previous index
							if index in self.indexmap:
													__magic_name__    :     int     		=      self.indexmap[index]
													if len(self.numbermap[n]				) == 1:
																			del self.numbermap[n]
													else:
																			__magic_name__    :     Union[str, Any]     		=      self.binary_search_delete(self.numbermap[n]    ,     lowerCAmelCase__				)

        # Set new index
							__magic_name__    :     List[Any]     		=      number

							# Number not seen before or empty so insert number value
							if number not in self.numbermap:
													__magic_name__    :     Any     		=      [index]

							# Here we need to perform a binary search insertion in order to insert
							# The item in the correct place
							else:
													__magic_name__    :     Optional[Any]     		=      self.binary_search_insert(
													    self.numbermap[number]    ,     lowerCAmelCase__				)







	def     __magic_name__       (       self    ,     lowerCAmelCase__				)							->   int:
							return self.numbermap.get(lowerCAmelCase__    ,     [-1]				)[0]


if __name__ == "__main__":
		import doctest

		doctest.testmod()
