






'''simple docstring'''







from collections import deque




class   __SCREAMING_SNAKE_CASE       :


  """simple docstring"""


  def __init__(			self   ,	UpperCamelCase__   ,	UpperCamelCase__   ,	UpperCamelCase__	):






         """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   __SCREAMING_SNAKE_CASE       :


  """simple docstring"""


  def __init__(			self   ,	UpperCamelCase__   ,	UpperCamelCase__   ,	UpperCamelCase__   ,	UpperCamelCase__   ,	):






         """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	):






         """simple docstring"""


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







  def    _a       (			self   ,	UpperCamelCase__	):






         """simple docstring"""


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







  def    _a       (			self   ,	UpperCamelCase__	):






         """simple docstring"""


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







  def    _a       (			self   ,	UpperCamelCase__	):






         """simple docstring"""


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







  def    _a       (			self   ,	UpperCamelCase__	):






         """simple docstring"""


         return [q.burst_time for q in queue]







  def    _a       (			self   ,	UpperCamelCase__	):






         """simple docstring"""


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







  def    _a       (			self   ,	UpperCamelCase__	):






         """simple docstring"""


         a_						     =			deque()  # sequence deque of finished process
         while len(UpperCamelCase__	) != 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(UpperCamelCase__	)
                # 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(UpperCamelCase__	)

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







  def    _a       (			self   ,	UpperCamelCase__   ,	UpperCamelCase__	):






         """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(UpperCamelCase__	)	):
                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(UpperCamelCase__	)
                # 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(UpperCamelCase__	)
                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(UpperCamelCase__	)

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







  def    _a       (			self	):






         """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

  __lowerCAmelCase        =							Process("P1", 0, 53)
  __lowerCAmelCase        =							Process("P2", 0, 17)
  __lowerCAmelCase        =							Process("P3", 0, 68)
  __lowerCAmelCase        =							Process("P4", 0, 24)
  __lowerCAmelCase        =							3
  __lowerCAmelCase        =							[17, 25]
  __lowerCAmelCase        =							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])})

  __lowerCAmelCase        =							Process("P1", 0, 53)
  __lowerCAmelCase        =							Process("P2", 0, 17)
  __lowerCAmelCase        =							Process("P3", 0, 68)
  __lowerCAmelCase        =							Process("P4", 0, 24)
  __lowerCAmelCase        =							3
  __lowerCAmelCase        =							[17, 25]
  __lowerCAmelCase        =							deque([Pa, Pa, Pa, Pa])
  __lowerCAmelCase        =							MLFQ(number_of_queues, time_slices, queue, 0)
  __lowerCAmelCase        =							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()}"""
  )












'''simple docstring'''







import gc
import unittest

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

from diffusers import TransformeraDModel, VQDiffusionPipeline, VQDiffusionScheduler, VQModel
from diffusers.pipelines.vq_diffusion.pipeline_vq_diffusion import LearnedClassifierFreeSamplingEmbeddings
from diffusers.utils import load_numpy, slow, torch_device
from diffusers.utils.testing_utils import require_torch_gpu


__lowerCAmelCase        =							False




class   __SCREAMING_SNAKE_CASE       (unittest.TestCase       ):


  """simple docstring"""


  def    _a       (			self	):






         """simple docstring"""


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







  @property
  def    _a       (			self	):






         """simple docstring"""


         return 12







  @property
  def    _a       (			self	):






         """simple docstring"""


         return 12







  @property
  def    _a       (			self	):






         """simple docstring"""


         return 32







  @property
  def    _a       (			self	):






         """simple docstring"""


         torch.manual_seed(0	)
         a_						     =			VQModel(
             block_out_channels=[32, 64]   ,	in_channels=3   ,	out_channels=3   ,	down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']   ,	up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']   ,	latent_channels=3   ,	num_vq_embeddings=self.num_embed   ,	vq_embed_dim=3   ,	)
         return model







  @property
  def    _a       (			self	):






         """simple docstring"""


         a_						     =			CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'	)
         return tokenizer







  @property
  def    _a       (			self	):






         """simple docstring"""


         torch.manual_seed(0	)
         a_						     =			CLIPTextConfig(
             bos_token_id=0   ,	eos_token_id=2   ,	hidden_size=self.text_embedder_hidden_size   ,	intermediate_size=37   ,	layer_norm_eps=1e-05   ,	num_attention_heads=4   ,	num_hidden_layers=5   ,	pad_token_id=1   ,	vocab_size=1_000   ,	)
         return CLIPTextModel(UpperCamelCase__	)







  @property
  def    _a       (			self	):






         """simple docstring"""


         torch.manual_seed(0	)

         a_						     =			12
         a_						     =			12

         a_						     =			{
             'attention_bias': True,
             'cross_attention_dim': 32,
             'attention_head_dim': height * width,
             'num_attention_heads': 1,
             'num_vector_embeds': self.num_embed,
             'num_embeds_ada_norm': self.num_embeds_ada_norm,
             'norm_num_groups': 32,
             'sample_size': width,
             'activation_fn': 'geglu-approximate',
         }

         a_						     =			TransformeraDModel(**UpperCamelCase__	)
         return model







  def    _a       (			self	):






         """simple docstring"""


         a_						     =			'cpu'

         a_						     =			self.dummy_vqvae
         a_						     =			self.dummy_text_encoder
         a_						     =			self.dummy_tokenizer
         a_						     =			self.dummy_transformer
         a_						     =			VQDiffusionScheduler(self.num_embed	)
         a_						     =			LearnedClassifierFreeSamplingEmbeddings(learnable=UpperCamelCase__	)

         a_						     =			VQDiffusionPipeline(
             vqvae=UpperCamelCase__   ,	text_encoder=UpperCamelCase__   ,	tokenizer=UpperCamelCase__   ,	transformer=UpperCamelCase__   ,	scheduler=UpperCamelCase__   ,	learned_classifier_free_sampling_embeddings=UpperCamelCase__   ,	)
         a_						     =			pipe.to(UpperCamelCase__	)
         pipe.set_progress_bar_config(disable=UpperCamelCase__	)

         a_						     =			'teddy bear playing in the pool'

         a_						     =			torch.Generator(device=UpperCamelCase__	).manual_seed(0	)
         a_						     =			pipe([prompt]   ,	generator=UpperCamelCase__   ,	num_inference_steps=2   ,	output_type='np'	)
         a_						     =			output.images

         a_						     =			torch.Generator(device=UpperCamelCase__	).manual_seed(0	)
         a_						     =			pipe(
             [prompt]   ,	generator=UpperCamelCase__   ,	output_type='np'   ,	return_dict=UpperCamelCase__   ,	num_inference_steps=2	)[0]

         a_						     =			image[0, -3:, -3:, -1]
         a_						     =			image_from_tuple[0, -3:, -3:, -1]

         assert image.shape == (1, 24, 24, 3)

         a_						     =			np.array([0.6_551, 0.6_168, 0.5_008, 0.5_676, 0.5_659, 0.4_295, 0.6_073, 0.5_599, 0.4_992]	)

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







  def    _a       (			self	):






         """simple docstring"""


         a_						     =			'cpu'

         a_						     =			self.dummy_vqvae
         a_						     =			self.dummy_text_encoder
         a_						     =			self.dummy_tokenizer
         a_						     =			self.dummy_transformer
         a_						     =			VQDiffusionScheduler(self.num_embed	)
         a_						     =			LearnedClassifierFreeSamplingEmbeddings(
             learnable=UpperCamelCase__   ,	hidden_size=self.text_embedder_hidden_size   ,	length=tokenizer.model_max_length	)

         a_						     =			VQDiffusionPipeline(
             vqvae=UpperCamelCase__   ,	text_encoder=UpperCamelCase__   ,	tokenizer=UpperCamelCase__   ,	transformer=UpperCamelCase__   ,	scheduler=UpperCamelCase__   ,	learned_classifier_free_sampling_embeddings=UpperCamelCase__   ,	)
         a_						     =			pipe.to(UpperCamelCase__	)
         pipe.set_progress_bar_config(disable=UpperCamelCase__	)

         a_						     =			'teddy bear playing in the pool'

         a_						     =			torch.Generator(device=UpperCamelCase__	).manual_seed(0	)
         a_						     =			pipe([prompt]   ,	generator=UpperCamelCase__   ,	num_inference_steps=2   ,	output_type='np'	)
         a_						     =			output.images

         a_						     =			torch.Generator(device=UpperCamelCase__	).manual_seed(0	)
         a_						     =			pipe(
             [prompt]   ,	generator=UpperCamelCase__   ,	output_type='np'   ,	return_dict=UpperCamelCase__   ,	num_inference_steps=2	)[0]

         a_						     =			image[0, -3:, -3:, -1]
         a_						     =			image_from_tuple[0, -3:, -3:, -1]

         assert image.shape == (1, 24, 24, 3)

         a_						     =			np.array([0.6_693, 0.6_075, 0.4_959, 0.5_701, 0.5_583, 0.4_333, 0.6_171, 0.5_684, 0.4_988]	)

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





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


  """simple docstring"""


  def    _a       (			self	):






         """simple docstring"""


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







  def    _a       (			self	):






         """simple docstring"""


         a_						     =			load_numpy(
             'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
             '/vq_diffusion/teddy_bear_pool_classifier_free_sampling.npy'	)

         a_						     =			VQDiffusionPipeline.from_pretrained('microsoft/vq-diffusion-ithq'	)
         a_						     =			pipeline.to(UpperCamelCase__	)
         pipeline.set_progress_bar_config(disable=UpperCamelCase__	)

         # requires GPU generator for gumbel softmax
         # don't use GPU generator in tests though
         a_						     =			torch.Generator(device=UpperCamelCase__	).manual_seed(0	)
         a_						     =			pipeline(
             'teddy bear playing in the pool'   ,	num_images_per_prompt=1   ,	generator=UpperCamelCase__   ,	output_type='np'   ,	)

         a_						     =			output.images[0]

         assert image.shape == (256, 256, 3)
         assert np.abs(expected_image - image	).max() < 2.0




