



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




	"""simple docstring"""






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






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

			_lowercase							:      Union[str, Any]			=					kwargs.pop("""alpha"""							,0.5					)
			_lowercase							:      List[Any]			=					kwargs.pop("""interp"""							,UpperCAmelCase_					)

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

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

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

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

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

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

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

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






	@staticmethod
	def 					lowerCamelCase__ (			UpperCAmelCase_							,UpperCAmelCase_							,UpperCAmelCase_							,UpperCAmelCase_					):
			_lowercase							:      Dict			=					alpha * alpha * (3 - (2 * alpha))
			return thetaa + ((thetaa - thetaa) * alpha)






	@staticmethod
	def 					lowerCamelCase__ (			UpperCAmelCase_							,UpperCAmelCase_							,UpperCAmelCase_							,UpperCAmelCase_					):
			import math

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






	@staticmethod
	def 					lowerCamelCase__ (			UpperCAmelCase_							,UpperCAmelCase_							,UpperCAmelCase_							,UpperCAmelCase_					):
			return thetaa + (thetaa - thetaa) * (1.0 - alpha)






"""simple docstring"""




import copy

from ...configuration_utils import PretrainedConfig
from ...utils import logging


__A     :    str    =		logging.get_logger(__name__)

__A     :    Union[str, Any]    =		{
    'facebook/sam-vit-huge': 'https://huggingface.co/facebook/sam-vit-huge/resolve/main/config.json',
    'facebook/sam-vit-large': 'https://huggingface.co/facebook/sam-vit-large/resolve/main/config.json',
    'facebook/sam-vit-base': 'https://huggingface.co/facebook/sam-vit-base/resolve/main/config.json',
}





class 			__UpperCamelCase (   _A						):


		def __init__(self :		int     ,     __SCREAMING_SNAKE_CASE :		Optional[int]=2_5_6     ,     __SCREAMING_SNAKE_CASE :		List[str]=1_0_2_4     ,     __SCREAMING_SNAKE_CASE :		Optional[int]=1_6     ,     __SCREAMING_SNAKE_CASE :		Optional[int]=1_6     ,     __SCREAMING_SNAKE_CASE :		List[str]=4     ,     __SCREAMING_SNAKE_CASE :		List[Any]="gelu"     ,     __SCREAMING_SNAKE_CASE :		str=1E-6     ,     **__SCREAMING_SNAKE_CASE :		Optional[int]     ,     ):
							super().__init__(**__SCREAMING_SNAKE_CASE)
							A			=					hidden_size
							A			=					image_size
							A			=					patch_size
							A			=					image_size // patch_size
							A			=					mask_input_channels
							A			=					num_point_embeddings
							A			=					hidden_act
							A			=					layer_norm_eps





class 			__UpperCamelCase (   _A						):


		def __init__(self :		List[Any]     ,     __SCREAMING_SNAKE_CASE :		Tuple=2_5_6     ,     __SCREAMING_SNAKE_CASE :		str="relu"     ,     __SCREAMING_SNAKE_CASE :		List[Any]=2_0_4_8     ,     __SCREAMING_SNAKE_CASE :		Tuple=2     ,     __SCREAMING_SNAKE_CASE :		Any=8     ,     __SCREAMING_SNAKE_CASE :		Union[str, Any]=2     ,     __SCREAMING_SNAKE_CASE :		Optional[int]=3     ,     __SCREAMING_SNAKE_CASE :		List[Any]=3     ,     __SCREAMING_SNAKE_CASE :		Tuple=2_5_6     ,     __SCREAMING_SNAKE_CASE :		int=1E-6     ,     **__SCREAMING_SNAKE_CASE :		int     ,     ):
							super().__init__(**__SCREAMING_SNAKE_CASE)
							A			=					hidden_size
							A			=					hidden_act
							A			=					mlp_dim
							A			=					num_hidden_layers
							A			=					num_attention_heads
							A			=					attention_downsample_rate
							A			=					num_multimask_outputs
							A			=					iou_head_depth
							A			=					iou_head_hidden_dim
							A			=					layer_norm_eps





