

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  snake_case_(_UpperCamelCase	)				->      str:



  """simple docstring"""






  if isinstance(_UpperCamelCase					,			torch.Tensor	):
    return image
  elif isinstance(_UpperCamelCase					,			PIL.Image.Image	):
    _snake_case		=				[image]

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

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

  return image



class     lowercase_       (    __lowercase	):



      def __init__(     self :							str     ,   A__ :							OnnxRuntimeModel     ,   A__ :							OnnxRuntimeModel     ,   A__ :							Any     ,   A__ :							OnnxRuntimeModel     ,   A__ :							DDPMScheduler     ,   A__ :							Any     ,   A__ :							int = 350     ,   )       ->	Dict:
        super().__init__(
            vae=A__     ,   text_encoder=A__     ,   tokenizer=A__     ,   unet=A__     ,   low_res_scheduler=A__     ,   scheduler=A__     ,   safety_checker=A__     ,   feature_extractor=A__     ,   watermarker=A__     ,   max_noise_level=A__     ,   )



      def __call__(     self :							List[str]     ,   A__ :							Union[str, List[str]]     ,   A__ :							Union[torch.FloatTensor, PIL.Image.Image, List[PIL.Image.Image]]     ,   A__ :							int = 75     ,   A__ :							float = 9.0     ,   A__ :							int = 20     ,   A__ :							Optional[Union[str, List[str]]] = None     ,   A__ :							Optional[int] = 1     ,   A__ :							float = 0.0     ,   A__ :							Optional[Union[torch.Generator, List[torch.Generator]]] = None     ,   A__ :							Optional[torch.FloatTensor] = None     ,   A__ :							Optional[np.ndarray] = None     ,   A__ :							Optional[np.ndarray] = None     ,   A__ :							Optional[str] = "pil"     ,   A__ :							bool = True     ,   A__ :							Optional[Callable[[int, int, torch.FloatTensor], None]] = None     ,   A__ :							Optional[int] = 1     ,   )       ->	List[Any]:
        self.check_inputs(A__     ,   A__     ,   A__     ,   A__						)

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

        _snake_case		=				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.
        _snake_case		=				guidance_scale > 1.0

        # 3. Encode input prompt
        _snake_case		=				self._encode_prompt(
            A__     ,   A__     ,   A__     ,   A__     ,   A__     ,   prompt_embeds=A__     ,   negative_prompt_embeds=A__     ,   )

        _snake_case		=				ORT_TO_PT_TYPE[str(text_embeddings.dtype						)]

        # 4. Preprocess image
        _snake_case		=				preprocess(A__						)
        _snake_case		=				image.cpu()

        # 5. set timesteps
        self.scheduler.set_timesteps(A__     ,   device=A__						)
        _snake_case		=				self.scheduler.timesteps

        # 5. Add noise to image
        _snake_case		=				torch.tensor([noise_level]     ,   dtype=torch.long     ,   device=A__						)
        _snake_case		=				torch.randn(image.shape     ,   generator=A__     ,   device=A__     ,   dtype=A__						)
        _snake_case		=				self.low_res_scheduler.add_noise(A__     ,   A__     ,   A__						)

        _snake_case		=				2 if do_classifier_free_guidance else 1
        _snake_case		=				np.concatenate([image] * batch_multiplier * num_images_per_prompt						)
        _snake_case		=				np.concatenate([noise_level] * image.shape[0]						)

        # 6. Prepare latent variables
        _snake_case,	_snake_case		=				image.shape[2:]
        _snake_case		=				self.prepare_latents(
            batch_size * num_images_per_prompt     ,   A__     ,   A__     ,   A__     ,   A__     ,   A__     ,   A__     ,   A__     ,   )

        # 7. Check that sizes of image and latents match
        _snake_case		=				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
        _snake_case		=				self.prepare_extra_step_kwargs(A__     ,   A__						)

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

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

            # concat latents, mask, masked_image_latents in the channel dimension
            _snake_case		=				self.scheduler.scale_model_input(A__     ,   A__						)
            _snake_case		=				np.concatenate([latent_model_input, image]     ,   axis=1						)

            # timestep to tensor
            _snake_case		=				np.array([t]     ,   dtype=A__						)

            # predict the noise residual
            _snake_case		=				self.unet(
                sample=A__     ,   timestep=A__     ,   encoder_hidden_states=A__     ,   class_labels=noise_level.astype(np.intaa						)     ,   )[0]

            # perform guidance
            if do_classifier_free_guidance:
              _snake_case,	_snake_case		=				np.split(A__     ,   2						)
            _snake_case		=				noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

            # compute the previous noisy sample x_t -> x_t-1
            _snake_case		=				self.scheduler.step(
                torch.from_numpy(A__						)     ,   A__     ,   A__     ,   **A__						).prev_sample

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

        # 10. Post-processing
        _snake_case		=				self.decode_latents(latents.float()						)

        # 11. Convert to PIL
        if output_type == "pil":
          _snake_case		=				self.numpy_to_pil(A__						)

        if not return_dict:
          return (image,)

        return ImagePipelineOutput(images=A__						)



      def 		UpperCamelCase_						(     self :							Optional[int]     ,   A__ :							Tuple						)       ->	Any:
        _snake_case		=				1 / 0.0_8333 * latents
        _snake_case		=				self.vae(latent_sample=A__						)[0]
        _snake_case		=				np.clip(image / 2 + 0.5     ,   0     ,   1						)
        _snake_case		=				image.transpose((0, 2, 3, 1)						)
        return image







      def 		UpperCamelCase_						(     self :							Dict     ,   A__ :							Union[str, List[str]]     ,   A__ :							Optional[int]     ,   A__ :							Optional[int]     ,   A__ :							bool     ,   A__ :							Optional[str]     ,   A__ :							Optional[torch.FloatTensor] = None     ,   A__ :							Optional[torch.FloatTensor] = None     ,   )       ->	str:
        if prompt is not None and isinstance(A__     ,   A__						):
          _snake_case		=				1
        elif prompt is not None and isinstance(A__     ,   A__						):
          _snake_case		=				len(A__						)
        else:
          _snake_case		=				prompt_embeds.shape[0]

        if prompt_embeds is None:
          _snake_case		=				self.tokenizer(
              A__     ,   padding='''max_length'''     ,   max_length=self.tokenizer.model_max_length     ,   truncation=A__     ,   return_tensors='''pt'''     ,   )
          _snake_case		=				text_inputs.input_ids
          _snake_case		=				self.tokenizer(A__     ,   padding='''longest'''     ,   return_tensors='''pt'''						).input_ids

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

        _snake_case,	_snake_case,	_snake_case		=				prompt_embeds.shape
        # duplicate text embeddings for each generation per prompt, using mps friendly method
        _snake_case		=				prompt_embeds.repeat(1     ,   A__						)
        _snake_case		=				prompt_embeds.reshape(bs_embed * num_images_per_prompt     ,   A__     ,   -1						)

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

          _snake_case		=				text_input_ids.shape[-1]
          _snake_case		=				self.tokenizer(
              A__     ,   padding='''max_length'''     ,   max_length=A__     ,   truncation=A__     ,   return_tensors='''pt'''     ,   )

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

          _snake_case		=				self.text_encoder(
              input_ids=uncond_input.input_ids.int().to(A__						)     ,   )
          _snake_case		=				uncond_embeddings[0]

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

        return prompt_embeds




