

'''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 							_snake_case	(    a_  ):
   def __init__(	self     : Optional[Any]							):
     self.register_to_config()
     super().__init__()
   def       __UpperCamelCase  (	self     : List[Any]       ,SCREAMING_SNAKE_CASE__     : Dict       ,SCREAMING_SNAKE_CASE__     : str							):


     if dicta == dicta:
       return True
     else:
       SCREAMING_SNAKE_CASE ,      SCREAMING_SNAKE_CASE:int   =   self._remove_meta_keys(snake_case__							)
       SCREAMING_SNAKE_CASE ,      SCREAMING_SNAKE_CASE:Dict   =   self._remove_meta_keys(snake_case__							)
       if configa == configa:
         print(F'''Warning !: Mismatch in keys {meta_keysa} and {meta_keysa}.'''							)
         return True
     return False
   def       __UpperCamelCase  (	self     : Dict       ,SCREAMING_SNAKE_CASE__     : Optional[int]							):
     SCREAMING_SNAKE_CASE:str   =   []
     SCREAMING_SNAKE_CASE:Union[str, Any]   =   config_dict.copy()
     for key in config_dict.keys():
       if key.startswith("_"							):
         temp_dict.pop(snake_case__							)
         meta_keys.append(snake_case__							)
     return (temp_dict, meta_keys)
   @torch.no_grad()
   def       __UpperCamelCase  (	self     : Any       ,SCREAMING_SNAKE_CASE__     : List[Any]       ,**SCREAMING_SNAKE_CASE__     : str							):
     SCREAMING_SNAKE_CASE:List[Any]   =   kwargs.pop("cache_dir"       ,snake_case__							)
     SCREAMING_SNAKE_CASE:Optional[int]   =   kwargs.pop("resume_download"       ,snake_case__							)
     SCREAMING_SNAKE_CASE:List[Any]   =   kwargs.pop("force_download"       ,snake_case__							)
     SCREAMING_SNAKE_CASE:int   =   kwargs.pop("proxies"       ,snake_case__							)
     SCREAMING_SNAKE_CASE:Optional[int]   =   kwargs.pop("local_files_only"       ,snake_case__							)
     SCREAMING_SNAKE_CASE:List[Any]   =   kwargs.pop("use_auth_token"       ,snake_case__							)
     SCREAMING_SNAKE_CASE:List[str]   =   kwargs.pop("revision"       ,snake_case__							)
     SCREAMING_SNAKE_CASE:Tuple   =   kwargs.pop("torch_dtype"       ,snake_case__							)
     SCREAMING_SNAKE_CASE:str   =   kwargs.pop("device_map"       ,snake_case__							)

     SCREAMING_SNAKE_CASE:Any   =   kwargs.pop("alpha"       ,0.5							)
     SCREAMING_SNAKE_CASE:str   =   kwargs.pop("interp"       ,snake_case__							)

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

     SCREAMING_SNAKE_CASE:int   =   len(snake_case__							)
     # Ignore result from model_index_json comparision of the two checkpoints
     SCREAMING_SNAKE_CASE:Optional[int]   =   kwargs.pop("force"       ,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 '_'
     SCREAMING_SNAKE_CASE:Optional[int]   =   []
     for pretrained_model_name_or_path in pretrained_model_name_or_path_list:
       SCREAMING_SNAKE_CASE:Dict   =   DiffusionPipeline.load_config(
           snake_case__       ,cache_dir=snake_case__       ,resume_download=snake_case__       ,force_download=snake_case__       ,proxies=snake_case__       ,local_files_only=snake_case__       ,use_auth_token=snake_case__       ,revision=snake_case__       ,)
       config_dicts.append(snake_case__							)

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

       SCREAMING_SNAKE_CASE:Union[str, Any]   =   (
           pretrained_model_name_or_path
           if os.path.isdir(snake_case__							)
           else snapshot_download(
               snake_case__       ,cache_dir=snake_case__       ,resume_download=snake_case__       ,proxies=snake_case__       ,local_files_only=snake_case__       ,use_auth_token=snake_case__       ,revision=snake_case__       ,allow_patterns=snake_case__       ,user_agent=snake_case__       ,)
       )
       print("Cached Folder"       ,snake_case__							)
       cached_folders.append(snake_case__							)

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

     SCREAMING_SNAKE_CASE:Optional[int]   =   None
     if len(snake_case__							) > 2:
       SCREAMING_SNAKE_CASE:Tuple   =   os.path.join(cached_folders[2]							)

     if interp == "sigmoid":
       SCREAMING_SNAKE_CASE:Any   =   CheckpointMergerPipeline.sigmoid
     elif interp == "inv_sigmoid":
       SCREAMING_SNAKE_CASE:str   =   CheckpointMergerPipeline.inv_sigmoid
     elif interp == "add_diff":
       SCREAMING_SNAKE_CASE:Optional[int]   =   CheckpointMergerPipeline.add_difference
     else:
       SCREAMING_SNAKE_CASE:Optional[Any]   =   CheckpointMergerPipeline.weighted_sum

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

           SCREAMING_SNAKE_CASE:Tuple   =   getattr(snake_case__       ,"load_state_dict"							)
           SCREAMING_SNAKE_CASE:str   =   (
               safetensors.torch.load_file(snake_case__							)
               if (is_safetensors_available() and checkpoint_path_a.endswith(".safetensors"							))
               else torch.load(snake_case__       ,map_location="cpu"							)
           )
           SCREAMING_SNAKE_CASE:List[Any]   =   None
           if checkpoint_path_a:
             SCREAMING_SNAKE_CASE:Tuple   =   (
                 safetensors.torch.load_file(snake_case__							)
                 if (is_safetensors_available() and checkpoint_path_a.endswith(".safetensors"							))
                 else torch.load(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(snake_case__							)}'''							)
           continue
         print(F'''MERGING {attr}'''							)

         for key in theta_a.keys():
           if theta_a:
             SCREAMING_SNAKE_CASE:List[str]   =   theta_func(theta_a[key]       ,theta_a[key]       ,theta_a[key]       ,snake_case__							)
           else:
             SCREAMING_SNAKE_CASE:Any   =   theta_func(theta_a[key]       ,theta_a[key]       ,snake_case__       ,snake_case__							)

         del theta_a
         del theta_a
         update_theta_a(snake_case__							)

         del theta_a
     return final_pipe
   @staticmethod
   def       __UpperCamelCase  (	SCREAMING_SNAKE_CASE__     : Any       ,SCREAMING_SNAKE_CASE__     : Optional[Any]       ,SCREAMING_SNAKE_CASE__     : Dict       ,SCREAMING_SNAKE_CASE__     : List[str]							):
     return ((1 - alpha) * thetaa) + (alpha * thetaa)
   @staticmethod
   def       __UpperCamelCase  (	SCREAMING_SNAKE_CASE__     : Optional[Any]       ,SCREAMING_SNAKE_CASE__     : Any       ,SCREAMING_SNAKE_CASE__     : Tuple       ,SCREAMING_SNAKE_CASE__     : str							):
     SCREAMING_SNAKE_CASE:Union[str, Any]   =   alpha * alpha * (3 - (2 * alpha))
     return thetaa + ((thetaa - thetaa) * alpha)
   @staticmethod
   def       __UpperCamelCase  (	SCREAMING_SNAKE_CASE__     : Any       ,SCREAMING_SNAKE_CASE__     : Optional[Any]       ,SCREAMING_SNAKE_CASE__     : int       ,SCREAMING_SNAKE_CASE__     : Optional[Any]							):
     import math

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







   @staticmethod
   def       __UpperCamelCase  (	SCREAMING_SNAKE_CASE__     : str       ,SCREAMING_SNAKE_CASE__     : Any       ,SCREAMING_SNAKE_CASE__     : Any       ,SCREAMING_SNAKE_CASE__     : List[Any]							):
     return thetaa + (thetaa - thetaa) * (1.0 - alpha)









