



'''simple docstring'''



import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DPMSolverMultistepScheduler,
    TextToVideoSDPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import is_xformers_available, load_numpy, skip_mps, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

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


enable_full_determinism()





@skip_mps
class _UpperCamelCase		(  a__   ,							unittest.TestCase		):





				'''simple docstring'''





				lowerCAmelCase__            =     TextToVideoSDPipeline
				lowerCAmelCase__            =     TEXT_TO_IMAGE_PARAMS
				lowerCAmelCase__            =     TEXT_TO_IMAGE_BATCH_PARAMS
				# No `output_type`.
				lowerCAmelCase__            =     frozenset(
				    [
				        """num_inference_steps""",
				        """generator""",
				        """latents""",
				        """return_dict""",
				        """callback""",
				        """callback_steps""",
				    ]		)

				def 					__lowerCamelCase  (				self	:	Dict):







											'''simple docstring'''

											torch.manual_seed(0)
											__lowercase													=UNetaDConditionModel(
											    block_out_channels=(3_2, 6_4, 6_4, 6_4)	,							layers_per_block=2	,							sample_size=3_2	,							in_channels=4	,							out_channels=4	,							down_block_types=('CrossAttnDownBlock3D', 'CrossAttnDownBlock3D', 'CrossAttnDownBlock3D', 'DownBlock3D')	,							up_block_types=('UpBlock3D', 'CrossAttnUpBlock3D', 'CrossAttnUpBlock3D', 'CrossAttnUpBlock3D')	,							cross_attention_dim=3_2	,							attention_head_dim=4	,							)
											__lowercase													=DDIMScheduler(
											    beta_start=0.0_0085	,							beta_end=0.012	,							beta_schedule='scaled_linear'	,							clip_sample=lowerCAmelCase__	,							set_alpha_to_one=lowerCAmelCase__	,							)
											torch.manual_seed(0)
											__lowercase													=AutoencoderKL(
											    block_out_channels=[3_2, 6_4]	,							in_channels=3	,							out_channels=3	,							down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']	,							up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']	,							latent_channels=4	,							sample_size=1_2_8	,							)
											torch.manual_seed(0)
											__lowercase													=CLIPTextConfig(
											    bos_token_id=0	,							eos_token_id=2	,							hidden_size=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	,							hidden_act='gelu'	,							projection_dim=5_1_2	,							)
											__lowercase													=CLIPTextModel(lowerCAmelCase__)
											__lowercase													=CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip')

											__lowercase													={
											    "unet": unet,
											    "scheduler": scheduler,
											    "vae": vae,
											    "text_encoder": text_encoder,
											    "tokenizer": tokenizer,
											}
											return components

				def 					__lowerCamelCase  (				self	:	Optional[int]	,							_lowerCAmelCase	:	Any	,							_lowerCAmelCase	:	Tuple=0):







											'''simple docstring'''

											if str(lowerCAmelCase__).startswith('mps'):
																		__lowercase													=torch.manual_seed(lowerCAmelCase__)
											else:
																		__lowercase													=torch.Generator(device=lowerCAmelCase__).manual_seed(lowerCAmelCase__)
											__lowercase													={
											    "prompt": "A painting of a squirrel eating a burger",
											    "generator": generator,
											    "num_inference_steps": 2,
											    "guidance_scale": 6.0,
											    "output_type": "pt",
											}
											return inputs

				def 					__lowerCamelCase  (				self	:	int):







											'''simple docstring'''

											__lowercase													="cpu"  # ensure determinism for the device-dependent torch.Generator
											__lowercase													=self.get_dummy_components()
											__lowercase													=TextToVideoSDPipeline(**lowerCAmelCase__)
											__lowercase													=sd_pipe.to(lowerCAmelCase__)
											sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__)

											__lowercase													=self.get_dummy_inputs(lowerCAmelCase__)
											__lowercase													="np"
											__lowercase													=sd_pipe(**lowerCAmelCase__).frames
											__lowercase													=frames[0][-3:, -3:, -1]

											assert frames[0].shape == (6_4, 6_4, 3)
											__lowercase													=np.array([158.0, 160.0, 153.0, 125.0, 100.0, 121.0, 111.0, 93.0, 113.0])

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

				def 					__lowerCamelCase  (				self	:	Dict):







											'''simple docstring'''

											self._test_attention_slicing_forward_pass(test_mean_pixel_difference=lowerCAmelCase__	,							expected_max_diff=3e-3)

				@unittest.skipIf(
				    torch_device != 'cuda' or not is_xformers_available()	,							reason='XFormers attention is only available with CUDA and `xformers` installed'	,							)
				def 					__lowerCamelCase  (				self	:	Optional[Any]):







											'''simple docstring'''

											self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=lowerCAmelCase__	,							expected_max_diff=1e-2)

				@unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.')
				def 					__lowerCamelCase  (				self	:	Optional[int]):







											'''simple docstring'''

											pass

				@unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.')
				def 					__lowerCamelCase  (				self	:	int):







											'''simple docstring'''

											pass

				@unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.')
				def 					__lowerCamelCase  (				self	:	Tuple):







											'''simple docstring'''

											pass

				def 					__lowerCamelCase  (				self	:	Tuple):







											'''simple docstring'''

											return super().test_progress_bar()





