


from typing import List, Optional, Union

import numpy as np
import PIL
import torch
from PIL import Image

from ...models import UNetaDConditionModel, VQModel
from ...pipelines import DiffusionPipeline
from ...pipelines.pipeline_utils import ImagePipelineOutput
from ...schedulers import DDPMScheduler
from ...utils import (
    is_accelerate_available,
    is_accelerate_version,
    logging,
    randn_tensor,
    replace_example_docstring,
)


snake_case							:	Optional[int]					      =				logging.get_logger(__name__)  # pylint: disable=invalid-name

snake_case							:	List[Any]					      =				'''
    Examples:
        ```py
        >>> from diffusers import KandinskyV22Img2ImgPipeline, KandinskyV22PriorPipeline
        >>> from diffusers.utils import load_image
        >>> import torch

        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained(
        ...     "kandinsky-community/kandinsky-2-2-prior", torch_dtype=torch.float16
        ... )
        >>> pipe_prior.to("cuda")

        >>> prompt = "A red cartoon frog, 4k"
        >>> image_emb, zero_image_emb = pipe_prior(prompt, return_dict=False)

        >>> pipe = KandinskyV22Img2ImgPipeline.from_pretrained(
        ...     "kandinsky-community/kandinsky-2-2-decoder", torch_dtype=torch.float16
        ... )
        >>> pipe.to("cuda")

        >>> init_image = load_image(
        ...     "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
        ...     "/kandinsky/frog.png"
        ... )

        >>> image = pipe(
        ...     image=init_image,
        ...     image_embeds=image_emb,
        ...     negative_image_embeds=zero_image_emb,
        ...     height=768,
        ...     width=768,
        ...     num_inference_steps=100,
        ...     strength=0.2,
        ... ).images

        >>> image[0].save("red_frog.png")
        ```
'''






def 							__lowercase       (			__lowerCAmelCase							:    List[Any] ,   __lowerCAmelCase							:    int ,   __lowerCAmelCase							:    Any=8       ):
							a__					=					height // scale_factor**2
							if height % scale_factor**2 != 0:
														new_height += 1
							a__					=					width // scale_factor**2
							if width % scale_factor**2 != 0:
														new_width += 1
							return new_height * scale_factor, new_width * scale_factor






def 							__lowercase       (			__lowerCAmelCase							:    List[Any] ,   __lowerCAmelCase							:    Tuple=5_1_2 ,   __lowerCAmelCase							:    Union[str, Any]=5_1_2       ):
							a__					=					pil_image.resize((w, h) ,   resample=Image.BICUBIC ,   reducing_gap=1       )
							a__					=					np.array(pil_image.convert('RGB'       )       )
							a__					=					arr.astype(np.floataa       ) / 127.5 - 1
							a__					=					np.transpose(__lowerCAmelCase ,   [2, 0, 1]       )
							a__					=					torch.from_numpy(__lowerCAmelCase       ).unsqueeze(0       )
							return image







