




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


_lowerCamelCase	:      Optional[Any]   =							4
_lowerCamelCase	:      str   =							7

_lowerCamelCase	:      Optional[Any]   =							{
    'float16': torch.floataa,
    'float32': torch.floataa,
}

def 		lowercase_   (		_UpperCAmelCase						):




	"""simple docstring"""






	if isinstance(_UpperCAmelCase	,   torch.Tensor						):
		return image
	elif isinstance(_UpperCAmelCase	,   PIL.Image.Image						):
		A_	:					Optional[Any]						 =		[image]

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

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

	return image

class 							lowercase   (     __UpperCAmelCase):







				def __init__(				self :       str   ,  _lowerCamelCase :       OnnxRuntimeModel   ,  _lowerCamelCase :       OnnxRuntimeModel   ,  _lowerCamelCase :       Any   ,  _lowerCamelCase :       OnnxRuntimeModel   ,  _lowerCamelCase :       DDPMScheduler   ,  _lowerCamelCase :       Any   ,  _lowerCamelCase :       int = 3_50   ,  ):



					"""simple docstring"""

					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 :       List[Any]   ,  _lowerCamelCase :       Union[str, List[str]]   ,  _lowerCamelCase :       Union[torch.FloatTensor, PIL.Image.Image, List[PIL.Image.Image]]   ,  _lowerCamelCase :       int = 75   ,  _lowerCamelCase :       float = 9.0   ,  _lowerCamelCase :       int = 20   ,  _lowerCamelCase :       Optional[Union[str, List[str]]] = None   ,  _lowerCamelCase :       Optional[int] = 1   ,  _lowerCamelCase :       float = 0.0   ,  _lowerCamelCase :       Optional[Union[torch.Generator, List[torch.Generator]]] = None   ,  _lowerCamelCase :       Optional[torch.FloatTensor] = None   ,  _lowerCamelCase :       Optional[np.ndarray] = None   ,  _lowerCamelCase :       Optional[np.ndarray] = None   ,  _lowerCamelCase :       Optional[str] = "pil"   ,  _lowerCamelCase :       bool = True   ,  _lowerCamelCase :       Optional[Callable[[int, int, torch.FloatTensor], None]] = None   ,  _lowerCamelCase :       Optional[int] = 1   ,  ):



					"""simple docstring"""

					self.check_inputs(_lowerCamelCase   ,  _lowerCamelCase   ,  _lowerCamelCase   ,  _lowerCamelCase							)

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

					A_	:					Optional[int]						 =		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.
					A_	:					List[str]						 =		guidance_scale > 1.0

					# 3. Encode input prompt
					A_	:					Optional[Any]						 =		self._encode_prompt(
					    _lowerCamelCase   ,  _lowerCamelCase   ,  _lowerCamelCase   ,  _lowerCamelCase   ,  _lowerCamelCase   ,  prompt_embeds=_lowerCamelCase   ,  negative_prompt_embeds=_lowerCamelCase   ,  )

					A_	:					Tuple						 =		ORT_TO_PT_TYPE[str(text_embeddings.dtype							)]

					# 4. Preprocess image
					A_	:					Optional[Any]						 =		preprocess(_lowerCamelCase							)
					A_	:					List[Any]						 =		image.cpu()

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

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

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

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

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

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

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

							# timestep to tensor
							A_	:					Optional[int]						 =		np.array([t]   ,  dtype=_lowerCamelCase							)

							# predict the noise residual
							A_	:					Any						 =		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:
								A_ ,			A_	:					List[Any]						 =		np.split(_lowerCamelCase   ,  2							)
							A_	:					Tuple						 =		noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

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

					# 11. Convert to PIL
					if output_type == "pil":
						A_	:					Any						 =		self.numpy_to_pil(_lowerCamelCase							)

					if not return_dict:
						return (image,)

					return ImagePipelineOutput(images=_lowerCamelCase							)







				def        a_	(				self :       str   ,  _lowerCamelCase :       List[Any]							):



					"""simple docstring"""

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







				def        a_	(				self :       Dict   ,  _lowerCamelCase :       Union[str, List[str]]   ,  _lowerCamelCase :       Tuple   ,  _lowerCamelCase :       Optional[int]   ,  _lowerCamelCase :       bool   ,  _lowerCamelCase :       Optional[str]   ,  _lowerCamelCase :       Optional[torch.FloatTensor] = None   ,  _lowerCamelCase :       Optional[torch.FloatTensor] = None   ,  ):



					"""simple docstring"""

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

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

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

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

						A_	:					List[str]						 =		self.text_encoder(
						    input_ids=uncond_input.input_ids.int().to(_lowerCamelCase							)   ,  )
						A_	:					Any						 =		uncond_embeddings[0]

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

					return prompt_embeds











"""simple docstring"""

from __future__ import annotations

def 		lowercase_   (		_UpperCAmelCase	,   _UpperCAmelCase	,   _UpperCAmelCase						):




	"""simple docstring"""






	assert (
	    c % greatest_common_divisor(_UpperCAmelCase	,   _UpperCAmelCase						) == 0
	)  # greatest_common_divisor(a,b) function implemented below
	((A_) ,			(A_) ,			(A_))	:					List[Any]						 =		extended_gcd(_UpperCAmelCase	,   _UpperCAmelCase						)  # extended_gcd(a,b) function implemented below
	A_	:					int						 =		c / d
	return (r * x, r * y)

def 		lowercase_   (		_UpperCAmelCase	,   _UpperCAmelCase	,   _UpperCAmelCase	,   _UpperCAmelCase = 2						):




	"""simple docstring"""






	((A_) ,			(A_))	:					Any						 =		diophantine(_UpperCAmelCase	,   _UpperCAmelCase	,   _UpperCAmelCase						)  # Initial value
	A_	:					Union[str, Any]						 =		greatest_common_divisor(_UpperCAmelCase	,   _UpperCAmelCase						)
	A_	:					List[str]						 =		a // d
	A_	:					str						 =		b // d

	for i in range(_UpperCAmelCase						):
		A_	:					int						 =		xa + i * q
		A_	:					str						 =		ya - i * p
		print(_UpperCAmelCase	,   _UpperCAmelCase						)

def 		lowercase_   (		_UpperCAmelCase	,   _UpperCAmelCase						):




	"""simple docstring"""






	if a < b:
		A_ ,			A_	:					Tuple						 =		b, a

	while a % b != 0:
		A_ ,			A_	:					int						 =		b, a % b

	return b

def 		lowercase_   (		_UpperCAmelCase	,   _UpperCAmelCase						):




	"""simple docstring"""






	assert a >= 0 and b >= 0

	if b == 0:
		A_ ,			A_ ,			A_	:					str						 =		a, 1, 0
	else:
		((A_) ,			(A_) ,			(A_))	:					List[str]						 =		extended_gcd(_UpperCAmelCase	,   a % b						)
		A_	:					Optional[int]						 =		q
		A_	:					Tuple						 =		p - q * (a // b)

	assert a % d == 0 and b % d == 0
	assert d == a * x + b * y

	return (d, x, y)


if __name__ == "__main__":
		from doctest import testmod

		testmod(name='diophantine', verbose=True)
		testmod(name='diophantine_all_soln', verbose=True)
		testmod(name='extended_gcd', verbose=True)
		testmod(name='greatest_common_divisor', verbose=True)





