






import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionSAGPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

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


enable_full_determinism()

class        lowerCAmelCase_				(      lowercase						,      lowercase						,      unittest.TestCase	):







				"""simple docstring"""




				_snake_case				:   Dict        =			StableDiffusionSAGPipeline
				_snake_case				:   Union[str, Any]        =			TEXT_TO_IMAGE_PARAMS
				_snake_case				:   int        =			TEXT_TO_IMAGE_BATCH_PARAMS
				_snake_case				:   List[str]        =			TEXT_TO_IMAGE_IMAGE_PARAMS
				_snake_case				:   Any        =			TEXT_TO_IMAGE_IMAGE_PARAMS
				_snake_case				:   Union[str, Any]        =			False







				def 				__a   ( self     :List[Any]     ):
					torch.manual_seed(0     )
					UpperCamelCase__ :Any						=      UNetaDConditionModel(
					    block_out_channels=(32, 64)  ,     layers_per_block=2  ,     sample_size=32  ,     in_channels=4  ,     out_channels=4  ,     down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")  ,     up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")  ,     cross_attention_dim=32  ,     )
					UpperCamelCase__ :Any						=      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     )
					UpperCamelCase__ :int						=      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  ,     )
					torch.manual_seed(0     )
					UpperCamelCase__ :str						=      CLIPTextConfig(
					    bos_token_id=0  ,     eos_token_id=2  ,     hidden_size=32  ,     intermediate_size=37  ,     layer_norm_eps=1e-05  ,     num_attention_heads=4  ,     num_hidden_layers=5  ,     pad_token_id=1  ,     vocab_size=10_00  ,     )
					UpperCamelCase__ :int						=      CLIPTextModel(lowerCamelCase__     )
					UpperCamelCase__ :str						=      CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""     )

					UpperCamelCase__ :Dict						=      {
					    """unet""": unet,
					    """scheduler""": scheduler,
					    """vae""": vae,
					    """text_encoder""": text_encoder,
					    """tokenizer""": tokenizer,
					    """safety_checker""": None,
					    """feature_extractor""": None,
					}
					return components







				def 				__a   ( self     :List[Any]  ,     lowerCamelCase__     :Dict  ,     lowerCamelCase__     :Union[str, Any]=0     ):
					if str(lowerCamelCase__     ).startswith("""mps"""     ):
						UpperCamelCase__ :Dict						=      torch.manual_seed(lowerCamelCase__     )
					else:
						UpperCamelCase__ :int						=      torch.Generator(device=lowerCamelCase__     ).manual_seed(lowerCamelCase__     )
					UpperCamelCase__ :Any						=      {
					    """prompt""": """.""",
					    """generator""": generator,
					    """num_inference_steps""": 2,
					    """guidance_scale""": 1.0,
					    """sag_scale""": 1.0,
					    """output_type""": """numpy""",
					}
					return inputs







				def 				__a   ( self     :Optional[Any]     ):
					super().test_inference_batch_single_identical(expected_max_diff=3e-3     )

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







				"""simple docstring"""




				def 				__a   ( self     :List[Any]     ):
					# clean up the VRAM after each test
					super().tearDown()
					gc.collect()
					torch.cuda.empty_cache()







				def 				__a   ( self     :int     ):
					UpperCamelCase__ :int						=      StableDiffusionSAGPipeline.from_pretrained("""CompVis/stable-diffusion-v1-4"""     )
					UpperCamelCase__ :Optional[int]						=      sag_pipe.to(lowerCamelCase__     )
					sag_pipe.set_progress_bar_config(disable=lowerCamelCase__     )

					UpperCamelCase__ :str						=      """."""
					UpperCamelCase__ :Dict						=      torch.manual_seed(0     )
					UpperCamelCase__ :str						=      sag_pipe(
					    [prompt]  ,     generator=lowerCamelCase__  ,     guidance_scale=7.5  ,     sag_scale=1.0  ,     num_inference_steps=20  ,     output_type="""np"""     )

					UpperCamelCase__ :Optional[int]						=      output.images

					UpperCamelCase__ :Optional[Any]						=      image[0, -3:, -3:, -1]

					assert image.shape == (1, 5_12, 5_12, 3)
					UpperCamelCase__ :Optional[Any]						=      np.array([0.1568, 0.1738, 0.1695, 0.1693, 0.1507, 0.1705, 0.1547, 0.1751, 0.1949]     )

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







				def 				__a   ( self     :Dict     ):
					UpperCamelCase__ :Tuple						=      StableDiffusionSAGPipeline.from_pretrained("""stabilityai/stable-diffusion-2-1-base"""     )
					UpperCamelCase__ :Dict						=      sag_pipe.to(lowerCamelCase__     )
					sag_pipe.set_progress_bar_config(disable=lowerCamelCase__     )

					UpperCamelCase__ :Dict						=      """."""
					UpperCamelCase__ :List[str]						=      torch.manual_seed(0     )
					UpperCamelCase__ :List[Any]						=      sag_pipe(
					    [prompt]  ,     generator=lowerCamelCase__  ,     guidance_scale=7.5  ,     sag_scale=1.0  ,     num_inference_steps=20  ,     output_type="""np"""     )

					UpperCamelCase__ :List[str]						=      output.images

					UpperCamelCase__ :Tuple						=      image[0, -3:, -3:, -1]

					assert image.shape == (1, 5_12, 5_12, 3)
					UpperCamelCase__ :List[str]						=      np.array([0.3459, 0.2876, 0.2537, 0.3002, 0.2671, 0.2160, 0.3026, 0.2262, 0.2371]     )

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







				def 				__a   ( self     :str     ):
					UpperCamelCase__ :str						=      StableDiffusionSAGPipeline.from_pretrained("""stabilityai/stable-diffusion-2-1-base"""     )
					UpperCamelCase__ :Tuple						=      sag_pipe.to(lowerCamelCase__     )
					sag_pipe.set_progress_bar_config(disable=lowerCamelCase__     )

					UpperCamelCase__ :str						=      """."""
					UpperCamelCase__ :int						=      torch.manual_seed(0     )
					UpperCamelCase__ :int						=      sag_pipe(
					    [prompt]  ,     width=7_68  ,     height=5_12  ,     generator=lowerCamelCase__  ,     guidance_scale=7.5  ,     sag_scale=1.0  ,     num_inference_steps=20  ,     output_type="""np"""  ,     )

					UpperCamelCase__ :str						=      output.images

					assert image.shape == (1, 5_12, 7_68, 3)


