
'''simple docstring'''




import gc
import random
import unittest

import torch

from diffusers import (
    IFImgaImgPipeline,
    IFImgaImgSuperResolutionPipeline,
    IFInpaintingPipeline,
    IFInpaintingSuperResolutionPipeline,
    IFPipeline,
    IFSuperResolutionPipeline,
)
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import floats_tensor, load_numpy, require_torch_gpu, skip_mps, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference
from . import IFPipelineTesterMixin





@skip_mps
class  A				(							__snake_case    ,			__snake_case    ,			unittest.TestCase						):
     __magic_name__						=		IFPipeline
     __magic_name__						=		TEXT_TO_IMAGE_PARAMS - {'''width''', '''height''', '''latents'''}
     __magic_name__						=		TEXT_TO_IMAGE_BATCH_PARAMS
     __magic_name__						=		PipelineTesterMixin.required_optional_params - {'''latents'''}





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






         """simple docstring"""






         return self._get_dummy_components()





     def 				__lowerCAmelCase							(							self						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE=0	) ->				Optional[int]:






         """simple docstring"""






         if str(SCREAMING_SNAKE_CASE	).startswith('''mps'''	):
             A       :   str					=						torch.manual_seed(SCREAMING_SNAKE_CASE	)
         else:
             A       :   List[Any]					=						torch.Generator(device=SCREAMING_SNAKE_CASE	).manual_seed(SCREAMING_SNAKE_CASE	)

         A       :   Any					=						{
             '''prompt''': '''A painting of a squirrel eating a burger''',
             '''generator''': generator,
             '''num_inference_steps''': 2,
             '''output_type''': '''numpy''',
         }

         return inputs





     def 				__lowerCAmelCase							(							self	) ->				Tuple:






         """simple docstring"""






         self._test_save_load_optional_components()





     @unittest.skipIf(torch_device != '''cuda'''						,			reason='''float16 requires CUDA'''	)
     def 				__lowerCAmelCase							(							self	) ->				Tuple:






         """simple docstring"""






         super().test_save_load_floataa(expected_max_diff=1e-1	)





     def 				__lowerCAmelCase							(							self	) ->				Tuple:






         """simple docstring"""






         self._test_attention_slicing_forward_pass(expected_max_diff=1e-2	)





     def 				__lowerCAmelCase							(							self	) ->				Optional[int]:






         """simple docstring"""






         self._test_save_load_local()





     def 				__lowerCAmelCase							(							self	) ->				Optional[int]:






         """simple docstring"""






         self._test_inference_batch_single_identical(
             expected_max_diff=1e-2						,			)





     @unittest.skipIf(
         torch_device != '''cuda''' or not is_xformers_available()						,			reason='''XFormers attention is only available with CUDA and `xformers` installed'''						,			)
     def 				__lowerCAmelCase							(							self	) ->				Tuple:






         """simple docstring"""






         self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1e-3	)





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





     def 				__lowerCAmelCase							(							self	) ->				Tuple:






         """simple docstring"""






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





     def 				__lowerCAmelCase							(							self	) ->				str:






         """simple docstring"""






         A       :   Optional[int]					=						IFPipeline.from_pretrained('''DeepFloyd/IF-I-XL-v1.0'''						,			variant='''fp16'''						,			torch_dtype=torch.floataa	)

         A       :   str					=						IFSuperResolutionPipeline.from_pretrained(
             '''DeepFloyd/IF-II-L-v1.0'''						,			variant='''fp16'''						,			torch_dtype=torch.floataa						,			text_encoder=SCREAMING_SNAKE_CASE						,			tokenizer=SCREAMING_SNAKE_CASE	)

         # pre compute text embeddings and remove T5 to save memory

         pipe_a.text_encoder.to('''cuda'''	)

         A,  A       :   Optional[Any]					=						pipe_a.encode_prompt('''anime turtle'''						,			device='''cuda'''	)

         del pipe_a.tokenizer
         del pipe_a.text_encoder
         gc.collect()

         A       :   List[Any]					=						None
         A       :   List[Any]					=						None

         pipe_a.enable_model_cpu_offload()
         pipe_a.enable_model_cpu_offload()

         pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()	)
         pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()	)

         self._test_if(SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE	)

         pipe_a.remove_all_hooks()
         pipe_a.remove_all_hooks()

         # img2img

         A       :   Optional[Any]					=						IFImgaImgPipeline(**pipe_a.components	)
         A       :   Union[str, Any]					=						IFImgaImgSuperResolutionPipeline(**pipe_a.components	)

         pipe_a.enable_model_cpu_offload()
         pipe_a.enable_model_cpu_offload()

         pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()	)
         pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()	)

         self._test_if_imgaimg(SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE	)

         pipe_a.remove_all_hooks()
         pipe_a.remove_all_hooks()

         # inpainting

         A       :   str					=						IFInpaintingPipeline(**pipe_a.components	)
         A       :   Any					=						IFInpaintingSuperResolutionPipeline(**pipe_a.components	)

         pipe_a.enable_model_cpu_offload()
         pipe_a.enable_model_cpu_offload()

         pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()	)
         pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()	)

         self._test_if_inpainting(SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE	)





     def 				__lowerCAmelCase							(							self						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE	) ->				str:






         """simple docstring"""






         _start_torch_memory_measurement()

         A       :   Any					=						torch.Generator(device='''cpu'''	).manual_seed(0	)
         A       :   Dict					=						pipe_a(
             prompt_embeds=SCREAMING_SNAKE_CASE						,			negative_prompt_embeds=SCREAMING_SNAKE_CASE						,			num_inference_steps=2						,			generator=SCREAMING_SNAKE_CASE						,			output_type='''np'''						,			)

         A       :   Any					=						output.images[0]

         assert image.shape == (64, 64, 3)

         A       :   Union[str, Any]					=						torch.cuda.max_memory_allocated()
         assert mem_bytes < 13 * 10**9

         A       :   Union[str, Any]					=						load_numpy(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if.npy'''	)
         assert_mean_pixel_difference(SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE	)

         # pipeline 2

         _start_torch_memory_measurement()

         A       :   str					=						torch.Generator(device='''cpu'''	).manual_seed(0	)

         A       :   Tuple					=						floats_tensor((1, 3, 64, 64)						,			rng=random.Random(0	)	).to(SCREAMING_SNAKE_CASE	)

         A       :   List[str]					=						pipe_a(
             prompt_embeds=SCREAMING_SNAKE_CASE						,			negative_prompt_embeds=SCREAMING_SNAKE_CASE						,			image=SCREAMING_SNAKE_CASE						,			generator=SCREAMING_SNAKE_CASE						,			num_inference_steps=2						,			output_type='''np'''						,			)

         A       :   List[str]					=						output.images[0]

         assert image.shape == (256, 256, 3)

         A       :   Tuple					=						torch.cuda.max_memory_allocated()
         assert mem_bytes < 4 * 10**9

         A       :   int					=						load_numpy(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_superresolution_stage_II.npy'''	)
         assert_mean_pixel_difference(SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE	)





     def 				__lowerCAmelCase							(							self						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE	) ->				List[str]:






         """simple docstring"""






         _start_torch_memory_measurement()

         A       :   Optional[Any]					=						floats_tensor((1, 3, 64, 64)						,			rng=random.Random(0	)	).to(SCREAMING_SNAKE_CASE	)

         A       :   Optional[Any]					=						torch.Generator(device='''cpu'''	).manual_seed(0	)

         A       :   Any					=						pipe_a(
             prompt_embeds=SCREAMING_SNAKE_CASE						,			negative_prompt_embeds=SCREAMING_SNAKE_CASE						,			image=SCREAMING_SNAKE_CASE						,			num_inference_steps=2						,			generator=SCREAMING_SNAKE_CASE						,			output_type='''np'''						,			)

         A       :   List[str]					=						output.images[0]

         assert image.shape == (64, 64, 3)

         A       :   Any					=						torch.cuda.max_memory_allocated()
         assert mem_bytes < 10 * 10**9

         A       :   Union[str, Any]					=						load_numpy(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_img2img.npy'''	)
         assert_mean_pixel_difference(SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE	)

         # pipeline 2

         _start_torch_memory_measurement()

         A       :   Any					=						torch.Generator(device='''cpu'''	).manual_seed(0	)

         A       :   Union[str, Any]					=						floats_tensor((1, 3, 256, 256)						,			rng=random.Random(0	)	).to(SCREAMING_SNAKE_CASE	)
         A       :   str					=						floats_tensor((1, 3, 64, 64)						,			rng=random.Random(0	)	).to(SCREAMING_SNAKE_CASE	)

         A       :   Tuple					=						pipe_a(
             prompt_embeds=SCREAMING_SNAKE_CASE						,			negative_prompt_embeds=SCREAMING_SNAKE_CASE						,			image=SCREAMING_SNAKE_CASE						,			original_image=SCREAMING_SNAKE_CASE						,			generator=SCREAMING_SNAKE_CASE						,			num_inference_steps=2						,			output_type='''np'''						,			)

         A       :   Any					=						output.images[0]

         assert image.shape == (256, 256, 3)

         A       :   Union[str, Any]					=						torch.cuda.max_memory_allocated()
         assert mem_bytes < 4 * 10**9

         A       :   List[str]					=						load_numpy(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_img2img_superresolution_stage_II.npy'''	)
         assert_mean_pixel_difference(SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE	)





     def 				__lowerCAmelCase							(							self						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE	) ->				Optional[Any]:






         """simple docstring"""






         _start_torch_memory_measurement()

         A       :   List[str]					=						floats_tensor((1, 3, 64, 64)						,			rng=random.Random(0	)	).to(SCREAMING_SNAKE_CASE	)
         A       :   Dict					=						floats_tensor((1, 3, 64, 64)						,			rng=random.Random(1	)	).to(SCREAMING_SNAKE_CASE	)

         A       :   Optional[Any]					=						torch.Generator(device='''cpu'''	).manual_seed(0	)
         A       :   List[Any]					=						pipe_a(
             prompt_embeds=SCREAMING_SNAKE_CASE						,			negative_prompt_embeds=SCREAMING_SNAKE_CASE						,			image=SCREAMING_SNAKE_CASE						,			mask_image=SCREAMING_SNAKE_CASE						,			num_inference_steps=2						,			generator=SCREAMING_SNAKE_CASE						,			output_type='''np'''						,			)

         A       :   Any					=						output.images[0]

         assert image.shape == (64, 64, 3)

         A       :   Optional[Any]					=						torch.cuda.max_memory_allocated()
         assert mem_bytes < 10 * 10**9

         A       :   int					=						load_numpy(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_inpainting.npy'''	)
         assert_mean_pixel_difference(SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE	)

         # pipeline 2

         _start_torch_memory_measurement()

         A       :   int					=						torch.Generator(device='''cpu'''	).manual_seed(0	)

         A       :   Union[str, Any]					=						floats_tensor((1, 3, 64, 64)						,			rng=random.Random(0	)	).to(SCREAMING_SNAKE_CASE	)
         A       :   Any					=						floats_tensor((1, 3, 256, 256)						,			rng=random.Random(0	)	).to(SCREAMING_SNAKE_CASE	)
         A       :   Optional[int]					=						floats_tensor((1, 3, 256, 256)						,			rng=random.Random(1	)	).to(SCREAMING_SNAKE_CASE	)

         A       :   Dict					=						pipe_a(
             prompt_embeds=SCREAMING_SNAKE_CASE						,			negative_prompt_embeds=SCREAMING_SNAKE_CASE						,			image=SCREAMING_SNAKE_CASE						,			mask_image=SCREAMING_SNAKE_CASE						,			original_image=SCREAMING_SNAKE_CASE						,			generator=SCREAMING_SNAKE_CASE						,			num_inference_steps=2						,			output_type='''np'''						,			)

         A       :   Optional[int]					=						output.images[0]

         assert image.shape == (256, 256, 3)

         A       :   List[Any]					=						torch.cuda.max_memory_allocated()
         assert mem_bytes < 4 * 10**9

         A       :   Any					=						load_numpy(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_inpainting_superresolution_stage_II.npy'''	)
         assert_mean_pixel_difference(SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE	)