import sys
from collections.abc import Mapping

import numpy as np
import pyarrow as pa

from .. import config
from ..utils.py_utils import map_nested
from .formatting import TensorFormatter



class     lowercase_       (    TensorFormatter[Mapping, np.ndarray, Mapping]	):



      def __init__(     self :							Union[str, Any]     ,   A__ :							Dict=None     ,   **A__ :							str						)       ->	Any:
        super().__init__(features=A__						)
        _snake_case		=				np_array_kwargs



      def 		UpperCamelCase_						(     self :							str     ,   A__ :							Dict						)       ->	Dict:
        if isinstance(A__     ,   A__						):
          if column and all(
              isinstance(A__     ,   np.ndarray						) and x.shape == column[0].shape and x.dtype == column[0].dtype for x in column						):
            return np.stack(A__						)
          else:
            # don't use np.array(column, dtype=object)
            # since it fails in certain cases
            # see https://stackoverflow.com/q/51005699
            _snake_case		=				np.empty(len(A__						)     ,   dtype=A__						)
            _snake_case		=				column
            return out
        return column



      def 		UpperCamelCase_						(     self :							Tuple     ,   A__ :							Optional[int]						)       ->	Optional[Any]:
        if isinstance(A__     ,   (str, bytes, type(A__						))						):
          return value
        elif isinstance(A__     ,   (np.character, np.ndarray)						) and np.issubdtype(value.dtype     ,   np.character						):
          return value
        elif isinstance(A__     ,   np.number						):
          return value

        _snake_case		=				{}

        if isinstance(A__     ,   np.ndarray						) and np.issubdtype(value.dtype     ,   np.integer						):
          _snake_case		=				{'''dtype''': np.intaa}
        elif isinstance(A__     ,   np.ndarray						) and np.issubdtype(value.dtype     ,   np.floating						):
          _snake_case		=				{'''dtype''': np.floataa}
        elif config.PIL_AVAILABLE and "PIL" in sys.modules:
          import PIL.Image

          if isinstance(A__     ,   PIL.Image.Image						):
            return np.asarray(A__     ,   **self.np_array_kwargs						)

        return np.asarray(A__     ,   **{**default_dtype, **self.np_array_kwargs}						)



      def 		UpperCamelCase_						(     self :							Union[str, Any]     ,   A__ :							Union[str, Any]						)       ->	int:
        # support for torch, tf, jax etc.
        if config.TORCH_AVAILABLE and "torch" in sys.modules:
          import torch

          if isinstance(A__     ,   torch.Tensor						):
            return self._tensorize(data_struct.detach().cpu().numpy()[()]						)
        if hasattr(A__     ,   '''__array__'''						) and not isinstance(A__     ,   (np.ndarray, np.character, np.number)						):
          _snake_case		=				data_struct.__array__()
        # support for nested types like struct of list of struct
        if isinstance(A__     ,   np.ndarray						):
          if data_struct.dtype == object:
            return self._consolidate([self.recursive_tensorize(A__						) for substruct in data_struct]						)
        if isinstance(A__     ,   (list, tuple)						):
          return self._consolidate([self.recursive_tensorize(A__						) for substruct in data_struct]						)
        return self._tensorize(A__						)



      def 		UpperCamelCase_						(     self :							List[Any]     ,   A__ :							dict						)       ->	Tuple:
        return map_nested(self._recursive_tensorize     ,   A__     ,   map_list=A__						)



      def 		UpperCamelCase_						(     self :							str     ,   A__ :							pa.Table						)       ->	Mapping:
        _snake_case		=				self.numpy_arrow_extractor().extract_row(A__						)
        _snake_case		=				self.python_features_decoder.decode_row(A__						)
        return self.recursive_tensorize(A__						)



      def 		UpperCamelCase_						(     self :							Optional[Any]     ,   A__ :							pa.Table						)       ->	np.ndarray:
        _snake_case		=				self.numpy_arrow_extractor().extract_column(A__						)
        _snake_case		=				self.python_features_decoder.decode_column(A__     ,   pa_table.column_names[0]						)
        _snake_case		=				self.recursive_tensorize(A__						)
        _snake_case		=				self._consolidate(A__						)
        return column







      def 		UpperCamelCase_						(     self :							Dict     ,   A__ :							pa.Table						)       ->	Mapping:
        _snake_case		=				self.numpy_arrow_extractor().extract_batch(A__						)
        _snake_case		=				self.python_features_decoder.decode_batch(A__						)
        _snake_case		=				self.recursive_tensorize(A__						)
        for column_name in batch:
          _snake_case		=				self._consolidate(batch[column_name]						)
        return batch

