from logging import getLogger
from typing import Any, Callable, List, Optional, Union

import numpy as np
import PIL
import torch

from ...schedulers import DDPMScheduler
from ..onnx_utils import ORT_TO_NP_TYPE, OnnxRuntimeModel
from ..pipeline_utils import ImagePipelineOutput
from . import StableDiffusionUpscalePipeline


lowerCAmelCase__  	=      getLogger(__name__)


lowerCAmelCase__  	=      4
lowerCAmelCase__  	=      7

lowerCAmelCase__  	=      {
    "float16": torch.floataa,
    "float32": torch.floataa,
}

def 	__lowerCamelCase							(				__a    : List[str]		)      ->       Optional[int]:
	if isinstance(__a      ,							torch.Tensor		):
		return image
	elif isinstance(__a      ,							PIL.Image.Image		):
		_lowercase   =[image]

	if isinstance(image[0]      ,							PIL.Image.Image		):
		_lowercase   ,  _lowercase   =image[0].size
		_lowercase   ,  _lowercase   =(x - x % 64 for x in (w, h))  # resize to integer multiple of 32

		_lowercase   =[np.array(i.resize((w, h)		)		)[None, :] for i in image]
		_lowercase   =np.concatenate(__a      ,							axis=0		)
		_lowercase   =np.array(__a		).astype(np.floataa		) / 255.0
		_lowercase   =image.transpose(0      ,							3      ,							1      ,							2		)
		_lowercase   =2.0 * image - 1.0
		_lowercase   =torch.from_numpy(__a		)
	elif isinstance(image[0]      ,							torch.Tensor		):
		_lowercase   =torch.cat(__a      ,							dim=0		)

	return image
