






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


			'''simple docstring'''





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

									'''simple docstring'''







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






			def 					A   ( self				:    str			,   lowercase				:    str			,   lowercase				:    Union[str, Any]      )				->     Optional[int]:

									'''simple docstring'''







									if dicta == dicta:
															return True
									else:
															UpperCamelCase__     , UpperCamelCase__       =			self._remove_meta_keys(lowercase      )
															UpperCamelCase__     , UpperCamelCase__       =			self._remove_meta_keys(lowercase      )
															if configa == configa:
																					print(f"Warning !: Mismatch in keys {meta_keysa} and {meta_keysa}."      )
																					return True
									return False






			def 					A   ( self				:    Dict			,   lowercase				:    Dict      )				->     List[str]:

									'''simple docstring'''







									UpperCamelCase__       =			[]
									UpperCamelCase__       =			config_dict.copy()
									for key in config_dict.keys():
															if key.startswith("""_"""      ):
																					temp_dict.pop(lowercase      )
																					meta_keys.append(lowercase      )
									return (temp_dict, meta_keys)






			@torch.no_grad()
			def 					A   ( self				:    Union[str, Any]			,   lowercase				:    List[Union[str, os.PathLike]]			,   **lowercase				:    List[Any]      )				->     str:

									'''simple docstring'''







									UpperCamelCase__       =			kwargs.pop("""cache_dir"""			,   lowercase      )
									UpperCamelCase__       =			kwargs.pop("""resume_download"""			,   lowercase      )
									UpperCamelCase__       =			kwargs.pop("""force_download"""			,   lowercase      )
									UpperCamelCase__       =			kwargs.pop("""proxies"""			,   lowercase      )
									UpperCamelCase__       =			kwargs.pop("""local_files_only"""			,   lowercase      )
									UpperCamelCase__       =			kwargs.pop("""use_auth_token"""			,   lowercase      )
									UpperCamelCase__       =			kwargs.pop("""revision"""			,   lowercase      )
									UpperCamelCase__       =			kwargs.pop("""torch_dtype"""			,   lowercase      )
									UpperCamelCase__       =			kwargs.pop("""device_map"""			,   lowercase      )

									UpperCamelCase__       =			kwargs.pop("""alpha"""			,   0.5      )
									UpperCamelCase__       =			kwargs.pop("""interp"""			,   lowercase      )

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

									UpperCamelCase__       =			len(lowercase      )
									# Ignore result from model_index_json comparision of the two checkpoints
									UpperCamelCase__       =			kwargs.pop("""force"""			,   lowercase      )

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

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

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

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

									UpperCamelCase__       =			None
									if len(lowercase      ) > 2:
															UpperCamelCase__       =			os.path.join(cached_folders[2]      )

									if interp == "sigmoid":
															UpperCamelCase__       =			CheckpointMergerPipeline.sigmoid
									elif interp == "inv_sigmoid":
															UpperCamelCase__       =			CheckpointMergerPipeline.inv_sigmoid
									elif interp == "add_diff":
															UpperCamelCase__       =			CheckpointMergerPipeline.add_difference
									else:
															UpperCamelCase__       =			CheckpointMergerPipeline.weighted_sum

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

																											UpperCamelCase__       =			getattr(lowercase			,   """load_state_dict"""      )
																											UpperCamelCase__       =			(
																											    safetensors.torch.load_file(lowercase      )
																											    if (is_safetensors_available() and checkpoint_path_a.endswith(""".safetensors"""      ))
																											    else torch.load(lowercase			,   map_location="""cpu"""      )
																											)
																											UpperCamelCase__       =			None
																											if checkpoint_path_a:
																																	UpperCamelCase__       =			(
																																	    safetensors.torch.load_file(lowercase      )
																																	    if (is_safetensors_available() and checkpoint_path_a.endswith(""".safetensors"""      ))
																																	    else torch.load(lowercase			,   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(lowercase      )}"      )
																											continue
																					print(f"MERGING {attr}"      )

																					for key in theta_a.keys():
																											if theta_a:
																																	UpperCamelCase__       =			theta_func(theta_a[key]			,   theta_a[key]			,   theta_a[key]			,   lowercase      )
																											else:
																																	UpperCamelCase__       =			theta_func(theta_a[key]			,   theta_a[key]			,   lowercase			,   lowercase      )

																					del theta_a
																					del theta_a
																					update_theta_a(lowercase      )

																					del theta_a
									return final_pipe






			@staticmethod
			def 					A   ( lowercase				:    Optional[Any]			,   lowercase				:    Optional[int]			,   lowercase				:    Union[str, Any]			,   lowercase				:    List[str]      )				->     str:

									'''simple docstring'''







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






			@staticmethod
			def 					A   ( lowercase				:    Union[str, Any]			,   lowercase				:    Union[str, Any]			,   lowercase				:    str			,   lowercase				:    Tuple      )				->     int:

									'''simple docstring'''







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






			@staticmethod
			def 					A   ( lowercase				:    int			,   lowercase				:    int			,   lowercase				:    Optional[Any]			,   lowercase				:    Union[str, Any]      )				->     Tuple:

									'''simple docstring'''







									import math

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







			@staticmethod
			def 					A   ( lowercase				:    int			,   lowercase				:    Optional[int]			,   lowercase				:    Optional[Any]			,   lowercase				:    int      )				->     Tuple:

									'''simple docstring'''







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








'''simple docstring'''


from dataclasses import dataclass
from typing import List, Optional, Union

import numpy as np
import PIL
from PIL import Image

from ...utils import BaseOutput, is_torch_available, is_transformers_available
@dataclass
# Copied from diffusers.pipelines.stable_diffusion.__init__.StableDiffusionPipelineOutput with Stable->Alt
class  _SCREAMING_SNAKE_CASE				(		SCREAMING_SNAKE_CASE						):


			'''simple docstring'''





			__a				:				Union[List[PIL.Image.Image], np.ndarray]
			__a				:				Optional[List[bool]]


if is_transformers_available() and is_torch_available():
					from .modeling_roberta_series import RobertaSeriesModelWithTransformation
					from .pipeline_alt_diffusion import AltDiffusionPipeline
					from .pipeline_alt_diffusion_imgaimg import AltDiffusionImgaImgPipeline
