



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







					def __init__(							self   :  Optional[int]):
											self.register_to_config()
											super().__init__()







					def 			_lowercase      (							self   :  List[Any]      ,		_A   :  Optional[int]      ,		_A   :  str):
											if dicta == dicta:
																	return True
											else:
																	A__     ,							A__						:				Optional[Any]         =			self._remove_meta_keys(lowercase_)
																	A__     ,							A__						:				Optional[Any]         =			self._remove_meta_keys(lowercase_)
																	if configa == configa:
																							print(F'Warning !: Mismatch in keys {meta_keysa} and {meta_keysa}.')
																							return True
											return False







					def 			_lowercase      (							self   :  List[Any]      ,		_A   :  List[Any]):
											A__						:				int         =			[]
											A__						:				Optional[Any]         =			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 			_lowercase      (							self   :  Optional[Any]      ,		_A   :  Dict      ,		**_A   :  str):
											A__						:				List[Any]         =			kwargs.pop("cache_dir"      ,		lowercase_)
											A__						:				List[Any]         =			kwargs.pop("resume_download"      ,		lowercase_)
											A__						:				Optional[int]         =			kwargs.pop("force_download"      ,		lowercase_)
											A__						:				str         =			kwargs.pop("proxies"      ,		lowercase_)
											A__						:				List[str]         =			kwargs.pop("local_files_only"      ,		lowercase_)
											A__						:				int         =			kwargs.pop("use_auth_token"      ,		lowercase_)
											A__						:				Union[str, Any]         =			kwargs.pop("revision"      ,		lowercase_)
											A__						:				Optional[Any]         =			kwargs.pop("torch_dtype"      ,		lowercase_)
											A__						:				str         =			kwargs.pop("device_map"      ,		lowercase_)

											A__						:				List[Any]         =			kwargs.pop("alpha"      ,		0.5)
											A__						:				List[str]         =			kwargs.pop("interp"      ,		lowercase_)

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

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

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

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

											A__						:				Optional[Any]         =			None
											if len(lowercase_) > 2:
																	A__						:				List[Any]         =			os.path.join(cached_folders[2])

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

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

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







					@staticmethod
					def 			_lowercase      (							_A   :  Optional[int]      ,		_A   :  List[Any]      ,		_A   :  Optional[int]      ,		_A   :  List[Any]):
											A__						:				Optional[Any]         =			alpha * alpha * (3 - (2 * alpha))
											return thetaa + ((thetaa - thetaa) * alpha)







					@staticmethod
					def 			_lowercase      (							_A   :  Optional[int]      ,		_A   :  Tuple      ,		_A   :  Dict      ,		_A   :  Tuple):
											import math

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







					@staticmethod
					def 			_lowercase      (							_A   :  int      ,		_A   :  Tuple      ,		_A   :  str      ,		_A   :  Tuple):
											return thetaa + (thetaa - thetaa) * (1.0 - alpha)


'''simple docstring'''

from __future__ import annotations

import inspect
import unittest
from typing import List, Tuple

from transformers import ConvNextConfig
from transformers.testing_utils import require_tf, require_vision, slow
from transformers.utils import cached_property, is_tf_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
						import tensorflow as tf

						from transformers import TFConvNextForImageClassification, TFConvNextModel


if is_vision_available():
						from PIL import Image

						from transformers import ConvNextImageProcessor







