


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



   '''simple docstring'''






   def __init__(	self	:					Optional[int]						)->	Union[str, Any]:
          self.register_to_config()
          super().__init__()




   def   __lowerCAmelCase						(	self	:					str					, lowerCamelCase	:					str					, lowerCamelCase	:					Union[str, Any]						)->	Optional[int]:







          if dicta == dicta:
                 return True
          else:
                 snake_case__       ,   snake_case__					:	List[str]          =					self._remove_meta_keys(lowerCamelCase						)
                 snake_case__       ,   snake_case__					:	Any          =					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	:					Dict					, lowerCamelCase	:					Dict						)->	List[str]:
          snake_case__					:	str          =					[]
          snake_case__					:	Tuple          =					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	:					Union[str, Any]					, lowerCamelCase	:					List[Union[str, os.PathLike]]					, **lowerCamelCase	:					List[Any]						)->	str:
          snake_case__					:	Any          =					kwargs.pop("""cache_dir"""					, lowerCamelCase						)
          snake_case__					:	int          =					kwargs.pop("""resume_download"""					, lowerCamelCase						)
          snake_case__					:	Dict          =					kwargs.pop("""force_download"""					, lowerCamelCase						)
          snake_case__					:	List[str]          =					kwargs.pop("""proxies"""					, lowerCamelCase						)
          snake_case__					:	Union[str, Any]          =					kwargs.pop("""local_files_only"""					, lowerCamelCase						)
          snake_case__					:	Any          =					kwargs.pop("""use_auth_token"""					, lowerCamelCase						)
          snake_case__					:	Dict          =					kwargs.pop("""revision"""					, lowerCamelCase						)
          snake_case__					:	Optional[Any]          =					kwargs.pop("""torch_dtype"""					, lowerCamelCase						)
          snake_case__					:	Dict          =					kwargs.pop("""device_map"""					, lowerCamelCase						)

          snake_case__					:	Union[str, Any]          =					kwargs.pop("""alpha"""					, 0.5						)
          snake_case__					:	Dict          =					kwargs.pop("""interp"""					, lowerCamelCase						)

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

          snake_case__					:	Optional[Any]          =					len(lowerCamelCase						)
          # Ignore result from model_index_json comparision of the two checkpoints
          snake_case__					:	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 '_'
          snake_case__					:	Optional[Any]          =					[]
          for pretrained_model_name_or_path in pretrained_model_name_or_path_list:
                 snake_case__					:	Any          =					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						)

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

                 snake_case__					:	List[str]          =					(
                     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
          snake_case__					:	int          =					DiffusionPipeline.from_pretrained(
              cached_folders[0]					, torch_dtype=lowerCamelCase					, device_map=lowerCamelCase						)
          final_pipe.to(self.device						)

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

          if interp == "sigmoid":
                 snake_case__					:	Dict          =					CheckpointMergerPipeline.sigmoid
          elif interp == "inv_sigmoid":
                 snake_case__					:	Union[str, Any]          =					CheckpointMergerPipeline.inv_sigmoid
          elif interp == "add_diff":
                 snake_case__					:	Optional[int]          =					CheckpointMergerPipeline.add_difference
          else:
                 snake_case__					:	str          =					CheckpointMergerPipeline.weighted_sum

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

                               snake_case__					:	Any          =					getattr(lowerCamelCase					, """load_state_dict"""						)
                               snake_case__					:	Optional[int]          =					(
                                   safetensors.torch.load_file(lowerCamelCase						)
                                   if (is_safetensors_available() and checkpoint_path_a.endswith(""".safetensors"""						))
                                   else torch.load(lowerCamelCase					, map_location="""cpu"""						)
                               )
                               snake_case__					:	int          =					None
                               if checkpoint_path_a:
                                      snake_case__					:	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"""						)
                                      )

                               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:
                                      snake_case__					:	Optional[Any]          =					theta_func(theta_a[key]					, theta_a[key]					, theta_a[key]					, lowerCamelCase						)
                               else:
                                      snake_case__					:	str          =					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	:					Optional[Any]					, lowerCamelCase	:					Optional[int]					, lowerCamelCase	:					Union[str, Any]					, lowerCamelCase	:					List[str]						)->	str:
          return ((1 - alpha) * thetaa) + (alpha * thetaa)




   @staticmethod
   def   __lowerCAmelCase						(	lowerCamelCase	:					Union[str, Any]					, lowerCamelCase	:					Union[str, Any]					, lowerCamelCase	:					str					, lowerCamelCase	:					Tuple						)->	int:
          snake_case__					:	Optional[Any]          =					alpha * alpha * (3 - (2 * alpha))
          return thetaa + ((thetaa - thetaa) * alpha)




   @staticmethod
   def   __lowerCAmelCase						(	lowerCamelCase	:					int					, lowerCamelCase	:					int					, lowerCamelCase	:					Optional[Any]					, lowerCamelCase	:					Union[str, Any]						)->	Tuple:
          import math

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




   @staticmethod
   def   __lowerCAmelCase						(	lowerCamelCase	:					int					, lowerCamelCase	:					Optional[int]					, lowerCamelCase	:					Optional[Any]					, lowerCamelCase	:					int						)->	Tuple:
          return thetaa + (thetaa - thetaa) * (1.0 - alpha)












'''simple docstring'''


def 						__magic_name__(			_A       ):
						'''simple docstring'''





						UpperCamelCase__       =			len(_A       )

						# Set the gap value to be decreased by a factor of 1.3
						# after each iteration
						UpperCamelCase__       =			1.3

						# Continue sorting until the gap is 1
						while gap > 1:
												# Decrease the gap value
												UpperCamelCase__       =			int(gap / shrink       )

												# Sort the elements using insertion sort
												for i in range(_A    ,   len(_A       )       ):
																		UpperCamelCase__       =			collection[i]
																		UpperCamelCase__       =			i
																		while j >= gap and collection[j - gap] > temp:
																								UpperCamelCase__       =			collection[j - gap]
																								j -= gap
																		UpperCamelCase__       =			temp

						return collection


if __name__ == "__main__":
					import doctest

					doctest.testmod()
