
'''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 	UpperCamelCase__					(      SCREAMING_SNAKE_CASE):






 def __init__(     self					:Tuple     )			->     str:







       '''simple docstring'''



       self.register_to_config()
       super().__init__()






 def 		lowercase_  (     self					:str		,			_A					:Union[str, Any]		,			_A					:List[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 		lowercase_  (     self					:Optional[int]		,			_A					:Dict     )			->     Any:







       '''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 		lowercase_  (     self					:Optional[Any]		,			_A					:List[Union[str, os.PathLike]]		,			**_A					:Tuple     )			->     Optional[Any]:







       '''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 		lowercase_  (     _A					:List[str]		,			_A					:List[Any]		,			_A					:Optional[Any]		,			_A					:Dict     )			->     str:







       '''simple docstring'''



       return ((1 - alpha) * thetaa) + (alpha * thetaa)






 @staticmethod
 def 		lowercase_  (     _A					:List[str]		,			_A					:int		,			_A					:int		,			_A					:List[str]     )			->     List[str]:







       '''simple docstring'''



       __A    		=     alpha * alpha * (3 - (2 * alpha))
       return thetaa + ((thetaa - thetaa) * alpha)






 @staticmethod
 def 		lowercase_  (     _A					:str		,			_A					:Union[str, Any]		,			_A					:str		,			_A					:List[str]     )			->     Any:







       '''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 		lowercase_  (     _A					:Tuple		,			_A					:int		,			_A					:int		,			_A					:Tuple     )			->     Optional[Any]:







       '''simple docstring'''



       return thetaa + (thetaa - thetaa) * (1.0 - alpha)








'''simple docstring'''





import collections
import inspect
import unittest

from transformers import DonutSwinConfig
from transformers.testing_utils import require_torch, slow, torch_device
from transformers.utils import is_torch_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, _config_zero_init, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
 import torch
 from torch import nn

 from transformers import DonutSwinModel
 from transformers.models.donut.modeling_donut_swin import DONUT_SWIN_PRETRAINED_MODEL_ARCHIVE_LIST


