




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

      """simple docstring"""

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







      def 					snake_case		(   self    :       Union[str, Any]       ,				snake_case    :       int       ,				snake_case    :       Optional[int] ):
          if dicta == dicta:
              return True
          else:
              __UpperCamelCase     						=  self._remove_meta_keys(UpperCAmelCase_ )
              __UpperCamelCase     						=  self._remove_meta_keys(UpperCAmelCase_ )
              if configa == configa:
                  print(F"Warning !: Mismatch in keys {meta_keysa} and {meta_keysa}." )
                  return True
          return False







      def 					snake_case		(   self    :       int       ,				snake_case    :       List[str] ):
          __UpperCamelCase     						=  []
          __UpperCamelCase     						=  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 					snake_case		(   self    :       List[str]       ,				snake_case    :       str       ,				**snake_case    :       List[Any] ):
          __UpperCamelCase     						=  kwargs.pop('''cache_dir'''       ,				UpperCAmelCase_ )
          __UpperCamelCase     						=  kwargs.pop('''resume_download'''       ,				UpperCAmelCase_ )
          __UpperCamelCase     						=  kwargs.pop('''force_download'''       ,				UpperCAmelCase_ )
          __UpperCamelCase     						=  kwargs.pop('''proxies'''       ,				UpperCAmelCase_ )
          __UpperCamelCase     						=  kwargs.pop('''local_files_only'''       ,				UpperCAmelCase_ )
          __UpperCamelCase     						=  kwargs.pop('''use_auth_token'''       ,				UpperCAmelCase_ )
          __UpperCamelCase     						=  kwargs.pop('''revision'''       ,				UpperCAmelCase_ )
          __UpperCamelCase     						=  kwargs.pop('''torch_dtype'''       ,				UpperCAmelCase_ )
          __UpperCamelCase     						=  kwargs.pop('''device_map'''       ,				UpperCAmelCase_ )

          __UpperCamelCase     						=  kwargs.pop('''alpha'''       ,				0.5 )
          __UpperCamelCase     						=  kwargs.pop('''interp'''       ,				UpperCAmelCase_ )

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

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

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

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

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

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

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

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







      @staticmethod
      def 					snake_case		(   snake_case    :       Tuple       ,				snake_case    :       Optional[Any]       ,				snake_case    :       Dict       ,				snake_case    :       Optional[Any] ):
          __UpperCamelCase     						=  alpha * alpha * (3 - (2 * alpha))
          return thetaa + ((thetaa - thetaa) * alpha)







      @staticmethod
      def 					snake_case		(   snake_case    :       List[Any]       ,				snake_case    :       Optional[Any]       ,				snake_case    :       int       ,				snake_case    :       Tuple ):
          import math

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







      @staticmethod
      def 					snake_case		(   snake_case    :       str       ,				snake_case    :       Tuple       ,				snake_case    :       Tuple       ,				snake_case    :       Any ):
          return thetaa + (thetaa - thetaa) * (1.0 - alpha)











"""simple docstring"""







import numpy as np
import torch
import torch.nn as nn
from transformers import CLIPConfig, CLIPVisionModel, PreTrainedModel

from ...utils import logging


UpperCAmelCase:   Dict       			=							logging.get_logger(__name__)


def       __SCREAMING_SNAKE_CASE      (   __UpperCAmelCase					,							__UpperCAmelCase	):
		_lowercase							:      List[str]			=					nn.functional.normalize(__UpperCAmelCase	)
		_lowercase							:      int			=					nn.functional.normalize(__UpperCAmelCase	)
		return torch.mm(__UpperCAmelCase					,							normalized_text_embeds.t()	)