class 			__UpperCamelCase (   _A						):


		def __init__(self :		List[str]     ,     __SCREAMING_SNAKE_CASE :		List[Any]=7_6_8     ,     __SCREAMING_SNAKE_CASE :		int=2_5_6     ,     __SCREAMING_SNAKE_CASE :		Tuple=1_2     ,     __SCREAMING_SNAKE_CASE :		str=1_2     ,     __SCREAMING_SNAKE_CASE :		Any=3     ,     __SCREAMING_SNAKE_CASE :		Dict=1_0_2_4     ,     __SCREAMING_SNAKE_CASE :		str=1_6     ,     __SCREAMING_SNAKE_CASE :		Tuple="gelu"     ,     __SCREAMING_SNAKE_CASE :		int=1E-06     ,     __SCREAMING_SNAKE_CASE :		Dict=0.0     ,     __SCREAMING_SNAKE_CASE :		str=1E-10     ,     __SCREAMING_SNAKE_CASE :		List[str]=True     ,     __SCREAMING_SNAKE_CASE :		Union[str, Any]=4.0     ,     __SCREAMING_SNAKE_CASE :		str=True     ,     __SCREAMING_SNAKE_CASE :		Any=True     ,     __SCREAMING_SNAKE_CASE :		Optional[int]=1_4     ,     __SCREAMING_SNAKE_CASE :		Tuple=[2, 5, 8, 1_1]     ,     __SCREAMING_SNAKE_CASE :		str=1_2_8     ,     __SCREAMING_SNAKE_CASE :		Optional[int]=None     ,     **__SCREAMING_SNAKE_CASE :		Optional[Any]     ,     ):
							super().__init__(**__SCREAMING_SNAKE_CASE)

							A			=					hidden_size
							A			=					output_channels
							A			=					num_hidden_layers
							A			=					num_attention_heads
							A			=					num_channels
							A			=					image_size
							A			=					patch_size
							A			=					hidden_act
							A			=					layer_norm_eps
							A			=					attention_dropout
							A			=					initializer_range
							A			=					qkv_bias
							A			=					mlp_ratio
							A			=					use_abs_pos
							A			=					use_rel_pos
							A			=					window_size
							A			=					global_attn_indexes
							A			=					num_pos_feats
							A			=					int(hidden_size * mlp_ratio) if mlp_dim is None else mlp_dim





class 			__UpperCamelCase (   _A						):
		SCREAMING_SNAKE_CASE								=    "sam"
		SCREAMING_SNAKE_CASE								=    True


		def __init__(self :		Optional[int]     ,     __SCREAMING_SNAKE_CASE :		Dict=None     ,     __SCREAMING_SNAKE_CASE :		Union[str, Any]=None     ,     __SCREAMING_SNAKE_CASE :		Optional[int]=None     ,     __SCREAMING_SNAKE_CASE :		List[Any]=0.0_2     ,     **__SCREAMING_SNAKE_CASE :		Optional[Any]     ,     ):
							super().__init__(**__SCREAMING_SNAKE_CASE)
							A			=					vision_config if vision_config is not None else {}
							A			=					prompt_encoder_config if prompt_encoder_config is not None else {}
							A			=					mask_decoder_config if mask_decoder_config is not None else {}

							if isinstance(__SCREAMING_SNAKE_CASE     ,     __SCREAMING_SNAKE_CASE):
												A			=					vision_config.to_dict()
							if isinstance(__SCREAMING_SNAKE_CASE     ,     __SCREAMING_SNAKE_CASE):
												A			=					prompt_encoder_config.to_dict()
							if isinstance(__SCREAMING_SNAKE_CASE     ,     __SCREAMING_SNAKE_CASE):
												A			=					mask_decoder_config.to_dict()

							A			=					SamVisionConfig(**__SCREAMING_SNAKE_CASE)
							A			=					SamPromptEncoderConfig(**__SCREAMING_SNAKE_CASE)
							A			=					SamMaskDecoderConfig(**__SCREAMING_SNAKE_CASE)
							A			=					initializer_range


		def 					SCREAMING_SNAKE_CASE__   (self :		List[str]):
							A			=					copy.deepcopy(self.__dict__)
							A			=					self.vision_config.to_dict()
							A			=					self.prompt_encoder_config.to_dict()
							A			=					self.mask_decoder_config.to_dict()
							A			=					self.__class__.model_type
							return output