class 	UpperCamelCase__					:






 def __init__(     self					:int		,			_A					:int		,			_A					:List[Any]=13		,			_A					:Any=32		,			_A					:List[str]=2		,			_A					:str=3		,			_A					:Optional[Any]=16		,			_A					:List[Any]=[1, 2, 1]		,			_A					:int=[2, 2, 4]		,			_A					:Optional[Any]=2		,			_A					:Optional[Any]=2.0		,			_A					:int=True		,			_A					:Any=0.0		,			_A					:Tuple=0.0		,			_A					:Union[str, Any]=0.1		,			_A					:str="gelu"		,			_A					:Tuple=False		,			_A					:List[str]=True		,			_A					:Union[str, Any]=0.02		,			_A					:List[str]=1E-5		,			_A					:List[Any]=True		,			_A					:int=None		,			_A					:Union[str, Any]=True		,			_A					:int=10		,			_A					:Optional[int]=8		,			)			->     Optional[int]:







       '''simple docstring'''



       __A    		=     parent
       __A    		=     batch_size
       __A    		=     image_size
       __A    		=     patch_size
       __A    		=     num_channels
       __A    		=     embed_dim
       __A    		=     depths
       __A    		=     num_heads
       __A    		=     window_size
       __A    		=     mlp_ratio
       __A    		=     qkv_bias
       __A    		=     hidden_dropout_prob
       __A    		=     attention_probs_dropout_prob
       __A    		=     drop_path_rate
       __A    		=     hidden_act
       __A    		=     use_absolute_embeddings
       __A    		=     patch_norm
       __A    		=     layer_norm_eps
       __A    		=     initializer_range
       __A    		=     is_training
       __A    		=     scope
       __A    		=     use_labels
       __A    		=     type_sequence_label_size
       __A    		=     encoder_stride






 def 		lowercase_  (     self					:Any     )			->     Optional[Any]:







       '''simple docstring'''



       __A    		=     floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]     )

       __A    		=     None
       if self.use_labels:
             __A    		=     ids_tensor([self.batch_size]		,			self.type_sequence_label_size     )

       __A    		=     self.get_config()

       return config, pixel_values, labels






 def 		lowercase_  (     self					:List[str]     )			->     str:







       '''simple docstring'''



       return DonutSwinConfig(
           image_size=self.image_size		,			patch_size=self.patch_size		,			num_channels=self.num_channels		,			embed_dim=self.embed_dim		,			depths=self.depths		,			num_heads=self.num_heads		,			window_size=self.window_size		,			mlp_ratio=self.mlp_ratio		,			qkv_bias=self.qkv_bias		,			hidden_dropout_prob=self.hidden_dropout_prob		,			attention_probs_dropout_prob=self.attention_probs_dropout_prob		,			drop_path_rate=self.drop_path_rate		,			hidden_act=self.hidden_act		,			use_absolute_embeddings=self.use_absolute_embeddings		,			path_norm=self.patch_norm		,			layer_norm_eps=self.layer_norm_eps		,			initializer_range=self.initializer_range		,			encoder_stride=self.encoder_stride		,			)






 def 		lowercase_  (     self					:List[str]		,			_A					:Any		,			_A					:List[str]		,			_A					:List[str]     )			->     int:







       '''simple docstring'''



       __A    		=     DonutSwinModel(config=_A     )
       model.to(_A     )
       model.eval()
       __A    		=     model(_A     )

       __A    		=     ((config.image_size // config.patch_size) ** 2) // (4 ** (len(config.depths     ) - 1))
       __A    		=     int(config.embed_dim * 2 ** (len(config.depths     ) - 1)     )

       self.parent.assertEqual(result.last_hidden_state.shape		,			(self.batch_size, expected_seq_len, expected_dim)     )






 def 		lowercase_  (     self					:List[str]     )			->     List[str]:







       '''simple docstring'''



       __A    		=     self.prepare_config_and_inputs()
       (
           (
           __A
       )		, (
           __A
       )		, (
           __A
       )		, 
       )    		=     config_and_inputs
       __A    		=     {'pixel_values': pixel_values}
       return config, inputs_dict


@require_torch
class 	UpperCamelCase__					(      SCREAMING_SNAKE_CASE ,       SCREAMING_SNAKE_CASE ,       unittest.TestCase):
 UpperCAmelCase__      :      List[str]  			=   (DonutSwinModel,) if is_torch_available() else ()
 UpperCAmelCase__      :      Optional[Any]  			=   {'feature-extraction': DonutSwinModel} if is_torch_available() else {}
 UpperCAmelCase__      :      Dict  			=   True

 UpperCAmelCase__      :      int  			=   False
 UpperCAmelCase__      :      List[Any]  			=   False
 UpperCAmelCase__      :      Tuple  			=   False






 def 		lowercase_  (     self					:Union[str, Any]     )			->     int:







       '''simple docstring'''



       __A    		=     DonutSwinModelTester(self     )
       __A    		=     ConfigTester(self		,			config_class=_A		,			embed_dim=37     )






 def 		lowercase_  (     self					:Dict     )			->     Union[str, Any]:







       '''simple docstring'''



       self.create_and_test_config_common_properties()
       self.config_tester.create_and_test_config_to_json_string()
       self.config_tester.create_and_test_config_to_json_file()
       self.config_tester.create_and_test_config_from_and_save_pretrained()
       self.config_tester.create_and_test_config_with_num_labels()
       self.config_tester.check_config_can_be_init_without_params()
       self.config_tester.check_config_arguments_init()






 def 		lowercase_  (     self					:Tuple     )			->     str:







       '''simple docstring'''



       return






 def 		lowercase_  (     self					:List[Any]     )			->     Optional[Any]:







       '''simple docstring'''



       __A    		=     self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_model(*_A     )






 def 		lowercase_  (     self					:int     )			->     int:







       '''simple docstring'''



       pass






 def 		lowercase_  (     self					:Any     )			->     Tuple:







       '''simple docstring'''



       __A		, __A    		=     self.model_tester.prepare_config_and_inputs_for_common()

       for model_class in self.all_model_classes:
             __A    		=     model_class(_A     )
             self.assertIsInstance(model.get_input_embeddings()		,			(nn.Module)     )
             __A    		=     model.get_output_embeddings()
             self.assertTrue(x is None or isinstance(_A		,			nn.Linear     )     )






 def 		lowercase_  (     self					:Any     )			->     Dict:







       '''simple docstring'''



       __A		, __A    		=     self.model_tester.prepare_config_and_inputs_for_common()

       for model_class in self.all_model_classes:
             __A    		=     model_class(_A     )
             __A    		=     inspect.signature(model.forward     )
             # signature.parameters is an OrderedDict => so arg_names order is deterministic
             __A    		=     [*signature.parameters.keys()]

             __A    		=     ['pixel_values']
             self.assertListEqual(arg_names[:1]		,			_A     )






 def 		lowercase_  (     self					:List[str]     )			->     Optional[int]:







       '''simple docstring'''



       __A		, __A    		=     self.model_tester.prepare_config_and_inputs_for_common()
       __A    		=     True

       for model_class in self.all_model_classes:
             __A    		=     True
             __A    		=     False
             __A    		=     True
             __A    		=     model_class(_A     )
             model.to(_A     )
             model.eval()
             with torch.no_grad():
                   __A    		=     model(**self._prepare_for_class(_A		,			_A     )     )
             __A    		=     outputs.attentions
             __A    		=     len(self.model_tester.depths     )
             self.assertEqual(len(_A     )		,			_A     )

             # check that output_attentions also work using config
             del inputs_dict["output_attentions"]
             __A    		=     True
             __A    		=     config.window_size**2
             __A    		=     model_class(_A     )
             model.to(_A     )
             model.eval()
             with torch.no_grad():
                   __A    		=     model(**self._prepare_for_class(_A		,			_A     )     )
             __A    		=     outputs.attentions
             self.assertEqual(len(_A     )		,			_A     )

             self.assertListEqual(
                 list(attentions[0].shape[-3:]     )		,			[self.model_tester.num_heads[0], window_size_squared, window_size_squared]		,			)
             __A    		=     len(_A     )

             # Check attention is always last and order is fine
             __A    		=     True
             __A    		=     True
             __A    		=     model_class(_A     )
             model.to(_A     )
             model.eval()
             with torch.no_grad():
                   __A    		=     model(**self._prepare_for_class(_A		,			_A     )     )

             if hasattr(self.model_tester		,			'num_hidden_states_types'     ):
                   __A    		=     self.model_tester.num_hidden_states_types
             else:
                   # also another +1 for reshaped_hidden_states
                   __A    		=     2
             self.assertEqual(out_len + added_hidden_states		,			len(_A     )     )

             __A    		=     outputs.attentions

             self.assertEqual(len(_A     )		,			_A     )

             self.assertListEqual(
                 list(self_attentions[0].shape[-3:]     )		,			[self.model_tester.num_heads[0], window_size_squared, window_size_squared]		,			)






 def 		lowercase_  (     self					:Dict		,			_A					:Optional[int]		,			_A					:Tuple		,			_A					:List[Any]		,			_A					:List[Any]     )			->     Union[str, Any]:







       '''simple docstring'''



       __A    		=     model_class(_A     )
       model.to(_A     )
       model.eval()

       with torch.no_grad():
             __A    		=     model(**self._prepare_for_class(_A		,			_A     )     )

       __A    		=     outputs.hidden_states

       __A    		=     getattr(
           self.model_tester		,			'expected_num_hidden_layers'		,			len(self.model_tester.depths     ) + 1     )
       self.assertEqual(len(_A     )		,			_A     )

       # DonutSwin has a different seq_length
       __A    		=     (
           config.patch_size
           if isinstance(config.patch_size		,			collections.abc.Iterable     )
           else (config.patch_size, config.patch_size)
       )

       __A    		=     (image_size[1] // patch_size[1]) * (image_size[0] // patch_size[0])

       self.assertListEqual(
           list(hidden_states[0].shape[-2:]     )		,			[num_patches, self.model_tester.embed_dim]		,			)

       __A    		=     outputs.reshaped_hidden_states
       self.assertEqual(len(_A     )		,			_A     )

       __A		, __A		, __A		, __A    		=     reshaped_hidden_states[0].shape
       __A    		=     (
           reshaped_hidden_states[0].view(_A		,			_A		,			height * width     ).permute(0		,			2		,			1     )
       )
       self.assertListEqual(
           list(reshaped_hidden_states.shape[-2:]     )		,			[num_patches, self.model_tester.embed_dim]		,			)






 def 		lowercase_  (     self					:List[Any]     )			->     Any:







       '''simple docstring'''



       __A		, __A    		=     self.model_tester.prepare_config_and_inputs_for_common()

       __A    		=     (
           self.model_tester.image_size
           if isinstance(self.model_tester.image_size		,			collections.abc.Iterable     )
           else (self.model_tester.image_size, self.model_tester.image_size)
       )

       for model_class in self.all_model_classes:
             __A    		=     True
             self.check_hidden_states_output(_A		,			_A		,			_A		,			_A     )

             # check that output_hidden_states also work using config
             del inputs_dict["output_hidden_states"]
             __A    		=     True

             self.check_hidden_states_output(_A		,			_A		,			_A		,			_A     )






 def 		lowercase_  (     self					:Dict     )			->     Optional[int]:







       '''simple docstring'''



       __A		, __A    		=     self.model_tester.prepare_config_and_inputs_for_common()
       __A    		=     3

       __A    		=     (
           self.model_tester.image_size
           if isinstance(self.model_tester.image_size		,			collections.abc.Iterable     )
           else (self.model_tester.image_size, self.model_tester.image_size)
       )
       __A    		=     (
           config.patch_size
           if isinstance(config.patch_size		,			collections.abc.Iterable     )
           else (config.patch_size, config.patch_size)
       )

       __A    		=     image_size[0] + patch_size[0] - (image_size[0] % patch_size[0])
       __A    		=     image_size[1] + patch_size[1] - (image_size[1] % patch_size[1])

       for model_class in self.all_model_classes:
             __A    		=     True
             self.check_hidden_states_output(_A		,			_A		,			_A		,			(padded_height, padded_width)     )

             # check that output_hidden_states also work using config
             del inputs_dict["output_hidden_states"]
             __A    		=     True
             self.check_hidden_states_output(_A		,			_A		,			_A		,			(padded_height, padded_width)     )






 @slow
 def 		lowercase_  (     self					:Optional[int]     )			->     int:







       '''simple docstring'''



       for model_name in DONUT_SWIN_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
             __A    		=     DonutSwinModel.from_pretrained(_A     )
             self.assertIsNotNone(_A     )






 def 		lowercase_  (     self					:Tuple     )			->     Union[str, Any]:







       '''simple docstring'''



       __A		, __A    		=     self.model_tester.prepare_config_and_inputs_for_common()

       __A    		=     _config_zero_init(_A     )
       for model_class in self.all_model_classes:
             __A    		=     model_class(config=_A     )
             for name, param in model.named_parameters():
                   if "embeddings" not in name and param.requires_grad:
                         self.assertIn(
                             ((param.data.mean() * 1E9).round() / 1E9).item()		,			[0.0, 1.0]		,			msg=F'Parameter {name} of model {model_class} seems not properly initialized'		,			)