class 	snake_case_       (lowerCamelCase_				):
					def __init__(				self       :List[Any]      ,__snake_case       :UNetaDConditionModel      ,__snake_case       :DDPMScheduler      ,__snake_case       :VQModel      ,) ->   Tuple:
												super().__init__()

												self.register_modules(
												    unet=__snake_case      ,scheduler=__snake_case      ,movq=__snake_case      ,)
												a__					=					2 ** (len(self.movq.config.block_out_channels				) - 1)
					def 			lowerCamelCase__(				self       :Optional[int]      ,__snake_case       :Dict      ,__snake_case       :Tuple      ,__snake_case       :Union[str, Any]				) ->   Optional[int]:
												# get the original timestep using init_timestep
												a__					=					min(int(num_inference_steps * strength				)      ,__snake_case				)

												a__					=					max(num_inference_steps - init_timestep      ,0				)
												a__					=					self.scheduler.timesteps[t_start:]

												return timesteps, num_inference_steps - t_start
					def 			lowerCamelCase__(				self       :Dict      ,__snake_case       :List[str]      ,__snake_case       :Union[str, Any]      ,__snake_case       :int      ,__snake_case       :Union[str, Any]      ,__snake_case       :Tuple      ,__snake_case       :int      ,__snake_case       :Tuple=None				) ->   Union[str, Any]:
												if not isinstance(__snake_case      ,(torch.Tensor, PIL.Image.Image, list)				):
																			raise ValueError(
																			    F'`image` has to be of type `torch.Tensor`, `PIL.Image.Image` or list but is {type(__snake_case				)}'				)

												a__					=					image.to(device=__snake_case      ,dtype=__snake_case				)

												a__					=					batch_size * num_images_per_prompt

												if image.shape[1] == 4:
																			a__					=					image

												else:
																			if isinstance(__snake_case      ,__snake_case				) and len(__snake_case				) != batch_size:
																										raise ValueError(
																										    F'You have passed a list of generators of length {len(__snake_case				)}, but requested an effective batch'
																										    F' size of {batch_size}. Make sure the batch size matches the length of the generators.'				)

																			elif isinstance(__snake_case      ,__snake_case				):
																										a__					=					[
																										    self.movq.encode(image[i : i + 1]				).latent_dist.sample(generator[i]				) for i in range(__snake_case				)
																										]
																										a__					=					torch.cat(__snake_case      ,dim=0				)
																			else:
																										a__					=					self.movq.encode(__snake_case				).latent_dist.sample(__snake_case				)

																			a__					=					self.movq.config.scaling_factor * init_latents

												a__					=					torch.cat([init_latents]      ,dim=0				)

												a__					=					init_latents.shape
												a__					=					randn_tensor(__snake_case      ,generator=__snake_case      ,device=__snake_case      ,dtype=__snake_case				)

												# get latents
												a__					=					self.scheduler.add_noise(__snake_case      ,__snake_case      ,__snake_case				)

												a__					=					init_latents

												return latents
					def 			lowerCamelCase__(				self       :List[Any]      ,__snake_case       :List[str]=0				) ->   str:
												if is_accelerate_available():
																			from accelerate import cpu_offload
												else:
																			raise ImportError('Please install accelerate via `pip install accelerate`'				)

												a__					=					torch.device(F'cuda:{gpu_id}'				)

												a__					=					[
												    self.unet,
												    self.movq,
												]
												for cpu_offloaded_model in models:
																			if cpu_offloaded_model is not None:
																										cpu_offload(__snake_case      ,__snake_case				)
					def 			lowerCamelCase__(				self       :List[str]      ,__snake_case       :Optional[int]=0				) ->   int:
												if is_accelerate_available() and is_accelerate_version('>='      ,'0.17.0.dev0'				):
																			from accelerate import cpu_offload_with_hook
												else:
																			raise ImportError('`enable_model_cpu_offload` requires `accelerate v0.17.0` or higher.'				)

												a__					=					torch.device(F'cuda:{gpu_id}'				)

												if self.device.type != "cpu":
																			self.to('cpu'      ,silence_dtype_warnings=__snake_case				)
																			torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

												a__					=					None
												for cpu_offloaded_model in [self.unet, self.movq]:
																			a__				, a__					=					cpu_offload_with_hook(__snake_case      ,__snake_case      ,prev_module_hook=__snake_case				)

												# We'll offload the last model manually.
												a__					=					hook
					@property
					# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
					def 			lowerCamelCase__(				self       :Optional[Any]				) ->   Any:
												if not hasattr(self.unet      ,'_hf_hook'				):
																			return self.device
												for module in self.unet.modules():
																			if (
																			    hasattr(__snake_case      ,'_hf_hook'				)
																			    and hasattr(module._hf_hook      ,'execution_device'				)
																			    and module._hf_hook.execution_device is not None
																			):
																										return torch.device(module._hf_hook.execution_device				)
												return self.device



					@torch.no_grad()
					@replace_example_docstring(__snake_case				)
					def __call__(				self       :str      ,__snake_case       :Union[torch.FloatTensor, List[torch.FloatTensor]]      ,__snake_case       :Union[torch.FloatTensor, PIL.Image.Image, List[torch.FloatTensor], List[PIL.Image.Image]]      ,__snake_case       :Union[torch.FloatTensor, List[torch.FloatTensor]]      ,__snake_case       :int = 5_12      ,__snake_case       :int = 5_12      ,__snake_case       :int = 1_00      ,__snake_case       :float = 4.0      ,__snake_case       :float = 0.3      ,__snake_case       :int = 1      ,__snake_case       :Optional[Union[torch.Generator, List[torch.Generator]]] = None      ,__snake_case       :Optional[str] = "pil"      ,__snake_case       :bool = True      ,) ->   Dict:
												a__					=					self._execution_device

												a__					=					guidance_scale > 1.0

												if isinstance(__snake_case      ,__snake_case				):
																			a__					=					torch.cat(__snake_case      ,dim=0				)
												a__					=					image_embeds.shape[0]
												if isinstance(__snake_case      ,__snake_case				):
																			a__					=					torch.cat(__snake_case      ,dim=0				)

												if do_classifier_free_guidance:
																			a__					=					image_embeds.repeat_interleave(__snake_case      ,dim=0				)
																			a__					=					negative_image_embeds.repeat_interleave(__snake_case      ,dim=0				)

												a__					=					torch.cat([negative_image_embeds, image_embeds]      ,dim=0				).to(dtype=self.unet.dtype      ,device=__snake_case				)

												if not isinstance(__snake_case      ,__snake_case				):
																			a__					=					[image]
												if not all(isinstance(__snake_case      ,(PIL.Image.Image, torch.Tensor)				) for i in image				):
																			raise ValueError(
																			    F'Input is in incorrect format: {[type(__snake_case				) for i in image]}. Currently, we only support  PIL image and pytorch tensor'				)

												a__					=					torch.cat([prepare_image(__snake_case      ,__snake_case      ,__snake_case				) for i in image]      ,dim=0				)
												a__					=					image.to(dtype=image_embeds.dtype      ,device=__snake_case				)

												a__					=					self.movq.encode(__snake_case				)['latents']
												a__					=					latents.repeat_interleave(__snake_case      ,dim=0				)
												self.scheduler.set_timesteps(__snake_case      ,device=__snake_case				)
												a__				, a__					=					self.get_timesteps(__snake_case      ,__snake_case      ,__snake_case				)
												a__					=					timesteps[:1].repeat(batch_size * num_images_per_prompt				)
												a__				, a__					=					downscale_height_and_width(__snake_case      ,__snake_case      ,self.movq_scale_factor				)
												a__					=					self.prepare_latents(
												    __snake_case      ,__snake_case      ,__snake_case      ,__snake_case      ,image_embeds.dtype      ,__snake_case      ,__snake_case				)
												for i, t in enumerate(self.progress_bar(__snake_case				)				):
																			# expand the latents if we are doing classifier free guidance
																			a__					=					torch.cat([latents] * 2				) if do_classifier_free_guidance else latents

																			a__					=					{'image_embeds': image_embeds}
																			a__					=					self.unet(
																			    sample=__snake_case      ,timestep=__snake_case      ,encoder_hidden_states=__snake_case      ,added_cond_kwargs=__snake_case      ,return_dict=__snake_case      ,)[0]

																			if do_classifier_free_guidance:
																										a__				, a__					=					noise_pred.split(latents.shape[1]      ,dim=1				)
																										a__				, a__					=					noise_pred.chunk(2				)
																										a__				, a__					=					variance_pred.chunk(2				)
																										a__					=					noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
																										a__					=					torch.cat([noise_pred, variance_pred_text]      ,dim=1				)

																			if not (
																			    hasattr(self.scheduler.config      ,'variance_type'				)
																			    and self.scheduler.config.variance_type in ["learned", "learned_range"]
																			):
																										a__				, a__					=					noise_pred.split(latents.shape[1]      ,dim=1				)

																			# compute the previous noisy sample x_t -> x_t-1
																			a__					=					self.scheduler.step(
																			    __snake_case      ,__snake_case      ,__snake_case      ,generator=__snake_case      ,)[0]

												# post-processing
												a__					=					self.movq.decode(__snake_case      ,force_not_quantize=__snake_case				)['sample']

												if output_type not in ["pt", "np", "pil"]:
																			raise ValueError(F'Only the output types `pt`, `pil` and `np` are supported not output_type={output_type}'				)

												if output_type in ["np", "pil"]:
																			a__					=					image * 0.5 + 0.5
																			a__					=					image.clamp(0      ,1				)
																			a__					=					image.cpu().permute(0      ,2      ,3      ,1				).float().numpy()

												if output_type == "pil":
																			a__					=					self.numpy_to_pil(__snake_case				)

												if not return_dict:
																			return (image,)

												return ImagePipelineOutput(images=__snake_case				)








