


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


__lowerCamelCase   	=  4
__lowerCamelCase   	=  7

__lowerCamelCase   	=  {
    '''float16''': torch.floataa,
    '''float32''': torch.floataa,
}



def 			_a      (    __UpperCamelCase			):
     if isinstance(__UpperCamelCase	,	torch.Tensor			):
          return image
     elif isinstance(__UpperCamelCase	,	PIL.Image.Image			):
          a_     :  int				      =     [image]

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

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

     return image


class 	a__						(    lowerCAmelCase_     ):

       def __init__( self   :  Any   ,				lowerCamelCase_   :  OnnxRuntimeModel   ,				lowerCamelCase_   :  OnnxRuntimeModel   ,				lowerCamelCase_   :  Any   ,				lowerCamelCase_   :  OnnxRuntimeModel   ,				lowerCamelCase_   :  DDPMScheduler   ,				lowerCamelCase_   :  Any   ,				lowerCamelCase_   :  int = 3_5_0   ,				):
            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   :  Tuple   ,				lowerCamelCase_   :  Union[str, List[str]]   ,				lowerCamelCase_   :  Union[torch.FloatTensor, PIL.Image.Image, List[PIL.Image.Image]]   ,				lowerCamelCase_   :  int = 7_5   ,				lowerCamelCase_   :  float = 9.0   ,				lowerCamelCase_   :  int = 2_0   ,				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   ,				):
            self.check_inputs(lowerCamelCase_   ,				lowerCamelCase_   ,				lowerCamelCase_   ,				lowerCamelCase_     )

            # 2. Define call parameters
            if prompt is not None and isinstance(lowerCamelCase_   ,				lowerCamelCase_     ):
                 a_     :  Dict				      =     1
            elif prompt is not None and isinstance(lowerCamelCase_   ,				lowerCamelCase_     ):
                 a_     :  Optional[Any]				      =     len(lowerCamelCase_     )
            else:
                 a_     :  Tuple				      =     prompt_embeds.shape[0]

            a_     :  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_     :  Union[str, Any]				      =     guidance_scale > 1.0

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

            a_     :  Optional[int]				      =     ORT_TO_PT_TYPE[str(text_embeddings.dtype     )]

            # 4. Preprocess image
            a_     :  Tuple				      =     preprocess(lowerCamelCase_     )
            a_     :  Dict				      =     image.cpu()

            # 5. set timesteps
            self.scheduler.set_timesteps(lowerCamelCase_   ,				device=lowerCamelCase_     )
            a_     :  Union[str, Any]				      =     self.scheduler.timesteps

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

            a_     :  str				      =     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_     :  Union[str, Any]				      =     image.shape[2:]
            a_     :  List[Any]				      =     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_     :  Dict				      =     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_     :  Optional[int]				      =     self.prepare_extra_step_kwargs(lowerCamelCase_   ,				lowerCamelCase_     )

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

            # 9. Denoising loop
            a_     :  Union[str, Any]				      =     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_     :  List[Any]				      =     np.concatenate([latents] * 2     ) if do_classifier_free_guidance else latents

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

                      # timestep to tensor
                      a_     :  Dict				      =     np.array([t]   ,				dtype=lowerCamelCase_     )

                      # predict the noise residual
                      a_     :  Tuple				      =     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_     :  Dict				      =     np.split(lowerCamelCase_   ,				2     )
                      a_     :  int				      =     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_     :  List[str]				      =     self.decode_latents(latents.float()     )

            # 11. Convert to PIL
            if output_type == "pil":
                 a_     :  int				      =     self.numpy_to_pil(lowerCamelCase_     )

            if not return_dict:
                 return (image,)

            return ImagePipelineOutput(images=lowerCamelCase_     )

       def     UpperCAmelCase( self   :  Any   ,				lowerCamelCase_   :  int     ):
            a_     :  List[Any]				      =     1 / 0.0_8_3_3_3 * latents
            a_     :  Optional[int]				      =     self.vae(latent_sample=lowerCamelCase_     )[0]
            a_     :  Tuple				      =     np.clip(image / 2 + 0.5   ,				0   ,				1     )
            a_     :  Optional[int]				      =     image.transpose((0, 2, 3, 1)     )
            return image

       def     UpperCAmelCase( self   :  Dict   ,				lowerCamelCase_   :  Union[str, List[str]]   ,				lowerCamelCase_   :  Optional[Any]   ,				lowerCamelCase_   :  Optional[int]   ,				lowerCamelCase_   :  bool   ,				lowerCamelCase_   :  Optional[str]   ,				lowerCamelCase_   :  Optional[torch.FloatTensor] = None   ,				lowerCamelCase_   :  Optional[torch.FloatTensor] = None   ,				):
            if prompt is not None and isinstance(lowerCamelCase_   ,				lowerCamelCase_     ):
                 a_     :  Optional[int]				      =     1
            elif prompt is not None and isinstance(lowerCamelCase_   ,				lowerCamelCase_     ):
                 a_     :  Optional[int]				      =     len(lowerCamelCase_     )
            else:
                 a_     :  str				      =     prompt_embeds.shape[0]

            if prompt_embeds is None:
                 a_     :  List[Any]				      =     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_     :  Optional[Any]				      =     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_     :  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_     :  Optional[Any]				      =     self.text_encoder(
                     input_ids=text_input_ids.int().to(lowerCamelCase_     )   ,				)
                 a_     :  Dict				      =     prompt_embeds[0]

            a_      ,      a_      ,      a_     :  List[str]				      =     prompt_embeds.shape
            # duplicate text embeddings for each generation per prompt, using mps friendly method
            a_     :  List[str]				      =     prompt_embeds.repeat(1   ,				lowerCamelCase_     )
            a_     :  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_     :  List[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_     :  Optional[Any]				      =     [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_     :  Union[str, Any]				      =     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_     :  Union[str, Any]				      =     self.text_encoder(
                     input_ids=uncond_input.input_ids.int().to(lowerCamelCase_     )   ,				)
                 a_     :  str				      =     uncond_embeddings[0]

            if do_classifier_free_guidance:
                 a_     :  Optional[int]				      =     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_     :  List[str]				      =     np.concatenate([uncond_embeddings, prompt_embeds]     )

            return prompt_embeds







import unittest

from transformers import SpeechaTextaConfig
from transformers.testing_utils import is_torch_available, require_torch, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
       import torch

       from transformers.models.speech_to_text_a.modeling_speech_to_text_a import (
           SpeechaTextaDecoder,
           SpeechaTextaForCausalLM,
       )


@require_torch
class 	a__						:

       def __init__( self   :  Dict   ,				lowerCamelCase_   :  Any   ,				lowerCamelCase_   :  Dict=9_9   ,				lowerCamelCase_   :  List[Any]=1_3   ,				lowerCamelCase_   :  Dict=1_6   ,				lowerCamelCase_   :  Any=7   ,				lowerCamelCase_   :  Dict=True   ,				lowerCamelCase_   :  List[str]=True   ,				lowerCamelCase_   :  Tuple=True   ,				lowerCamelCase_   :  Any=False   ,				lowerCamelCase_   :  Union[str, Any]=True   ,				lowerCamelCase_   :  int=2   ,				lowerCamelCase_   :  Tuple=3_2   ,				lowerCamelCase_   :  int=4   ,				lowerCamelCase_   :  Dict=4   ,				lowerCamelCase_   :  str=3_0   ,				lowerCamelCase_   :  Optional[int]=0   ,				lowerCamelCase_   :  Optional[Any]=1   ,				lowerCamelCase_   :  Any=2   ,				lowerCamelCase_   :  Optional[int]=None   ,				):
            a_     :  Optional[Any]				      =     parent
            a_     :  Union[str, Any]				      =     batch_size
            a_     :  Tuple				      =     decoder_seq_length
            # For common tests
            a_     :  Union[str, Any]				      =     self.decoder_seq_length
            a_     :  List[str]				      =     is_training
            a_     :  int				      =     use_attention_mask
            a_     :  Dict				      =     use_labels

            a_     :  List[str]				      =     vocab_size
            a_     :  Optional[int]				      =     d_model
            a_     :  str				      =     d_model
            a_     :  Any				      =     decoder_layers
            a_     :  Tuple				      =     decoder_layers
            a_     :  Dict				      =     decoder_ffn_dim
            a_     :  List[str]				      =     decoder_attention_heads
            a_     :  List[Any]				      =     decoder_attention_heads
            a_     :  List[str]				      =     eos_token_id
            a_     :  Any				      =     bos_token_id
            a_     :  List[Any]				      =     pad_token_id
            a_     :  Dict				      =     decoder_start_token_id
            a_     :  Tuple				      =     use_cache
            a_     :  List[Any]				      =     max_position_embeddings

            a_     :  List[str]				      =     None
            a_     :  Any				      =     decoder_seq_length
            a_     :  List[Any]				      =     2
            a_     :  Optional[Any]				      =     1

       def     UpperCAmelCase( self   :  str     ):
            a_     :  Any				      =     ids_tensor([self.batch_size, self.decoder_seq_length]   ,				self.vocab_size     )

            a_     :  Any				      =     None
            if self.use_attention_mask:
                 a_     :  Optional[Any]				      =     ids_tensor([self.batch_size, self.decoder_seq_length]   ,				vocab_size=2     )

            a_     :  Tuple				      =     None
            if self.use_labels:
                 a_     :  int				      =     ids_tensor([self.batch_size, self.decoder_seq_length]   ,				self.vocab_size     )

            a_     :  int				      =     SpeechaTextaConfig(
                vocab_size=self.vocab_size   ,				d_model=self.d_model   ,				decoder_layers=self.decoder_layers   ,				decoder_ffn_dim=self.decoder_ffn_dim   ,				decoder_attention_heads=self.decoder_attention_heads   ,				eos_token_id=self.eos_token_id   ,				bos_token_id=self.bos_token_id   ,				use_cache=self.use_cache   ,				pad_token_id=self.pad_token_id   ,				decoder_start_token_id=self.decoder_start_token_id   ,				max_position_embeddings=self.max_position_embeddings   ,				)

            return (
                config,
                input_ids,
                attention_mask,
                lm_labels,
            )

       def     UpperCAmelCase( self   :  Tuple   ,				lowerCamelCase_   :  List[str]   ,				lowerCamelCase_   :  str   ,				lowerCamelCase_   :  Optional[int]   ,				lowerCamelCase_   :  List[str]   ,				):
            a_     :  List[Any]				      =     True
            a_     :  Optional[int]				      =     SpeechaTextaDecoder(config=lowerCamelCase_     ).to(lowerCamelCase_     ).eval()
            a_     :  Optional[int]				      =     input_ids[:2]

            input_ids[input_ids == 0] += 1
            # first forward pass
            a_     :  Union[str, Any]				      =     model(lowerCamelCase_   ,				use_cache=lowerCamelCase_     )
            a_     :  Tuple				      =     model(lowerCamelCase_     )
            a_     :  str				      =     model(lowerCamelCase_   ,				use_cache=lowerCamelCase_     )

            self.parent.assertTrue(len(lowerCamelCase_     ) == len(lowerCamelCase_     )     )
            self.parent.assertTrue(len(lowerCamelCase_     ) == len(lowerCamelCase_     ) + 1     )

            a_     :  Tuple				      =     outputs["""past_key_values"""]

            # create hypothetical next token and extent to next_input_ids
            a_     :  Any				      =     ids_tensor((2, 1)   ,				config.vocab_size - 1     ) + 1

            # append to next input_ids and
            a_     :  str				      =     torch.cat([input_ids, next_tokens]   ,				dim=-1     )

            print(lowerCamelCase_     )

            a_     :  Any				      =     model(lowerCamelCase_     )["""last_hidden_state"""]
            a_     :  int				      =     model(lowerCamelCase_   ,				past_key_values=lowerCamelCase_     )["""last_hidden_state"""]

            # select random slice
            a_     :  Tuple				      =     ids_tensor((1,)   ,				output_from_past.shape[-1]     ).item()
            a_     :  List[str]				      =     output_from_no_past[:, next_input_ids.shape[-1] - 1, random_slice_idx].detach()
            a_     :  List[Any]				      =     output_from_past[:, 0, random_slice_idx].detach()

            # test that outputs are equal for slice
            assert torch.allclose(lowerCamelCase_   ,				lowerCamelCase_   ,				atol=1E-3     )

       def     UpperCAmelCase( self   :  Any     ):
            a_     :  List[Any]				      =     self.prepare_config_and_inputs()
            (
                (
                a_
            )      ,      (
                a_
            )      ,      (
                a_
            )      ,      (
                a_
            )      ,      
            )     :  Optional[int]				      =     config_and_inputs

            a_     :  Tuple				      =     {
                """input_ids""": input_ids,
                """attention_mask""": attention_mask,
            }
            return config, inputs_dict


@require_torch
class 	a__						(    lowerCAmelCase_    ,							lowerCAmelCase_    ,							lowerCAmelCase_    ,							unittest.TestCase     ):
       lowerCamelCase__:  int				     =  (SpeechaTextaDecoder, SpeechaTextaForCausalLM) if is_torch_available() else ()
       lowerCamelCase__:  int				     =  (SpeechaTextaForCausalLM,) if is_torch_available() else ()
       lowerCamelCase__:  int				     =  {"""text-generation""": SpeechaTextaForCausalLM} if is_torch_available() else {}
       lowerCamelCase__:  Dict				     =  True
       lowerCamelCase__:  List[Any]				     =  False

       def     UpperCAmelCase( self   :  List[str]   ,				):
            a_     :  Optional[int]				      =     SpeechaTextaStandaloneDecoderModelTester(self   ,				is_training=lowerCamelCase_     )
            a_     :  str				      =     ConfigTester(self   ,				config_class=lowerCamelCase_     )

       def     UpperCAmelCase( self   :  Optional[Any]     ):
            pass

       def     UpperCAmelCase( self   :  Optional[Any]     ):
            pass

       def     UpperCAmelCase( self   :  Optional[Any]     ):
            pass

       def     UpperCAmelCase( self   :  str     ):
            self.config_tester.run_common_tests()

       def     UpperCAmelCase( self   :  Dict     ):
            a_     :  int				      =     self.model_tester.prepare_config_and_inputs()
            self.model_tester.create_and_check_decoder_model_past(*lowerCamelCase_     )

       def     UpperCAmelCase( self   :  int     ):
            return



