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


snake_case    =		getLogger(__name__)


snake_case    =		4
snake_case    =		7

snake_case    =		{
    'float16': torch.floataa,
    'float32': torch.floataa,
}

def   UpperCamelCase_  (		SCREAMING_SNAKE_CASE_			):
					if isinstance(SCREAMING_SNAKE_CASE_,     torch.Tensor			):
										return image
					elif isinstance(SCREAMING_SNAKE_CASE_,     PIL.Image.Image			):
										SCREAMING_SNAKE_CASE				=	[image]

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

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

					return image




class      UpperCamelCase	(       __magic_name__      ):

				"""simple docstring"""



				def __init__(				self		,	lowercase__		,	lowercase__		,	lowercase__		,	lowercase__		,	lowercase__		,	lowercase__		,	lowercase__ = 350		,	)    ->    Optional[Any]:
									"""simple docstring"""







									super().__init__(
									    vae=lowercase__		,	text_encoder=lowercase__		,	tokenizer=lowercase__		,	unet=lowercase__		,	low_res_scheduler=lowercase__		,	scheduler=lowercase__		,	safety_checker=lowercase__		,	feature_extractor=lowercase__		,	watermarker=lowercase__		,	max_noise_level=lowercase__		,	)


				def __call__(				self		,	lowercase__		,	lowercase__		,	lowercase__ = 75		,	lowercase__ = 9.0		,	lowercase__ = 20		,	lowercase__ = None		,	lowercase__ = 1		,	lowercase__ = 0.0		,	lowercase__ = None		,	lowercase__ = None		,	lowercase__ = None		,	lowercase__ = None		,	lowercase__ = "pil"		,	lowercase__ = True		,	lowercase__ = None		,	lowercase__ = 1		,	)    ->    Dict:
									"""simple docstring"""







									self.check_inputs(lowercase__		,	lowercase__		,	lowercase__		,	lowercase__			)

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

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

									# 3. Encode input prompt
									SCREAMING_SNAKE_CASE				=	self._encode_prompt(
									    lowercase__		,	lowercase__		,	lowercase__		,	lowercase__		,	lowercase__		,	prompt_embeds=lowercase__		,	negative_prompt_embeds=lowercase__		,	)

									SCREAMING_SNAKE_CASE				=	ORT_TO_PT_TYPE[str(text_embeddings.dtype			)]

									# 4. Preprocess image
									SCREAMING_SNAKE_CASE				=	preprocess(lowercase__			)
									SCREAMING_SNAKE_CASE				=	image.cpu()

									# 5. set timesteps
									self.scheduler.set_timesteps(lowercase__		,	device=lowercase__			)
									SCREAMING_SNAKE_CASE				=	self.scheduler.timesteps

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

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

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

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

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

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

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

																			# timestep to tensor
																			SCREAMING_SNAKE_CASE				=	np.array([t]		,	dtype=lowercase__			)

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

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

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

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

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

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

									if not return_dict:
														return (image,)

									return ImagePipelineOutput(images=lowercase__			)


				def 							A					(				self		,	lowercase__			)    ->    int:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE				=	1 / 0.08_333 * latents
									SCREAMING_SNAKE_CASE				=	self.vae(latent_sample=lowercase__			)[0]
									SCREAMING_SNAKE_CASE				=	np.clip(image / 2 + 0.5		,	0		,	1			)
									SCREAMING_SNAKE_CASE				=	image.transpose((0, 2, 3, 1)			)
									return image




				def 							A					(				self		,	lowercase__		,	lowercase__		,	lowercase__		,	lowercase__		,	lowercase__		,	lowercase__ = None		,	lowercase__ = None		,	)    ->    Optional[int]:
									"""simple docstring"""







									if prompt is not None and isinstance(lowercase__		,	lowercase__			):
														SCREAMING_SNAKE_CASE				=	1
									elif prompt is not None and isinstance(lowercase__		,	lowercase__			):
														SCREAMING_SNAKE_CASE				=	len(lowercase__			)
									else:
														SCREAMING_SNAKE_CASE				=	prompt_embeds.shape[0]

									if prompt_embeds is None:
														SCREAMING_SNAKE_CASE				=	self.tokenizer(
														    lowercase__		,	padding='max_length'		,	max_length=self.tokenizer.model_max_length		,	truncation=lowercase__		,	return_tensors='pt'		,	)
														SCREAMING_SNAKE_CASE				=	text_inputs.input_ids
														SCREAMING_SNAKE_CASE				=	self.tokenizer(lowercase__		,	padding='longest'		,	return_tensors='pt'			).input_ids

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

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

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

														SCREAMING_SNAKE_CASE				=	text_input_ids.shape[-1]
														SCREAMING_SNAKE_CASE				=	self.tokenizer(
														    lowercase__		,	padding='max_length'		,	max_length=lowercase__		,	truncation=lowercase__		,	return_tensors='pt'		,	)

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

														SCREAMING_SNAKE_CASE				=	self.text_encoder(
														    input_ids=uncond_input.input_ids.int().to(lowercase__			)		,	)
														SCREAMING_SNAKE_CASE				=	uncond_embeddings[0]

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

									return prompt_embeds




from __future__ import annotations

from collections.abc import Callable







def   lowerCamelCase(      a__		,a__		,a__		,a__ = 100		,):
				_SCREAMING_SNAKE_CASE	     =x_start
				_SCREAMING_SNAKE_CASE	     =fnc(a__)
				_SCREAMING_SNAKE_CASE	     =0.0

				for _ in range(a__):
								# Approximates small segments of curve as linear and solve
								# for trapezoidal area
								_SCREAMING_SNAKE_CASE	     =(x_end - x_start) / steps + xa
								_SCREAMING_SNAKE_CASE	     =fnc(a__)
								area += abs(fxa + fxa) * (xa - xa) / 2

								# Increment step
								_SCREAMING_SNAKE_CASE	     =xa
								_SCREAMING_SNAKE_CASE	     =fxa
				return area


if __name__ == "__main__":







							def   lowerCamelCase(      a__):
											return x**3

							print('''f(x) = x^3''')
							print('''The area between the curve, x = -10, x = 10 and the x axis is:''')
							snake_case_  :	Any							  =							10
							while i <= 10_00_00:
														snake_case_  :	Tuple							  =							trapezoidal_area(f, -5, 5, i)
														print(f"""with {i} steps: {area}""")
														i *= 10