import inspect
import os
import unittest
from pathlib import Path

import torch

import accelerate
from accelerate.test_utils import execute_subprocess_async
from accelerate.test_utils.testing import run_command







class 	snake_case_       (unittest.TestCase				):
					UpperCAmelCase__ :   Optional[Any]   				=      inspect.getfile(accelerate.test_utils				)
					UpperCAmelCase__ :   Tuple   				=      os.path.sep.join(mod_file.split(os.path.sep				)[:-1] + ['''scripts''', '''test_cli.py''']				)

					UpperCAmelCase__ :   Optional[int]   				=      ['''accelerate''', '''launch''']
					UpperCAmelCase__ :   List[str]   				=      Path.home() / '''.cache/huggingface/accelerate'''
					UpperCAmelCase__ :   Union[str, Any]   				=      '''default_config.yaml'''
					UpperCAmelCase__ :   List[Any]   				=      config_folder / config_file
					UpperCAmelCase__ :   Optional[int]   				=      config_folder / '''_default_config.yaml'''

					UpperCAmelCase__ :   int   				=      Path('''tests/test_configs'''				)
					@classmethod
					def 			lowerCamelCase__(				cls       :Any				) ->   int:
												if cls.config_path.is_file():
																			cls.config_path.rename(cls.changed_path				)
					@classmethod
					def 			lowerCamelCase__(				cls       :Dict				) ->   Union[str, Any]:
												if cls.changed_path.is_file():
																			cls.changed_path.rename(cls.config_path				)
					def 			lowerCamelCase__(				self       :Any				) ->   Optional[int]:
												a__					=					self.base_cmd
												if torch.cuda.is_available() and (torch.cuda.device_count() > 1):
																			cmd += ["--multi_gpu"]
												execute_subprocess_async(cmd + [self.test_file_path]      ,env=os.environ.copy()				)
					def 			lowerCamelCase__(				self       :str				) ->   List[Any]:
												for config in sorted(self.test_config_path.glob('**/*.yaml'				)				):
																			with self.subTest(config_file=__snake_case				):
																										execute_subprocess_async(
																										    self.base_cmd + ['--config_file', str(__snake_case				), self.test_file_path]      ,env=os.environ.copy()				)



					def 			lowerCamelCase__(				self       :int				) ->   List[str]:
												execute_subprocess_async(['accelerate', 'test']      ,env=os.environ.copy()				)







