


'''simple docstring'''




import gc
import tempfile
import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionPipeline
from diffusers.utils.testing_utils import load_image, nightly, require_torch_gpu, torch_device


__lowerCamelCase   :					int   							=					False







class    A_	(unittest.TestCase							):
							"""simple docstring"""




							pass







@nightly
@require_torch_gpu
class    A_	(unittest.TestCase							):
							"""simple docstring"""




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


										'''simple docstring'''




										super().tearDown()
										gc.collect()
										torch.cuda.empty_cache()

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


										'''simple docstring'''




										snake_case_				:					Optional[Any]   =       VersatileDiffusionPipeline.from_pretrained("shi-labs/versatile-diffusion"       ,							torch_dtype=torch.floataa				)
										pipe.to(A__				)
										pipe.set_progress_bar_config(disable=A__				)

										snake_case_				:					int   =       load_image(
										    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg"				)

										snake_case_				:					Optional[int]   =       torch.manual_seed(0				)
										snake_case_				:					Any   =       pipe.dual_guided(
										    prompt="first prompt"       ,							image=A__       ,							text_to_image_strength=0.7_5       ,							generator=A__       ,							guidance_scale=7.5       ,							num_inference_steps=2       ,							output_type="numpy"       ,							).images

										with tempfile.TemporaryDirectory() as tmpdirname:
													pipe.save_pretrained(A__				)
													snake_case_				:					Union[str, Any]   =       VersatileDiffusionPipeline.from_pretrained(A__       ,							torch_dtype=torch.floataa				)
										pipe.to(A__				)
										pipe.set_progress_bar_config(disable=A__				)

										snake_case_				:					Any   =       generator.manual_seed(0				)
										snake_case_				:					str   =       pipe.dual_guided(
										    prompt="first prompt"       ,							image=A__       ,							text_to_image_strength=0.7_5       ,							generator=A__       ,							guidance_scale=7.5       ,							num_inference_steps=2       ,							output_type="numpy"       ,							).images

										assert np.abs(image - new_image				).sum() < 1E-5, "Models don't have the same forward pass"






							def 							_A	(		self			:Tuple				)				->  Optional[int]:


										'''simple docstring'''




										snake_case_				:					Tuple   =       VersatileDiffusionPipeline.from_pretrained("shi-labs/versatile-diffusion"       ,							torch_dtype=torch.floataa				)
										pipe.to(A__				)
										pipe.set_progress_bar_config(disable=A__				)

										snake_case_				:					Optional[int]   =       "cyberpunk 2077"
										snake_case_				:					Dict   =       load_image(
										    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg"				)
										snake_case_				:					Optional[Any]   =       torch.manual_seed(0				)
										snake_case_				:					List[str]   =       pipe.dual_guided(
										    prompt=A__       ,							image=A__       ,							text_to_image_strength=0.7_5       ,							generator=A__       ,							guidance_scale=7.5       ,							num_inference_steps=50       ,							output_type="numpy"       ,							).images

										snake_case_				:					Union[str, Any]   =       image[0, 253:256, 253:256, -1]

										assert image.shape == (1, 512, 512, 3)
										snake_case_				:					Any   =       np.array([0.1_4_4_8, 0.1_6_1_9, 0.1_7_4_1, 0.1_0_8_6, 0.1_1_4_7, 0.1_1_2_8, 0.1_1_9_9, 0.1_1_6_5, 0.1_0_0_1]				)

										assert np.abs(image_slice.flatten() - expected_slice				).max() < 1E-1

										snake_case_				:					List[Any]   =       "A painting of a squirrel eating a burger "
										snake_case_				:					List[str]   =       torch.manual_seed(0				)
										snake_case_				:					str   =       pipe.text_to_image(
										    prompt=A__       ,							generator=A__       ,							guidance_scale=7.5       ,							num_inference_steps=50       ,							output_type="numpy"				).images

										snake_case_				:					Dict   =       image[0, 253:256, 253:256, -1]

										assert image.shape == (1, 512, 512, 3)
										snake_case_				:					List[str]   =       np.array([0.3_3_6_7, 0.3_1_6_9, 0.2_6_5_6, 0.3_8_7_0, 0.4_7_9_0, 0.3_7_9_6, 0.4_0_0_9, 0.4_8_7_8, 0.4_7_7_8]				)

										assert np.abs(image_slice.flatten() - expected_slice				).max() < 1E-1

										snake_case_				:					Optional[Any]   =       pipe.image_variation(A__       ,							generator=A__       ,							output_type="numpy"				).images

										snake_case_				:					Tuple   =       image[0, 253:256, 253:256, -1]

										assert image.shape == (1, 512, 512, 3)
										snake_case_				:					List[str]   =       np.array([0.3_0_7_6, 0.3_1_2_3, 0.3_2_8_4, 0.3_7_8_2, 0.3_7_7_0, 0.3_8_9_4, 0.4_2_9_7, 0.4_3_3_1, 0.4_4_5_6]				)

										assert np.abs(image_slice.flatten() - expected_slice				).max() < 1E-1












'''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 	__UpperCAmelCase       ( __magic_name__     ,__magic_name__=()     ,__magic_name__=None     ,__magic_name__="no"     ,__magic_name__="29500"						)->    Optional[int]:



			"""simple docstring"""



			snake_case_				:					str   =       False
			snake_case_				:					int   =       False
			if any(key.startswith("KAGGLE"						) for key in os.environ.keys()						):
						snake_case_				:					Any   =       True
			elif "IPython" in sys.modules:
						snake_case_				:					Union[str, Any]   =       "google.colab" in str(sys.modules["IPython"].get_ipython()						)

			try:
						snake_case_				:					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"     ,__magic_name__						) 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:
									snake_case_				:					Tuple   =       8

						snake_case_				:					Optional[int]   =       PrepareForLaunch(__magic_name__     ,distributed_type="TPU"						)
						print(F'''Launching a training on {num_processes} TPU cores.'''						)
						xmp.spawn(__magic_name__     ,args=__magic_name__     ,nprocs=__magic_name__     ,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(*__magic_name__						)
			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=__magic_name__     ,master_addr="127.0.01"     ,master_port=__magic_name__     ,mixed_precision=__magic_name__						):
												snake_case_				:					Optional[int]   =       PrepareForLaunch(__magic_name__     ,distributed_type="MULTI_GPU"						)
												print(F'''Launching training on {num_processes} GPUs.'''						)
												try:
															start_processes(__magic_name__     ,args=__magic_name__     ,nprocs=__magic_name__     ,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():
												snake_case_				:					Any   =       "1"
												print("Launching training on MPS."						)
									elif torch.cuda.is_available():
												print("Launching training on one GPU."						)
									else:
												print("Launching training on CPU."						)
									function(*__magic_name__						)






def 	__UpperCAmelCase       ( __magic_name__     ,__magic_name__=()     ,__magic_name__=2						)->    Dict:



			"""simple docstring"""



			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=__magic_name__     ,master_addr="127.0.01"     ,master_port="29500"     ,accelerate_mixed_precision="no"     ,accelerate_debug_rdv_file=tmp_file.name     ,accelerate_use_cpu="yes"     ,):
									snake_case_				:					Any   =       PrepareForLaunch(__magic_name__     ,debug=__magic_name__						)
									start_processes(__magic_name__     ,args=__magic_name__     ,nprocs=__magic_name__     ,start_method="fork"						)