class 	_a  (							lowerCamelCase_					):



	"""simple docstring"""



	def __init__(   self       ,  lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_ = 350       ,  ):
		super().__init__(
		    vae=lowerCAmelCase_       ,  text_encoder=lowerCAmelCase_       ,  tokenizer=lowerCAmelCase_       ,  unet=lowerCAmelCase_       ,  low_res_scheduler=lowerCAmelCase_       ,  scheduler=lowerCAmelCase_       ,  safety_checker=lowerCAmelCase_       ,  feature_extractor=lowerCAmelCase_       ,  watermarker=lowerCAmelCase_       ,  max_noise_level=lowerCAmelCase_       ,  )


	def __call__(   self       ,  lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_ = 75       ,  lowerCAmelCase_ = 9.0       ,  lowerCAmelCase_ = 20       ,  lowerCAmelCase_ = None       ,  lowerCAmelCase_ = 1       ,  lowerCAmelCase_ = 0.0       ,  lowerCAmelCase_ = None       ,  lowerCAmelCase_ = None       ,  lowerCAmelCase_ = None       ,  lowerCAmelCase_ = None       ,  lowerCAmelCase_ = "pil"       ,  lowerCAmelCase_ = True       ,  lowerCAmelCase_ = None       ,  lowerCAmelCase_ = 1       ,  ):
		self.check_inputs(lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_       )

		# 2. Define call parameters
		if prompt is not None and isinstance(lowerCAmelCase_       ,  lowerCAmelCase_       ):
			_lowercase   =1
		elif prompt is not None and isinstance(lowerCAmelCase_       ,  lowerCAmelCase_       ):
			_lowercase   =len(lowerCAmelCase_       )
		else:
			_lowercase   =prompt_embeds.shape[0]

		_lowercase   =self._execution_device
		# here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
		# of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
		# corresponds to doing no classifier free guidance.
		_lowercase   =guidance_scale > 1.0

		# 3. Encode input prompt
		_lowercase   =self._encode_prompt(
		    lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_       ,  prompt_embeds=lowerCAmelCase_       ,  negative_prompt_embeds=lowerCAmelCase_       ,  )

		_lowercase   =ORT_TO_PT_TYPE[str(text_embeddings.dtype       )]

		# 4. Preprocess image
		_lowercase   =preprocess(lowerCAmelCase_       )
		_lowercase   =image.cpu()

		# 5. set timesteps
		self.scheduler.set_timesteps(lowerCAmelCase_       ,  device=lowerCAmelCase_       )
		_lowercase   =self.scheduler.timesteps

		# 5. Add noise to image
		_lowercase   =torch.tensor([noise_level]       ,  dtype=torch.long       ,  device=lowerCAmelCase_       )
		_lowercase   =torch.randn(image.shape       ,  generator=lowerCAmelCase_       ,  device=lowerCAmelCase_       ,  dtype=lowerCAmelCase_       )
		_lowercase   =self.low_res_scheduler.add_noise(lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_       )

		_lowercase   =2 if do_classifier_free_guidance else 1
		_lowercase   =np.concatenate([image] * batch_multiplier * num_images_per_prompt       )
		_lowercase   =np.concatenate([noise_level] * image.shape[0]       )

		# 6. Prepare latent variables
		_lowercase   ,  _lowercase   =image.shape[2:]
		_lowercase   =self.prepare_latents(
		    batch_size * num_images_per_prompt       ,  lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_       ,  )

		# 7. Check that sizes of image and latents match
		_lowercase   =image.shape[1]
		if NUM_LATENT_CHANNELS + num_channels_image != NUM_UNET_INPUT_CHANNELS:
			raise ValueError(
			    "Incorrect configuration settings! The config of `pipeline.unet` expects"
			    F''' {NUM_UNET_INPUT_CHANNELS} but received `num_channels_latents`: {NUM_LATENT_CHANNELS} +'''
			    F''' `num_channels_image`: {num_channels_image} '''
			    F''' = {NUM_LATENT_CHANNELS+num_channels_image}. Please verify the config of'''
			    " `pipeline.unet` or your `image` input."       )

		# 8. Prepare extra step kwargs. TODO: Logic should ideally just be moved out of the pipeline
		_lowercase   =self.prepare_extra_step_kwargs(lowerCAmelCase_       ,  lowerCAmelCase_       )

		_lowercase   =next(
		    (input.type for input in self.unet.model.get_inputs() if input.name == "timestep")       ,  "tensor(float)"       )
		_lowercase   =ORT_TO_NP_TYPE[timestep_dtype]

		# 9. Denoising loop
		_lowercase   =len(lowerCAmelCase_       ) - num_inference_steps * self.scheduler.order
		with self.progress_bar(total=lowerCAmelCase_       ) as progress_bar:
			for i, t in enumerate(lowerCAmelCase_       ):
				# expand the latents if we are doing classifier free guidance
				_lowercase   =np.concatenate([latents] * 2       ) if do_classifier_free_guidance else latents

				# concat latents, mask, masked_image_latents in the channel dimension
				_lowercase   =self.scheduler.scale_model_input(lowerCAmelCase_       ,  lowerCAmelCase_       )
				_lowercase   =np.concatenate([latent_model_input, image]       ,  axis=1       )

				# timestep to tensor
				_lowercase   =np.array([t]       ,  dtype=lowerCAmelCase_       )

				# predict the noise residual
				_lowercase   =self.unet(
				    sample=lowerCAmelCase_       ,  timestep=lowerCAmelCase_       ,  encoder_hidden_states=lowerCAmelCase_       ,  class_labels=noise_level.astype(np.intaa       )       ,  )[0]

				# perform guidance
				if do_classifier_free_guidance:
					_lowercase   ,  _lowercase   =np.split(lowerCAmelCase_       ,  2       )
				_lowercase   =noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

				# compute the previous noisy sample x_t -> x_t-1
				_lowercase   =self.scheduler.step(
				    torch.from_numpy(lowerCAmelCase_       )       ,  lowerCAmelCase_       ,  lowerCAmelCase_       ,  **lowerCAmelCase_       ).prev_sample

				# call the callback, if provided
				if i == len(lowerCAmelCase_       ) - 1 or ((i + 1) > num_warmup_steps and (i + 1) % self.scheduler.order == 0):
					progress_bar.update()
					if callback is not None and i % callback_steps == 0:
						callback(lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_       )

        # 10. Post-processing
		_lowercase   =self.decode_latents(latents.float()       )

		# 11. Convert to PIL
		if output_type == "pil":
			_lowercase   =self.numpy_to_pil(lowerCAmelCase_       )

		if not return_dict:
			return (image,)

		return ImagePipelineOutput(images=lowerCAmelCase_       )


	def __lowerCAmelCase      (   self       ,  lowerCAmelCase_       ):
		_lowercase   =1 / 0.0_8_3_3_3 * latents
		_lowercase   =self.vae(latent_sample=lowerCAmelCase_       )[0]
		_lowercase   =np.clip(image / 2 + 0.5       ,  0       ,  1       )
		_lowercase   =image.transpose((0, 2, 3, 1)       )
		return image


	def __lowerCAmelCase      (   self       ,  lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_       ,  lowerCAmelCase_ = None       ,  lowerCAmelCase_ = None       ,  ):
		if prompt is not None and isinstance(lowerCAmelCase_       ,  lowerCAmelCase_       ):
			_lowercase   =1
		elif prompt is not None and isinstance(lowerCAmelCase_       ,  lowerCAmelCase_       ):
			_lowercase   =len(lowerCAmelCase_       )
		else:
			_lowercase   =prompt_embeds.shape[0]

		if prompt_embeds is None:
			_lowercase   =self.tokenizer(
			    lowerCAmelCase_       ,  padding="max_length"       ,  max_length=self.tokenizer.model_max_length       ,  truncation=lowerCAmelCase_       ,  return_tensors="pt"       ,  )
			_lowercase   =text_inputs.input_ids
			_lowercase   =self.tokenizer(lowerCAmelCase_       ,  padding="longest"       ,  return_tensors="pt"       ).input_ids

			if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(
			    lowerCAmelCase_       ,  lowerCAmelCase_       ):
				_lowercase   =self.tokenizer.batch_decode(
				    untruncated_ids[:, self.tokenizer.model_max_length - 1 : -1]       )
				logger.warning(
				    "The following part of your input was truncated because CLIP can only handle sequences up to"
				    F''' {self.tokenizer.model_max_length} tokens: {removed_text}'''       )

			# no positional arguments to text_encoder
			_lowercase   =self.text_encoder(
			    input_ids=text_input_ids.int().to(lowerCAmelCase_       )       ,  )
			_lowercase   =prompt_embeds[0]

		_lowercase   ,  _lowercase   ,  _lowercase   =prompt_embeds.shape
		# duplicate text embeddings for each generation per prompt, using mps friendly method
		_lowercase   =prompt_embeds.repeat(1       ,  lowerCAmelCase_       )
		_lowercase   =prompt_embeds.reshape(bs_embed * num_images_per_prompt       ,  lowerCAmelCase_       ,  -1       )

		# get unconditional embeddings for classifier free guidance
		if do_classifier_free_guidance and negative_prompt_embeds is None:
			_lowercase   =42
			if negative_prompt is None:
				_lowercase   =[""] * batch_size
			elif type(lowerCAmelCase_       ) is not type(lowerCAmelCase_       ):
				raise TypeError(
				    F'''`negative_prompt` should be the same type to `prompt`, but got {type(lowerCAmelCase_       )} !='''
				    F''' {type(lowerCAmelCase_       )}.'''       )
			elif isinstance(lowerCAmelCase_       ,  lowerCAmelCase_       ):
				_lowercase   =[negative_prompt]
			elif batch_size != len(lowerCAmelCase_       ):
				raise ValueError(
				    F'''`negative_prompt`: {negative_prompt} has batch size {len(lowerCAmelCase_       )}, but `prompt`:'''
				    F''' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'''
				    " the batch size of `prompt`."       )
			else:
				_lowercase   =negative_prompt

			_lowercase   =text_input_ids.shape[-1]
			_lowercase   =self.tokenizer(
			    lowerCAmelCase_       ,  padding="max_length"       ,  max_length=lowerCAmelCase_       ,  truncation=lowerCAmelCase_       ,  return_tensors="pt"       ,  )

			# if hasattr(uncond_input, "attention_mask"):
			#     attention_mask = uncond_input.attention_mask.to(device)
			# else:
			#     attention_mask = None

			_lowercase   =self.text_encoder(
			    input_ids=uncond_input.input_ids.int().to(lowerCAmelCase_       )       ,  )
			_lowercase   =uncond_embeddings[0]

		if do_classifier_free_guidance:
			_lowercase   =uncond_embeddings.shape[1]
			# duplicate unconditional embeddings for each generation per prompt, using mps friendly method
			_lowercase   =uncond_embeddings.repeat(1       ,  lowerCAmelCase_       )
			_lowercase   =uncond_embeddings.reshape(batch_size * num_images_per_prompt       ,  lowerCAmelCase_       ,  -1       )

			# For classifier free guidance, we need to do two forward passes.
			# Here we concatenate the unconditional and text embeddings into a single batch
			# to avoid doing two forward passes
			_lowercase   =np.concatenate([uncond_embeddings, prompt_embeds]       )

		return prompt_embeds





