





'''simple docstring'''






import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPImageProcessor, CLIPVisionConfig, CLIPVisionModelWithProjection

from diffusers import (
    AutoencoderKL,
    DPMSolverMultistepScheduler,
    PNDMScheduler,
    StableDiffusionImageVariationPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, nightly, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import IMAGE_VARIATION_BATCH_PARAMS, IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()







class       lowerCamelCase       (    __UpperCAmelCase  ,				__UpperCAmelCase  ,				__UpperCAmelCase  ,				unittest.TestCase	):
					_SCREAMING_SNAKE_CASE				 =     StableDiffusionImageVariationPipeline
					_SCREAMING_SNAKE_CASE				 =     IMAGE_VARIATION_PARAMS
					_SCREAMING_SNAKE_CASE				 =     IMAGE_VARIATION_BATCH_PARAMS
					_SCREAMING_SNAKE_CASE				 =     frozenset([]	)
					# TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
					_SCREAMING_SNAKE_CASE				 =     frozenset([]	)


					def  SCREAMING_SNAKE_CASE_		(       self						:	int							):





											'''simple docstring'''





											torch.manual_seed(0							)
											_snake_case: Tuple	       =		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							,  )
											_snake_case: Dict	       =		PNDMScheduler(skip_prk_steps=lowercase__							)
											torch.manual_seed(0							)
											_snake_case: List[str]	       =		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							)
											_snake_case: int	       =		CLIPVisionConfig(
											    hidden_size=32							,  projection_dim=32							,  intermediate_size=37							,  layer_norm_eps=1e-05							,  num_attention_heads=4							,  num_hidden_layers=5							,  image_size=32							,  patch_size=4							,  )
											_snake_case: Optional[int]	       =		CLIPVisionModelWithProjection(lowercase__							)
											_snake_case: Optional[Any]	       =		CLIPImageProcessor(crop_size=32							,  size=32							)

											_snake_case: Any	       =		{
											    'unet': unet,
											    'scheduler': scheduler,
											    'vae': vae,
											    'image_encoder': image_encoder,
											    'feature_extractor': feature_extractor,
											    'safety_checker': None,
											}
											return components


					def  SCREAMING_SNAKE_CASE_		(       self						:	Union[str, Any]							,  __snake_case						:	List[str]							,  __snake_case						:	Tuple=0							):





											'''simple docstring'''





											_snake_case: Dict	       =		floats_tensor((1, 3, 32, 32)							,  rng=random.Random(lowercase__							)							)
											_snake_case: Dict	       =		image.cpu().permute(0							,  2							,  3							,  1							)[0]
											_snake_case: Optional[Any]	       =		Image.fromarray(np.uinta(lowercase__							)							).convert('RGB'							).resize((32, 32)							)
											if str(lowercase__							).startswith('mps'							):
																	_snake_case: int	       =		torch.manual_seed(lowercase__							)
											else:
																	_snake_case: Tuple	       =		torch.Generator(device=lowercase__							).manual_seed(lowercase__							)
											_snake_case: str	       =		{
											    'image': image,
											    'generator': generator,
											    'num_inference_steps': 2,
											    'guidance_scale': 6.0,
											    'output_type': 'numpy',
											}
											return inputs


					def  SCREAMING_SNAKE_CASE_		(       self						:	Tuple							):





											'''simple docstring'''





											_snake_case: Tuple	       =		'cpu'  # ensure determinism for the device-dependent torch.Generator
											_snake_case: Tuple	       =		self.get_dummy_components()
											_snake_case: Dict	       =		StableDiffusionImageVariationPipeline(**lowercase__							)
											_snake_case: Tuple	       =		sd_pipe.to(lowercase__							)
											sd_pipe.set_progress_bar_config(disable=lowercase__							)

											_snake_case: List[str]	       =		self.get_dummy_inputs(lowercase__							)
											_snake_case: List[Any]	       =		sd_pipe(**lowercase__							).images
											_snake_case: Optional[Any]	       =		image[0, -3:, -3:, -1]

											assert image.shape == (1, 64, 64, 3)
											_snake_case: Union[str, Any]	       =		np.array([0.5_239, 0.5_723, 0.4_796, 0.5_049, 0.5_550, 0.4_685, 0.5_329, 0.4_891, 0.4_921]							)

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


					def  SCREAMING_SNAKE_CASE_		(       self						:	List[str]							):





											'''simple docstring'''





											_snake_case: int	       =		'cpu'  # ensure determinism for the device-dependent torch.Generator
											_snake_case: Any	       =		self.get_dummy_components()
											_snake_case: Dict	       =		StableDiffusionImageVariationPipeline(**lowercase__							)
											_snake_case: Optional[int]	       =		sd_pipe.to(lowercase__							)
											sd_pipe.set_progress_bar_config(disable=lowercase__							)

											_snake_case: Dict	       =		self.get_dummy_inputs(lowercase__							)
											_snake_case: Union[str, Any]	       =		2 * [inputs['image']]
											_snake_case: Union[str, Any]	       =		sd_pipe(**lowercase__							)

											_snake_case: Optional[int]	       =		output.images

											_snake_case: Optional[int]	       =		image[-1, -3:, -3:, -1]

											assert image.shape == (2, 64, 64, 3)
											_snake_case: List[Any]	       =		np.array([0.6_892, 0.5_637, 0.5_836, 0.5_771, 0.6_254, 0.6_409, 0.5_580, 0.5_569, 0.5_289]							)

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




					def  SCREAMING_SNAKE_CASE_		(       self						:	Optional[int]							):





											'''simple docstring'''





											super().test_inference_batch_single_identical(expected_max_diff=3e-3							)







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


					def  SCREAMING_SNAKE_CASE_		(       self						:	Any							):





											'''simple docstring'''





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


					def  SCREAMING_SNAKE_CASE_		(       self						:	str							,  __snake_case						:	Optional[int]							,  __snake_case						:	Union[str, Any]="cpu"							,  __snake_case						:	Dict=torch.floataa							,  __snake_case						:	Optional[int]=0							):





											'''simple docstring'''





											_snake_case: str	       =		torch.Generator(device=lowercase__							).manual_seed(lowercase__							)
											_snake_case: Tuple	       =		load_image(
											    'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
											    '/stable_diffusion_imgvar/input_image_vermeer.png'							)
											_snake_case: List[Any]	       =		np.random.RandomState(lowercase__							).standard_normal((1, 4, 64, 64)							)
											_snake_case: int	       =		torch.from_numpy(lowercase__							).to(device=lowercase__							,  dtype=lowercase__							)
											_snake_case: List[Any]	       =		{
											    'image': init_image,
											    'latents': latents,
											    'generator': generator,
											    'num_inference_steps': 3,
											    'guidance_scale': 7.5,
											    'output_type': 'numpy',
											}
											return inputs


					def  SCREAMING_SNAKE_CASE_		(       self						:	int							):





											'''simple docstring'''





											_snake_case: Optional[Any]	       =		StableDiffusionImageVariationPipeline.from_pretrained(
											    'lambdalabs/sd-image-variations-diffusers'							,  safety_checker=lowercase__							)
											_snake_case: Optional[Any]	       =		sd_pipe.to(lowercase__							)
											sd_pipe.set_progress_bar_config(disable=lowercase__							)

											_snake_case: Any	       =		self.get_inputs(lowercase__							)
											_snake_case: Any	       =		sd_pipe(**lowercase__							).images
											_snake_case: Tuple	       =		image[0, -3:, -3:, -1].flatten()

											assert image.shape == (1, 5_12, 5_12, 3)
											_snake_case: List[str]	       =		np.array([0.84_491, 0.90_789, 0.75_708, 0.78_734, 0.83_485, 0.70_099, 0.66_938, 0.68_727, 0.61_379]							)
											assert np.abs(image_slice - expected_slice							).max() < 6e-3


					def  SCREAMING_SNAKE_CASE_		(       self						:	Union[str, Any]							):





											'''simple docstring'''





											_snake_case: Any	       =		0

											def callback_fn(__snake_case						:	int							,  __snake_case						:	int							,  __snake_case						:	torch.FloatTensor							) -> None:
																	_snake_case: Tuple	       =		True
																	nonlocal number_of_steps
																	number_of_steps += 1
																	if step == 1:
																							_snake_case: Dict	       =		latents.detach().cpu().numpy()
																							assert latents.shape == (1, 4, 64, 64)
																							_snake_case: Union[str, Any]	       =		latents[0, -3:, -3:, -1]
																							_snake_case: Union[str, Any]	       =		np.array(
																							    [-0.1_621, 0.2_837, -0.7_979, -0.1_221, -1.3_057, 0.7_681, -2.1_191, 0.0_464, 1.6_309]							)

																							assert np.abs(latents_slice.flatten() - expected_slice							).max() < 5e-2
																	elif step == 2:
																							_snake_case: str	       =		latents.detach().cpu().numpy()
																							assert latents.shape == (1, 4, 64, 64)
																							_snake_case: List[Any]	       =		latents[0, -3:, -3:, -1]
																							_snake_case: Optional[int]	       =		np.array([0.6_299, 1.7_500, 1.1_992, -2.1_582, -1.8_994, 0.7_334, -0.7_090, 1.0_137, 1.5_273]							)

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

											_snake_case: List[Any]	       =		False

											_snake_case: Optional[Any]	       =		StableDiffusionImageVariationPipeline.from_pretrained(
											    'fusing/sd-image-variations-diffusers'							,  safety_checker=lowercase__							,  torch_dtype=torch.floataa							,  )
											pipe.to(lowercase__							)
											pipe.set_progress_bar_config(disable=lowercase__							)
											pipe.enable_attention_slicing()

											_snake_case: int	       =		self.get_inputs(lowercase__							,  dtype=torch.floataa							)
											pipe(**lowercase__							,  callback=lowercase__							,  callback_steps=1							)
											assert callback_fn.has_been_called
											assert number_of_steps == inputs["num_inference_steps"]




					def  SCREAMING_SNAKE_CASE_		(       self						:	str							):





											'''simple docstring'''





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

											_snake_case: str	       =		'fusing/sd-image-variations-diffusers'
											_snake_case: Tuple	       =		StableDiffusionImageVariationPipeline.from_pretrained(
											    lowercase__							,  safety_checker=lowercase__							,  torch_dtype=torch.floataa							)
											_snake_case: Optional[Any]	       =		pipe.to(lowercase__							)
											pipe.set_progress_bar_config(disable=lowercase__							)
											pipe.enable_attention_slicing(1							)
											pipe.enable_sequential_cpu_offload()

											_snake_case: Optional[Any]	       =		self.get_inputs(lowercase__							,  dtype=torch.floataa							)
											_snake_case: Any	       =		pipe(**lowercase__							)

											_snake_case: Tuple	       =		torch.cuda.max_memory_allocated()
											# make sure that less than 2.6 GB is allocated
											assert mem_bytes < 2.6 * 10**9







