


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


UpperCAmelCase_		:       Dict							       =		getLogger(__name__)


UpperCAmelCase_		:       List[str]							       =		4
UpperCAmelCase_		:       Any							       =		7

UpperCAmelCase_		:       Union[str, Any]							       =		{
    'float16': torch.floataa,
    'float32': torch.floataa,
}







def       snake_case_       (  SCREAMING_SNAKE_CASE__					):







				"""simple docstring"""





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

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

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

				return image







class    lowercase__  (						_snake_case ):





							'''simple docstring'''


							def __init__(   self   , __snake_case   , __snake_case   , __snake_case   , __snake_case   , __snake_case   , __snake_case   , __snake_case = 350   , ):
											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 = 75   , __snake_case = 9.0   , __snake_case = 20   , __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   , ):
											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  ):
															_SCREAMING_SNAKE_CASE			:	Any							=				1
											elif prompt is not None and isinstance(__snake_case   , __snake_case  ):
															_SCREAMING_SNAKE_CASE			:	Union[str, Any]							=				len(__snake_case  )
											else:
															_SCREAMING_SNAKE_CASE			:	List[str]							=				prompt_embeds.shape[0]

											_SCREAMING_SNAKE_CASE			:	Tuple							=				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			:	List[str]							=				guidance_scale > 1.0

											# 3. Encode input prompt
											_SCREAMING_SNAKE_CASE			:	Dict							=				self._encode_prompt(
											    __snake_case   , __snake_case   , __snake_case   , __snake_case   , __snake_case   , prompt_embeds=__snake_case   , negative_prompt_embeds=__snake_case   , )

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

											# 4. Preprocess image
											_SCREAMING_SNAKE_CASE			:	int							=				preprocess(__snake_case  )
											_SCREAMING_SNAKE_CASE			:	Any							=				image.cpu()

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

											# 5. Add noise to image
											_SCREAMING_SNAKE_CASE			:	Dict							=				torch.tensor([noise_level]   , dtype=torch.long   , device=__snake_case  )
											_SCREAMING_SNAKE_CASE			:	Union[str, Any]							=				torch.randn(image.shape   , generator=__snake_case   , device=__snake_case   , dtype=__snake_case  )
											_SCREAMING_SNAKE_CASE			:	Optional[Any]							=				self.low_res_scheduler.add_noise(__snake_case   , __snake_case   , __snake_case  )

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

											# 6. Prepare latent variables
											_SCREAMING_SNAKE_CASE		,	_SCREAMING_SNAKE_CASE			:	int							=				image.shape[2:]
											_SCREAMING_SNAKE_CASE			:	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
											_SCREAMING_SNAKE_CASE			:	Optional[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
											_SCREAMING_SNAKE_CASE			:	List[Any]							=				self.prepare_extra_step_kwargs(__snake_case   , __snake_case  )

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

											# 9. Denoising loop
											_SCREAMING_SNAKE_CASE			:	Optional[int]							=				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
																			_SCREAMING_SNAKE_CASE			:	str							=				np.concatenate([latents] * 2  ) if do_classifier_free_guidance else latents

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

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

																			# predict the noise residual
																			_SCREAMING_SNAKE_CASE			:	Union[str, Any]							=				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:
																							_SCREAMING_SNAKE_CASE		,	_SCREAMING_SNAKE_CASE			:	Optional[int]							=				np.split(__snake_case   , 2  )
																			_SCREAMING_SNAKE_CASE			:	Optional[Any]							=				noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

																			# compute the previous noisy sample x_t -> x_t-1
																			_SCREAMING_SNAKE_CASE			:	Dict							=				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
											_SCREAMING_SNAKE_CASE			:	Union[str, Any]							=				self.decode_latents(latents.float()  )

											# 11. Convert to PIL
											if output_type == "pil":
															_SCREAMING_SNAKE_CASE			:	str							=				self.numpy_to_pil(__snake_case  )

											if not return_dict:
															return (image,)

											return ImagePipelineOutput(images=__snake_case  )






							def UpperCAmelCase_     (   self   , __snake_case  ):
											_SCREAMING_SNAKE_CASE			:	str							=				1 / 0.08333 * latents
											_SCREAMING_SNAKE_CASE			:	str							=				self.vae(latent_sample=__snake_case  )[0]
											_SCREAMING_SNAKE_CASE			:	Optional[Any]							=				np.clip(image / 2 + 0.5   , 0   , 1  )
											_SCREAMING_SNAKE_CASE			:	Union[str, Any]							=				image.transpose((0, 2, 3, 1)  )
											return image






							def UpperCAmelCase_     (   self   , __snake_case   , __snake_case   , __snake_case   , __snake_case   , __snake_case   , __snake_case = None   , __snake_case = None   , ):
											if prompt is not None and isinstance(__snake_case   , __snake_case  ):
															_SCREAMING_SNAKE_CASE			:	Any							=				1
											elif prompt is not None and isinstance(__snake_case   , __snake_case  ):
															_SCREAMING_SNAKE_CASE			:	List[Any]							=				len(__snake_case  )
											else:
															_SCREAMING_SNAKE_CASE			:	Optional[Any]							=				prompt_embeds.shape[0]

											if prompt_embeds is None:
															_SCREAMING_SNAKE_CASE			:	Optional[Any]							=				self.tokenizer(
															    __snake_case   , padding="""max_length"""   , max_length=self.tokenizer.model_max_length   , truncation=__snake_case   , return_tensors="""pt"""   , )
															_SCREAMING_SNAKE_CASE			:	Tuple							=				text_inputs.input_ids
															_SCREAMING_SNAKE_CASE			:	Tuple							=				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  ):
																			_SCREAMING_SNAKE_CASE			:	str							=				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			:	List[Any]							=				self.text_encoder(
															    input_ids=text_input_ids.int().to(__snake_case  )   , )
															_SCREAMING_SNAKE_CASE			:	Union[str, Any]							=				prompt_embeds[0]

											_SCREAMING_SNAKE_CASE		,	_SCREAMING_SNAKE_CASE		,	_SCREAMING_SNAKE_CASE			:	Any							=				prompt_embeds.shape
											# duplicate text embeddings for each generation per prompt, using mps friendly method
											_SCREAMING_SNAKE_CASE			:	str							=				prompt_embeds.repeat(1   , __snake_case  )
											_SCREAMING_SNAKE_CASE			:	Any							=				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:
															_SCREAMING_SNAKE_CASE			:	List[str]
															if negative_prompt is None:
																			_SCREAMING_SNAKE_CASE			:	str							=				[""""""] * 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  ):
																			_SCREAMING_SNAKE_CASE			:	Optional[Any]							=				[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:
																			_SCREAMING_SNAKE_CASE			:	Tuple							=				negative_prompt

															_SCREAMING_SNAKE_CASE			:	Any							=				text_input_ids.shape[-1]
															_SCREAMING_SNAKE_CASE			:	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

															_SCREAMING_SNAKE_CASE			:	Union[str, Any]							=				self.text_encoder(
															    input_ids=uncond_input.input_ids.int().to(__snake_case  )   , )
															_SCREAMING_SNAKE_CASE			:	Any							=				uncond_embeddings[0]

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

											return prompt_embeds








'''simple docstring'''


import sys







def       snake_case_       (  SCREAMING_SNAKE_CASE__							,				SCREAMING_SNAKE_CASE__					):







				"""simple docstring"""





				if not numbers:
								return 0
				if target == 0 and target in numbers:
								return 0
				if not isinstance(SCREAMING_SNAKE_CASE__							,				(list, tuple)					) or not all(
				    isinstance(SCREAMING_SNAKE_CASE__							,				SCREAMING_SNAKE_CASE__					) for number in numbers					):
								raise ValueError("""numbers must be an iterable of integers"""					)

				_SCREAMING_SNAKE_CASE			:	str							=				0
				_SCREAMING_SNAKE_CASE			:	int							=				sys.maxsize

				while right < len(SCREAMING_SNAKE_CASE__					):
								curr_sum += numbers[right]
								while curr_sum >= target and left <= right:
												_SCREAMING_SNAKE_CASE			:	Any							=				min(SCREAMING_SNAKE_CASE__							,				right - left + 1					)
												curr_sum -= numbers[left]
												left += 1
								right += 1

				return 0 if min_len == sys.maxsize else min_len