import gc
import tempfile
import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionDualGuidedPipeline
from diffusers.utils.testing_utils import load_image, nightly, require_torch_gpu, torch_device


lowerCAmelCase__	:  Union[str, Any]						=  False





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




					def      __a     (  self						)	->    str:



									'''simple docstring'''







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




					def      __a     (  self						)	->    Optional[int]:



									'''simple docstring'''







									snake_case__   :      str   				= VersatileDiffusionDualGuidedPipeline.from_pretrained('shi-labs/versatile-diffusion'						)
									# remove text_unet
									pipe.remove_unused_weights()
									pipe.to(__UpperCamelCase						)
									pipe.set_progress_bar_config(disable=__UpperCamelCase						)

									snake_case__   :      Union[str, Any]   				= load_image(
									    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg'						)

									snake_case__   :      Dict   				= torch.manual_seed(0						)
									snake_case__   :      Tuple   				= pipe(
									    prompt='first prompt'       ,			image=__UpperCamelCase       ,			text_to_image_strength=0.7_5       ,			generator=__UpperCamelCase       ,			guidance_scale=7.5       ,			num_inference_steps=2       ,			output_type='numpy'       ,			).images

									with tempfile.TemporaryDirectory() as tmpdirname:
													pipe.save_pretrained(__UpperCamelCase						)
													snake_case__   :      List[Any]   				= VersatileDiffusionDualGuidedPipeline.from_pretrained(__UpperCamelCase						)

									pipe.to(__UpperCamelCase						)
									pipe.set_progress_bar_config(disable=__UpperCamelCase						)

									snake_case__   :      str   				= generator.manual_seed(0						)
									snake_case__   :      List[str]   				= pipe(
									    prompt='first prompt'       ,			image=__UpperCamelCase       ,			text_to_image_strength=0.7_5       ,			generator=__UpperCamelCase       ,			guidance_scale=7.5       ,			num_inference_steps=2       ,			output_type='numpy'       ,			).images

									assert np.abs(image - new_image						).sum() < 1E-5, "Models don't have the same forward pass"




					def      __a     (  self						)	->    Optional[Any]:



									'''simple docstring'''







									snake_case__   :      Union[str, Any]   				= VersatileDiffusionDualGuidedPipeline.from_pretrained('shi-labs/versatile-diffusion'						)
									pipe.remove_unused_weights()
									pipe.to(__UpperCamelCase						)
									pipe.set_progress_bar_config(disable=__UpperCamelCase						)

									snake_case__   :      str   				= 'cyberpunk 2077'
									snake_case__   :      Dict   				= load_image(
									    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg'						)
									snake_case__   :      Union[str, Any]   				= torch.manual_seed(0						)
									snake_case__   :      Union[str, Any]   				= pipe(
									    prompt=__UpperCamelCase       ,			image=__UpperCamelCase       ,			text_to_image_strength=0.7_5       ,			generator=__UpperCamelCase       ,			guidance_scale=7.5       ,			num_inference_steps=50       ,			output_type='numpy'       ,			).images

									snake_case__   :      int   				= image[0, 253:256, 253:256, -1]

									assert image.shape == (1, 512, 512, 3)
									snake_case__   :      List[str]   				= np.array([0.0_7_8_7, 0.0_8_4_9, 0.0_8_2_6, 0.0_8_1_2, 0.0_8_0_7, 0.0_7_9_5, 0.0_8_1_8, 0.0_7_9_8, 0.0_7_7_9]						)

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


