


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_						( _A       ):

 '''simple docstring'''

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



 def 						UpperCamelCase__     (	self		,      _UpperCAmelCase		,      _UpperCAmelCase					):
       if dicta == dicta:
             return True
       else:
             snake_case_       ,    snake_case_       =      self._remove_meta_keys(__SCREAMING_SNAKE_CASE					)
             snake_case_       ,    snake_case_       =      self._remove_meta_keys(__SCREAMING_SNAKE_CASE					)
             if configa == configa:
                   print(F'''Warning !: Mismatch in keys {meta_keysa} and {meta_keysa}.'''					)
                   return True
       return False



 def 						UpperCamelCase__     (	self		,      _UpperCAmelCase					):
       snake_case_       =      []
       snake_case_       =      config_dict.copy()
       for key in config_dict.keys():
             if key.startswith('''_'''					):
                   temp_dict.pop(__SCREAMING_SNAKE_CASE					)
                   meta_keys.append(__SCREAMING_SNAKE_CASE					)
       return (temp_dict, meta_keys)



 @torch.no_grad()
 def 						UpperCamelCase__     (	self		,      _UpperCAmelCase		,      **_UpperCAmelCase					):
       snake_case_       =      kwargs.pop('''cache_dir'''		,      __SCREAMING_SNAKE_CASE					)
       snake_case_       =      kwargs.pop('''resume_download'''		,      __SCREAMING_SNAKE_CASE					)
       snake_case_       =      kwargs.pop('''force_download'''		,      __SCREAMING_SNAKE_CASE					)
       snake_case_       =      kwargs.pop('''proxies'''		,      __SCREAMING_SNAKE_CASE					)
       snake_case_       =      kwargs.pop('''local_files_only'''		,      __SCREAMING_SNAKE_CASE					)
       snake_case_       =      kwargs.pop('''use_auth_token'''		,      __SCREAMING_SNAKE_CASE					)
       snake_case_       =      kwargs.pop('''revision'''		,      __SCREAMING_SNAKE_CASE					)
       snake_case_       =      kwargs.pop('''torch_dtype'''		,      __SCREAMING_SNAKE_CASE					)
       snake_case_       =      kwargs.pop('''device_map'''		,      __SCREAMING_SNAKE_CASE					)

       snake_case_       =      kwargs.pop('''alpha'''		,      0.5					)
       snake_case_       =      kwargs.pop('''interp'''		,      __SCREAMING_SNAKE_CASE					)

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

       snake_case_       =      len(__SCREAMING_SNAKE_CASE					)
       # Ignore result from model_index_json comparision of the two checkpoints
       snake_case_       =      kwargs.pop('''force'''		,      __SCREAMING_SNAKE_CASE					)

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

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

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

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

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

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

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

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

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

                   del theta_a
                   del theta_a
                   update_theta_a(__SCREAMING_SNAKE_CASE					)

                   del theta_a
       return final_pipe



 @staticmethod
 def 						UpperCamelCase__     (	_UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase					):
       return ((1 - alpha) * thetaa) + (alpha * thetaa)



 @staticmethod
 def 						UpperCamelCase__     (	_UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase					):
       snake_case_       =      alpha * alpha * (3 - (2 * alpha))
       return thetaa + ((thetaa - thetaa) * alpha)



 @staticmethod
 def 						UpperCamelCase__     (	_UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase					):
       import math

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





 @staticmethod
 def 						UpperCamelCase__     (	_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