@slow
@skip_mps
class _UpperCamelCase		(  unittest.TestCase		):





				'''simple docstring'''





				def 					__lowerCamelCase  (				self	:	List[str]):







											'''simple docstring'''

											__lowercase													=load_numpy(
											    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy')

											__lowercase													=TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b')
											__lowercase													=DPMSolverMultistepScheduler.from_config(pipe.scheduler.config)
											__lowercase													=pipe.to('cuda')

											__lowercase													="Spiderman is surfing"
											__lowercase													=torch.Generator(device='cpu').manual_seed(0)

											__lowercase													=pipe(lowerCAmelCase__	,							generator=lowerCAmelCase__	,							num_inference_steps=2_5	,							output_type='pt').frames
											__lowercase													=video_frames.cpu().numpy()

											assert np.abs(expected_video - video).mean() < 5e-2

				def 					__lowerCamelCase  (				self	:	int):







											'''simple docstring'''

											__lowercase													=load_numpy(
											    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy')

											__lowercase													=TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b')
											__lowercase													=pipe.to('cuda')

											__lowercase													="Spiderman is surfing"
											__lowercase													=torch.Generator(device='cpu').manual_seed(0)

											__lowercase													=pipe(lowerCAmelCase__	,							generator=lowerCAmelCase__	,							num_inference_steps=2	,							output_type='pt').frames
											__lowercase													=video_frames.cpu().numpy()

											assert np.abs(expected_video - video).mean() < 5e-2










"""simple docstring"""




from __future__ import annotations

from statistics import mean






def  _SCREAMING_SNAKE_CASE						(      _lowercase		:       list[int]       ,			_lowercase		:       list[int]       ,			_lowercase		:       int   )					->list[int]:






	'''simple docstring'''






	a :						Union[str, Any]      =							[0] * no_of_processes
	a :						int      =							[0] * no_of_processes

	# Initialize remaining_time to waiting_time.

	for i in range(_lowercase   ):
		a :						List[Any]      =							burst_time[i]
	a :						list[int]      =							[]

	a :						List[Any]      =							0
	a :						Any      =							0

	# When processes are not completed,
	# A process whose arrival time has passed \
	# and has remaining execution time is put into the ready_process.
	# The shortest process in the ready_process, target_process is executed.

	while completed != no_of_processes:
		a :						str      =							[]
		a :						Tuple      =							-1

		for i in range(_lowercase   ):
			if (arrival_time[i] <= total_time) and (remaining_time[i] > 0):
				ready_process.append(_lowercase   )

		if len(_lowercase   ) > 0:
			a :						List[Any]      =							ready_process[0]
			for i in ready_process:
				if remaining_time[i] < remaining_time[target_process]:
					a :						Optional[Any]      =							i
			total_time += burst_time[target_process]
			completed += 1
			a :						str      =							0
			a :						Union[str, Any]      =							(
			    total_time - arrival_time[target_process] - burst_time[target_process]
			)
		else:
			total_time += 1

	return waiting_time






def  _SCREAMING_SNAKE_CASE						(      _lowercase		:       list[int]       ,			_lowercase		:       int       ,			_lowercase		:       list[int]   )					->list[int]:






	'''simple docstring'''






	a :						List[str]      =							[0] * no_of_processes
	for i in range(_lowercase   ):
		a :						Optional[int]      =							burst_time[i] + waiting_time[i]
	return turn_around_time


if __name__ == "__main__":
		print('''[TEST CASE 01]''')

		a		:					Dict	      =     4
		a		:					Tuple	      =     [2, 5, 3, 7]
		a		:					List[Any]	      =     [0, 0, 0, 0]
		a		:					int	      =     calculate_waitingtime(arrival_time, burst_time, no_of_processes)
		a		:					List[Any]	      =     calculate_turnaroundtime(
		    burst_time, no_of_processes, waiting_time
		)

		# Printing the Result
		print('''PID\tBurst Time\tArrival Time\tWaiting Time\tTurnaround Time''')
		for i, process_id in enumerate(list(range(1, 5))):
				print(
				    F'''{process_id}\t{burst_time[i]}\t\t\t{arrival_time[i]}\t\t\t\t'''
				    F'''{waiting_time[i]}\t\t\t\t{turn_around_time[i]}'''
				)
		print(F'''\nAverage waiting time = {mean(waiting_time):.5f}''')
		print(F'''Average turnaround time = {mean(turn_around_time):.5f}''')