@nightly
@require_torch_gpu
class       lowerCamelCase       (    unittest.TestCase	):


					def  SCREAMING_SNAKE_CASE_		(       self						:	Union[str, Any]							):





											'''simple docstring'''





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


					def  SCREAMING_SNAKE_CASE_		(       self						:	int							,  __snake_case						:	Any							,  __snake_case						:	Dict="cpu"							,  __snake_case						:	str=torch.floataa							,  __snake_case						:	List[Any]=0							):





											'''simple docstring'''





											_snake_case: Union[str, Any]	       =		torch.Generator(device=lowercase__							).manual_seed(lowercase__							)
											_snake_case: Optional[Any]	       =		load_image(
											    'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
											    '/stable_diffusion_imgvar/input_image_vermeer.png'							)
											_snake_case: int	       =		np.random.RandomState(lowercase__							).standard_normal((1, 4, 64, 64)							)
											_snake_case: str	       =		torch.from_numpy(lowercase__							).to(device=lowercase__							,  dtype=lowercase__							)
											_snake_case: Any	       =		{
											    'image': init_image,
											    'latents': latents,
											    'generator': generator,
											    'num_inference_steps': 50,
											    'guidance_scale': 7.5,
											    'output_type': 'numpy',
											}
											return inputs


					def  SCREAMING_SNAKE_CASE_		(       self						:	int							):





											'''simple docstring'''





											_snake_case: str	       =		StableDiffusionImageVariationPipeline.from_pretrained('fusing/sd-image-variations-diffusers'							)
											sd_pipe.to(lowercase__							)
											sd_pipe.set_progress_bar_config(disable=lowercase__							)

											_snake_case: Union[str, Any]	       =		self.get_inputs(lowercase__							)
											_snake_case: Any	       =		sd_pipe(**lowercase__							).images[0]

											_snake_case: Any	       =		load_numpy(
											    'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
											    '/stable_diffusion_imgvar/lambdalabs_variations_pndm.npy'							)
											_snake_case: Union[str, Any]	       =		np.abs(expected_image - image							).max()
											assert max_diff < 1e-3




					def  SCREAMING_SNAKE_CASE_		(       self						:	Optional[Any]							):





											'''simple docstring'''





											_snake_case: Dict	       =		StableDiffusionImageVariationPipeline.from_pretrained('fusing/sd-image-variations-diffusers'							)
											_snake_case: str	       =		DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config							)
											sd_pipe.to(lowercase__							)
											sd_pipe.set_progress_bar_config(disable=lowercase__							)

											_snake_case: Dict	       =		self.get_inputs(lowercase__							)
											_snake_case: Optional[Any]	       =		25
											_snake_case: List[Any]	       =		sd_pipe(**lowercase__							).images[0]

											_snake_case: int	       =		load_numpy(
											    'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
											    '/stable_diffusion_imgvar/lambdalabs_variations_dpm_multi.npy'							)
											_snake_case: List[str]	       =		np.abs(expected_image - image							).max()
											assert max_diff < 1e-3













