


"""simple docstring"""







import gc
import random
import unittest

import numpy as np
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModel,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import AutoencoderKL, DDIMScheduler, DDPMScheduler, StableUnCLIPImgaImgPipeline, UNetaDConditionModel
from diffusers.pipelines.pipeline_utils import DiffusionPipeline
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    floats_tensor,
    load_image,
    load_numpy,
    require_torch_gpu,
    skip_mps,
    slow,
    torch_device,
)

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()




class  UpperCamelCase_	( UpperCamelCase			,  UpperCamelCase			,  UpperCamelCase			,  unittest.TestCase):

	"""simple docstring"""





	snake_case__				:  Dict    					=			StableUnCLIPImgaImgPipeline
	snake_case__				:  Dict    					=			TEXT_GUIDED_IMAGE_VARIATION_PARAMS
	snake_case__				:  str    					=			TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
	snake_case__				:  Union[str, Any]    					=			frozenset(
	    [])  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
	snake_case__				:  Union[str, Any]    					=			frozenset([])




	def   UpperCAmelCase_      ( self			:				Tuple   )       ->	Dict:
						__SCREAMING_SNAKE_CASE						   =				3_2
						__SCREAMING_SNAKE_CASE						   =				embedder_hidden_size

						# image encoding components

						__SCREAMING_SNAKE_CASE						   =				CLIPImageProcessor(crop_size=3_2			,			size=3_2   )

						torch.manual_seed(0   )
						__SCREAMING_SNAKE_CASE						   =				CLIPVisionModelWithProjection(
						    CLIPVisionConfig(
						        hidden_size=UpperCAmelCase__			,			projection_dim=UpperCAmelCase__			,			num_hidden_layers=5			,			num_attention_heads=4			,			image_size=3_2			,			intermediate_size=3_7			,			patch_size=1			,			)   )

						# regular denoising components

						torch.manual_seed(0   )
						__SCREAMING_SNAKE_CASE						   =				StableUnCLIPImageNormalizer(embedding_dim=UpperCAmelCase__   )
						__SCREAMING_SNAKE_CASE						   =				DDPMScheduler(beta_schedule="squaredcos_cap_v2"   )

						torch.manual_seed(0   )
						__SCREAMING_SNAKE_CASE						   =				CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"   )

						torch.manual_seed(0   )
						__SCREAMING_SNAKE_CASE						   =				CLIPTextModel(
						    CLIPTextConfig(
						        bos_token_id=0			,			eos_token_id=2			,			hidden_size=UpperCAmelCase__			,			projection_dim=3_2			,			intermediate_size=3_7			,			layer_norm_eps=1E-05			,			num_attention_heads=4			,			num_hidden_layers=5			,			pad_token_id=1			,			vocab_size=1_0_0_0			,			)   )

						torch.manual_seed(0   )
						__SCREAMING_SNAKE_CASE						   =				UNetaDConditionModel(
						    sample_size=3_2			,			in_channels=4			,			out_channels=4			,			down_block_types=("CrossAttnDownBlock2D", "DownBlock2D")			,			up_block_types=("UpBlock2D", "CrossAttnUpBlock2D")			,			block_out_channels=(3_2, 6_4)			,			attention_head_dim=(2, 4)			,			class_embed_type="projection"			,			projection_class_embeddings_input_dim=embedder_projection_dim * 2			,			cross_attention_dim=UpperCAmelCase__			,			layers_per_block=1			,			upcast_attention=UpperCAmelCase__			,			use_linear_projection=UpperCAmelCase__			,			)

						torch.manual_seed(0   )
						__SCREAMING_SNAKE_CASE						   =				DDIMScheduler(
						    beta_schedule="scaled_linear"			,			beta_start=0.00_085			,			beta_end=0.012			,			prediction_type="v_prediction"			,			set_alpha_to_one=UpperCAmelCase__			,			steps_offset=1			,			)

						torch.manual_seed(0   )
						__SCREAMING_SNAKE_CASE						   =				AutoencoderKL()

						__SCREAMING_SNAKE_CASE						   =				{
						    # image encoding components
						    "feature_extractor": feature_extractor,
						    "image_encoder": image_encoder.eval(),
						    # image noising components
						    "image_normalizer": image_normalizer.eval(),
						    "image_noising_scheduler": image_noising_scheduler,
						    # regular denoising components
						    "tokenizer": tokenizer,
						    "text_encoder": text_encoder.eval(),
						    "unet": unet.eval(),
						    "scheduler": scheduler,
						    "vae": vae.eval(),
						}

						return components




	def   UpperCAmelCase_      ( self			:				str			,			UpperCAmelCase__			:				Any			,			UpperCAmelCase__			:				int=0			,			UpperCAmelCase__			:				Union[str, Any]=True   )       ->	Optional[int]:
						if str(UpperCAmelCase__   ).startswith("mps"   ):
											__SCREAMING_SNAKE_CASE						   =				torch.manual_seed(UpperCAmelCase__   )
						else:
											__SCREAMING_SNAKE_CASE						   =				torch.Generator(device=UpperCAmelCase__   ).manual_seed(UpperCAmelCase__   )

						__SCREAMING_SNAKE_CASE						   =				floats_tensor((1, 3, 3_2, 3_2)			,			rng=random.Random(UpperCAmelCase__   )   ).to(UpperCAmelCase__   )

						if pil_image:
											__SCREAMING_SNAKE_CASE						   =				input_image * 0.5 + 0.5
											__SCREAMING_SNAKE_CASE						   =				input_image.clamp(0			,			1   )
											__SCREAMING_SNAKE_CASE						   =				input_image.cpu().permute(0			,			2			,			3			,			1   ).float().numpy()
											__SCREAMING_SNAKE_CASE						   =				DiffusionPipeline.numpy_to_pil(UpperCAmelCase__   )[0]

						return {
						    "prompt": "An anime racoon running a marathon",
						    "image": input_image,
						    "generator": generator,
						    "num_inference_steps": 2,
						    "output_type": "np",
						}




	@skip_mps
	def   UpperCAmelCase_      ( self			:				Optional[int]   )       ->	List[str]:
						__SCREAMING_SNAKE_CASE						   =				"cpu"  # ensure determinism for the device-dependent torch.Generator
						__SCREAMING_SNAKE_CASE						   =				self.get_dummy_components()
						__SCREAMING_SNAKE_CASE						   =				StableUnCLIPImgaImgPipeline(**UpperCAmelCase__   )
						__SCREAMING_SNAKE_CASE						   =				sd_pipe.to(UpperCAmelCase__   )
						sd_pipe.set_progress_bar_config(disable=UpperCAmelCase__   )

						__SCREAMING_SNAKE_CASE						   =				self.get_dummy_inputs(UpperCAmelCase__   )
						inputs.update({"image_embeds": None}   )
						__SCREAMING_SNAKE_CASE						   =				sd_pipe(**UpperCAmelCase__   ).images
						__SCREAMING_SNAKE_CASE						   =				image[0, -3:, -3:, -1]

						assert image.shape == (1, 3_2, 3_2, 3)
						__SCREAMING_SNAKE_CASE						   =				np.array([0.3_872, 0.7_224, 0.5_601, 0.4_741, 0.6_872, 0.5_814, 0.4_636, 0.3_867, 0.5_078]   )

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




	def   UpperCAmelCase_      ( self			:				Optional[int]   )       ->	Union[str, Any]:
						__SCREAMING_SNAKE_CASE						   =				torch_device in ["cpu", "mps"]

						self._test_attention_slicing_forward_pass(test_max_difference=UpperCAmelCase__   )




	def   UpperCAmelCase_      ( self			:				Optional[Any]   )       ->	Union[str, Any]:
						__SCREAMING_SNAKE_CASE						   =				torch_device in ["cpu", "mps"]

						self._test_inference_batch_single_identical(test_max_difference=UpperCAmelCase__   )




	@unittest.skipIf(
	    torch_device != "cuda" or not is_xformers_available()			,			reason="XFormers attention is only available with CUDA and `xformers` installed"			,			)
	def   UpperCAmelCase_      ( self			:				Any   )       ->	Any:
						self._test_xformers_attention_forwardGenerator_pass(test_max_difference=UpperCAmelCase__   )




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

	"""simple docstring"""





	def   UpperCAmelCase_      ( self			:				Optional[Any]   )       ->	int:
						# clean up the VRAM after each test
						super().tearDown()
						gc.collect()
						torch.cuda.empty_cache()




	def   UpperCAmelCase_      ( self			:				Any   )       ->	List[str]:
						__SCREAMING_SNAKE_CASE						   =				load_image(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"   )

						__SCREAMING_SNAKE_CASE						   =				load_numpy(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_img2img_anime_turtle_fp16.npy"   )

						__SCREAMING_SNAKE_CASE						   =				StableUnCLIPImgaImgPipeline.from_pretrained(
						    "fusing/stable-unclip-2-1-l-img2img"			,			torch_dtype=torch.floataa   )
						pipe.to(UpperCAmelCase__   )
						pipe.set_progress_bar_config(disable=UpperCAmelCase__   )
						# stable unclip will oom when integration tests are run on a V100,
						# so turn on memory savings
						pipe.enable_attention_slicing()
						pipe.enable_sequential_cpu_offload()

						__SCREAMING_SNAKE_CASE						   =				torch.Generator(device="cpu"   ).manual_seed(0   )
						__SCREAMING_SNAKE_CASE						   =				pipe(UpperCAmelCase__			,			"anime turle"			,			generator=UpperCAmelCase__			,			output_type="np"   )

						__SCREAMING_SNAKE_CASE						   =				output.images[0]

						assert image.shape == (7_6_8, 7_6_8, 3)

						assert_mean_pixel_difference(UpperCAmelCase__			,			UpperCAmelCase__   )




	def   UpperCAmelCase_      ( self			:				Dict   )       ->	Dict:
						__SCREAMING_SNAKE_CASE						   =				load_image(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"   )

						__SCREAMING_SNAKE_CASE						   =				load_numpy(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_h_img2img_anime_turtle_fp16.npy"   )

						__SCREAMING_SNAKE_CASE						   =				StableUnCLIPImgaImgPipeline.from_pretrained(
						    "fusing/stable-unclip-2-1-h-img2img"			,			torch_dtype=torch.floataa   )
						pipe.to(UpperCAmelCase__   )
						pipe.set_progress_bar_config(disable=UpperCAmelCase__   )
						# stable unclip will oom when integration tests are run on a V100,
						# so turn on memory savings
						pipe.enable_attention_slicing()
						pipe.enable_sequential_cpu_offload()

						__SCREAMING_SNAKE_CASE						   =				torch.Generator(device="cpu"   ).manual_seed(0   )
						__SCREAMING_SNAKE_CASE						   =				pipe(UpperCAmelCase__			,			"anime turle"			,			generator=UpperCAmelCase__			,			output_type="np"   )

						__SCREAMING_SNAKE_CASE						   =				output.images[0]

						assert image.shape == (7_6_8, 7_6_8, 3)

						assert_mean_pixel_difference(UpperCAmelCase__			,			UpperCAmelCase__   )




	def   UpperCAmelCase_      ( self			:				int   )       ->	Any:
						__SCREAMING_SNAKE_CASE						   =				load_image(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"   )

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

						__SCREAMING_SNAKE_CASE						   =				StableUnCLIPImgaImgPipeline.from_pretrained(
						    "fusing/stable-unclip-2-1-h-img2img"			,			torch_dtype=torch.floataa   )
						__SCREAMING_SNAKE_CASE						   =				pipe.to(UpperCAmelCase__   )
						pipe.set_progress_bar_config(disable=UpperCAmelCase__   )
						pipe.enable_attention_slicing()
						pipe.enable_sequential_cpu_offload()

						__SCREAMING_SNAKE_CASE						   =				pipe(
						    UpperCAmelCase__			,			"anime turtle"			,			num_inference_steps=2			,			output_type="np"			,			)

						__SCREAMING_SNAKE_CASE						   =				torch.cuda.max_memory_allocated()
						# make sure that less than 7 GB is allocated
						assert mem_bytes < 7 * 1_0**9







from collections import deque


class   A			:
			'''simple docstring'''




			def __init__(			self   :    List[str]							,	__lowerCAmelCase   :    str							,	__lowerCAmelCase   :    int							,	__lowerCAmelCase   :    int   )	->	None:







					"""simple docstring"""






					A__        =   process_name  # process name
					A__        =   arrival_time  # arrival time of the process
					# completion time of finished process or last interrupted time
					A__        =   arrival_time
					A__        =   burst_time  # remaining burst time
					A__        =   0  # total time of the process wait in ready queue
					A__        =   0  # time from arrival time to completion time


class   A			:
			'''simple docstring'''




			def __init__(			self   :    Tuple							,	__lowerCAmelCase   :    int							,	__lowerCAmelCase   :    list[int]							,	__lowerCAmelCase   :    deque[Process]							,	__lowerCAmelCase   :    int							,	)	->	None:







					"""simple docstring"""






					A__        =   number_of_queues
					# time slice of queues that round robin algorithm applied
					A__        =   time_slices
					# unfinished process is in this ready_queue
					A__        =   queue
					# current time
					A__        =   current_time
					# finished process is in this sequence queue
					A__        =   deque()






			def    a_       (			self   :    Dict   )	->	list[str]:







					"""simple docstring"""






					A__        =   []
					for i in range(len(self.finish_queue   )   ):
							sequence.append(self.finish_queue[i].process_name   )
					return sequence






			def    a_       (			self   :    Tuple							,	__lowerCAmelCase   :    list[Process]   )	->	list[int]:







					"""simple docstring"""






					A__        =   []
					for i in range(len(__lowerCAmelCase   )   ):
							waiting_times.append(queue[i].waiting_time   )
					return waiting_times






			def    a_       (			self   :    Optional[Any]							,	__lowerCAmelCase   :    list[Process]   )	->	list[int]:







					"""simple docstring"""






					A__        =   []
					for i in range(len(__lowerCAmelCase   )   ):
							turnaround_times.append(queue[i].turnaround_time   )
					return turnaround_times






			def    a_       (			self   :    Dict							,	__lowerCAmelCase   :    list[Process]   )	->	list[int]:







					"""simple docstring"""






					A__        =   []
					for i in range(len(__lowerCAmelCase   )   ):
							completion_times.append(queue[i].stop_time   )
					return completion_times






			def    a_       (			self   :    int							,	__lowerCAmelCase   :    deque[Process]   )	->	list[int]:







					"""simple docstring"""






					return [q.burst_time for q in queue]






			def    a_       (			self   :    Any							,	__lowerCAmelCase   :    Process   )	->	int:







					"""simple docstring"""






					process.waiting_time += self.current_time - process.stop_time
					return process.waiting_time






			def    a_       (			self   :    Union[str, Any]							,	__lowerCAmelCase   :    deque[Process]   )	->	deque[Process]:







					"""simple docstring"""






					A__        =   deque()  # sequence deque of finished process
					while len(__lowerCAmelCase   ) != 0:
							A__        =   ready_queue.popleft()  # current process

							# if process's arrival time is later than current time, update current time
							if self.current_time < cp.arrival_time:
									self.current_time += cp.arrival_time

							# update waiting time of current process
							self.update_waiting_time(__lowerCAmelCase   )
							# update current time
							self.current_time += cp.burst_time
							# finish the process and set the process's burst-time 0
							A__        =   0
							# set the process's turnaround time because it is finished
							A__        =   self.current_time - cp.arrival_time
							# set the completion time
							A__        =   self.current_time
							# add the process to queue that has finished queue
							finished.append(__lowerCAmelCase   )

					self.finish_queue.extend(__lowerCAmelCase   )  # add finished process to finish queue
					# FCFS will finish all remaining processes
					return finished






			def    a_       (			self   :    Optional[Any]							,	__lowerCAmelCase   :    deque[Process]							,	__lowerCAmelCase   :    int   )	->	tuple[deque[Process], deque[Process]]:







					"""simple docstring"""






					A__        =   deque()  # sequence deque of terminated process
					# just for 1 cycle and unfinished processes will go back to queue
					for _ in range(len(__lowerCAmelCase   )   ):
							A__        =   ready_queue.popleft()  # current process

							# if process's arrival time is later than current time, update current time
							if self.current_time < cp.arrival_time:
									self.current_time += cp.arrival_time

							# update waiting time of unfinished processes
							self.update_waiting_time(__lowerCAmelCase   )
							# if the burst time of process is bigger than time-slice
							if cp.burst_time > time_slice:
									# use CPU for only time-slice
									self.current_time += time_slice
									# update remaining burst time
									cp.burst_time -= time_slice
									# update end point time
									A__        =   self.current_time
									# locate the process behind the queue because it is not finished
									ready_queue.append(__lowerCAmelCase   )
							else:
									# use CPU for remaining burst time
									self.current_time += cp.burst_time
									# set burst time 0 because the process is finished
									A__        =   0
									# set the finish time
									A__        =   self.current_time
									# update the process' turnaround time because it is finished
									A__        =   self.current_time - cp.arrival_time
									# add the process to queue that has finished queue
									finished.append(__lowerCAmelCase   )

					self.finish_queue.extend(__lowerCAmelCase   )  # add finished process to finish queue
					# return finished processes queue and remaining processes queue
					return finished, ready_queue







			def    a_       (			self   :    List[Any]   )	->	deque[Process]:







					"""simple docstring"""






					for i in range(self.number_of_queues - 1   ):
							A__      ,			A__        =   self.round_robin(
							    self.ready_queue							,	self.time_slices[i]   )
					#  the last queue has first_come_first_served algorithm
					self.first_come_first_served(self.ready_queue   )

					return self.finish_queue


if __name__ == "__main__":
					import doctest

					A    :					Union[str, Any] =      Process('''P1''', 0, 5_3)
					A    :					Optional[Any] =      Process('''P2''', 0, 1_7)
					A    :					Optional[int] =      Process('''P3''', 0, 6_8)
					A    :					int =      Process('''P4''', 0, 2_4)
					A    :					Any =      3
					A    :					List[Any] =      [1_7, 2_5]
					A    :					Optional[Any] =      deque([Pa, Pa, Pa, Pa])

					if len(time_slices) != number_of_queues - 1:
										raise SystemExit(0)

					doctest.testmod(extraglobs={'''queue''': deque([Pa, Pa, Pa, Pa])})

					A    :					Optional[Any] =      Process('''P1''', 0, 5_3)
					A    :					int =      Process('''P2''', 0, 1_7)
					A    :					Optional[int] =      Process('''P3''', 0, 6_8)
					A    :					Tuple =      Process('''P4''', 0, 2_4)
					A    :					Union[str, Any] =      3
					A    :					Optional[Any] =      [1_7, 2_5]
					A    :					Tuple =      deque([Pa, Pa, Pa, Pa])
					A    :					Optional[int] =      MLFQ(number_of_queues, time_slices, queue, 0)
					A    :					Dict =      mlfq.multi_level_feedback_queue()

					# print total waiting times of processes(P1, P2, P3, P4)
					print(
					    F'''waiting time:\
        \t\t\t{MLFQ.calculate_waiting_time(mlfq, [Pa, Pa, Pa, Pa])}'''
					)
					# print completion times of processes(P1, P2, P3, P4)
					print(
					    F'''completion time:\
        \t\t{MLFQ.calculate_completion_time(mlfq, [Pa, Pa, Pa, Pa])}'''
					)
					# print total turnaround times of processes(P1, P2, P3, P4)
					print(
					    F'''turnaround time:\
        \t\t{MLFQ.calculate_turnaround_time(mlfq, [Pa, Pa, Pa, Pa])}'''
					)
					# print sequence of finished processes
					print(
					    F'''sequence of finished processes:\
        {mlfq.calculate_sequence_of_finish_queue()}'''
					)



