





"""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    _UpperCAmelCase  (      lowercase_					):


					def __init__(   self     :Any						):
												self.register_to_config()
												super().__init__()


					def       lowerCamelCase			(   self     :Tuple	,			__UpperCamelCase     :List[Any]	,			__UpperCamelCase     :Any						):
												if dicta == dicta:
																			return True
												else:
																			A,      A						=    self._remove_meta_keys(__UpperCamelCase						)
																			A,      A						=    self._remove_meta_keys(__UpperCamelCase						)
																			if configa == configa:
																										print(f"Warning !: Mismatch in keys {meta_keysa} and {meta_keysa}."						)
																										return True
												return False


					def       lowerCamelCase			(   self     :Tuple	,			__UpperCamelCase     :Dict						):
												A						=    []
												A						=    config_dict.copy()
												for key in config_dict.keys():
																			if key.startswith("_"						):
																										temp_dict.pop(__UpperCamelCase						)
																										meta_keys.append(__UpperCamelCase						)
												return (temp_dict, meta_keys)


					@torch.no_grad()
					def       lowerCamelCase			(   self     :Optional[Any]	,			__UpperCamelCase     :List[Union[str, os.PathLike]]	,			**__UpperCamelCase     :List[str]						):
												A						=    kwargs.pop("cache_dir"	,			__UpperCamelCase						)
												A						=    kwargs.pop("resume_download"	,			__UpperCamelCase						)
												A						=    kwargs.pop("force_download"	,			__UpperCamelCase						)
												A						=    kwargs.pop("proxies"	,			__UpperCamelCase						)
												A						=    kwargs.pop("local_files_only"	,			__UpperCamelCase						)
												A						=    kwargs.pop("use_auth_token"	,			__UpperCamelCase						)
												A						=    kwargs.pop("revision"	,			__UpperCamelCase						)
												A						=    kwargs.pop("torch_dtype"	,			__UpperCamelCase						)
												A						=    kwargs.pop("device_map"	,			__UpperCamelCase						)

												A						=    kwargs.pop("alpha"	,			0.5						)
												A						=    kwargs.pop("interp"	,			__UpperCamelCase						)

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

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

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

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

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

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

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

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

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

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

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

																										del theta_a
																										del theta_a
																										update_theta_a(__UpperCamelCase						)

																										del theta_a
												return final_pipe


					@staticmethod
					def       lowerCamelCase			(   __UpperCamelCase     :str	,			__UpperCamelCase     :Optional[Any]	,			__UpperCamelCase     :Union[str, Any]	,			__UpperCamelCase     :List[Any]						):
												return ((1 - alpha) * thetaa) + (alpha * thetaa)


					@staticmethod
					def       lowerCamelCase			(   __UpperCamelCase     :List[str]	,			__UpperCamelCase     :Optional[int]	,			__UpperCamelCase     :str	,			__UpperCamelCase     :str						):
												A						=    alpha * alpha * (3 - (2 * alpha))
												return thetaa + ((thetaa - thetaa) * alpha)


					@staticmethod
					def       lowerCamelCase			(   __UpperCamelCase     :Tuple	,			__UpperCamelCase     :List[Any]	,			__UpperCamelCase     :int	,			__UpperCamelCase     :Tuple						):
												import math

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


					@staticmethod
					def       lowerCamelCase			(   __UpperCamelCase     :int	,			__UpperCamelCase     :Optional[int]	,			__UpperCamelCase     :Union[str, Any]	,			__UpperCamelCase     :List[str]						):
												return thetaa + (thetaa - thetaa) * (1.0 - alpha)









"""simple docstring"""
import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin


if is_torch_available():
		import torch

if is_vision_available():
		from PIL import Image

		from transformers import BlipImageProcessor






class    _UpperCAmelCase  (      unittest.TestCase					):


					def __init__(   self     :Optional[Any]	,			__UpperCamelCase     :Optional[int]	,			__UpperCamelCase     :Dict=7	,			__UpperCamelCase     :Optional[Any]=3	,			__UpperCamelCase     :Tuple=18	,			__UpperCamelCase     :List[Any]=30	,			__UpperCamelCase     :Optional[Any]=4_00	,			__UpperCamelCase     :Union[str, Any]=True	,			__UpperCamelCase     :Tuple=None	,			__UpperCamelCase     :List[str]=True	,			__UpperCamelCase     :Any=False	,			__UpperCamelCase     :str=[0.48_145_466, 0.4_578_275, 0.40_821_073]	,			__UpperCamelCase     :Optional[Any]=[0.26_862_954, 0.26_130_258, 0.27_577_711]	,			__UpperCamelCase     :Optional[Any]=True	,			):
												A						=    size if size is not None else {"height": 20, "width": 20}
												A						=    parent
												A						=    batch_size
												A						=    num_channels
												A						=    image_size
												A						=    min_resolution
												A						=    max_resolution
												A						=    do_resize
												A						=    size
												A						=    do_normalize
												A						=    image_mean
												A						=    image_std
												A						=    do_pad
												A						=    do_convert_rgb


					def       lowerCamelCase			(   self     :List[str]						):
												return {
												    "do_resize": self.do_resize,
												    "size": self.size,
												    "do_normalize": self.do_normalize,
												    "image_mean": self.image_mean,
												    "image_std": self.image_std,
												    "do_convert_rgb": self.do_convert_rgb,
												    "do_pad": self.do_pad,
												}


					def       lowerCamelCase			(   self     :Optional[Any]	,			__UpperCamelCase     :Optional[int]=False	,			__UpperCamelCase     :Dict=False	,			__UpperCamelCase     :Dict=False						):
												assert not (numpify and torchify), "You cannot specify both numpy and PyTorch tensors at the same time"

												if equal_resolution:
																			A						=    []
																			for i in range(self.batch_size						):
																										image_inputs.append(
																										    np.random.randint(
																										        2_55	,			size=(self.num_channels, self.max_resolution, self.max_resolution)	,			dtype=np.uinta						)						)
												else:
																			A						=    []
																			for i in range(self.batch_size						):
																										A,      A						=    np.random.choice(np.arange(self.min_resolution	,			self.max_resolution						)	,			2						)
																										image_inputs.append(np.random.randint(2_55	,			size=(self.num_channels, width, height)	,			dtype=np.uinta						)						)

												if not numpify and not torchify:
																			# PIL expects the channel dimension as last dimension
																			A						=    [Image.fromarray(np.moveaxis(__UpperCamelCase	,			0	,			-1						)						) for x in image_inputs]

												if torchify:
																			A						=    [torch.from_numpy(__UpperCamelCase						) for x in image_inputs]

												return image_inputs







@require_torch
@require_vision
class    _UpperCAmelCase  (      lowercase_	,				unittest.TestCase					):
					UpperCamelCase						=				BlipImageProcessor if is_vision_available() else None


					def       lowerCamelCase			(   self     :Optional[Any]						):
												A						=    BlipImageProcessingTester(self						)


					@property
					def       lowerCamelCase			(   self     :List[str]						):
												return self.image_processor_tester.prepare_image_processor_dict()


					def       lowerCamelCase			(   self     :Any						):
												A						=    self.image_processing_class(**self.image_processor_dict						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"do_resize"						)						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"size"						)						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"do_normalize"						)						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"image_mean"						)						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"image_std"						)						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"do_convert_rgb"						)						)


					def       lowerCamelCase			(   self     :List[Any]						):
												pass


					def       lowerCamelCase			(   self     :Optional[Any]						):
												# Initialize image_processor
												A						=    self.image_processing_class(**self.image_processor_dict						)
												# create random PIL images
												A						=    self.image_processor_tester.prepare_inputs(equal_resolution=__UpperCamelCase						)
												for image in image_inputs:
																			self.assertIsInstance(__UpperCamelCase	,			Image.Image						)

												# Test not batched input
												A						=    image_processor(image_inputs[0]	,			return_tensors="pt"						).pixel_values
												self.assertEqual(
												    encoded_images.shape	,			(
												        1,
												        self.image_processor_tester.num_channels,
												        self.image_processor_tester.size["height"],
												        self.image_processor_tester.size["width"],
												    )	,			)

												# Test batched
												A						=    image_processor(__UpperCamelCase	,			return_tensors="pt"						).pixel_values
												self.assertEqual(
												    encoded_images.shape	,			(
												        self.image_processor_tester.batch_size,
												        self.image_processor_tester.num_channels,
												        self.image_processor_tester.size["height"],
												        self.image_processor_tester.size["width"],
												    )	,			)


					def       lowerCamelCase			(   self     :Union[str, Any]						):
												# Initialize image_processor
												A						=    self.image_processing_class(**self.image_processor_dict						)
												# create random numpy tensors
												A						=    self.image_processor_tester.prepare_inputs(equal_resolution=__UpperCamelCase	,			numpify=__UpperCamelCase						)
												for image in image_inputs:
																			self.assertIsInstance(__UpperCamelCase	,			np.ndarray						)

												# Test not batched input
												A						=    image_processor(image_inputs[0]	,			return_tensors="pt"						).pixel_values
												self.assertEqual(
												    encoded_images.shape	,			(
												        1,
												        self.image_processor_tester.num_channels,
												        self.image_processor_tester.size["height"],
												        self.image_processor_tester.size["width"],
												    )	,			)

												# Test batched
												A						=    image_processor(__UpperCamelCase	,			return_tensors="pt"						).pixel_values
												self.assertEqual(
												    encoded_images.shape	,			(
												        self.image_processor_tester.batch_size,
												        self.image_processor_tester.num_channels,
												        self.image_processor_tester.size["height"],
												        self.image_processor_tester.size["width"],
												    )	,			)


					def       lowerCamelCase			(   self     :Tuple						):
												# Initialize image_processor
												A						=    self.image_processing_class(**self.image_processor_dict						)
												# create random PyTorch tensors
												A						=    self.image_processor_tester.prepare_inputs(equal_resolution=__UpperCamelCase	,			torchify=__UpperCamelCase						)
												for image in image_inputs:
																			self.assertIsInstance(__UpperCamelCase	,			torch.Tensor						)

												# Test not batched input
												A						=    image_processor(image_inputs[0]	,			return_tensors="pt"						).pixel_values
												self.assertEqual(
												    encoded_images.shape	,			(
												        1,
												        self.image_processor_tester.num_channels,
												        self.image_processor_tester.size["height"],
												        self.image_processor_tester.size["width"],
												    )	,			)

												# Test batched
												A						=    image_processor(__UpperCamelCase	,			return_tensors="pt"						).pixel_values
												self.assertEqual(
												    encoded_images.shape	,			(
												        self.image_processor_tester.batch_size,
												        self.image_processor_tester.num_channels,
												        self.image_processor_tester.size["height"],
												        self.image_processor_tester.size["width"],
												    )	,			)







@require_torch
@require_vision
class    _UpperCAmelCase  (      lowercase_	,				unittest.TestCase					):
					UpperCamelCase						=				BlipImageProcessor if is_vision_available() else None


					def       lowerCamelCase			(   self     :int						):
												A						=    BlipImageProcessingTester(self	,			num_channels=4						)
												A						=    3


					@property
					def       lowerCamelCase			(   self     :Optional[Any]						):
												return self.image_processor_tester.prepare_image_processor_dict()


					def       lowerCamelCase			(   self     :int						):
												A						=    self.image_processing_class(**self.image_processor_dict						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"do_resize"						)						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"size"						)						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"do_normalize"						)						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"image_mean"						)						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"image_std"						)						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"do_convert_rgb"						)						)


					def       lowerCamelCase			(   self     :Any						):
												pass


					def       lowerCamelCase			(   self     :Optional[Any]						):
												# Initialize image_processor
												A						=    self.image_processing_class(**self.image_processor_dict						)
												# create random PIL images
												A						=    self.image_processor_tester.prepare_inputs(equal_resolution=__UpperCamelCase						)
												for image in image_inputs:
																			self.assertIsInstance(__UpperCamelCase	,			Image.Image						)

												# Test not batched input
												A						=    image_processor(image_inputs[0]	,			return_tensors="pt"						).pixel_values
												self.assertEqual(
												    encoded_images.shape	,			(
												        1,
												        self.expected_encoded_image_num_channels,
												        self.image_processor_tester.size["height"],
												        self.image_processor_tester.size["width"],
												    )	,			)

												# Test batched
												A						=    image_processor(__UpperCamelCase	,			return_tensors="pt"						).pixel_values
												self.assertEqual(
												    encoded_images.shape	,			(
												        self.image_processor_tester.batch_size,
												        self.expected_encoded_image_num_channels,
												        self.image_processor_tester.size["height"],
												        self.image_processor_tester.size["width"],
												    )	,			)


