'''simple docstring'''




import os
import sys
import tempfile

import torch

from .state import AcceleratorState
from .utils import PrecisionType, PrepareForLaunch, is_mps_available, patch_environment




def 							lowerCamelCase	(      UpperCAmelCase__       :					Dict      ,      UpperCAmelCase__       :					Union[str, Any]=()      ,      UpperCAmelCase__       :					Dict=None      ,      UpperCAmelCase__       :					Tuple="no"      ,      UpperCAmelCase__       :					List[str]="29500" )			->      Optional[int]:
							lowercase_					:			Union[str, Any]										=       False
							lowercase_					:			Optional[Any]										=       False
							if any(key.startswith("""KAGGLE""" ) for key in os.environ.keys() ):
														lowercase_					:			List[str]										=       True
							elif "IPython" in sys.modules:
														lowercase_					:			Optional[int]										=       """google.colab""" in str(sys.modules["""IPython"""].get_ipython() )

							try:
														lowercase_					:			List[Any]										=       PrecisionType(mixed_precision.lower() )
							except ValueError:
														raise ValueError(
														    F'''Unknown mixed_precision mode: {args.mixed_precision.lower()}. Choose between {PrecisionType.list()}.''' )

							if (in_colab or in_kaggle) and (os.environ.get("""TPU_NAME"""      ,      UpperCAmelCase__ ) is not None):
														# TPU launch
														import torch_xla.distributed.xla_multiprocessing as xmp

														if len(AcceleratorState._shared_state ) > 0:
																					raise ValueError(
																					    """To train on TPU in Colab or Kaggle Kernel, the `Accelerator` should only be initialized inside """
																					    """your training function. Restart your notebook and make sure no cells initializes an """
																					    """`Accelerator`.""" )
														if num_processes is None:
																					lowercase_					:			int										=       8

														lowercase_					:			Optional[int]										=       PrepareForLaunch(UpperCAmelCase__      ,      distributed_type="""TPU""" )
														print(F'''Launching a training on {num_processes} TPU cores.''' )
														xmp.spawn(UpperCAmelCase__      ,      args=UpperCAmelCase__      ,      nprocs=UpperCAmelCase__      ,      start_method="""fork""" )
							elif in_colab:
														# No need for a distributed launch otherwise as it's either CPU or one GPU.
														if torch.cuda.is_available():
																					print("""Launching training on one GPU.""" )
														else:
																					print("""Launching training on one CPU.""" )
														function(*UpperCAmelCase__ )
							else:
														if num_processes is None:
																					raise ValueError(
																					    """You have to specify the number of GPUs you would like to use, add `num_processes=...` to your call.""" )

														if num_processes > 1:
																					# Multi-GPU launch
																					from torch.multiprocessing import start_processes
																					from torch.multiprocessing.spawn import ProcessRaisedException

																					if len(AcceleratorState._shared_state ) > 0:
																												raise ValueError(
																												    """To launch a multi-GPU training from your notebook, the `Accelerator` should only be initialized """
																												    """inside your training function. Restart your notebook and make sure no cells initializes an """
																												    """`Accelerator`.""" )

																					if torch.cuda.is_initialized():
																												raise ValueError(
																												    """To launch a multi-GPU training from your notebook, you need to avoid running any instruction """
																												    """using `torch.cuda` in any cell. Restart your notebook and make sure no cells use any CUDA """
																												    """function.""" )

																					# torch.distributed will expect a few environment variable to be here. We set the ones common to each
																					# process here (the other ones will be set be the launcher).
																					with patch_environment(
																					    world_size=UpperCAmelCase__      ,      master_addr="""127.0.01"""      ,      master_port=UpperCAmelCase__      ,      mixed_precision=UpperCAmelCase__ ):
																												lowercase_					:			Any										=       PrepareForLaunch(UpperCAmelCase__      ,      distributed_type="""MULTI_GPU""" )
																												print(F'''Launching training on {num_processes} GPUs.''' )
																												try:
																																			start_processes(UpperCAmelCase__      ,      args=UpperCAmelCase__      ,      nprocs=UpperCAmelCase__      ,      start_method="""fork""" )
																												except ProcessRaisedException as e:
																																			if "Cannot re-initialize CUDA in forked subprocess" in e.args[0]:
																																										raise RuntimeError(
																																										    """CUDA has been initialized before the `notebook_launcher` could create a forked subprocess. """
																																										    """This likely stems from an outside import causing issues once the `notebook_launcher()` is called. """
																																										    """Please review your imports and test them when running the `notebook_launcher()` to identify """
																																										    """which one is problematic.""" ) from e

														else:
																					# No need for a distributed launch otherwise as it's either CPU, GPU or MPS.
																					if is_mps_available():
																												lowercase_					:			Tuple										=       """1"""
																												print("""Launching training on MPS.""" )
																					elif torch.cuda.is_available():
																												print("""Launching training on one GPU.""" )
																					else:
																												print("""Launching training on CPU.""" )
																					function(*UpperCAmelCase__ )




