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


_A       =					4
_A       =					7

_A       =					{
    "float16": torch.floataa,
    "float32": torch.floataa,
}






def        lowerCamelCase__    (    __lowerCAmelCase  :  List[Any]    ):




			"""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    ) / 255.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      _lowerCAmelCase (     __a	):




				def __init__( self    ,    _UpperCamelCase    ,    _UpperCamelCase    ,    _UpperCamelCase    ,    _UpperCamelCase    ,    _UpperCamelCase    ,    _UpperCamelCase    ,    _UpperCamelCase = 350    ,    )    ->				Any:
							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 					__a						( self    ,    _UpperCamelCase					)    ->				Tuple:
							lowerCAmelCase_      =     1 / 0.08333 * 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 					__a						( self    ,    _UpperCamelCase    ,    _UpperCamelCase    ,    _UpperCamelCase    ,    _UpperCamelCase    ,    _UpperCamelCase    ,    _UpperCamelCase = None    ,    _UpperCamelCase = None    ,    )    ->				List[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





import math






def        lowerCamelCase__    (    __lowerCAmelCase  :  int    ):




			"""simple docstring"""


			if 1 < number < 4:
						# 2 and 3 are primes
						return True
			elif number < 2 or number % 2 == 0 or number % 3 == 0:
						# Negatives, 0, 1, all even numbers, all multiples of 3 are not primes
						return False

			# All primes number are in format of 6k +/- 1
			for i in range(5   ,							int(math.sqrt(__lowerCAmelCase    ) + 1    )   ,							6    ):
						if number % i == 0 or number % (i + 2) == 0:
									return False
			return True






def        lowerCamelCase__    (    __lowerCAmelCase  :  int = 600851475143    ):




			"""simple docstring"""


			try:
						lowerCAmelCase_      =     int(__lowerCAmelCase    )
			except (TypeError, ValueError):
						raise TypeError("Parameter n must be int or castable to int."    )
			if n <= 0:
						raise ValueError("Parameter n must be greater than or equal to one."    )
			lowerCAmelCase_      =     0
			if is_prime(__lowerCAmelCase    ):
						return n
			while n % 2 == 0:
						n //= 2
			if is_prime(__lowerCAmelCase    ):
						return n
			for i in range(3   ,							int(math.sqrt(__lowerCAmelCase    )    ) + 1   ,							2    ):
						if n % i == 0:
									if is_prime(n // i    ):
												lowerCAmelCase_      =     n // i
												break
									elif is_prime(__lowerCAmelCase    ):
												lowerCAmelCase_      =     i
			return max_number


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