from __future__ import annotations

from statistics import mean
def 	a (     A__      :  list[int] ,  A__      :  list[int] ,  A__      :  int	)       -> list[int]:





      """simple docstring"""



      _lowercase      			=[0] * no_of_processes
      _lowercase      			=[0] * no_of_processes

      # Initialize remaining_time to waiting_time.

      for i in range(A__	):
            _lowercase      			=burst_time[i]
      _lowercase      			=[]

      _lowercase      			=0
      _lowercase      			=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:
            _lowercase      			=[]
            _lowercase      			=-1

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

            if len(A__	) > 0:
                  _lowercase      			=ready_process[0]
                  for i in ready_process:
                        if remaining_time[i] < remaining_time[target_process]:
                              _lowercase      			=i
                  total_time += burst_time[target_process]
                  completed += 1
                  _lowercase      			=0
                  _lowercase      			=(
                      total_time - arrival_time[target_process] - burst_time[target_process]
                  )
            else:
                  total_time += 1

      return waiting_time
def 	a (     A__      :  list[int] ,  A__      :  int ,  A__      :  list[int]	)       -> list[int]:





      """simple docstring"""



      _lowercase      			=[0] * no_of_processes
      for i in range(A__	):
            _lowercase      			=burst_time[i] + waiting_time[i]
      return turn_around_time


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

  lowercase_   							=  4
  lowercase_   							=  [2, 5, 3, 7]
  lowercase_   							=  [0, 0, 0, 0]
  lowercase_   							=  calculate_waitingtime(arrival_time, burst_time, no_of_processes)
  lowercase_   							=  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}")


