






'''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 	SCREAMING_SNAKE_CASE	(      lowerCAmelCase_    ,   unittest.TestCase  ):
       snake_case__    :     Any   		=			TextToVideoSDPipeline
       snake_case__    :     Dict   		=			TEXT_TO_IMAGE_PARAMS
       snake_case__    :     Union[str, Any]   		=			TEXT_TO_IMAGE_BATCH_PARAMS
       # No `output_type`.
       snake_case__    :     Optional[int]   		=			frozenset(
           [
               'num_inference_steps',
               'generator',
               'latents',
               'return_dict',
               'callback',
               'callback_steps',
           ]  )




       def 					a_					(					self      : Any					):


            """simple docstring"""







            torch.manual_seed(0					)
            __lowerCamelCase						:       Optional[int]				   =				UNetaDConditionModel(
                block_out_channels=(32, 64, 64, 64)	,    layers_per_block=2	,    sample_size=32	,    in_channels=4	,    out_channels=4	,    down_block_types=("""CrossAttnDownBlock3D""", """CrossAttnDownBlock3D""", """CrossAttnDownBlock3D""", """DownBlock3D""")	,    up_block_types=("""UpBlock3D""", """CrossAttnUpBlock3D""", """CrossAttnUpBlock3D""", """CrossAttnUpBlock3D""")	,    cross_attention_dim=32	,    attention_head_dim=4	,    )
            __lowerCamelCase						:       Union[str, Any]				   =				DDIMScheduler(
                beta_start=0.0_0085	,    beta_end=0.012	,    beta_schedule="""scaled_linear"""	,    clip_sample=A__	,    set_alpha_to_one=A__	,    )
            torch.manual_seed(0					)
            __lowerCamelCase						:       Union[str, Any]				   =				AutoencoderKL(
                block_out_channels=[32, 64]	,    in_channels=3	,    out_channels=3	,    down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]	,    up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]	,    latent_channels=4	,    sample_size=128	,    )
            torch.manual_seed(0					)
            __lowerCamelCase						:       Tuple				   =				CLIPTextConfig(
                bos_token_id=0	,    eos_token_id=2	,    hidden_size=32	,    intermediate_size=37	,    layer_norm_eps=1e-0_5	,    num_attention_heads=4	,    num_hidden_layers=5	,    pad_token_id=1	,    vocab_size=1000	,    hidden_act="""gelu"""	,    projection_dim=512	,    )
            __lowerCamelCase						:       Optional[int]				   =				CLIPTextModel(A__					)
            __lowerCamelCase						:       str				   =				CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""					)

            __lowerCamelCase						:       List[str]				   =				{
                """unet""": unet,
                """scheduler""": scheduler,
                """vae""": vae,
                """text_encoder""": text_encoder,
                """tokenizer""": tokenizer,
            }
            return components




       def 					a_					(					self      : int	,    A__      : int	,    A__      : Dict=0					):


            """simple docstring"""







            if str(A__					).startswith("""mps"""					):
                 __lowerCamelCase						:       Optional[Any]				   =				torch.manual_seed(A__					)
            else:
                 __lowerCamelCase						:       List[str]				   =				torch.Generator(device=A__					).manual_seed(A__					)
            __lowerCamelCase						:       Tuple				   =				{
                """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 					a_					(					self      : List[Any]					):


            """simple docstring"""







            __lowerCamelCase						:       Dict				   =				"""cpu"""  # ensure determinism for the device-dependent torch.Generator
            __lowerCamelCase						:       int				   =				self.get_dummy_components()
            __lowerCamelCase						:       Optional[Any]				   =				TextToVideoSDPipeline(**A__					)
            __lowerCamelCase						:       Union[str, Any]				   =				sd_pipe.to(A__					)
            sd_pipe.set_progress_bar_config(disable=A__					)

            __lowerCamelCase						:       int				   =				self.get_dummy_inputs(A__					)
            __lowerCamelCase						:       List[str]				   =				"""np"""
            __lowerCamelCase						:       Dict				   =				sd_pipe(**A__					).frames
            __lowerCamelCase						:       Any				   =				frames[0][-3:, -3:, -1]

            assert frames[0].shape == (64, 64, 3)
            __lowerCamelCase						:       Union[str, Any]				   =				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 					a_					(					self      : List[str]					):


            """simple docstring"""







            self._test_attention_slicing_forward_pass(test_mean_pixel_difference=A__	,    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 					a_					(					self      : int					):


            """simple docstring"""







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




       @unittest.skip(reason="""Batching needs to be properly figured out first for this pipeline."""					)
       def 					a_					(					self      : List[str]					):


            """simple docstring"""







            pass




       @unittest.skip(reason="""Batching needs to be properly figured out first for this pipeline."""					)
       def 					a_					(					self      : Tuple					):


            """simple docstring"""







            pass




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


            """simple docstring"""







            pass




       def 					a_					(					self      : Tuple					):


            """simple docstring"""







            return super().test_progress_bar()







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




       def 					a_					(					self      : Dict					):


            """simple docstring"""







            __lowerCamelCase						:       Optional[int]				   =				load_numpy(
                """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy"""					)

            __lowerCamelCase						:       Tuple				   =				TextToVideoSDPipeline.from_pretrained("""damo-vilab/text-to-video-ms-1.7b"""					)
            __lowerCamelCase						:       List[Any]				   =				DPMSolverMultistepScheduler.from_config(pipe.scheduler.config					)
            __lowerCamelCase						:       int				   =				pipe.to("""cuda"""					)

            __lowerCamelCase						:       str				   =				"""Spiderman is surfing"""
            __lowerCamelCase						:       Union[str, Any]				   =				torch.Generator(device="""cpu"""					).manual_seed(0					)

            __lowerCamelCase						:       List[str]				   =				pipe(A__	,    generator=A__	,    num_inference_steps=25	,    output_type="""pt"""					).frames
            __lowerCamelCase						:       Union[str, Any]				   =				video_frames.cpu().numpy()

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




       def 					a_					(					self      : int					):


            """simple docstring"""







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

            __lowerCamelCase						:       str				   =				TextToVideoSDPipeline.from_pretrained("""damo-vilab/text-to-video-ms-1.7b"""					)
            __lowerCamelCase						:       int				   =				pipe.to("""cuda"""					)

            __lowerCamelCase						:       Union[str, Any]				   =				"""Spiderman is surfing"""
            __lowerCamelCase						:       Optional[Any]				   =				torch.Generator(device="""cpu"""					).manual_seed(0					)

            __lowerCamelCase						:       Optional[int]				   =				pipe(A__	,    generator=A__	,    num_inference_steps=2	,    output_type="""pt"""					).frames
            __lowerCamelCase						:       Tuple				   =				video_frames.cpu().numpy()

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











'''simple docstring'''

from statistics import mean

import numpy as np
def     __lowercase  (_lowercase,						_lowercase,						_lowercase,						_lowercase     )   ->     list:


     """simple docstring"""
     __lowerCamelCase						:       str				   =				0
     # Number of processes finished
     __lowerCamelCase						:       List[Any]				   =				0
     # Displays the finished process.
     # If it is 0, the performance is completed if it is 1, before the performance.
     __lowerCamelCase						:       int				   =				[0] * no_of_process
     # List to include calculation results
     __lowerCamelCase						:       Optional[int]				   =				[0] * no_of_process

     # Sort by arrival time.
     __lowerCamelCase						:       int				   =				[burst_time[i] for i in np.argsort(_lowercase     )]
     __lowerCamelCase						:       Union[str, Any]				   =				[process_name[i] for i in np.argsort(_lowercase     )]
     arrival_time.sort()

     while no_of_process > finished_process_count:
          __lowerCamelCase						:       Tuple				   =				0
          while finished_process[i] == 1:
               i += 1
          if current_time < arrival_time[i]:
               __lowerCamelCase						:       List[str]				   =				arrival_time[i]

          __lowerCamelCase						:       Dict				   =				0
          # Index showing the location of the process being performed
          __lowerCamelCase						:       Optional[Any]				   =				0
          # Saves the current response ratio.
          __lowerCamelCase						:       List[str]				   =				0
          for i in range(0,						_lowercase     ):
               if finished_process[i] == 0 and arrival_time[i] <= current_time:
                    __lowerCamelCase						:       Optional[int]				   =				(burst_time[i] + (current_time - arrival_time[i])) / burst_time[
                        i
                    ]
               if response_ratio < temp:
                    __lowerCamelCase						:       List[Any]				   =				temp
                    __lowerCamelCase						:       str				   =				i

        # Calculate the turn around time
          __lowerCamelCase						:       Optional[Any]				   =				current_time + burst_time[loc] - arrival_time[loc]
          current_time += burst_time[loc]
          # Indicates that the process has been performed.
          __lowerCamelCase						:       Optional[Any]				   =				1
          # Increase finished_process_count by 1
          finished_process_count += 1

     return turn_around_time
def     __lowercase  (_lowercase,						_lowercase,						_lowercase,						_lowercase     )   ->     list:


     """simple docstring"""
     __lowerCamelCase						:       List[Any]				   =				[0] * no_of_process
     for i in range(0,						_lowercase     ):
          __lowerCamelCase						:       Tuple				   =				turn_around_time[i] - burst_time[i]
     return waiting_time


if __name__ == "__main__":
   UpperCAmelCase__						:List[Any]          =			5
   UpperCAmelCase__						:Tuple          =			["""A""", """B""", """C""", """D""", """E"""]
   UpperCAmelCase__						:Tuple          =			[1, 2, 3, 4, 5]
   UpperCAmelCase__						:List[Any]          =			[1, 2, 3, 4, 5]

   UpperCAmelCase__						:Union[str, Any]          =			calculate_turn_around_time(
       process_name, arrival_time, burst_time, no_of_process
   )
   UpperCAmelCase__						:Optional[int]          =			calculate_waiting_time(
       process_name, turn_around_time, burst_time, no_of_process
   )

   print("""Process name \tArrival time \tBurst time \tTurn around time \tWaiting time""")
   for i in range(0, no_of_process):
      print(
          f'''{process_name[i]}\t\t{arrival_time[i]}\t\t{burst_time[i]}\t\t'''
          f'''{turn_around_time[i]}\t\t\t{waiting_time[i]}'''
      )

   print(f'''average waiting time : {mean(waiting_time):.5f}''')
   print(f'''average turn around time : {mean(turn_around_time):.5f}''')



