

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


a       :      List[Any]  	=			getLogger(__name__)


a       :      int  	=			4
a       :      Tuple  	=			7

a       :      Dict  	=			{
    "float16": torch.floataa,
    "float32": torch.floataa,
}





def 				lowercase (					__magic_name__							):



					'''simple docstring'''







					if isinstance(__magic_name__			,   torch.Tensor							):
										return image
					elif isinstance(__magic_name__			,   PIL.Image.Image							):
										UpperCAmelCase		:							Any        =       [image]

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

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

					return image







class   UpperCamelCase__   (       lowercase__       ):


		"""simple docstring"""

		def __init__(			self   ,      snake_case   ,      snake_case   ,      snake_case   ,      snake_case   ,      snake_case   ,      snake_case   ,      snake_case = 3_5_0   ,      ):


							'''simple docstring'''

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







		def __call__(			self   ,      snake_case   ,      snake_case   ,      snake_case = 7_5   ,      snake_case = 9.0   ,      snake_case = 2_0   ,      snake_case = None   ,      snake_case = 1   ,      snake_case = 0.0   ,      snake_case = None   ,      snake_case = None   ,      snake_case = None   ,      snake_case = None   ,      snake_case = "pil"   ,      snake_case = True   ,      snake_case = None   ,      snake_case = 1   ,      ):


							'''simple docstring'''

							self.check_inputs(snake_case   ,      snake_case   ,      snake_case   ,      snake_case      )

							# 2. Define call parameters
							if prompt is not None and isinstance(snake_case   ,      snake_case      ):
												UpperCAmelCase		:							Optional[int]        =       1
							elif prompt is not None and isinstance(snake_case   ,      snake_case      ):
												UpperCAmelCase		:							List[Any]        =       len(snake_case      )
							else:
												UpperCAmelCase		:							Dict        =       prompt_embeds.shape[0]

							UpperCAmelCase		:							Union[str, Any]        =       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.
							UpperCAmelCase		:							Union[str, Any]        =       guidance_scale > 1.0

							# 3. Encode input prompt
							UpperCAmelCase		:							int        =       self._encode_prompt(
							    snake_case   ,      snake_case   ,      snake_case   ,      snake_case   ,      snake_case   ,      prompt_embeds=snake_case   ,      negative_prompt_embeds=snake_case   ,      )

							UpperCAmelCase		:							Any        =       ORT_TO_PT_TYPE[str(text_embeddings.dtype      )]

							# 4. Preprocess image
							UpperCAmelCase		:							List[Any]        =       preprocess(snake_case      )
							UpperCAmelCase		:							Tuple        =       image.cpu()

							# 5. set timesteps
							self.scheduler.set_timesteps(snake_case   ,      device=snake_case      )
							UpperCAmelCase		:							List[Any]        =       self.scheduler.timesteps

							# 5. Add noise to image
							UpperCAmelCase		:							int        =       torch.tensor([noise_level]   ,      dtype=torch.long   ,      device=snake_case      )
							UpperCAmelCase		:							Any        =       torch.randn(image.shape   ,      generator=snake_case   ,      device=snake_case   ,      dtype=snake_case      )
							UpperCAmelCase		:							Any        =       self.low_res_scheduler.add_noise(snake_case   ,      snake_case   ,      snake_case      )

							UpperCAmelCase		:							List[str]        =       2 if do_classifier_free_guidance else 1
							UpperCAmelCase		:							int        =       np.concatenate([image] * batch_multiplier * num_images_per_prompt      )
							UpperCAmelCase		:							Optional[int]        =       np.concatenate([noise_level] * image.shape[0]      )

							# 6. Prepare latent variables
							UpperCAmelCase ,			UpperCAmelCase		:							Any        =       image.shape[2:]
							UpperCAmelCase		:							int        =       self.prepare_latents(
							    batch_size * num_images_per_prompt   ,      snake_case   ,      snake_case   ,      snake_case   ,      snake_case   ,      snake_case   ,      snake_case   ,      snake_case   ,      )

							# 7. Check that sizes of image and latents match
							UpperCAmelCase		:							Union[str, Any]        =       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
							UpperCAmelCase		:							str        =       self.prepare_extra_step_kwargs(snake_case   ,      snake_case      )

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

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

																	# concat latents, mask, masked_image_latents in the channel dimension
																	UpperCAmelCase		:							List[Any]        =       self.scheduler.scale_model_input(snake_case   ,      snake_case      )
																	UpperCAmelCase		:							int        =       np.concatenate([latent_model_input, image]   ,      axis=1      )

																	# timestep to tensor
																	UpperCAmelCase		:							Optional[Any]        =       np.array([t]   ,      dtype=snake_case      )

																	# predict the noise residual
																	UpperCAmelCase		:							Dict        =       self.unet(
																	    sample=snake_case   ,      timestep=snake_case   ,      encoder_hidden_states=snake_case   ,      class_labels=noise_level.astype(np.intaa      )   ,      )[0]

																	# perform guidance
																	if do_classifier_free_guidance:
																						UpperCAmelCase ,			UpperCAmelCase		:							Any        =       np.split(snake_case   ,      2      )
																	UpperCAmelCase		:							List[Any]        =       noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

																	# compute the previous noisy sample x_t -> x_t-1
																	UpperCAmelCase		:							List[str]        =       self.scheduler.step(
																	    torch.from_numpy(snake_case      )   ,      snake_case   ,      snake_case   ,      **snake_case      ).prev_sample

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

        # 10. Post-processing
							UpperCAmelCase		:							List[Any]        =       self.decode_latents(latents.float()      )

							# 11. Convert to PIL
							if output_type == "pil":
												UpperCAmelCase		:							int        =       self.numpy_to_pil(snake_case      )

							if not return_dict:
												return (image,)

							return ImagePipelineOutput(images=snake_case      )







		def 				A_	(			self   ,      snake_case      ):


							'''simple docstring'''

							UpperCAmelCase		:							Any        =       1 / 0.0_8333 * latents
							UpperCAmelCase		:							Any        =       self.vae(latent_sample=snake_case      )[0]
							UpperCAmelCase		:							List[str]        =       np.clip(image / 2 + 0.5   ,      0   ,      1      )
							UpperCAmelCase		:							Tuple        =       image.transpose((0, 2, 3, 1)      )
							return image







		def 				A_	(			self   ,      snake_case   ,      snake_case   ,      snake_case   ,      snake_case   ,      snake_case   ,      snake_case = None   ,      snake_case = None   ,      ):


							'''simple docstring'''

							if prompt is not None and isinstance(snake_case   ,      snake_case      ):
												UpperCAmelCase		:							Dict        =       1
							elif prompt is not None and isinstance(snake_case   ,      snake_case      ):
												UpperCAmelCase		:							Optional[Any]        =       len(snake_case      )
							else:
												UpperCAmelCase		:							Tuple        =       prompt_embeds.shape[0]

							if prompt_embeds is None:
												UpperCAmelCase		:							str        =       self.tokenizer(
												    snake_case   ,      padding="max_length"   ,      max_length=self.tokenizer.model_max_length   ,      truncation=snake_case   ,      return_tensors="pt"   ,      )
												UpperCAmelCase		:							Tuple        =       text_inputs.input_ids
												UpperCAmelCase		:							Dict        =       self.tokenizer(snake_case   ,      padding="longest"   ,      return_tensors="pt"      ).input_ids

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

							UpperCAmelCase ,			UpperCAmelCase ,			UpperCAmelCase		:							Any        =       prompt_embeds.shape
							# duplicate text embeddings for each generation per prompt, using mps friendly method
							UpperCAmelCase		:							List[Any]        =       prompt_embeds.repeat(1   ,      snake_case      )
							UpperCAmelCase		:							int        =       prompt_embeds.reshape(bs_embed * num_images_per_prompt   ,      snake_case   ,      -1      )

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

												UpperCAmelCase		:							int        =       text_input_ids.shape[-1]
												UpperCAmelCase		:							Optional[Any]        =       self.tokenizer(
												    snake_case   ,      padding="max_length"   ,      max_length=snake_case   ,      truncation=snake_case   ,      return_tensors="pt"   ,      )

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

												UpperCAmelCase		:							Optional[Any]        =       self.text_encoder(
												    input_ids=uncond_input.input_ids.int().to(snake_case      )   ,      )
												UpperCAmelCase		:							Any        =       uncond_embeddings[0]

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

							return prompt_embeds



'''simple docstring'''

def 				lowercase (					__magic_name__			,   __magic_name__							):



					'''simple docstring'''







					UpperCAmelCase		:							Dict        =       1.0
					for i in range(1			,   r + 1							):
										ret *= (n + 1 - i) / i
					return round(__magic_name__							)





def 				lowercase (					__magic_name__							):



					'''simple docstring'''







					return choose(8 + n			,   __magic_name__							) + choose(9 + n			,   __magic_name__							) - 10





def 				lowercase (					__magic_name__							):



					'''simple docstring'''







					return sum(non_bouncy_exact(__magic_name__							) for i in range(1			,   n + 1							)							)





def 				lowercase (					__magic_name__ = 100							):



					'''simple docstring'''







					return non_bouncy_upto(__magic_name__							)


if __name__ == "__main__":
				print(F'{solution() = }')
