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


lowercase		:    List[str]			=       getLogger(__name__)


lowercase		:    Optional[int]			=       4
lowercase		:    Any			=       7

lowercase		:    Any			=       {
    'float16': torch.floataa,
    'float32': torch.floataa,
}


def      lowerCamelCase_     (						lowerCamelCase_			)  ->      Any:


					'''simple docstring'''
					if isinstance(lowerCamelCase_						,					torch.Tensor			):
										return image
					elif isinstance(lowerCamelCase_						,					PIL.Image.Image			):
										__a:    str		     =       [image]

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

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

					return image




class  _snake_case	(    _lowercase							):






							def __init__(       self    :      List[str]    ,			snake_case__    :      OnnxRuntimeModel    ,			snake_case__    :      OnnxRuntimeModel    ,			snake_case__    :      Any    ,			snake_case__    :      OnnxRuntimeModel    ,			snake_case__    :      DDPMScheduler    ,			snake_case__    :      Any    ,			snake_case__    :      int = 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    :      int    ,			snake_case__    :      Union[str, List[str]]    ,			snake_case__    :      Union[torch.FloatTensor, PIL.Image.Image, List[PIL.Image.Image]]    ,			snake_case__    :      int = 7_5    ,			snake_case__    :      float = 9.0    ,			snake_case__    :      int = 2_0    ,			snake_case__    :      Optional[Union[str, List[str]]] = None    ,			snake_case__    :      Optional[int] = 1    ,			snake_case__    :      float = 0.0    ,			snake_case__    :      Optional[Union[torch.Generator, List[torch.Generator]]] = None    ,			snake_case__    :      Optional[torch.FloatTensor] = None    ,			snake_case__    :      Optional[np.ndarray] = None    ,			snake_case__    :      Optional[np.ndarray] = None    ,			snake_case__    :      Optional[str] = "pil"    ,			snake_case__    :      bool = True    ,			snake_case__    :      Optional[Callable[[int, int, torch.FloatTensor], None]] = None    ,			snake_case__    :      Optional[int] = 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__     ):
																	__a:    List[Any]		     =       1
												elif prompt is not None and isinstance(snake_case__    ,			snake_case__     ):
																	__a:    Optional[int]		     =       len(snake_case__     )
												else:
																	__a:    Optional[Any]		     =       prompt_embeds.shape[0]

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

												# 3. Encode input prompt
												__a:    Union[str, Any]		     =       self._encode_prompt(
												    snake_case__    ,			snake_case__    ,			snake_case__    ,			snake_case__    ,			snake_case__    ,			prompt_embeds=snake_case__    ,			negative_prompt_embeds=snake_case__    ,			)

												__a:    Dict		     =       ORT_TO_PT_TYPE[str(text_embeddings.dtype     )]

												# 4. Preprocess image
												__a:    Union[str, Any]		     =       preprocess(snake_case__     )
												__a:    Optional[Any]		     =       image.cpu()

												# 5. set timesteps
												self.scheduler.set_timesteps(snake_case__    ,			device=snake_case__     )
												__a:    int		     =       self.scheduler.timesteps

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

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

												# 6. Prepare latent variables
												__a     ,					__a:    Any		     =       image.shape[2:]
												__a:    Optional[Any]		     =       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
												__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:    Tuple		     =       self.prepare_extra_step_kwargs(snake_case__    ,			snake_case__     )

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

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

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

																						# timestep to tensor
																						__a:    Union[str, Any]		     =       np.array([t]    ,			dtype=snake_case__     )

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

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

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

												if not return_dict:
																	return (image,)

												return ImagePipelineOutput(images=snake_case__     )






							def 				__lowercase	(       self    :      List[str]    ,			snake_case__    :      str     ):






												"""simple docstring"""



												__a:    int		     =       1 / 0.08_333 * latents
												__a:    Optional[Any]		     =       self.vae(latent_sample=snake_case__     )[0]
												__a:    List[str]		     =       np.clip(image / 2 + 0.5    ,			0    ,			1     )
												__a:    int		     =       image.transpose((0, 2, 3, 1)     )
												return image






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






												"""simple docstring"""



												if prompt is not None and isinstance(snake_case__    ,			snake_case__     ):
																	__a:    Dict		     =       1
												elif prompt is not None and isinstance(snake_case__    ,			snake_case__     ):
																	__a:    int		     =       len(snake_case__     )
												else:
																	__a:    str		     =       prompt_embeds.shape[0]

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

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

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

																	__a:    List[str]		     =       self.text_encoder(
																	    input_ids=uncond_input.input_ids.int().to(snake_case__     )    ,			)
																	__a:    Optional[int]		     =       uncond_embeddings[0]

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

												return prompt_embeds


"""simple docstring"""
from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available, is_vision_available


lowercase		:    Dict			=       {
    'configuration_mobilenet_v1': [
        'MOBILENET_V1_PRETRAINED_CONFIG_ARCHIVE_MAP',
        'MobileNetV1Config',
        'MobileNetV1OnnxConfig',
    ],
}

try:
			if not is_vision_available():
						raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
			pass
else:
			lowercase		:    Optional[int]			=       ['MobileNetV1FeatureExtractor']
			lowercase		:    Union[str, Any]			=       ['MobileNetV1ImageProcessor']

try:
			if not is_torch_available():
						raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
			pass
else:
			lowercase		:    Tuple			=       [
			    'MOBILENET_V1_PRETRAINED_MODEL_ARCHIVE_LIST',
			    'MobileNetV1ForImageClassification',
			    'MobileNetV1Model',
			    'MobileNetV1PreTrainedModel',
			    'load_tf_weights_in_mobilenet_v1',
			]


if TYPE_CHECKING:
			from .configuration_mobilenet_va import (
			    MOBILENET_V1_PRETRAINED_CONFIG_ARCHIVE_MAP,
			    MobileNetVaConfig,
			    MobileNetVaOnnxConfig,
			)

			try:
						if not is_vision_available():
									raise OptionalDependencyNotAvailable()
			except OptionalDependencyNotAvailable:
						pass
			else:
						from .feature_extraction_mobilenet_va import MobileNetVaFeatureExtractor
						from .image_processing_mobilenet_va import MobileNetVaImageProcessor

			try:
						if not is_torch_available():
									raise OptionalDependencyNotAvailable()
			except OptionalDependencyNotAvailable:
						pass
			else:
						from .modeling_mobilenet_va import (
						    MOBILENET_V1_PRETRAINED_MODEL_ARCHIVE_LIST,
						    MobileNetVaForImageClassification,
						    MobileNetVaModel,
						    MobileNetVaPreTrainedModel,
						    load_tf_weights_in_mobilenet_va,
						)


else:
			import sys

			lowercase		:    Union[str, Any]			=       _LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)
