


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						(   _snake_case							):




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




	def  SCREAMING_SNAKE_CASE__       (	self      ,						_lowerCamelCase      ,						_lowerCamelCase					):
							if dicta == dicta:
													return True
							else:
													a		:Any											= self._remove_meta_keys(_lowerCamelCase					)
													a		: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  SCREAMING_SNAKE_CASE__       (	self      ,						_lowerCamelCase					):
							a		:Dict											= []
							a		:Any											= 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  SCREAMING_SNAKE_CASE__       (	self      ,						_lowerCamelCase      ,						**_lowerCamelCase					):
							a		:str											= kwargs.pop('''cache_dir'''      ,						_lowerCamelCase					)
							a		:Dict											= kwargs.pop('''resume_download'''      ,						_lowerCamelCase					)
							a		:List[str]											= kwargs.pop('''force_download'''      ,						_lowerCamelCase					)
							a		:str											= kwargs.pop('''proxies'''      ,						_lowerCamelCase					)
							a		:str											= kwargs.pop('''local_files_only'''      ,						_lowerCamelCase					)
							a		:Optional[Any]											= kwargs.pop('''use_auth_token'''      ,						_lowerCamelCase					)
							a		:Dict											= kwargs.pop('''revision'''      ,						_lowerCamelCase					)
							a		:Dict											= kwargs.pop('''torch_dtype'''      ,						_lowerCamelCase					)
							a		:Optional[Any]											= kwargs.pop('''device_map'''      ,						_lowerCamelCase					)

							a		:int											= kwargs.pop('''alpha'''      ,						0.5					)
							a		:Dict											= kwargs.pop('''interp'''      ,						_lowerCamelCase					)

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

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

							a		:Optional[int]											= 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.
							a		:Optional[Any]											= []
							for pretrained_model_name_or_path, config_dict in zip(_lowerCamelCase      ,						_lowerCamelCase					):
													a		:List[Any]											= [k for k in config_dict.keys() if not k.startswith('''_'''					)]
													a		:List[str]											= [os.path.join(_lowerCamelCase      ,						'''*'''					) 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		:Optional[int]											= {'''diffusers''': __version__, '''pipeline_class''': requested_pipeline_class}

													a		:Optional[int]											= (
													    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
							a		:Tuple											= DiffusionPipeline.from_pretrained(
							    cached_folders[0]      ,						torch_dtype=_lowerCamelCase      ,						device_map=_lowerCamelCase					)
							final_pipe.to(self.device					)

							a		:List[Any]											= None
							if len(_lowerCamelCase					) > 2:
													a		:str											= os.path.join(cached_folders[2]					)

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

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

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




	@staticmethod
	def  SCREAMING_SNAKE_CASE__       (	_lowerCamelCase      ,						_lowerCamelCase      ,						_lowerCamelCase      ,						_lowerCamelCase					):
							a		:Union[str, Any]											= alpha * alpha * (3 - (2 * alpha))
							return thetaa + ((thetaa - thetaa) * alpha)




	@staticmethod
	def  SCREAMING_SNAKE_CASE__       (	_lowerCamelCase      ,						_lowerCamelCase      ,						_lowerCamelCase      ,						_lowerCamelCase					):
							import math

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




	@staticmethod
	def  SCREAMING_SNAKE_CASE__       (	_lowerCamelCase      ,						_lowerCamelCase      ,						_lowerCamelCase      ,						_lowerCamelCase					):
							return thetaa + (thetaa - thetaa) * (1.0 - alpha)









from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_torch_available,
)


snake_case       :					Dict        =  {
    '''configuration_align''': [
        '''ALIGN_PRETRAINED_CONFIG_ARCHIVE_MAP''',
        '''AlignConfig''',
        '''AlignTextConfig''',
        '''AlignVisionConfig''',
    ],
    '''processing_align''': ['''AlignProcessor'''],
}

try:
     if not is_torch_available():
          raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
     pass
else:
     snake_case       :					List[str]        =  [
         '''ALIGN_PRETRAINED_MODEL_ARCHIVE_LIST''',
         '''AlignModel''',
         '''AlignPreTrainedModel''',
         '''AlignTextModel''',
         '''AlignVisionModel''',
     ]

if TYPE_CHECKING:
     from .configuration_align import (
         ALIGN_PRETRAINED_CONFIG_ARCHIVE_MAP,
         AlignConfig,
         AlignTextConfig,
         AlignVisionConfig,
     )
     from .processing_align import AlignProcessor

     try:
          if not is_torch_available():
               raise OptionalDependencyNotAvailable()
     except OptionalDependencyNotAvailable:
          pass
     else:
          from .modeling_align import (
              ALIGN_PRETRAINED_MODEL_ARCHIVE_LIST,
              AlignModel,
              AlignPreTrainedModel,
              AlignTextModel,
              AlignVisionModel,
          )

else:
     import sys

     snake_case       :					int        =  _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)