'''simple docstring'''







from __future__ import annotations


def     _UpperCamelCase				(  snake_case__				):
      __A		 =   [0] * len(snake_case__				)
      for i in range(1				,      len(snake_case__				)				):
            __A		 =   duration_times[i - 1] + waiting_times[i - 1]
      return waiting_times


def     _UpperCamelCase				(  snake_case__				,      snake_case__				):
      return [
          duration_time + waiting_times[i]
          for i, duration_time in enumerate(snake_case__				)
      ]


def     _UpperCamelCase				(  snake_case__				):
      return sum(snake_case__				) / len(snake_case__				)


def     _UpperCamelCase				(  snake_case__				):
      return sum(snake_case__				) / len(snake_case__				)


if __name__ == "__main__":
       # process id's
       __A   :							List[str]     	=						[1, 2, 3]

       # ensure that we actually have processes
       if len(processes) == 0:
              print('Zero amount of processes')
              raise SystemExit(0)

       # duration time of all processes
       __A   :							List[Any]     	=						[1_9, 8, 9]

       # ensure we can match each id to a duration time
       if len(duration_times) != len(processes):
              print('Unable to match all id\'s with their duration time')
              raise SystemExit(0)

       # get the waiting times and the turnaround times
       __A   :							Dict     	=						calculate_waiting_times(duration_times)
       __A   :							Optional[int]     	=						calculate_turnaround_times(duration_times, waiting_times)

       # get the average times
       __A   :							Dict     	=						calculate_average_waiting_time(waiting_times)
       __A   :							Tuple     	=						calculate_average_turnaround_time(turnaround_times)

       # print all the results
       print('Process ID\tDuration Time\tWaiting Time\tTurnaround Time')
       for i, process in enumerate(processes):
              print(
                  F"""{process}\t\t{duration_times[i]}\t\t{waiting_times[i]}\t\t"""
                  F"""{turnaround_times[i]}"""
              )
       print(F"""Average waiting time = {average_waiting_time}""")
       print(F"""Average turn around time = {average_turnaround_time}""")