def 							lowerCamelCase	(      UpperCAmelCase__       :					List[Any]      ,      UpperCAmelCase__       :					str=()      ,      UpperCAmelCase__       :					Tuple=2 )			->      int:
							from torch.multiprocessing import start_processes

							with tempfile.NamedTemporaryFile() as tmp_file:
														# torch.distributed will expect a few environment variable to be here. We set the ones common to each
														# process here (the other ones will be set be the launcher).
														with patch_environment(
														    world_size=UpperCAmelCase__      ,      master_addr="""127.0.01"""      ,      master_port="""29500"""      ,      accelerate_mixed_precision="""no"""      ,      accelerate_debug_rdv_file=tmp_file.name      ,      accelerate_use_cpu="""yes"""      ,      ):
																					lowercase_					:			int										=       PrepareForLaunch(UpperCAmelCase__      ,      debug=UpperCAmelCase__ )
																					start_processes(UpperCAmelCase__      ,      args=UpperCAmelCase__      ,      nprocs=UpperCAmelCase__      ,      start_method="""fork""" )






'''simple docstring'''




import gc
import inspect
import unittest

import torch
from parameterized import parameterized

from diffusers import PriorTransformer
from diffusers.utils import floats_tensor, slow, torch_all_close, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from .test_modeling_common import ModelTesterMixin


enable_full_determinism()





class     __magic_name__						(    _UpperCAmelCase,     unittest.TestCase):
			UpperCamelCase__				=       PriorTransformer
			UpperCamelCase__				=       '''hidden_states'''







			@property
			def 	SCREAMING_SNAKE_CASE_     (	self				:  int    ):
										lowercase_					:			Optional[Any]										=       4
										lowercase_					:			int										=       8
										lowercase_					:			Union[str, Any]										=       7

										lowercase_					:			List[str]										=       floats_tensor((batch_size, embedding_dim)    ).to(lowercase_    )

										lowercase_					:			Tuple										=       floats_tensor((batch_size, embedding_dim)    ).to(lowercase_    )
										lowercase_					:			List[str]										=       floats_tensor((batch_size, num_embeddings, embedding_dim)    ).to(lowercase_    )

										return {
										    "hidden_states": hidden_states,
										    "timestep": 2,
										    "proj_embedding": proj_embedding,
										    "encoder_hidden_states": encoder_hidden_states,
										}







			def 	SCREAMING_SNAKE_CASE_     (	self				:  List[str]				,			lowercase_				:  Optional[Any]=0    ):
										torch.manual_seed(lowercase_    )
										lowercase_					:			int										=       4
										lowercase_					:			Any										=       8
										lowercase_					:			Tuple										=       7

										lowercase_					:			Union[str, Any]										=       torch.randn((batch_size, embedding_dim)    ).to(lowercase_    )

										lowercase_					:			Tuple										=       torch.randn((batch_size, embedding_dim)    ).to(lowercase_    )
										lowercase_					:			Union[str, Any]										=       torch.randn((batch_size, num_embeddings, embedding_dim)    ).to(lowercase_    )

										return {
										    "hidden_states": hidden_states,
										    "timestep": 2,
										    "proj_embedding": proj_embedding,
										    "encoder_hidden_states": encoder_hidden_states,
										}







			@property
			def 	SCREAMING_SNAKE_CASE_     (	self				:  str    ):
										return (4, 8)







			@property
			def 	SCREAMING_SNAKE_CASE_     (	self				:  Union[str, Any]    ):
										return (4, 8)







			def 	SCREAMING_SNAKE_CASE_     (	self				:  Tuple    ):
										lowercase_					:			Optional[Any]										=       {
										    """num_attention_heads""": 2,
										    """attention_head_dim""": 4,
										    """num_layers""": 2,
										    """embedding_dim""": 8,
										    """num_embeddings""": 7,
										    """additional_embeddings""": 4,
										}
										lowercase_					:			List[Any]										=       self.dummy_input
										return init_dict, inputs_dict







			def 	SCREAMING_SNAKE_CASE_     (	self				:  List[str]    ):
										lowercase_						,						lowercase_					:			Tuple										=       PriorTransformer.from_pretrained(
										    """hf-internal-testing/prior-dummy"""				,			output_loading_info=lowercase_    )
										self.assertIsNotNone(lowercase_    )
										self.assertEqual(len(loading_info["""missing_keys"""]    )				,			0    )

										model.to(lowercase_    )
										lowercase_					:			Optional[int]										=       model(**self.dummy_input    )[0]

										assert hidden_states is not None, "Make sure output is not None"







			def 	SCREAMING_SNAKE_CASE_     (	self				:  Any    ):
										lowercase_						,						lowercase_					:			str										=       self.prepare_init_args_and_inputs_for_common()

										lowercase_					:			List[Any]										=       self.model_class(**lowercase_    )
										lowercase_					:			Any										=       inspect.signature(model.forward    )
										# signature.parameters is an OrderedDict => so arg_names order is deterministic
										lowercase_					:			Any										=       [*signature.parameters.keys()]

										lowercase_					:			Optional[Any]										=       ["""hidden_states""", """timestep"""]
										self.assertListEqual(arg_names[:2]				,			lowercase_    )







			def 	SCREAMING_SNAKE_CASE_     (	self				:  List[str]    ):
										lowercase_					:			Optional[Any]										=       PriorTransformer.from_pretrained("""hf-internal-testing/prior-dummy"""    )
										lowercase_					:			Any										=       model.to(lowercase_    )

										if hasattr(lowercase_				,			"""set_default_attn_processor"""    ):
																	model.set_default_attn_processor()

										lowercase_					:			Any										=       self.get_dummy_seed_input()

										with torch.no_grad():
																	lowercase_					:			List[str]										=       model(**lowercase_    )[0]

										lowercase_					:			Tuple										=       output[0, :5].flatten().cpu()
										print(lowercase_    )

										# Since the VAE Gaussian prior's generator is seeded on the appropriate device,
										# the expected output slices are not the same for CPU and GPU.
										lowercase_					:			Dict										=       torch.tensor([-1.34_36, -0.28_70, 0.75_38, 0.43_68, -0.02_39]    )
										self.assertTrue(torch_all_close(lowercase_				,			lowercase_				,			rtol=1E-2    )    )





