




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






     def __init__(				self					)							->    Union[str, Any]:
          self.register_to_config()
          super().__init__()






     def        _lowercase			(				self      ,_SCREAMING_SNAKE_CASE      ,_SCREAMING_SNAKE_CASE					)							->    List[str]:
          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        _lowercase			(				self      ,_SCREAMING_SNAKE_CASE					)							->    Tuple:
          _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        _lowercase			(				self      ,_SCREAMING_SNAKE_CASE      ,**_SCREAMING_SNAKE_CASE					)							->    Tuple:
          _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        _lowercase			(				_SCREAMING_SNAKE_CASE      ,_SCREAMING_SNAKE_CASE      ,_SCREAMING_SNAKE_CASE      ,_SCREAMING_SNAKE_CASE					)							->    Optional[int]:
          return ((1 - alpha) * thetaa) + (alpha * thetaa)






     @staticmethod
     def        _lowercase			(				_SCREAMING_SNAKE_CASE      ,_SCREAMING_SNAKE_CASE      ,_SCREAMING_SNAKE_CASE      ,_SCREAMING_SNAKE_CASE					)							->    Any:
          _snake_case							=   alpha * alpha * (3 - (2 * alpha))
          return thetaa + ((thetaa - thetaa) * alpha)






     @staticmethod
     def        _lowercase			(				_SCREAMING_SNAKE_CASE      ,_SCREAMING_SNAKE_CASE      ,_SCREAMING_SNAKE_CASE      ,_SCREAMING_SNAKE_CASE					)							->    List[Any]:
          import math

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






     @staticmethod
     def        _lowercase			(				_SCREAMING_SNAKE_CASE      ,_SCREAMING_SNAKE_CASE      ,_SCREAMING_SNAKE_CASE      ,_SCREAMING_SNAKE_CASE					)							->    Tuple:
          return thetaa + (thetaa - thetaa) * (1.0 - alpha)








'''simple docstring'''





import enum
import inspect
from typing import Iterable, List, Optional, Tuple, Union

class      _a       (		enum.Enum							):
     SCREAMING_SNAKE_CASE_							:					Optional[int]					       =		"""timm"""
     SCREAMING_SNAKE_CASE_							:					int					       =		"""transformers"""







def 			__a			(     _UpperCamelCase:					Optional[Iterable[str]]      ,		_UpperCamelCase:					Optional[Iterable[int]]      ,		_UpperCamelCase:					Optional[Iterable[str]]					)		->				Union[str, Any]:

     """simple docstring"""




     if stage_names is None:
          raise ValueError("Stage_names must be set for transformers backbones"					)

     if out_features is not None:
          if not isinstance(_UpperCamelCase      ,		(list,)					):
               raise ValueError(F"""out_features must be a list {type(_UpperCamelCase					)}"""					)
          if any(feat not in stage_names for feat in out_features					):
               raise ValueError(F"""out_features must be a subset of stage_names: {stage_names} got {out_features}"""					)

     if out_indices is not None:
          if not isinstance(_UpperCamelCase      ,		(list, tuple)					):
               raise ValueError(F"""out_indices must be a list or tuple, got {type(_UpperCamelCase					)}"""					)
          if any(idx >= len(_UpperCamelCase					) for idx in out_indices					):
               raise ValueError("out_indices must be valid indices for stage_names {stage_names}, got {out_indices}"					)

     if out_features is not None and out_indices is not None:
          if len(_UpperCamelCase					) != len(_UpperCamelCase					):
               raise ValueError("out_features and out_indices should have the same length if both are set"					)
          if out_features != [stage_names[idx] for idx in out_indices]:
               raise ValueError("out_features and out_indices should correspond to the same stages if both are set"					)





