



'''simple docstring'''



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						:		Tuple						      = getLogger(__name__)


lowerCAmelCase						:		Tuple						      = 4
lowerCAmelCase						:		Tuple						      = 7

lowerCAmelCase						:		Optional[int]						      = {
    '''float16''': torch.floataa,
    '''float32''': torch.floataa,
}



def 	__lowerCAmelCase       (				lowerCamelCase :    List[str]       ):






	'''simple docstring'''




	if isinstance(lowerCamelCase					,		torch.Tensor       ):
		return image
	elif isinstance(lowerCamelCase					,		PIL.Image.Image       ):
		__lowerCAmelCase       				=	[image]

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

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

	return image






class        UpperCAmelCase__						(		UpperCamelCase__			):



					def __init__(     self      ,					UpperCamelCase      ,					UpperCamelCase      ,					UpperCamelCase      ,					UpperCamelCase      ,					UpperCamelCase      ,					UpperCamelCase      ,					UpperCamelCase = 350      ,					)  ->      Optional[int]:
						super().__init__(
						    vae=UpperCamelCase      ,					text_encoder=UpperCamelCase      ,					tokenizer=UpperCamelCase      ,					unet=UpperCamelCase      ,					low_res_scheduler=UpperCamelCase      ,					scheduler=UpperCamelCase      ,					safety_checker=UpperCamelCase      ,					feature_extractor=UpperCamelCase      ,					watermarker=UpperCamelCase      ,					max_noise_level=UpperCamelCase      ,					)



					def __call__(     self      ,					UpperCamelCase      ,					UpperCamelCase      ,					UpperCamelCase = 75      ,					UpperCamelCase = 9.0      ,					UpperCamelCase = 20      ,					UpperCamelCase = None      ,					UpperCamelCase = 1      ,					UpperCamelCase = 0.0      ,					UpperCamelCase = None      ,					UpperCamelCase = None      ,					UpperCamelCase = None      ,					UpperCamelCase = None      ,					UpperCamelCase = "pil"      ,					UpperCamelCase = True      ,					UpperCamelCase = None      ,					UpperCamelCase = 1      ,					)  ->      Optional[int]:
						self.check_inputs(UpperCamelCase      ,					UpperCamelCase      ,					UpperCamelCase      ,					UpperCamelCase  )

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

						__lowerCAmelCase       				=	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.
						__lowerCAmelCase       				=	guidance_scale > 1.0

						# 3. Encode input prompt
						__lowerCAmelCase       				=	self._encode_prompt(
						    UpperCamelCase      ,					UpperCamelCase      ,					UpperCamelCase      ,					UpperCamelCase      ,					UpperCamelCase      ,					prompt_embeds=UpperCamelCase      ,					negative_prompt_embeds=UpperCamelCase      ,					)

						__lowerCAmelCase       				=	ORT_TO_PT_TYPE[str(text_embeddings.dtype  )]

						# 4. Preprocess image
						__lowerCAmelCase       				=	preprocess(UpperCamelCase  )
						__lowerCAmelCase       				=	image.cpu()

						# 5. set timesteps
						self.scheduler.set_timesteps(UpperCamelCase      ,					device=UpperCamelCase  )
						__lowerCAmelCase       				=	self.scheduler.timesteps

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

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

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

						# 7. Check that sizes of image and latents match
						__lowerCAmelCase       				=	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
						__lowerCAmelCase       				=	self.prepare_extra_step_kwargs(UpperCamelCase      ,					UpperCamelCase  )

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

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

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

								# timestep to tensor
								__lowerCAmelCase       				=	np.array([t]      ,					dtype=UpperCamelCase  )

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

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

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

								# call the callback, if provided
								if i == len(UpperCamelCase  ) - 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(UpperCamelCase      ,					UpperCamelCase      ,					UpperCamelCase  )

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

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

						if not return_dict:
							return (image,)

						return ImagePipelineOutput(images=UpperCamelCase  )



					def 							UpperCAmelCase_			(     self      ,					UpperCamelCase  )  ->      Dict:
						__lowerCAmelCase       				=	1 / 0.0_83_33 * latents
						__lowerCAmelCase       				=	self.vae(latent_sample=UpperCamelCase  )[0]
						__lowerCAmelCase       				=	np.clip(image / 2 + 0.5      ,					0      ,					1  )
						__lowerCAmelCase       				=	image.transpose((0, 2, 3, 1)  )
						return image



					def 							UpperCAmelCase_			(     self      ,					UpperCamelCase      ,					UpperCamelCase      ,					UpperCamelCase      ,					UpperCamelCase      ,					UpperCamelCase      ,					UpperCamelCase = None      ,					UpperCamelCase = None      ,					)  ->      Optional[Any]:
						if prompt is not None and isinstance(UpperCamelCase      ,					UpperCamelCase  ):
							__lowerCAmelCase       				=	1
						elif prompt is not None and isinstance(UpperCamelCase      ,					UpperCamelCase  ):
							__lowerCAmelCase       				=	len(UpperCamelCase  )
						else:
							__lowerCAmelCase       				=	prompt_embeds.shape[0]

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

							if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(
							    UpperCamelCase      ,					UpperCamelCase  ):
								__lowerCAmelCase       				=	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
							__lowerCAmelCase       				=	self.text_encoder(
							    input_ids=text_input_ids.int().to(UpperCamelCase  )      ,					)
							__lowerCAmelCase       				=	prompt_embeds[0]

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

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

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

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

							__lowerCAmelCase       				=	self.text_encoder(
							    input_ids=uncond_input.input_ids.int().to(UpperCamelCase  )      ,					)
							__lowerCAmelCase       				=	uncond_embeddings[0]

						if do_classifier_free_guidance:
							__lowerCAmelCase       				=	uncond_embeddings.shape[1]
							# duplicate unconditional embeddings for each generation per prompt, using mps friendly method
							__lowerCAmelCase       				=	uncond_embeddings.repeat(1      ,					UpperCamelCase  )
							__lowerCAmelCase       				=	uncond_embeddings.reshape(batch_size * num_images_per_prompt      ,					UpperCamelCase      ,					-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
							__lowerCAmelCase       				=	np.concatenate([uncond_embeddings, prompt_embeds]  )

						return prompt_embeds




'''simple docstring'''



def 	__lowerCAmelCase       (				lowerCamelCase :    float					,		lowerCamelCase :    float       ):






	'''simple docstring'''




	__lowerCAmelCase       				=	float()
	try:
		__lowerCAmelCase       				=	mass * acceleration
	except Exception:
		return -0.0
	return force


if __name__ == "__main__":
	import doctest

	# run doctest
	doctest.testmod()

	# demo
	lowerCAmelCase						:		List[Any]						      = 12.5
	lowerCAmelCase						:		Optional[Any]						      = 1_0
	lowerCAmelCase						:		str						      = newtons_second_law_of_motion(mass, acceleration)
	print('''The force is ''', force, '''N''')