


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 	_lowercase				(      a_	):






				'''simple docstring'''



				def __init__(	self      :List[Any]     )		->      int:
										self.register_to_config()
										super().__init__()


				def 		__magic_name__(	self      :Tuple      ,      lowerCAmelCase__      :List[str]      ,      lowerCAmelCase__      :Optional[Any]     )		->      Optional[Any]:


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

										__SCREAMING_SNAKE_CASE						:					str		      =		kwargs.pop('''alpha'''      ,      0.5     )
										__SCREAMING_SNAKE_CASE						:					Union[str, Any]		      =		kwargs.pop('''interp'''      ,      snake_case__     )

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

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

										__SCREAMING_SNAKE_CASE						:					Tuple		      =		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.
										__SCREAMING_SNAKE_CASE						:					Optional[int]		      =		[]
										for pretrained_model_name_or_path, config_dict in zip(snake_case__      ,      snake_case__     ):
																__SCREAMING_SNAKE_CASE						:					List[str]		      =		[k for k in config_dict.keys() if not k.startswith('''_'''     )]
																__SCREAMING_SNAKE_CASE						:					List[Any]		      =		[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,
																]
																__SCREAMING_SNAKE_CASE						:					Union[str, Any]		      =		config_dict.get('''_class_name'''     )
																__SCREAMING_SNAKE_CASE						:					List[str]		      =		{'''diffusers''': __version__, '''pipeline_class''': requested_pipeline_class}

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

										__SCREAMING_SNAKE_CASE						:					str		      =		None
										if len(snake_case__     ) > 2:
																__SCREAMING_SNAKE_CASE						:					List[Any]		      =		os.path.join(cached_folders[2]     )

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

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

																												__SCREAMING_SNAKE_CASE						:					Optional[int]		      =		getattr(snake_case__      ,      '''load_state_dict'''     )
																												__SCREAMING_SNAKE_CASE						:					Union[str, 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'''     )
																												)
																												__SCREAMING_SNAKE_CASE						:					List[Any]		      =		None
																												if checkpoint_path_a:
																																		__SCREAMING_SNAKE_CASE						:					Tuple		      =		(
																																		    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:
																																		__SCREAMING_SNAKE_CASE						:					Union[str, Any]		      =		theta_func(theta_a[key]      ,      theta_a[key]      ,      theta_a[key]      ,      snake_case__     )
																												else:
																																		__SCREAMING_SNAKE_CASE						:					List[Any]		      =		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 		__magic_name__(	lowerCAmelCase__      :List[Any]      ,      lowerCAmelCase__      :Union[str, Any]      ,      lowerCAmelCase__      :List[str]      ,      lowerCAmelCase__      :List[Any]     )		->      List[Any]:
										return ((1 - alpha) * thetaa) + (alpha * thetaa)


				@staticmethod
				def 		__magic_name__(	lowerCAmelCase__      :Any      ,      lowerCAmelCase__      :Optional[int]      ,      lowerCAmelCase__      :Optional[Any]      ,      lowerCAmelCase__      :Dict     )		->      Tuple:
										__SCREAMING_SNAKE_CASE						:					Any		      =		alpha * alpha * (3 - (2 * alpha))
										return thetaa + ((thetaa - thetaa) * alpha)


				@staticmethod
				def 		__magic_name__(	lowerCAmelCase__      :List[Any]      ,      lowerCAmelCase__      :Dict      ,      lowerCAmelCase__      :List[str]      ,      lowerCAmelCase__      :Tuple     )		->      Optional[int]:
										import math

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



				@staticmethod
				def 		__magic_name__(	lowerCAmelCase__      :Any      ,      lowerCAmelCase__      :str      ,      lowerCAmelCase__      :Union[str, Any]      ,      lowerCAmelCase__      :Optional[int]     )		->      Tuple:
										return thetaa + (thetaa - thetaa) * (1.0 - alpha)




from typing import Any, Callable, Dict, List, Optional, Union

import PIL.Image
import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionImgaImgPipeline,
    StableDiffusionInpaintPipelineLegacy,
    StableDiffusionPipeline,
    UNetaDConditionModel,
)
from diffusers.configuration_utils import FrozenDict
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.utils import deprecate, logging


__lowerCAmelCase     :			Tuple      =logging.get_logger(__name__)  # pylint: disable=invalid-name
class 	_lowercase				(      A__	):






				'''simple docstring'''



				SCREAMING_SNAKE_CASE__    :      Union[str, Any]      =		['''safety_checker''', '''feature_extractor''']


				def __init__(	self      :int      ,      lowerCAmelCase__      :AutoencoderKL      ,      lowerCAmelCase__      :CLIPTextModel      ,      lowerCAmelCase__      :CLIPTokenizer      ,      lowerCAmelCase__      :UNetaDConditionModel      ,      lowerCAmelCase__      :Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]      ,      lowerCAmelCase__      :StableDiffusionSafetyChecker      ,      lowerCAmelCase__      :CLIPImageProcessor      ,      lowerCAmelCase__      :bool = True      ,      )		->      List[Any]:
										super().__init__()
										if hasattr(scheduler.config      ,      '''steps_offset'''     ) and scheduler.config.steps_offset != 1:
																__SCREAMING_SNAKE_CASE						:					Union[str, Any]		      =		(
																    f'''The configuration file of this scheduler: {scheduler} is outdated. `steps_offset`'''
																    f''' should be set to 1 instead of {scheduler.config.steps_offset}. Please make sure '''
																    '''to update the config accordingly as leaving `steps_offset` might led to incorrect results'''
																    ''' in future versions. If you have downloaded this checkpoint from the Hugging Face Hub,'''
																    ''' it would be very nice if you could open a Pull request for the `scheduler/scheduler_config.json`'''
																    ''' file'''
																)
																deprecate('''steps_offset!=1'''      ,      '''1.0.0'''      ,      lowerCAmelCase__      ,      standard_warn=lowerCAmelCase__     )
																__SCREAMING_SNAKE_CASE						:					Dict		      =		dict(scheduler.config     )
																__SCREAMING_SNAKE_CASE						:					Tuple		      =		1
																__SCREAMING_SNAKE_CASE						:					Dict		      =		FrozenDict(lowerCAmelCase__     )

										self.register_modules(
										    vae=lowerCAmelCase__      ,      text_encoder=lowerCAmelCase__      ,      tokenizer=lowerCAmelCase__      ,      unet=lowerCAmelCase__      ,      scheduler=lowerCAmelCase__      ,      safety_checker=lowerCAmelCase__      ,      feature_extractor=lowerCAmelCase__      ,      )
										self.register_to_config(requires_safety_checker=lowerCAmelCase__     )


				@property
				def 		__magic_name__(	self      :Union[str, Any]     )		->      Dict[str, Any]:
										return {k: getattr(self      ,      lowerCAmelCase__     ) for k in self.config.keys() if not k.startswith('''_'''     )}


				def 		__magic_name__(	self      :str      ,      lowerCAmelCase__      :Optional[Union[str, int]] = "auto"     )		->      str:
										if slice_size == "auto":
																# half the attention head size is usually a good trade-off between
																# speed and memory
																__SCREAMING_SNAKE_CASE						:					Union[str, Any]		      =		self.unet.config.attention_head_dim // 2
										self.unet.set_attention_slice(lowerCAmelCase__     )


				def 		__magic_name__(	self      :Optional[int]     )		->      Union[str, Any]:
										self.enable_attention_slicing(lowerCAmelCase__     )


				@torch.no_grad()
				def 		__magic_name__(	self      :str      ,      lowerCAmelCase__      :Union[str, List[str]]      ,      lowerCAmelCase__      :Union[torch.FloatTensor, PIL.Image.Image]      ,      lowerCAmelCase__      :Union[torch.FloatTensor, PIL.Image.Image]      ,      lowerCAmelCase__      :float = 0.8      ,      lowerCAmelCase__      :Optional[int] = 50      ,      lowerCAmelCase__      :Optional[float] = 7.5      ,      lowerCAmelCase__      :Optional[Union[str, List[str]]] = None      ,      lowerCAmelCase__      :Optional[int] = 1      ,      lowerCAmelCase__      :Optional[float] = 0.0      ,      lowerCAmelCase__      :Optional[torch.Generator] = None      ,      lowerCAmelCase__      :Optional[str] = "pil"      ,      lowerCAmelCase__      :bool = True      ,      lowerCAmelCase__      :Optional[Callable[[int, int, torch.FloatTensor], None]] = None      ,      lowerCAmelCase__      :int = 1      ,      )		->      Any:
										# For more information on how this function works, please see: https://huggingface.co/docs/diffusers/api/pipelines/stable_diffusion#diffusers.StableDiffusionImg2ImgPipeline
										return StableDiffusionInpaintPipelineLegacy(**self.components     )(
										    prompt=lowerCAmelCase__      ,      image=lowerCAmelCase__      ,      mask_image=lowerCAmelCase__      ,      strength=lowerCAmelCase__      ,      num_inference_steps=lowerCAmelCase__      ,      guidance_scale=lowerCAmelCase__      ,      negative_prompt=lowerCAmelCase__      ,      num_images_per_prompt=lowerCAmelCase__      ,      eta=lowerCAmelCase__      ,      generator=lowerCAmelCase__      ,      output_type=lowerCAmelCase__      ,      return_dict=lowerCAmelCase__      ,      callback=lowerCAmelCase__      ,      )


				@torch.no_grad()
				def 		__magic_name__(	self      :Dict      ,      lowerCAmelCase__      :Union[str, List[str]]      ,      lowerCAmelCase__      :Union[torch.FloatTensor, PIL.Image.Image]      ,      lowerCAmelCase__      :float = 0.8      ,      lowerCAmelCase__      :Optional[int] = 50      ,      lowerCAmelCase__      :Optional[float] = 7.5      ,      lowerCAmelCase__      :Optional[Union[str, List[str]]] = None      ,      lowerCAmelCase__      :Optional[int] = 1      ,      lowerCAmelCase__      :Optional[float] = 0.0      ,      lowerCAmelCase__      :Optional[torch.Generator] = None      ,      lowerCAmelCase__      :Optional[str] = "pil"      ,      lowerCAmelCase__      :bool = True      ,      lowerCAmelCase__      :Optional[Callable[[int, int, torch.FloatTensor], None]] = None      ,      lowerCAmelCase__      :int = 1      ,      **lowerCAmelCase__      :Optional[Any]      ,      )		->      Dict:
										# For more information on how this function works, please see: https://huggingface.co/docs/diffusers/api/pipelines/stable_diffusion#diffusers.StableDiffusionImg2ImgPipeline
										return StableDiffusionImgaImgPipeline(**self.components     )(
										    prompt=lowerCAmelCase__      ,      image=lowerCAmelCase__      ,      strength=lowerCAmelCase__      ,      num_inference_steps=lowerCAmelCase__      ,      guidance_scale=lowerCAmelCase__      ,      negative_prompt=lowerCAmelCase__      ,      num_images_per_prompt=lowerCAmelCase__      ,      eta=lowerCAmelCase__      ,      generator=lowerCAmelCase__      ,      output_type=lowerCAmelCase__      ,      return_dict=lowerCAmelCase__      ,      callback=lowerCAmelCase__      ,      callback_steps=lowerCAmelCase__      ,      )



				@torch.no_grad()
				def 		__magic_name__(	self      :Tuple      ,      lowerCAmelCase__      :Union[str, List[str]]      ,      lowerCAmelCase__      :int = 512      ,      lowerCAmelCase__      :int = 512      ,      lowerCAmelCase__      :int = 50      ,      lowerCAmelCase__      :float = 7.5      ,      lowerCAmelCase__      :Optional[Union[str, List[str]]] = None      ,      lowerCAmelCase__      :Optional[int] = 1      ,      lowerCAmelCase__      :float = 0.0      ,      lowerCAmelCase__      :Optional[torch.Generator] = None      ,      lowerCAmelCase__      :Optional[torch.FloatTensor] = None      ,      lowerCAmelCase__      :Optional[str] = "pil"      ,      lowerCAmelCase__      :bool = True      ,      lowerCAmelCase__      :Optional[Callable[[int, int, torch.FloatTensor], None]] = None      ,      lowerCAmelCase__      :int = 1      ,      )		->      Optional[int]:
										# For more information on how this function https://huggingface.co/docs/diffusers/api/pipelines/stable_diffusion#diffusers.StableDiffusionPipeline
										return StableDiffusionPipeline(**self.components     )(
										    prompt=lowerCAmelCase__      ,      height=lowerCAmelCase__      ,      width=lowerCAmelCase__      ,      num_inference_steps=lowerCAmelCase__      ,      guidance_scale=lowerCAmelCase__      ,      negative_prompt=lowerCAmelCase__      ,      num_images_per_prompt=lowerCAmelCase__      ,      eta=lowerCAmelCase__      ,      generator=lowerCAmelCase__      ,      latents=lowerCAmelCase__      ,      output_type=lowerCAmelCase__      ,      return_dict=lowerCAmelCase__      ,      callback=lowerCAmelCase__      ,      callback_steps=lowerCAmelCase__      ,      )