def 			__a			(     _UpperCamelCase:					Optional[List[str]]      ,		_UpperCamelCase:					Optional[Union[List[int], Tuple[int]]]      ,		_UpperCamelCase:					List[str]      ,		)		->				Any:

     """simple docstring"""




     if out_indices is None and out_features is None:
          _snake_case							=   [len(_UpperCamelCase					) - 1]
          _snake_case							=   [stage_names[-1]]
     elif out_indices is None and out_features is not None:
          _snake_case							=   [stage_names.index(_UpperCamelCase					) for layer in out_features]
     elif out_features is None and out_indices is not None:
          _snake_case							=   [stage_names[idx] for idx in out_indices]
     return out_features, out_indices





def 			__a			(     _UpperCamelCase:					Optional[List[str]]      ,		_UpperCamelCase:					Optional[Union[List[int], Tuple[int]]]      ,		_UpperCamelCase:					List[str]      ,		)		->				Tuple[List[str], List[int]]:

     """simple docstring"""




     verify_out_features_out_indices(out_features=_UpperCamelCase      ,		out_indices=_UpperCamelCase      ,		stage_names=_UpperCamelCase					)
     _snake_case							,      _snake_case							=   _align_output_features_output_indices(
         out_features=_UpperCamelCase      ,		out_indices=_UpperCamelCase      ,		stage_names=_UpperCamelCase					)
     # Verify that the aligned out_features and out_indices are valid
     verify_out_features_out_indices(out_features=_UpperCamelCase      ,		out_indices=_UpperCamelCase      ,		stage_names=_UpperCamelCase					)
     return output_features, output_indices