def 		lowerCAmelCase_				(   ):




    '''simple docstring'''



    torch.cuda.empty_cache()
    torch.cuda.reset_max_memory_allocated()
    torch.cuda.reset_peak_memory_stats()











from multiprocessing import Lock, Pipe, Process

# lock used to ensure that two processes do not access a pipe at the same time
A    :					Dict =      Lock()







def    __lowerCamelCase					(							__a   :Dict    ,     __a   :List[str]    ,     __a   :Optional[int]    ,     __a   :Optional[int]    ,     __a   :Optional[Any]    ,     __a   :Optional[int]    ,     __a   :int    )		->				Dict:






		"""simple docstring"""

		global process_lock

		# we perform n swaps since after n swaps we know we are sorted
		# we *could* stop early if we are sorted already, but it takes as long to
		# find out we are sorted as it does to sort the list with this algorithm
		for i in range(0    ,     1_0    ):
				if (i + position) % 2 == 0 and r_send is not None:
						# send your value to your right neighbor
						process_lock.acquire()
						r_send[1].send(__a    )
						process_lock.release()

						# receive your right neighbor's value
						process_lock.acquire()
						A__        =   rr_cv[0].recv()
						process_lock.release()

						# take the lower value since you are on the left
						A__        =   min(__a    ,     __a    )
				elif (i + position) % 2 != 0 and l_send is not None:
						# send your value to your left neighbor
						process_lock.acquire()
						l_send[1].send(__a    )
						process_lock.release()

						# receive your left neighbor's value
						process_lock.acquire()
						A__        =   lr_cv[0].recv()
						process_lock.release()

						# take the higher value since you are on the right
						A__        =   max(__a    ,     __a    )
    # after all swaps are performed, send the values back to main
		result_pipe[1].send(__a    )