class    UpperCamelCase     (			snake_case    ):




	"""simple docstring"""
	SCREAMING_SNAKE_CASE_				:    List[str]		    =   CLIPConfig

	SCREAMING_SNAKE_CASE_				:    List[Any]		    =   ["CLIPEncoderLayer"]






	def __init__(			self							,UpperCAmelCase_					):
			super().__init__(UpperCAmelCase_					)

			_lowercase							:      Any			=					CLIPVisionModel(config.vision_config					)
			_lowercase							:      Any			=					nn.Linear(config.vision_config.hidden_size							,config.projection_dim							,bias=UpperCAmelCase_					)

			_lowercase							:      Dict			=					nn.Parameter(torch.ones(17							,config.projection_dim					)							,requires_grad=UpperCAmelCase_					)
			_lowercase							:      Tuple			=					nn.Parameter(torch.ones(3							,config.projection_dim					)							,requires_grad=UpperCAmelCase_					)

			_lowercase							:      Optional[int]			=					nn.Parameter(torch.ones(17					)							,requires_grad=UpperCAmelCase_					)
			_lowercase							:      List[Any]			=					nn.Parameter(torch.ones(3					)							,requires_grad=UpperCAmelCase_					)






	@torch.no_grad()
	def 					lowerCamelCase__ (			self							,UpperCAmelCase_							,UpperCAmelCase_					):
			_lowercase							:      List[str]			=					self.vision_model(UpperCAmelCase_					)[1]  # pooled_output
			_lowercase							:      str			=					self.visual_projection(UpperCAmelCase_					)

			# we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
			_lowercase							:      Union[str, Any]			=					cosine_distance(UpperCAmelCase_							,self.special_care_embeds					).cpu().float().numpy()
			_lowercase							:      str			=					cosine_distance(UpperCAmelCase_							,self.concept_embeds					).cpu().float().numpy()

			_lowercase							:      List[str]			=					[]
			_lowercase							:      Optional[Any]			=					image_embeds.shape[0]
			for i in range(UpperCAmelCase_					):
					_lowercase							:      int			=					{"""special_scores""": {}, """special_care""": [], """concept_scores""": {}, """bad_concepts""": []}

					# increase this value to create a stronger `nfsw` filter
					# at the cost of increasing the possibility of filtering benign images
					_lowercase							:      str			=					0.0

					for concept_idx in range(len(special_cos_dist[0]					)					):
							_lowercase							:      Optional[Any]			=					special_cos_dist[i][concept_idx]
							_lowercase							:      str			=					self.special_care_embeds_weights[concept_idx].item()
							_lowercase							:      Tuple			=					round(concept_cos - concept_threshold + adjustment							,3					)
							if result_img["special_scores"][concept_idx] > 0:
									result_img["special_care"].append({concept_idx, result_img["""special_scores"""][concept_idx]}					)
									_lowercase							:      Union[str, Any]			=					0.01

					for concept_idx in range(len(cos_dist[0]					)					):
							_lowercase							:      Union[str, Any]			=					cos_dist[i][concept_idx]
							_lowercase							:      str			=					self.concept_embeds_weights[concept_idx].item()
							_lowercase							:      str			=					round(concept_cos - concept_threshold + adjustment							,3					)
							if result_img["concept_scores"][concept_idx] > 0:
									result_img["bad_concepts"].append(UpperCAmelCase_					)

					result.append(UpperCAmelCase_					)

			_lowercase							:      List[str]			=					[len(res["""bad_concepts"""]					) > 0 for res in result]

			for idx, has_nsfw_concept in enumerate(UpperCAmelCase_					):
					if has_nsfw_concept:
							if torch.is_tensor(UpperCAmelCase_					) or torch.is_tensor(images[0]					):
									_lowercase							:      List[Any]			=					torch.zeros_like(images[idx]					)  # black image
							else:
									_lowercase							:      Any			=					np.zeros(images[idx].shape					)  # black image

			if any(UpperCAmelCase_					):
					logger.warning(
					    """Potential NSFW content was detected in one or more images. A black image will be returned instead."""
					    """ Try again with a different prompt and/or seed."""					)

			return images, has_nsfw_concepts






	@torch.no_grad()
	def 					lowerCamelCase__ (			self							,UpperCAmelCase_							,UpperCAmelCase_					):
			_lowercase							:      List[Any]			=					self.vision_model(UpperCAmelCase_					)[1]  # pooled_output
			_lowercase							:      Optional[int]			=					self.visual_projection(UpperCAmelCase_					)

			_lowercase							:      Union[str, Any]			=					cosine_distance(UpperCAmelCase_							,self.special_care_embeds					)
			_lowercase							:      Tuple			=					cosine_distance(UpperCAmelCase_							,self.concept_embeds					)

			# increase this value to create a stronger `nsfw` filter
			# at the cost of increasing the possibility of filtering benign images
			_lowercase							:      Any			=					0.0

			_lowercase							:      Tuple			=					special_cos_dist - self.special_care_embeds_weights + adjustment
			# special_scores = special_scores.round(decimals=3)
			_lowercase							:      Dict			=					torch.any(special_scores > 0							,dim=1					)
			_lowercase							:      Any			=					special_care * 0.01
			_lowercase							:      Optional[int]			=					special_adjustment.unsqueeze(1					).expand(-1							,cos_dist.shape[1]					)

			_lowercase							:      Optional[Any]			=					(cos_dist - self.concept_embeds_weights) + special_adjustment
			# concept_scores = concept_scores.round(decimals=3)
			_lowercase							:      Optional[int]			=					torch.any(concept_scores > 0							,dim=1					)

			_lowercase							:      Tuple			=					0.0  # black image

			return images, has_nsfw_concepts