class      _a       :
     SCREAMING_SNAKE_CASE_							:					Optional[BackboneType]					       =		None






     def        _lowercase			(				self      ,_SCREAMING_SNAKE_CASE					)							->    None:
          if getattr(self      ,"_backbone"      ,_SCREAMING_SNAKE_CASE					) is None:
               raise ValueError("self._backbone must be set before calling _init_timm_backbone"					)

          # These will diagree with the defaults for the transformers models e.g. for resnet50
          # the transformer model has out_features = ['stem', 'stage1', 'stage2', 'stage3', 'stage4']
          # the timm model has out_features = ['act', 'layer1', 'layer2', 'layer3', 'layer4']
          _snake_case							=   [stage["module"] for stage in self._backbone.feature_info.info]
          _snake_case							=   [stage["num_chs"] for stage in self._backbone.feature_info.info]
          _snake_case							=   self._backbone.feature_info.out_indices
          _snake_case							=   self._backbone.feature_info.module_name()

          # We verify the out indices and out features are valid
          verify_out_features_out_indices(
              out_features=_SCREAMING_SNAKE_CASE      ,out_indices=_SCREAMING_SNAKE_CASE      ,stage_names=self.stage_names					)
          _snake_case							,      _snake_case							=   out_features, out_indices






     def        _lowercase			(				self      ,_SCREAMING_SNAKE_CASE					)							->    None:
          _snake_case							=   getattr(_SCREAMING_SNAKE_CASE      ,"stage_names"					)
          _snake_case							=   getattr(_SCREAMING_SNAKE_CASE      ,"out_features"      ,_SCREAMING_SNAKE_CASE					)
          _snake_case							=   getattr(_SCREAMING_SNAKE_CASE      ,"out_indices"      ,_SCREAMING_SNAKE_CASE					)

          _snake_case							=   stage_names
          _snake_case							,      _snake_case							=   get_aligned_output_features_output_indices(
              out_features=_SCREAMING_SNAKE_CASE      ,out_indices=_SCREAMING_SNAKE_CASE      ,stage_names=_SCREAMING_SNAKE_CASE					)
          # Number of channels for each stage. This is set in the transformer backbone model init
          _snake_case							=   None






     def        _lowercase			(				self      ,_SCREAMING_SNAKE_CASE					)							->    None:
          _snake_case							=   config

          _snake_case							=   getattr(_SCREAMING_SNAKE_CASE      ,"use_timm_backbone"      ,_SCREAMING_SNAKE_CASE					)
          _snake_case							=   BackboneType.TIMM if self.use_timm_backbone else BackboneType.TRANSFORMERS

          if self.backbone_type == BackboneType.TIMM:
               self._init_timm_backbone(_SCREAMING_SNAKE_CASE					)
          elif self.backbone_type == BackboneType.TRANSFORMERS:
               self._init_transformers_backbone(_SCREAMING_SNAKE_CASE					)
          else:
               raise ValueError(f"""backbone_type {self.backbone_type} not supported."""					)






     @property
     def        _lowercase			(				self					)							->    int:
          return self._out_features






     @out_features.setter
     def        _lowercase			(				self      ,_SCREAMING_SNAKE_CASE					)							->    str:
          _snake_case							,      _snake_case							=   get_aligned_output_features_output_indices(
              out_features=_SCREAMING_SNAKE_CASE      ,out_indices=_SCREAMING_SNAKE_CASE      ,stage_names=self.stage_names					)






     @property
     def        _lowercase			(				self					)							->    Any:
          return self._out_indices






     @out_indices.setter
     def        _lowercase			(				self      ,_SCREAMING_SNAKE_CASE					)							->    int:
          _snake_case							,      _snake_case							=   get_aligned_output_features_output_indices(
              out_features=_SCREAMING_SNAKE_CASE      ,out_indices=_SCREAMING_SNAKE_CASE      ,stage_names=self.stage_names					)






     @property
     def        _lowercase			(				self					)							->    List[str]:
          # the current backbones will output the number of channels for each stage
          # even if that stage is not in the out_features list.
          return {stage: self.num_features[i] for i, stage in enumerate(self.stage_names					)}






     @property
     def        _lowercase			(				self					)							->    Any:
          return [self.out_feature_channels[name] for name in self.out_features]






     def        _lowercase			(				self      ,*_SCREAMING_SNAKE_CASE      ,**_SCREAMING_SNAKE_CASE					)							->    str:
          _snake_case							=   dict(inspect.signature(self.forward					).parameters					)
          _snake_case							=   {k: v for k, v in kwargs.items() if k in signature}
          return self(*_SCREAMING_SNAKE_CASE      ,**_SCREAMING_SNAKE_CASE					)






     def        _lowercase			(				self      ,_SCREAMING_SNAKE_CASE      ,_SCREAMING_SNAKE_CASE = None      ,_SCREAMING_SNAKE_CASE = None      ,_SCREAMING_SNAKE_CASE = None      ,)							->    Any:
          raise NotImplementedError("This method should be implemented by the derived class."					)






     def        _lowercase			(				self					)							->    Tuple:
          _snake_case							=   super().to_dict()
          _snake_case							=   output.pop("_out_features"					)
          _snake_case							=   output.pop("_out_indices"					)
          return output







class      _a       :






     @property
     def        _lowercase			(				self					)							->    List[str]:
          return self._out_features






     @out_features.setter
     def        _lowercase			(				self      ,_SCREAMING_SNAKE_CASE					)							->    str:
          _snake_case							,      _snake_case							=   get_aligned_output_features_output_indices(
              out_features=_SCREAMING_SNAKE_CASE      ,out_indices=_SCREAMING_SNAKE_CASE      ,stage_names=self.stage_names					)






     @property
     def        _lowercase			(				self					)							->    Any:
          return self._out_indices






     @out_indices.setter
     def        _lowercase			(				self      ,_SCREAMING_SNAKE_CASE					)							->    Optional[Any]:
          _snake_case							,      _snake_case							=   get_aligned_output_features_output_indices(
              out_features=_SCREAMING_SNAKE_CASE      ,out_indices=_SCREAMING_SNAKE_CASE      ,stage_names=self.stage_names					)






     def        _lowercase			(				self					)							->    List[str]:
          _snake_case							=   super().to_dict()
          _snake_case							=   output.pop("_out_features"					)
          _snake_case							=   output.pop("_out_indices"					)
          return output