def    __lowerCamelCase					(							__a   :List[str]    )		->				int:






		"""simple docstring"""

		A__        =   []
		A__        =   []
		# initialize the list of pipes where the values will be retrieved
		for _ in arr:
				result_pipe.append(Pipe()    )
		# creates the processes
		# the first and last process only have one neighbor so they are made outside
		# of the loop
		A__        =   Pipe()
		A__        =   Pipe()
		process_array_.append(
		    Process(
		        target=__a    ,     args=(0, arr[0], None, temp_rs, None, temp_rr, result_pipe[0])    ,     )    )
		A__        =   temp_rs
		A__        =   temp_rr

		for i in range(1    ,     len(__a    ) - 1    ):
				A__        =   Pipe()
				A__        =   Pipe()
				process_array_.append(
				    Process(
				        target=__a    ,     args=(i, arr[i], temp_ls, temp_rs, temp_lr, temp_rr, result_pipe[i])    ,     )    )
				A__        =   temp_rs
				A__        =   temp_rr

		process_array_.append(
		    Process(
		        target=__a    ,     args=(
		            len(__a    ) - 1,
		            arr[len(__a    ) - 1],
		            temp_ls,
		            None,
		            temp_lr,
		            None,
		            result_pipe[len(__a    ) - 1],
		        )    ,     )    )

		# start the processes
		for p in process_array_:
				p.start()

		# wait for the processes to end and write their values to the list
		for p in range(0    ,     len(__a    )    ):
				A__        =   result_pipe[p][0].recv()
				process_array_[p].join()
		return arr







def    __lowerCamelCase					(							)		->				str:






		"""simple docstring"""

		A__        =   list(range(1_0    ,     0    ,     -1    )    )
		print("""Initial List"""    )
		print(*__a    )
		A__        =   odd_even_transposition(__a    )
		print("""Sorted List\n"""    )
		print(*__a    )


if __name__ == "__main__":
					main()