class lowerCamelCase   :

				def __init__(       self,						lowercase_,						lowercase_=13,						lowercase_=32,						lowercase_=3,						lowercase_=4,						lowercase_=[10, 20, 30, 40],						lowercase_=[2, 2, 3, 2],						lowercase_=True,						lowercase_=True,						lowercase_=37,						lowercase_="gelu",						lowercase_=10,						lowercase_=0.02,						lowercase_=3,						lowercase_=None,						)    ->							Dict:
					snake_case							=					parent
					snake_case							=					batch_size
					snake_case							=					image_size
					snake_case							=					num_channels
					snake_case							=					num_stages
					snake_case							=					hidden_sizes
					snake_case							=					depths
					snake_case							=					is_training
					snake_case							=					use_labels
					snake_case							=					intermediate_size
					snake_case							=					hidden_act
					snake_case							=					type_sequence_label_size
					snake_case							=					initializer_range
					snake_case							=					scope

				def 					_lowerCamelCase    (       self )    ->							List[Any]:
					snake_case							=					floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size] )

					snake_case							=					None
					if self.use_labels:
						snake_case							=					ids_tensor([self.batch_size],						self.type_sequence_label_size )

					snake_case							=					self.get_config()

					return config, pixel_values, labels

				def 					_lowerCamelCase    (       self )    ->							Any:
					return ConvNextConfig(
					    num_channels=self.num_channels,						hidden_sizes=self.hidden_sizes,						depths=self.depths,						num_stages=self.num_stages,						hidden_act=self.hidden_act,						is_decoder=lowercase_,						initializer_range=self.initializer_range,						)

				def 					_lowerCamelCase    (       self,						lowercase_,						lowercase_,						lowercase_ )    ->							List[Any]:
					snake_case							=					TFConvNextModel(config=lowercase_ )
					snake_case							=					model(lowercase_,						training=lowercase_ )
					# expected last hidden states: B, C, H // 32, W // 32
					self.parent.assertEqual(
					    result.last_hidden_state.shape,						(self.batch_size, self.hidden_sizes[-1], self.image_size // 32, self.image_size // 32),						)

				def 					_lowerCamelCase    (       self,						lowercase_,						lowercase_,						lowercase_ )    ->							Union[str, Any]:
					snake_case							=					self.type_sequence_label_size
					snake_case							=					TFConvNextForImageClassification(lowercase_ )
					snake_case							=					model(lowercase_,						labels=lowercase_,						training=lowercase_ )
					self.parent.assertEqual(result.logits.shape,						(self.batch_size, self.type_sequence_label_size) )






				def 					_lowerCamelCase    (       self )    ->							Optional[int]:
					snake_case							=					self.prepare_config_and_inputs()
					snake_case			,			snake_case			,			snake_case							=					config_and_inputs
					snake_case							=					{'pixel_values': pixel_values}
					return config, inputs_dict







@require_tf
class lowerCamelCase   (    __lowerCAmelCase				,				__lowerCAmelCase				,				unittest.TestCase				):
				snake_case_							     = (TFConvNextModel, TFConvNextForImageClassification) if is_tf_available() else ()
				snake_case_							     = (
				    {'''feature-extraction''': TFConvNextModel, '''image-classification''': TFConvNextForImageClassification}
				    if is_tf_available()
				    else {}
				)

				snake_case_							     = False
				snake_case_							     = False
				snake_case_							     = False
				snake_case_							     = False
				snake_case_							     = False

				def 					_lowerCamelCase    (       self )    ->							str:
					snake_case							=					TFConvNextModelTester(self )
					snake_case							=					ConfigTester(
					    self,						config_class=lowercase_,						has_text_modality=lowercase_,						hidden_size=37,						)

				@unittest.skip(reason='ConvNext does not use inputs_embeds' )
				def 					_lowerCamelCase    (       self )    ->							Tuple:
					pass

				@unittest.skipIf(
				    not is_tf_available() or len(tf.config.list_physical_devices('GPU' ) ) == 0,						reason='TF does not support backprop for grouped convolutions on CPU.',						)
				@slow
				def 					_lowerCamelCase    (       self )    ->							Optional[Any]:
					super().test_keras_fit()

				@unittest.skip(reason='ConvNext does not support input and output embeddings' )
				def 					_lowerCamelCase    (       self )    ->							Dict:
					pass

				def 					_lowerCamelCase    (       self )    ->							List[Any]:
					snake_case			,			snake_case							=					self.model_tester.prepare_config_and_inputs_for_common()

					for model_class in self.all_model_classes:
						snake_case							=					model_class(lowercase_ )
						snake_case							=					inspect.signature(model.call )
						# signature.parameters is an OrderedDict => so arg_names order is deterministic
						snake_case							=					[*signature.parameters.keys()]

						snake_case							=					['pixel_values']
						self.assertListEqual(arg_names[:1],						lowercase_ )

				def 					_lowerCamelCase    (       self )    ->							str:
					snake_case							=					self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_model(*lowercase_ )

				@unittest.skipIf(
				    not is_tf_available() or len(tf.config.list_physical_devices('GPU' ) ) == 0,						reason='TF does not support backprop for grouped convolutions on CPU.',						)
				def 					_lowerCamelCase    (       self )    ->							Optional[int]:
					super().test_dataset_conversion()

				def 					_lowerCamelCase    (       self )    ->							Union[str, Any]:
					def check_hidden_states_output(lowercase_,						lowercase_,						lowercase_ ):
						snake_case							=					model_class(lowercase_ )

						snake_case							=					model(**self._prepare_for_class(lowercase_,						lowercase_ ) )
						snake_case							=					outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states

						snake_case							=					self.model_tester.num_stages
						self.assertEqual(len(lowercase_ ),						expected_num_stages + 1 )

						# ConvNext's feature maps are of shape (batch_size, num_channels, height, width)
						self.assertListEqual(
						    list(hidden_states[0].shape[-2:] ),						[self.model_tester.image_size // 4, self.model_tester.image_size // 4],						)

					snake_case			,			snake_case							=					self.model_tester.prepare_config_and_inputs_for_common()

					for model_class in self.all_model_classes:
						snake_case							=					True
						check_hidden_states_output(lowercase_,						lowercase_,						lowercase_ )

						# check that output_hidden_states also work using config
						del inputs_dict["output_hidden_states"]
						snake_case							=					True

						check_hidden_states_output(lowercase_,						lowercase_,						lowercase_ )

				def 					_lowerCamelCase    (       self )    ->							Dict:
					snake_case			,			snake_case							=					self.model_tester.prepare_config_and_inputs_for_common()

					def check_equivalence(lowercase_,						lowercase_,						lowercase_,						lowercase_={} ):
						snake_case							=					model(lowercase_,						return_dict=lowercase_,						**lowercase_ )
						snake_case							=					model(lowercase_,						return_dict=lowercase_,						**lowercase_ ).to_tuple()

						def recursive_check(lowercase_,						lowercase_ ):
							if isinstance(lowercase_,						(List, Tuple) ):
								for tuple_iterable_value, dict_iterable_value in zip(lowercase_,						lowercase_ ):
									recursive_check(lowercase_,						lowercase_ )
							elif tuple_object is None:
								return
							else:
								self.assertTrue(
								    all(tf.equal(lowercase_,						lowercase_ ) ),						msg=(
								        'Tuple and dict output are not equal. Difference:'
								        F''' {tf.math.reduce_max(tf.abs(tuple_object - dict_object ) )}'''
								    ),						)

							recursive_check(lowercase_,						lowercase_ )

					for model_class in self.all_model_classes:
						snake_case							=					model_class(lowercase_ )

						snake_case							=					self._prepare_for_class(lowercase_,						lowercase_ )
						snake_case							=					self._prepare_for_class(lowercase_,						lowercase_ )
						check_equivalence(lowercase_,						lowercase_,						lowercase_ )

						snake_case							=					self._prepare_for_class(lowercase_,						lowercase_,						return_labels=lowercase_ )
						snake_case							=					self._prepare_for_class(lowercase_,						lowercase_,						return_labels=lowercase_ )
						check_equivalence(lowercase_,						lowercase_,						lowercase_ )

						snake_case							=					self._prepare_for_class(lowercase_,						lowercase_ )
						snake_case							=					self._prepare_for_class(lowercase_,						lowercase_ )
						check_equivalence(lowercase_,						lowercase_,						lowercase_,						{'output_hidden_states': True} )

						snake_case							=					self._prepare_for_class(lowercase_,						lowercase_,						return_labels=lowercase_ )
						snake_case							=					self._prepare_for_class(lowercase_,						lowercase_,						return_labels=lowercase_ )
						check_equivalence(lowercase_,						lowercase_,						lowercase_,						{'output_hidden_states': True} )

				def 					_lowerCamelCase    (       self )    ->							List[str]:
					snake_case							=					self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_for_image_classification(*lowercase_ )






				@slow
				def 					_lowerCamelCase    (       self )    ->							str:
					snake_case							=					TFConvNextModel.from_pretrained('facebook/convnext-tiny-224' )
					self.assertIsNotNone(lowercase_ )

def      __magic_name__							(			)      ->		Optional[int]:
	snake_case							=					Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png'				)
	return image







@require_tf
@require_vision
class lowerCamelCase   (    unittest.TestCase				):

				@cached_property
				def 					_lowerCamelCase    (       self )    ->							List[str]:
					return ConvNextImageProcessor.from_pretrained('facebook/convnext-tiny-224' ) if is_vision_available() else None






				@slow
				def 					_lowerCamelCase    (       self )    ->							Dict:
					snake_case							=					TFConvNextForImageClassification.from_pretrained('facebook/convnext-tiny-224' )

					snake_case							=					self.default_image_processor
					snake_case							=					prepare_img()
					snake_case							=					image_processor(images=lowercase_,						return_tensors='tf' )

					# forward pass
					snake_case							=					model(**lowercase_ )

					# verify the logits
					snake_case							=					tf.TensorShape((1, 1000) )
					self.assertEqual(outputs.logits.shape,						lowercase_ )

					snake_case							=					tf.constant([-0.0_260, -0.4_739, 0.1_911] )

					tf.debugging.assert_near(outputs.logits[0, :3],						lowercase_,						atol=1E-4 )





