import importlib
import sys
from argparse import REMAINDER, ArgumentParser
from pathlib import Path

import torch_xla.distributed.xla_multiprocessing as xmp







def       A__   (						):
 SCREAMING_SNAKE_CASE__: str=   ArgumentParser(
     description=(
         '''PyTorch TPU distributed training launch '''
         '''helper utility that will spawn up '''
         '''multiple distributed processes'''
     )   )

 # Optional arguments for the launch helper
 parser.add_argument('''--num_cores'''							,					type=snake_case_							,					default=1							,					help='''Number of TPU cores to use (1 or 8).'''   )

 # positional
 parser.add_argument(
     '''training_script'''							,					type=snake_case_							,					help=(
         '''The full path to the single TPU training '''
         '''program/script to be launched in parallel, '''
         '''followed by all the arguments for the '''
         '''training script'''
     )							,					)

 # rest from the training program
 parser.add_argument('''training_script_args'''							,					nargs=snake_case_   )

 return parser.parse_args()







def       A__   (						):
 SCREAMING_SNAKE_CASE__: Dict=   parse_args()

 # Import training_script as a module.
 SCREAMING_SNAKE_CASE__: List[Any]=   Path(args.training_script   )
 sys.path.append(str(script_fpath.parent.resolve()   )   )
 SCREAMING_SNAKE_CASE__: Optional[Any]=   script_fpath.stem
 SCREAMING_SNAKE_CASE__: Dict=   importlib.import_module(snake_case_   )

 # Patch sys.argv
 SCREAMING_SNAKE_CASE__: List[str]=   [args.training_script] + args.training_script_args + ['''--tpu_num_cores''', str(args.num_cores   )]
 xmp.spawn(mod._mp_fn							,					args=()							,					nprocs=args.num_cores   )


if __name__ == "__main__":
      main()








import gc
import unittest

import numpy as np
import torch

from diffusers import StableDiffusionKDiffusionPipeline
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu


enable_full_determinism()




@slow
@require_torch_gpu
class 	UpperCamelCase_							(  unittest.TestCase     ):


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



									"""simple docstring"""



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


							def   _snake_case	(				self     :Any		) -> Optional[int]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   StableDiffusionKDiffusionPipeline.from_pretrained("""CompVis/stable-diffusion-v1-4"""		)
									SCREAMING_SNAKE_CASE__								=   sd_pipe.to(__A		)
									sd_pipe.set_progress_bar_config(disable=__A		)

									sd_pipe.set_scheduler("""sample_euler"""		)

									SCREAMING_SNAKE_CASE__								=   """A painting of a squirrel eating a burger"""
									SCREAMING_SNAKE_CASE__								=   torch.manual_seed(0		)
									SCREAMING_SNAKE_CASE__								=   sd_pipe([prompt]				,					generator=__A				,					guidance_scale=9.0				,					num_inference_steps=20				,					output_type="""np"""		)

									SCREAMING_SNAKE_CASE__								=   output.images

									SCREAMING_SNAKE_CASE__								=   image[0, -3:, -3:, -1]

									assert image.shape == (1, 512, 512, 3)
									SCREAMING_SNAKE_CASE__								=   np.array([0.0_4_4_7, 0.0_4_9_2, 0.0_4_6_8, 0.0_4_0_8, 0.0_3_8_3, 0.0_4_0_8, 0.0_3_5_4, 0.0_3_8_0, 0.0_3_3_9]		)

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


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



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   StableDiffusionKDiffusionPipeline.from_pretrained("""stabilityai/stable-diffusion-2-1-base"""		)
									SCREAMING_SNAKE_CASE__								=   sd_pipe.to(__A		)
									sd_pipe.set_progress_bar_config(disable=__A		)

									sd_pipe.set_scheduler("""sample_euler"""		)

									SCREAMING_SNAKE_CASE__								=   """A painting of a squirrel eating a burger"""
									SCREAMING_SNAKE_CASE__								=   torch.manual_seed(0		)
									SCREAMING_SNAKE_CASE__								=   sd_pipe([prompt]				,					generator=__A				,					guidance_scale=9.0				,					num_inference_steps=20				,					output_type="""np"""		)

									SCREAMING_SNAKE_CASE__								=   output.images

									SCREAMING_SNAKE_CASE__								=   image[0, -3:, -3:, -1]

									assert image.shape == (1, 512, 512, 3)
									SCREAMING_SNAKE_CASE__								=   np.array([0.1_2_3_7, 0.1_3_2_0, 0.1_4_3_8, 0.1_3_5_9, 0.1_3_9_0, 0.1_1_3_2, 0.1_2_7_7, 0.1_1_7_5, 0.1_1_1_2]		)

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


							def   _snake_case	(				self     :Tuple		) -> Optional[Any]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   StableDiffusionKDiffusionPipeline.from_pretrained("""stabilityai/stable-diffusion-2-1-base"""		)
									SCREAMING_SNAKE_CASE__								=   sd_pipe.to(__A		)
									sd_pipe.set_progress_bar_config(disable=__A		)

									sd_pipe.set_scheduler("""sample_dpmpp_2m"""		)

									SCREAMING_SNAKE_CASE__								=   """A painting of a squirrel eating a burger"""
									SCREAMING_SNAKE_CASE__								=   torch.manual_seed(0		)
									SCREAMING_SNAKE_CASE__								=   sd_pipe(
									    [prompt]				,					generator=__A				,					guidance_scale=7.5				,					num_inference_steps=15				,					output_type="""np"""				,					use_karras_sigmas=__A				,					)

									SCREAMING_SNAKE_CASE__								=   output.images

									SCREAMING_SNAKE_CASE__								=   image[0, -3:, -3:, -1]

									assert image.shape == (1, 512, 512, 3)
									SCREAMING_SNAKE_CASE__								=   np.array(
									    [0.1_1_3_8_1_6_8_9, 0.1_2_1_1_2_9_2_1, 0.1_3_8_9_4_5_7, 0.1_2_5_4_9_6_0_6, 0.1_2_4_4_9_6_4, 0.1_0_8_3_1_5_1_7, 0.1_1_5_6_2_8_6_6, 0.1_0_8_6_7_8_1_6, 0.1_0_4_9_9_0_4_8]		)

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