




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





     """simple docstring"""



     def __init__(   self			:							Dict     )      ->      Tuple:






         """simple docstring"""
         self.register_to_config()
         super().__init__()

     def 					__SCREAMING_SNAKE_CASE		(   self			:							Any    ,					A			:							Any    ,					A			:							str     )      ->      int:






         """simple docstring"""
         if dicta == dicta:
             return True
         else:
             a__			,		a__  			=							self._remove_meta_keys(A     )
             a__			,		a__  			=							self._remove_meta_keys(A     )
             if configa == configa:
                 print(f"""Warning !: Mismatch in keys {meta_keysa} and {meta_keysa}."""     )
                 return True
         return False

     def 					__SCREAMING_SNAKE_CASE		(   self			:							Dict    ,					A			:							Dict     )      ->      List[str]:






         """simple docstring"""
         a__  			=							[]
         a__  			=							config_dict.copy()
         for key in config_dict.keys():
             if key.startswith("""_"""     ):
                 temp_dict.pop(A     )
                 meta_keys.append(A     )
         return (temp_dict, meta_keys)

     @torch.no_grad()
     def 					__SCREAMING_SNAKE_CASE		(   self			:							int    ,					A			:							List[Union[str, os.PathLike]]    ,					**A			:							Optional[Any]     )      ->      Tuple:






         """simple docstring"""
         a__  			=							kwargs.pop("""cache_dir"""    ,					A     )
         a__  			=							kwargs.pop("""resume_download"""    ,					A     )
         a__  			=							kwargs.pop("""force_download"""    ,					A     )
         a__  			=							kwargs.pop("""proxies"""    ,					A     )
         a__  			=							kwargs.pop("""local_files_only"""    ,					A     )
         a__  			=							kwargs.pop("""use_auth_token"""    ,					A     )
         a__  			=							kwargs.pop("""revision"""    ,					A     )
         a__  			=							kwargs.pop("""torch_dtype"""    ,					A     )
         a__  			=							kwargs.pop("""device_map"""    ,					A     )

         a__  			=							kwargs.pop("""alpha"""    ,					0.5     )
         a__  			=							kwargs.pop("""interp"""    ,					A     )

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

         a__  			=							len(A     )
         # Ignore result from model_index_json comparision of the two checkpoints
         a__  			=							kwargs.pop("""force"""    ,					A     )

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

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

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

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

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

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

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

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

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

                 del theta_a
                 del theta_a
                 update_theta_a(A     )

                 del theta_a
         return final_pipe

     @staticmethod
     def 					__SCREAMING_SNAKE_CASE		(   A			:							Optional[int]    ,					A			:							int    ,					A			:							Optional[Any]    ,					A			:							Union[str, Any]     )      ->      str:






         """simple docstring"""
         return ((1 - alpha) * thetaa) + (alpha * thetaa)

     @staticmethod
     def 					__SCREAMING_SNAKE_CASE		(   A			:							Dict    ,					A			:							List[Any]    ,					A			:							Tuple    ,					A			:							Dict     )      ->      Tuple:






         """simple docstring"""
         a__  			=							alpha * alpha * (3 - (2 * alpha))
         return thetaa + ((thetaa - thetaa) * alpha)

     @staticmethod
     def 					__SCREAMING_SNAKE_CASE		(   A			:							int    ,					A			:							Tuple    ,					A			:							List[Any]    ,					A			:							List[Any]     )      ->      str:






         """simple docstring"""
         import math

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






     @staticmethod
     def 					__SCREAMING_SNAKE_CASE		(   A			:							int    ,					A			:							Any    ,					A			:							Optional[int]    ,					A			:							int     )      ->      str:






         """simple docstring"""
         return thetaa + (thetaa - thetaa) * (1.0 - alpha)












from __future__ import annotations

from typing import Generic, TypeVar

SCREAMING_SNAKE_CASE_    :       List[str]       =							TypeVar('''T''')







class 		__UpperCAmelCase(		lowercase__			):





     """simple docstring"""



     pass







class 		__UpperCAmelCase(		lowercase__			):





     """simple docstring"""



     pass







class 		__UpperCAmelCase(		Generic[T]			):





     """simple docstring"""



     def __init__(   self			:							str    ,					A			:							int = 10     )      ->      str:






         """simple docstring"""
         a__  			=							[]
         a__  			=							limit

     def __bool__(   self			:							Dict     )      ->      bool:






         """simple docstring"""
         return bool(self.stack     )

     def __str__(   self			:							str     )      ->      str:






         """simple docstring"""
         return str(self.stack     )

     def 					__SCREAMING_SNAKE_CASE		(   self			:							List[str]    ,					A			:							T     )      ->      None:






         """simple docstring"""
         if len(self.stack     ) >= self.limit:
             raise StackOverflowError
         self.stack.append(A     )

     def 					__SCREAMING_SNAKE_CASE		(   self			:							Any     )      ->      T:






         """simple docstring"""
         if not self.stack:
             raise StackUnderflowError
         return self.stack.pop()

     def 					__SCREAMING_SNAKE_CASE		(   self			:							Optional[int]     )      ->      T:






         """simple docstring"""
         if not self.stack:
             raise StackUnderflowError
         return self.stack[-1]

     def 					__SCREAMING_SNAKE_CASE		(   self			:							str     )      ->      bool:






         """simple docstring"""
         return not bool(self.stack     )

     def 					__SCREAMING_SNAKE_CASE		(   self			:							Dict     )      ->      bool:






         """simple docstring"""
         return self.size() == self.limit

     def 					__SCREAMING_SNAKE_CASE		(   self			:							int     )      ->      int:






         """simple docstring"""
         return len(self.stack     )






     def __contains__(   self			:							int    ,					A			:							T     )      ->      bool:






         """simple docstring"""
         return item in self.stack






def 							__lowerCAmelCase			():
    a__  			=							Stack(10		)
    assert bool(UpperCAmelCase_		) is False
    assert stack.is_empty() is True
    assert stack.is_full() is False
    assert str(UpperCAmelCase_		) == "[]"

    try:
        a__  			=							stack.pop()
        raise AssertionError  # This should not happen
    except StackUnderflowError:
        assert True  # This should happen

    try:
        a__  			=							stack.peek()
        raise AssertionError  # This should not happen
    except StackUnderflowError:
        assert True  # This should happen

    for i in range(10		):
        assert stack.size() == i
        stack.push(UpperCAmelCase_		)

    assert bool(UpperCAmelCase_		)
    assert not stack.is_empty()
    assert stack.is_full()
    assert str(UpperCAmelCase_		) == str(list(range(10		)		)		)
    assert stack.pop() == 9
    assert stack.peek() == 8

    stack.push(100		)
    assert str(UpperCAmelCase_		) == str([0, 1, 2, 3, 4, 5, 6, 7, 8, 100]		)

    try:
        stack.push(200		)
        raise AssertionError  # This should not happen
    except StackOverflowError:
        assert True  # This should happen

    assert not stack.is_empty()
    assert stack.size() == 10

    assert 5 in stack
    assert 55 not in stack


if __name__ == "__main__":
 test_stack()