'''simple docstring'''

from __future__ import annotations

from collections import Counter





def 				A_					(	snake_case	):
  SCREAMING_SNAKE_CASE:int   =   [True] * n
  SCREAMING_SNAKE_CASE:List[str]   =   False
  SCREAMING_SNAKE_CASE:Tuple   =   False
  SCREAMING_SNAKE_CASE:int   =   True

  for i in range(3 ,    int(n**0.5 + 1	) ,    2	):
    SCREAMING_SNAKE_CASE:str   =   i * 2
    while index < n:
      SCREAMING_SNAKE_CASE:Any   =   False
      SCREAMING_SNAKE_CASE:List[Any]   =   index + i

  SCREAMING_SNAKE_CASE:Union[str, Any]   =   [2]

  for i in range(3 ,    snake_case ,    2	):
    if is_prime[i]:
      primes.append(snake_case	)

  return primes





def 				A_					(	snake_case	):
  SCREAMING_SNAKE_CASE:List[str]   =   str(snake_case	)
  SCREAMING_SNAKE_CASE:Union[str, Any]   =   []
  SCREAMING_SNAKE_CASE:List[str]   =   ["0", "1", "2", "3", "4", "5", "6", "7", "8", "9"]

  for duplicate in Counter(snake_case	) - Counter(set(snake_case	)	):
    SCREAMING_SNAKE_CASE:List[str]   =   [int(number_str.replace(snake_case ,    snake_case	)	) for digit in digits]
    replacements.append(snake_case	)

  return replacements





def 				A_					(	snake_case = 8	):
  SCREAMING_SNAKE_CASE:str   =   set()

  # Filter primes with less than 3 replaceable digits
  SCREAMING_SNAKE_CASE:Tuple   =   {
      x for x in set(prime_sieve(1000000	)	) if len(str(snake_case	)	) - len(set(str(snake_case	)	)	) >= 3
  }

  for prime in primes:
    if prime in numbers_checked:
      continue

    SCREAMING_SNAKE_CASE:List[Any]   =   digit_replacements(snake_case	)

    for family in replacements:
      numbers_checked.update(snake_case	)
      SCREAMING_SNAKE_CASE:Optional[Any]   =   primes.intersection(snake_case	)

      if len(snake_case	) != family_length:
        continue

      return min(snake_case	)

  return -1


if __name__ == "__main__":
    print(solution())