class 	snake_case_       (unittest.TestCase				):
					UpperCAmelCase__ :   List[Any]   				=      '''test-tpu'''
					UpperCAmelCase__ :   str   				=      '''us-central1-a'''
					UpperCAmelCase__ :   Optional[Any]   				=      '''ls'''
					UpperCAmelCase__ :   Optional[int]   				=      ['''accelerate''', '''tpu-config''']
					UpperCAmelCase__ :   Any   				=      '''cd /usr/share'''
					UpperCAmelCase__ :   Tuple   				=      '''tests/test_samples/test_command_file.sh'''
					UpperCAmelCase__ :   List[Any]   				=      '''Running gcloud compute tpus tpu-vm ssh'''
					def 			lowerCamelCase__(				self       :Optional[Any]				) ->   Tuple:
												a__					=					run_command(
												    self.cmd
												    + ['--command', self.command, '--tpu_zone', self.tpu_zone, '--tpu_name', self.tpu_name, '--debug']      ,return_stdout=__snake_case      ,)
												self.assertIn(
												    F'{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; ls --worker all'      ,__snake_case      ,)
					def 			lowerCamelCase__(				self       :int				) ->   str:
												a__					=					run_command(
												    self.cmd
												    + [
												        '--config_file',
												        'tests/test_configs/0_12_0.yaml',
												        '--command',
												        self.command,
												        '--tpu_zone',
												        self.tpu_zone,
												        '--tpu_name',
												        self.tpu_name,
												        '--debug',
												    ]      ,return_stdout=__snake_case      ,)
												self.assertIn(
												    F'{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; ls --worker all'      ,__snake_case      ,)
					def 			lowerCamelCase__(				self       :Any				) ->   Dict:
												a__					=					run_command(
												    self.cmd + ['--config_file', 'tests/test_configs/latest.yaml', '--debug']      ,return_stdout=__snake_case				)
												self.assertIn(
												    F'{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; echo "hello world"; echo "this is a second command" --worker all'      ,__snake_case      ,)
					def 			lowerCamelCase__(				self       :Tuple				) ->   str:
												a__					=					run_command(
												    self.cmd + ['--config_file', 'tests/test_configs/latest.yaml', '--command', self.command, '--debug']      ,return_stdout=__snake_case      ,)
												self.assertIn(
												    F'{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; ls --worker all'      ,__snake_case      ,)
					def 			lowerCamelCase__(				self       :Dict				) ->   int:
												a__					=					run_command(
												    self.cmd
												    + [
												        '--config_file',
												        'tests/test_configs/latest.yaml',
												        '--command',
												        self.command,
												        '--command',
												        'echo "Hello World"',
												        '--debug',
												    ]      ,return_stdout=__snake_case      ,)
												self.assertIn(
												    F'{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; ls; echo "Hello World" --worker all'      ,__snake_case      ,)
					def 			lowerCamelCase__(				self       :Union[str, Any]				) ->   List[Any]:
												a__					=					run_command(
												    self.cmd
												    + ['--config_file', 'tests/test_configs/latest.yaml', '--command_file', self.command_file, '--debug']      ,return_stdout=__snake_case      ,)
												self.assertIn(
												    F'{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; echo "hello world"; echo "this is a second command" --worker all'      ,__snake_case      ,)
					def 			lowerCamelCase__(				self       :int				) ->   int:
												a__					=					run_command(
												    self.cmd
												    + [
												        '--config_file',
												        'tests/test_configs/0_12_0.yaml',
												        '--command_file',
												        self.command_file,
												        '--tpu_zone',
												        self.tpu_zone,
												        '--tpu_name',
												        self.tpu_name,
												        '--debug',
												    ]      ,return_stdout=__snake_case      ,)
												self.assertIn(
												    F'{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; echo "hello world"; echo "this is a second command" --worker all'      ,__snake_case      ,)
					def 			lowerCamelCase__(				self       :Union[str, Any]				) ->   List[Any]:
												a__					=					run_command(
												    self.cmd + ['--config_file', 'tests/test_configs/latest.yaml', '--install_accelerate', '--debug']      ,return_stdout=__snake_case      ,)
												self.assertIn(
												    F'{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; pip install accelerate -U; echo "hello world"; echo "this is a second command" --worker all'      ,__snake_case      ,)



					def 			lowerCamelCase__(				self       :Any				) ->   Optional[int]:
												a__					=					run_command(
												    self.cmd
												    + [
												        '--config_file',
												        'tests/test_configs/latest.yaml',
												        '--install_accelerate',
												        '--accelerate_version',
												        '12.0.0',
												        '--debug',
												    ]      ,return_stdout=__snake_case      ,)
												self.assertIn(
												    F'{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; pip install accelerate==12.0.0; echo "hello world"; echo "this is a second command" --worker all'      ,__snake_case      ,)




