






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




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




					def 						A     (       self       :						Optional[int]							,				_A       :						List[str]							,				_A       :						Dict     )  ->    List[Any]:


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

								UpperCAmelCase_     :   Optional[int]       =   kwargs.pop('''alpha'''							,				0.5     )
								UpperCAmelCase_     :   Dict       =   kwargs.pop('''interp'''							,				__lowerCAmelCase     )

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

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

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

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

								UpperCAmelCase_     :   List[str]       =   None
								if len(__lowerCAmelCase     ) > 2:
											UpperCAmelCase_     :   int       =   os.path.join(cached_folders[2]     )

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

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

																	UpperCAmelCase_     :   Union[str, Any]       =   getattr(__lowerCAmelCase							,				'''load_state_dict'''     )
																	UpperCAmelCase_     :   Dict       =   (
																	    safetensors.torch.load_file(__lowerCAmelCase     )
																	    if (is_safetensors_available() and checkpoint_path_a.endswith('''.safetensors'''     ))
																	    else torch.load(__lowerCAmelCase							,				map_location='''cpu'''     )
																	)
																	UpperCAmelCase_     :   Optional[int]       =   None
																	if checkpoint_path_a:
																				UpperCAmelCase_     :   Optional[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:
																				UpperCAmelCase_     :   Union[str, Any]       =   theta_func(theta_a[key]							,				theta_a[key]							,				theta_a[key]							,				__lowerCAmelCase     )
																	else:
																				UpperCAmelCase_     :   List[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 						A     (       _A       :						Optional[Any]							,				_A       :						str							,				_A       :						Optional[Any]							,				_A       :						List[str]     )  ->    Dict:
								return ((1 - alpha) * thetaa) + (alpha * thetaa)




					@staticmethod
					def 						A     (       _A       :						List[Any]							,				_A       :						Optional[Any]							,				_A       :						List[Any]							,				_A       :						int     )  ->    Any:
								UpperCAmelCase_     :   Tuple       =   alpha * alpha * (3 - (2 * alpha))
								return thetaa + ((thetaa - thetaa) * alpha)




					@staticmethod
					def 						A     (       _A       :						Optional[Any]							,				_A       :						Optional[Any]							,				_A       :						Union[str, Any]							,				_A       :						str     )  ->    int:
								import math

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





					@staticmethod
					def 						A     (       _A       :						List[str]							,				_A       :						Any							,				_A       :						Tuple							,				_A       :						Optional[Any]     )  ->    Tuple:
								return thetaa + (thetaa - thetaa) * (1.0 - alpha)














'''simple docstring'''





import unittest

import torch

from diffusers import DDIMScheduler, TextToVideoZeroPipeline
from diffusers.utils import load_pt, require_torch_gpu, slow

from ..test_pipelines_common import assert_mean_pixel_difference


@slow
@require_torch_gpu
class 				snake_case__     (			unittest.TestCase):




					def 						A     (       self       :						Optional[Any]     )  ->    List[str]:
								UpperCAmelCase_     :   Optional[Any]       =   '''runwayml/stable-diffusion-v1-5'''
								UpperCAmelCase_     :   Optional[int]       =   TextToVideoZeroPipeline.from_pretrained(_A							,				torch_dtype=torch.floataa     ).to('''cuda'''     )
								UpperCAmelCase_     :   Tuple       =   DDIMScheduler.from_config(pipe.scheduler.config     )
								UpperCAmelCase_     :   Tuple       =   torch.Generator(device='''cuda'''     ).manual_seed(0     )

								UpperCAmelCase_     :   List[Any]       =   '''A bear is playing a guitar on Times Square'''
								UpperCAmelCase_     :   List[Any]       =   pipe(prompt=_A							,				generator=_A     ).images

								UpperCAmelCase_     :   Any       =   load_pt(
								    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text-to-video/A bear is playing a guitar on Times Square.pt'''     )

								assert_mean_pixel_difference(_A							,				_A     )