@slow
class     __magic_name__						(    unittest.TestCase):







			def 	SCREAMING_SNAKE_CASE_     (	self				:  List[Any]				,			lowercase_				:  Union[str, Any]=1				,			lowercase_				:  Optional[Any]=768				,			lowercase_				:  Tuple=77				,			lowercase_				:  Optional[int]=0    ):
										torch.manual_seed(lowercase_    )
										lowercase_					:			Optional[Any]										=       batch_size
										lowercase_					:			int										=       embedding_dim
										lowercase_					:			int										=       num_embeddings

										lowercase_					:			Dict										=       torch.randn((batch_size, embedding_dim)    ).to(lowercase_    )

										lowercase_					:			Any										=       torch.randn((batch_size, embedding_dim)    ).to(lowercase_    )
										lowercase_					:			List[Any]										=       torch.randn((batch_size, num_embeddings, embedding_dim)    ).to(lowercase_    )

										return {
										    "hidden_states": hidden_states,
										    "timestep": 2,
										    "proj_embedding": proj_embedding,
										    "encoder_hidden_states": encoder_hidden_states,
										}







			def 	SCREAMING_SNAKE_CASE_     (	self				:  Any    ):
										# clean up the VRAM after each test
										super().tearDown()
										gc.collect()
										torch.cuda.empty_cache()







			@parameterized.expand(
			    [
			        # fmt: off
			        [13, [-0.58_61,  0.12_83, -0.09_31,  0.08_82,  0.44_76,  0.13_29, -0.04_98,  0.06_40]],
			        [37, [-0.49_13,  0.01_10, -0.04_83,  0.05_41,  0.49_54, -0.01_70,  0.03_54,  0.16_51]],
			        # fmt: on
			    ]    )
			def 	SCREAMING_SNAKE_CASE_     (	self				:  int				,			lowercase_				:  Any				,			lowercase_				:  List[str]    ):
										lowercase_					:			List[Any]										=       PriorTransformer.from_pretrained("""kandinsky-community/kandinsky-2-1-prior"""				,			subfolder="""prior"""    )
										model.to(lowercase_    )
										lowercase_					:			Optional[Any]										=       self.get_dummy_seed_input(seed=lowercase_    )

										with torch.no_grad():
																	lowercase_					:			Tuple										=       model(**lowercase_    )[0]

										assert list(sample.shape    ) == [1, 768]

										lowercase_					:			Union[str, Any]										=       sample[0, :8].flatten().cpu()
										print(lowercase_    )
										lowercase_					:			Optional[Any]										=       torch.tensor(lowercase_    )

										assert torch_all_close(lowercase_				,			lowercase_				,			atol=1E-3    )





