'''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     _lowercase					(    A__      ):


 '''simple docstring'''






 def __init__(				self       :     Optional[int]							)  ->   Optional[int]:
  self.register_to_config()
  super().__init__()
 def a (				self       :     Optional[int]				,	SCREAMING_SNAKE_CASE__       :     Any				,	SCREAMING_SNAKE_CASE__       :     str							)  ->   Optional[Any]:
  if dicta == dicta:
   return True
  else:
   __lowerCAmelCase							,	__lowerCAmelCase    							=						self._remove_meta_keys(SCREAMING_SNAKE_CASE__							)
   __lowerCAmelCase							,	__lowerCAmelCase    							=						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 a (				self       :     List[str]				,	SCREAMING_SNAKE_CASE__       :     Dict							)  ->   Optional[int]:
  __lowerCAmelCase    							=						[]
  __lowerCAmelCase    							=						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 a (				self       :     int				,	SCREAMING_SNAKE_CASE__       :     List[Union[str, os.PathLike]]				,	**SCREAMING_SNAKE_CASE__       :     str							)  ->   Optional[int]:
  __lowerCAmelCase    							=						kwargs.pop("""cache_dir"""				,	SCREAMING_SNAKE_CASE__							)
  __lowerCAmelCase    							=						kwargs.pop("""resume_download"""				,	SCREAMING_SNAKE_CASE__							)
  __lowerCAmelCase    							=						kwargs.pop("""force_download"""				,	SCREAMING_SNAKE_CASE__							)
  __lowerCAmelCase    							=						kwargs.pop("""proxies"""				,	SCREAMING_SNAKE_CASE__							)
  __lowerCAmelCase    							=						kwargs.pop("""local_files_only"""				,	SCREAMING_SNAKE_CASE__							)
  __lowerCAmelCase    							=						kwargs.pop("""use_auth_token"""				,	SCREAMING_SNAKE_CASE__							)
  __lowerCAmelCase    							=						kwargs.pop("""revision"""				,	SCREAMING_SNAKE_CASE__							)
  __lowerCAmelCase    							=						kwargs.pop("""torch_dtype"""				,	SCREAMING_SNAKE_CASE__							)
  __lowerCAmelCase    							=						kwargs.pop("""device_map"""				,	SCREAMING_SNAKE_CASE__							)

  __lowerCAmelCase    							=						kwargs.pop("""alpha"""				,	0.5							)
  __lowerCAmelCase    							=						kwargs.pop("""interp"""				,	SCREAMING_SNAKE_CASE__							)

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

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

  __lowerCAmelCase    							=						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.
  __lowerCAmelCase    							=						[]
  for pretrained_model_name_or_path, config_dict in zip(SCREAMING_SNAKE_CASE__				,	SCREAMING_SNAKE_CASE__							):
   __lowerCAmelCase    							=						[k for k in config_dict.keys() if not k.startswith("""_"""							)]
   __lowerCAmelCase    							=						[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,
   ]
   __lowerCAmelCase    							=						config_dict.get("""_class_name"""							)
   __lowerCAmelCase    							=						{"""diffusers""": __version__, """pipeline_class""": requested_pipeline_class}

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

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

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

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

     __lowerCAmelCase    							=						getattr(SCREAMING_SNAKE_CASE__				,	"""load_state_dict"""							)
     __lowerCAmelCase    							=						(
         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"""							)
     )
     __lowerCAmelCase    							=						None
     if checkpoint_path_a:
      __lowerCAmelCase    							=						(
          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:
      __lowerCAmelCase    							=						theta_func(theta_a[key]				,	theta_a[key]				,	theta_a[key]				,	SCREAMING_SNAKE_CASE__							)
     else:
      __lowerCAmelCase    							=						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 a (				SCREAMING_SNAKE_CASE__       :     List[Any]				,	SCREAMING_SNAKE_CASE__       :     Union[str, Any]				,	SCREAMING_SNAKE_CASE__       :     Optional[Any]				,	SCREAMING_SNAKE_CASE__       :     List[str]							)  ->   str:
  return ((1 - alpha) * thetaa) + (alpha * thetaa)
 @staticmethod
 def a (				SCREAMING_SNAKE_CASE__       :     int				,	SCREAMING_SNAKE_CASE__       :     Any				,	SCREAMING_SNAKE_CASE__       :     str				,	SCREAMING_SNAKE_CASE__       :     Tuple							)  ->   Optional[Any]:
  __lowerCAmelCase    							=						alpha * alpha * (3 - (2 * alpha))
  return thetaa + ((thetaa - thetaa) * alpha)
 @staticmethod
 def a (				SCREAMING_SNAKE_CASE__       :     List[str]				,	SCREAMING_SNAKE_CASE__       :     str				,	SCREAMING_SNAKE_CASE__       :     Optional[int]				,	SCREAMING_SNAKE_CASE__       :     str							)  ->   Any:
  import math

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





 @staticmethod
 def a (				SCREAMING_SNAKE_CASE__       :     Tuple				,	SCREAMING_SNAKE_CASE__       :     int				,	SCREAMING_SNAKE_CASE__       :     Optional[int]				,	SCREAMING_SNAKE_CASE__       :     int							)  ->   List[str]:
  return thetaa + (thetaa - thetaa) * (1.0 - alpha)







'''simple docstring'''

def 					UpperCamelCase_						( snake_case_	:	int				) ->					int:






 '''simple docstring'''

 if number != int(snake_case_				):
  raise ValueError("""factorial() only accepts integral values"""				)
 if number < 0:
  raise ValueError("""factorial() not defined for negative values"""				)
 __lowerCAmelCase    							=						1
 for i in range(1						,		number + 1				):
  value *= i
 return value







def 					UpperCamelCase_						( snake_case_	:	int				) ->					int:






 '''simple docstring'''

 if not isinstance(snake_case_						,		snake_case_				):
  raise ValueError("""factorial() only accepts integral values"""				)
 if n < 0:
  raise ValueError("""factorial() not defined for negative values"""				)
 return 1 if n == 0 or n == 1 else n * factorial(n - 1				)


if __name__ == "__main__":
 import doctest

 doctest.testmod()

 _A      :				List[Any] 				=    int(input('''Enter a positive integer: ''').strip() or 0)
 print(f'factorial{n} is {factorial(n)}')






