


from __future__ import annotations

import unittest

import numpy as np

from transformers import OPTConfig, is_tf_available
from transformers.testing_utils import require_sentencepiece, require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
     import tensorflow as tf

     from transformers import GPTaTokenizer, TFOPTForCausalLM, TFOPTModel



def  _a			(						__UpperCamelCase  ,  __UpperCamelCase  ,  __UpperCamelCase=None  ,  __UpperCamelCase=None ):



     if attention_mask is None:
          a_    :     Optional[int]										=    tf.cast(tf.math.not_equal(__UpperCamelCase  ,  config.pad_token_id )  ,  tf.inta )
     return {"input_ids": input_ids, "attention_mask": attention_mask}


@require_tf
class   a__	:
       lowerCamelCase__:     int							    =     OPTConfig
       lowerCamelCase__:     Any							    =     {}
       lowerCamelCase__:     List[str]							    =     """gelu"""

       def __init__(   self      :     int      ,							lowerCamelCase_      :     Union[str, Any]      ,							lowerCamelCase_      :     Union[str, Any]=1_3      ,							lowerCamelCase_      :     str=7      ,							lowerCamelCase_      :     Any=True      ,							lowerCamelCase_      :     Tuple=False      ,							lowerCamelCase_      :     Any=9_9      ,							lowerCamelCase_      :     str=1_6      ,							lowerCamelCase_      :     Union[str, Any]=2      ,							lowerCamelCase_      :     str=4      ,							lowerCamelCase_      :     Optional[int]=4      ,							lowerCamelCase_      :     List[Any]="gelu"      ,							lowerCamelCase_      :     str=0.1      ,							lowerCamelCase_      :     Dict=0.1      ,							lowerCamelCase_      :     List[Any]=2_0      ,							lowerCamelCase_      :     int=2      ,							lowerCamelCase_      :     Union[str, Any]=1      ,							lowerCamelCase_      :     Optional[int]=0      ,							lowerCamelCase_      :     Tuple=1_6      ,							lowerCamelCase_      :     int=1_6      ,							):
            a_    :     Union[str, Any]										=    parent
            a_    :     int										=    batch_size
            a_    :     int										=    seq_length
            a_    :     str										=    is_training
            a_    :     int										=    use_labels
            a_    :     List[str]										=    vocab_size
            a_    :     Dict										=    hidden_size
            a_    :     int										=    num_hidden_layers
            a_    :     str										=    num_attention_heads
            a_    :     Optional[Any]										=    intermediate_size
            a_    :     List[str]										=    hidden_act
            a_    :     Dict										=    hidden_dropout_prob
            a_    :     str										=    attention_probs_dropout_prob
            a_    :     Optional[Any]										=    max_position_embeddings
            a_    :     Optional[Any]										=    eos_token_id
            a_    :     Any										=    pad_token_id
            a_    :     List[str]										=    bos_token_id
            a_    :     str										=    embed_dim
            a_    :     Union[str, Any]										=    word_embed_proj_dim
            a_    :     int										=    False

       def 						UpperCAmelCase(   self      :     int    ):
            a_    :     int										=    ids_tensor([self.batch_size, self.seq_length - 1]      ,							self.vocab_size    )
            a_    :     Union[str, Any]										=    tf.expand_dims(tf.constant([self.eos_token_id] * self.batch_size    )      ,							1    )
            a_    :     List[str]										=    tf.concat([input_ids, eos_tensor]      ,							axis=1    )

            a_    :     str										=    self.config_cls(
                vocab_size=self.vocab_size      ,							hidden_size=self.hidden_size      ,							num_hidden_layers=self.num_hidden_layers      ,							num_attention_heads=self.num_attention_heads      ,							ffn_dim=self.intermediate_size      ,							dropout=self.hidden_dropout_prob      ,							attention_dropout=self.attention_probs_dropout_prob      ,							max_position_embeddings=self.max_position_embeddings      ,							eos_token_id=self.eos_token_id      ,							bos_token_id=self.bos_token_id      ,							pad_token_id=self.pad_token_id      ,							embed_dim=self.embed_dim      ,							word_embed_proj_dim=self.word_embed_proj_dim      ,							is_encoder_decoder=snake_case__      ,							**self.config_updates      ,							)
            a_    :     Optional[Any]										=    prepare_opt_inputs_dict(snake_case__      ,							snake_case__    )
            return config, inputs_dict

       def 						UpperCAmelCase(   self      :     Dict      ,							lowerCamelCase_      :     Optional[int]      ,							lowerCamelCase_      :     Optional[Any]    ):
            a_    :     Union[str, Any]										=    TFOPTModel(config=snake_case__    )
            a_    :     Any										=    inputs_dict["input_ids"]

            a_    :     Optional[int]										=    input_ids[:1, :]
            a_    :     Dict										=    inputs_dict["attention_mask"][:1, :]
            a_    :     int										=    1

            # first forward pass
            a_    :     int										=    model(snake_case__      ,							attention_mask=snake_case__      ,							use_cache=snake_case__    )

            a_    :     str										=    outputs.to_tuple()

            # create hypothetical next token and extent to next_input_ids
            a_    :     int										=    ids_tensor((self.batch_size, 3)      ,							config.vocab_size    )
            a_    :     Optional[int]										=    tf.cast(ids_tensor((self.batch_size, 3)      ,							2    )      ,							tf.inta    )

            # append to next input_ids and
            a_    :     List[Any]										=    tf.concat([input_ids, next_tokens]      ,							axis=-1    )
            a_    :     List[Any]										=    tf.concat([attention_mask, next_attn_mask]      ,							axis=-1    )

            a_    :     List[str]										=    model(snake_case__      ,							attention_mask=snake_case__    )[0]
            a_    :     Union[str, Any]										=    model(snake_case__      ,							attention_mask=snake_case__      ,							past_key_values=snake_case__    )[0]

            self.parent.assertEqual(next_tokens.shape[1]      ,							output_from_past.shape[1]    )

            # select random slice
            a_    :     List[Any]										=    int(ids_tensor((1,)      ,							output_from_past.shape[-1]    )    )
            a_    :     Any										=    output_from_no_past[:, -3:, random_slice_idx]
            a_    :     Any										=    output_from_past[:, :, random_slice_idx]

            # test that outputs are equal for slice
            tf.debugging.assert_near(snake_case__      ,							snake_case__      ,							rtol=1E-3    )


@require_tf
class   a__	(						SCREAMING_SNAKE_CASE__						,       SCREAMING_SNAKE_CASE__						,       unittest.TestCase    ):
       lowerCamelCase__:     List[str]							    =     (TFOPTModel, TFOPTForCausalLM) if is_tf_available() else ()
       lowerCamelCase__:     int							    =     (TFOPTForCausalLM,) if is_tf_available() else ()
       lowerCamelCase__:     Optional[Any]							    =     (
           {"""feature-extraction""": TFOPTModel, """text-generation""": TFOPTForCausalLM} if is_tf_available() else {}
       )
       lowerCamelCase__:     Dict							    =     False
       lowerCamelCase__:     List[Any]							    =     False
       lowerCamelCase__:     Tuple							    =     False
       lowerCamelCase__:     List[Any]							    =     10

       def 						UpperCAmelCase(   self      :     Optional[Any]    ):
            a_    :     List[str]										=    TFOPTModelTester(self    )
            a_    :     Optional[Any]										=    ConfigTester(self      ,							config_class=snake_case__    )

       def 						UpperCAmelCase(   self      :     List[str]    ):
            self.config_tester.run_common_tests()

       def 						UpperCAmelCase(   self      :     Optional[int]    ):
            a_    :     Optional[int]										=    self.model_tester.prepare_config_and_inputs_for_common()
            self.model_tester.check_decoder_model_past_large_inputs(*snake_case__    )

       def 						UpperCAmelCase(   self      :     Optional[Any]    ):
            a_    :     int										=    self.model_tester.prepare_config_and_inputs_for_common()

            def _get_word_embedding_weight(lowerCamelCase_      :     Any      ,							lowerCamelCase_      :     int    ):
                 if hasattr(snake_case__      ,							"""weight"""    ):
                      return embedding_layer.weight
                 else:
                      # Here we build the word embeddings weights if not exists.
                      # And then we retry to get the attribute once built.
                      model.build()
                      if hasattr(snake_case__      ,							"""weight"""    ):
                           return embedding_layer.weight
                      else:
                           return None

            for model_class in self.all_model_classes:
                 for size in [config.vocab_size - 1_0, config.vocab_size + 1_0]:
                      # build the embeddings
                      a_    :     List[Any]										=    model_class(config=snake_case__    )
                      a_    :     Optional[int]										=    _get_word_embedding_weight(snake_case__      ,							model.get_input_embeddings()    )
                      a_    :     Union[str, Any]										=    _get_word_embedding_weight(snake_case__      ,							model.get_output_embeddings()    )

                      # reshape the embeddings
                      model.resize_token_embeddings(snake_case__    )
                      a_    :     List[str]										=    _get_word_embedding_weight(snake_case__      ,							model.get_input_embeddings()    )
                      a_    :     List[Any]										=    _get_word_embedding_weight(snake_case__      ,							model.get_output_embeddings()    )

                      # check that the resized embeddings size matches the desired size.
                      a_    :     str										=    size if size is not None else config.vocab_size

                      self.assertEqual(new_input_embeddings.shape[0]      ,							snake_case__    )

                      # check that weights remain the same after resizing
                      a_    :     Any										=    True
                      for pa, pa in zip(old_input_embeddings.value()      ,							new_input_embeddings.value()    ):
                           if tf.math.reduce_sum(tf.math.abs(pa - pa    )    ) > 0:
                                a_    :     Dict										=    False
                      self.assertTrue(snake_case__    )

                      if old_output_embeddings is not None and new_output_embeddings is not None:
                           self.assertEqual(new_output_embeddings.shape[0]      ,							snake_case__    )

                           a_    :     List[str]										=    True
                           for pa, pa in zip(old_output_embeddings.value()      ,							new_output_embeddings.value()    ):
                                if tf.math.reduce_sum(tf.math.abs(pa - pa    )    ) > 0:
                                     a_    :     str										=    False
                           self.assertTrue(snake_case__    )



def  _a			(						__UpperCamelCase ):
     return tf.constant(__UpperCamelCase  ,  dtype=tf.intaa )


@require_tf
class   a__	(						unittest.TestCase    ):
       lowerCamelCase__:     Union[str, Any]							    =     99

       def 						UpperCAmelCase(   self      :     int    ):
            a_    :     str										=    tf.ones((4, 1)      ,							dtype=tf.intaa    ) * 2
            a_    :     Dict										=    tf.concat([ids_tensor((4, 6)      ,							self.vocab_size - 3    ) + 3, eos_column_vector]      ,							axis=1    )
            a_    :     str										=    input_ids.shape[0]
            a_    :     Optional[int]										=    OPTConfig(
                vocab_size=self.vocab_size      ,							hidden_size=2_4      ,							num_hidden_layers=2      ,							num_attention_heads=2      ,							ffn_dim=3_2      ,							max_position_embeddings=4_8      ,							eos_token_id=2      ,							pad_token_id=1      ,							bos_token_id=0      ,							)
            return config, input_ids, batch_size


@require_sentencepiece
@require_tf
class   a__	(						unittest.TestCase    ):

       @slow
       def 						UpperCAmelCase(   self      :     Tuple    ):
            a_    :     Tuple										=    TFOPTModel.from_pretrained("""facebook/opt-350m"""    )
            a_    :     str										=    _long_tensor([[0, 3_1_4_1_4, 2_3_2, 3_2_8, 7_4_0, 1_1_4_0, 1_2_6_9_5, 6_9, 4_6_0_7_8, 1_5_8_8, 2]]    )
            a_    :     Tuple										=    tf.not_equal(snake_case__      ,							model.config.pad_token_id    )
            with tf.GradientTape():
                 a_    :     Tuple										=    model(input_ids=snake_case__      ,							attention_mask=snake_case__    ).last_hidden_state
            a_    :     List[str]										=    (1, 1_1, 5_1_2)
            self.assertEqual(output.shape      ,							snake_case__    )
            a_    :     Dict										=    tf.constant(
                [[-0.2_8_7_3, -1.9_2_1_8, -0.3_0_3_3], [-1.2_7_1_0, -0.1_3_3_8, -0.1_9_0_2], [0.4_0_9_5, 0.1_2_1_4, -1.3_1_2_1]]    )
            self.assertTrue(np.allclose(output[:, :3, :3]      ,							snake_case__      ,							atol=4E-3    )    )

            a_    :     int										=    tf.function(snake_case__      ,							jit_compile=snake_case__    )
            a_    :     List[str]										=    xla_generate(snake_case__      ,							snake_case__    )[0]
            self.assertTrue(np.allclose(output[:, :3, :3]      ,							snake_case__      ,							atol=4E-2    )    )


@require_tf
@slow
class   a__	(						unittest.TestCase    ):

       def 						UpperCAmelCase(   self      :     List[Any]    ):
            super().setUp()
            a_    :     List[str]										=    "facebook/opt-350m"

       def 						UpperCAmelCase(   self      :     str    ):
            a_    :     Optional[int]										=    TFOPTForCausalLM.from_pretrained(self.path_model    )
            a_    :     str										=    GPTaTokenizer.from_pretrained(self.path_model    )

            a_    :     Tuple										=    [
                "Today is a beautiful day and I want to",
                "In the city of",
                "Paris is the capital of France and",
                "Computers and mobile phones have taken",
            ]
            # verify that prompt without BOS token is identical to Metaseq -> add_special_tokens=False
            a_    :     int										=    tokenizer(snake_case__      ,							return_tensors="""tf"""      ,							padding=snake_case__      ,							add_special_tokens=snake_case__    )
            a_    :     int										=    tf.math.reduce_mean(model(inputs.input_ids      ,							attention_mask=inputs.attention_mask    )[0]      ,							axis=-1    )
            a_    :     int										=    tf.constant(
                [
                    [1.3_8_5_1, -1_3.8_9_2_3, -1_0.5_2_2_9, -1_0.7_5_3_3, -0.2_3_0_9, -1_0.2_3_8_4, -0.5_3_6_5, -9.0_9_4_7, -5.1_6_7_0],
                    [-4.7_0_7_3, -1_0.6_2_7_6, -3.9_4_1_5, -2_1.5_2_4_2, -0.2_8_2_2, -0.2_8_2_2, -0.2_8_2_2, -0.2_8_2_2, -0.2_8_2_2],
                    [0.6_2_4_7, -3.4_2_2_9, -8.9_1_7_9, -1.4_2_9_7, -1_4.1_6_5_0, 1.4_1_4_6, -9.0_2_1_8, -0.2_7_0_3, -0.2_7_0_3],
                    [6.4_7_8_3, -1.9_9_1_3, -1_0.7_9_2_6, -2.3_3_3_6, 1.5_0_9_2, -0.9_9_7_4, -6.8_2_1_3, 1.3_4_7_7, 1.3_4_7_7],
                ]    )
            self.assertTrue(np.allclose(snake_case__      ,							snake_case__      ,							atol=1E-4    )    )

            a_    :     int										=    tf.function(snake_case__      ,							jit_compile=snake_case__    )
            a_    :     Optional[Any]										=    tf.math.reduce_mean(xla_generate(inputs.input_ids      ,							attention_mask=inputs.attention_mask    )[0]      ,							axis=-1    )
            self.assertTrue(np.allclose(snake_case__      ,							snake_case__      ,							atol=1E-4    )    )


@require_tf
@slow
class   a__	(						unittest.TestCase    ):

       @property
       def 						UpperCAmelCase(   self      :     List[Any]    ):
            return [
                "Today is a beautiful day and I want",
                "In the city of",
                "Paris is the capital of France and",
                "Computers and mobile phones have taken",
            ]

       def 						UpperCAmelCase(   self      :     Dict    ):
            a_    :     Tuple										=    "facebook/opt-125m"

            a_    :     List[str]										=    [
                "Today is a beautiful day and I want to",
                "In the city of New York, the city",
                "Paris is the capital of France and the capital",
                "Computers and mobile phones have taken over the",
            ]

            a_    :     Union[str, Any]										=    []
            a_    :     Optional[Any]										=    GPTaTokenizer.from_pretrained(snake_case__    )
            a_    :     Any										=    TFOPTForCausalLM.from_pretrained(snake_case__    )

            for prompt in self.prompts:
                 a_    :     Optional[int]										=    tokenizer(snake_case__      ,							return_tensors="""tf"""    ).input_ids

                 a_    :     List[str]										=    model.generate(snake_case__      ,							max_length=1_0    )

                 a_    :     Union[str, Any]										=    tokenizer.batch_decode(snake_case__      ,							skip_special_tokens=snake_case__    )
                 predicted_outputs += generated_string

            self.assertListEqual(snake_case__      ,							snake_case__    )

       def 						UpperCAmelCase(   self      :     Optional[int]    ):
            a_    :     List[str]										=    "facebook/opt-350m"

            a_    :     Optional[Any]										=    GPTaTokenizer.from_pretrained(snake_case__    )
            a_    :     Optional[int]										=    TFOPTForCausalLM.from_pretrained(snake_case__    )

            a_    :     List[str]										=    "left"

            # use different length sentences to test batching
            a_    :     Dict										=    [
                "Hello, my dog is a little",
                "Today, I",
            ]

            a_    :     Optional[int]										=    tokenizer(snake_case__      ,							return_tensors="""tf"""      ,							padding=snake_case__    )
            a_    :     int										=    inputs["input_ids"]

            a_    :     int										=    model.generate(input_ids=snake_case__      ,							attention_mask=inputs["""attention_mask"""]    )

            a_    :     Dict										=    tokenizer(sentences[0]      ,							return_tensors="""tf"""    ).input_ids
            a_    :     Optional[Any]										=    model.generate(input_ids=snake_case__    )

            a_    :     Tuple										=    inputs_non_padded.shape[-1] - tf.math.reduce_sum(
                tf.cast(inputs["""attention_mask"""][-1]      ,							tf.intaa    )    )
            a_    :     Optional[Any]										=    tokenizer(sentences[1]      ,							return_tensors="""tf"""    ).input_ids
            a_    :     Union[str, Any]										=    model.generate(input_ids=snake_case__      ,							max_length=model.config.max_length - num_paddings    )

            a_    :     Tuple										=    tokenizer.batch_decode(snake_case__      ,							skip_special_tokens=snake_case__    )
            a_    :     Optional[Any]										=    tokenizer.decode(output_non_padded[0]      ,							skip_special_tokens=snake_case__    )
            a_    :     int										=    tokenizer.decode(output_padded[0]      ,							skip_special_tokens=snake_case__    )

            a_    :     List[Any]										=    [
                "Hello, my dog is a little bit of a dork.\nI'm a little bit",
                "Today, I was in the middle of a conversation with a friend about the",
            ]
            self.assertListEqual(snake_case__      ,							snake_case__    )
            self.assertListEqual(snake_case__      ,							[non_padded_sentence, padded_sentence]    )

       def 						UpperCAmelCase(   self      :     List[str]    ):
            a_    :     List[str]										=    "facebook/opt-350m"

            a_    :     Optional[int]										=    [
                "Today is a beautiful day and I want to",
                "In the city of San Francisco, the city",
                "Paris is the capital of France and the capital",
                "Computers and mobile phones have taken over the",
            ]

            a_    :     Dict										=    []
            a_    :     Optional[int]										=    GPTaTokenizer.from_pretrained(snake_case__    )
            a_    :     int										=    TFOPTForCausalLM.from_pretrained(snake_case__    )

            for prompt in self.prompts:
                 a_    :     Dict										=    tokenizer(snake_case__      ,							return_tensors="""tf"""    ).input_ids

                 a_    :     Any										=    model.generate(snake_case__      ,							max_length=1_0    )

                 a_    :     Tuple										=    tokenizer.batch_decode(snake_case__      ,							skip_special_tokens=snake_case__    )
                 predicted_outputs += generated_string

            self.assertListEqual(snake_case__      ,							snake_case__    )







from math import acos, sin
from typing import List, Tuple, Union

import numpy as np
import torch
from PIL import Image

from ...models import AutoencoderKL, UNetaDConditionModel
from ...schedulers import DDIMScheduler, DDPMScheduler
from ...utils import randn_tensor
from ..pipeline_utils import AudioPipelineOutput, BaseOutput, DiffusionPipeline, ImagePipelineOutput
from .mel import Mel


class   a__	(						lowerCAmelCase_    ):
       lowerCamelCase__:     List[str]							    =     ["""vqvae"""]

       def __init__(   self      :     str      ,							lowerCamelCase_      :     AutoencoderKL      ,							lowerCamelCase_      :     UNetaDConditionModel      ,							lowerCamelCase_      :     Mel      ,							lowerCamelCase_      :     Union[DDIMScheduler, DDPMScheduler]      ,							):
            super().__init__()
            self.register_modules(unet=lowerCamelCase_      ,							scheduler=lowerCamelCase_      ,							mel=lowerCamelCase_      ,							vqvae=lowerCamelCase_    )

       def 						UpperCAmelCase(   self      :     Any    ):
            return 5_0 if isinstance(self.scheduler      ,							lowerCamelCase_    ) else 1_0_0_0

       @torch.no_grad()
       def __call__(   self      :     List[str]      ,							lowerCamelCase_      :     int = 1      ,							lowerCamelCase_      :     str = None      ,							lowerCamelCase_      :     np.ndarray = None      ,							lowerCamelCase_      :     int = 0      ,							lowerCamelCase_      :     int = 0      ,							lowerCamelCase_      :     int = None      ,							lowerCamelCase_      :     torch.Generator = None      ,							lowerCamelCase_      :     float = 0      ,							lowerCamelCase_      :     float = 0      ,							lowerCamelCase_      :     torch.Generator = None      ,							lowerCamelCase_      :     float = 0      ,							lowerCamelCase_      :     torch.Tensor = None      ,							lowerCamelCase_      :     torch.Tensor = None      ,							lowerCamelCase_      :     List[Any]=True      ,							):
            a_    :     Optional[Any]										=    steps or self.get_default_steps()
            self.scheduler.set_timesteps(lowerCamelCase_    )
            a_    :     Optional[int]										=    step_generator or generator
            # For backwards compatibility
            if type(self.unet.config.sample_size    ) == int:
                 a_    :     str										=    (self.unet.config.sample_size, self.unet.config.sample_size)
            if noise is None:
                 a_    :     Dict										=    randn_tensor(
                     (
                         batch_size,
                         self.unet.config.in_channels,
                         self.unet.config.sample_size[0],
                         self.unet.config.sample_size[1],
                     )      ,							generator=lowerCamelCase_      ,							device=self.device      ,							)
            a_    :     Tuple										=    noise
            a_    :     Union[str, Any]										=    None

            if audio_file is not None or raw_audio is not None:
                 self.mel.load_audio(lowerCamelCase_      ,							lowerCamelCase_    )
                 a_    :     Union[str, Any]										=    self.mel.audio_slice_to_image(lowerCamelCase_    )
                 a_    :     Union[str, Any]										=    np.frombuffer(input_image.tobytes()      ,							dtype="""uint8"""    ).reshape(
                     (input_image.height, input_image.width)    )
                 a_    :     Union[str, Any]										=    (input_image / 2_5_5) * 2 - 1
                 a_    :     Tuple										=    torch.tensor(input_image[np.newaxis, :, :]      ,							dtype=torch.float    ).to(self.device    )

                 if self.vqvae is not None:
                      a_    :     Dict										=    self.vqvae.encode(torch.unsqueeze(lowerCamelCase_      ,							0    )    ).latent_dist.sample(
                          generator=lowerCamelCase_    )[0]
                      a_    :     List[Any]										=    self.vqvae.config.scaling_factor * input_images

                 if start_step > 0:
                      a_    :     Tuple										=    self.scheduler.add_noise(lowerCamelCase_      ,							lowerCamelCase_      ,							self.scheduler.timesteps[start_step - 1]    )

                 a_    :     Union[str, Any]										=    (
                     self.unet.config.sample_size[1] * self.mel.get_sample_rate() / self.mel.x_res / self.mel.hop_length
                 )
                 a_    :     int										=    int(mask_start_secs * pixels_per_second    )
                 a_    :     Tuple										=    int(mask_end_secs * pixels_per_second    )
                 a_    :     Tuple										=    self.scheduler.add_noise(lowerCamelCase_      ,							lowerCamelCase_      ,							torch.tensor(self.scheduler.timesteps[start_step:]    )    )

            for step, t in enumerate(self.progress_bar(self.scheduler.timesteps[start_step:]    )    ):
                 if isinstance(self.unet      ,							lowerCamelCase_    ):
                      a_    :     Dict										=    self.unet(lowerCamelCase_      ,							lowerCamelCase_      ,							lowerCamelCase_    )["""sample"""]
                 else:
                      a_    :     Dict										=    self.unet(lowerCamelCase_      ,							lowerCamelCase_    )["""sample"""]

                 if isinstance(self.scheduler      ,							lowerCamelCase_    ):
                      a_    :     Any										=    self.scheduler.step(
                          model_output=lowerCamelCase_      ,							timestep=lowerCamelCase_      ,							sample=lowerCamelCase_      ,							eta=lowerCamelCase_      ,							generator=lowerCamelCase_      ,							)["""prev_sample"""]
                 else:
                      a_    :     Optional[Any]										=    self.scheduler.step(
                          model_output=lowerCamelCase_      ,							timestep=lowerCamelCase_      ,							sample=lowerCamelCase_      ,							generator=lowerCamelCase_      ,							)["""prev_sample"""]

                 if mask is not None:
                      if mask_start > 0:
                           a_    :     Any										=    mask[:, step, :, :mask_start]
                      if mask_end > 0:
                           a_    :     Dict										=    mask[:, step, :, -mask_end:]

            if self.vqvae is not None:
                 # 0.18215 was scaling factor used in training to ensure unit variance
                 a_    :     str										=    1 / self.vqvae.config.scaling_factor * images
                 a_    :     List[Any]										=    self.vqvae.decode(lowerCamelCase_    )["""sample"""]

            a_    :     Union[str, Any]										=    (images / 2 + 0.5).clamp(0      ,							1    )
            a_    :     Dict										=    images.cpu().permute(0      ,							2      ,							3      ,							1    ).numpy()
            a_    :     Any										=    (images * 2_5_5).round().astype("""uint8"""    )
            a_    :     int										=    list(
                (Image.fromarray(_[:, :, 0]    ) for _ in images)
                if images.shape[3] == 1
                else (Image.fromarray(lowerCamelCase_      ,							mode="""RGB"""    ).convert("""L"""    ) for _ in images)    )

            a_    :     str										=    [self.mel.image_to_audio(lowerCamelCase_    ) for _ in images]
            if not return_dict:
                 return images, (self.mel.get_sample_rate(), audios)

            return BaseOutput(**AudioPipelineOutput(np.array(lowerCamelCase_    )[:, np.newaxis, :]    )      ,							**ImagePipelineOutput(lowerCamelCase_    )    )

       @torch.no_grad()
       def 						UpperCAmelCase(   self      :     List[str]      ,							lowerCamelCase_      :     List[Image.Image]      ,							lowerCamelCase_      :     int = 5_0    ):
            assert isinstance(self.scheduler      ,							lowerCamelCase_    )
            self.scheduler.set_timesteps(lowerCamelCase_    )
            a_    :     Any										=    np.array(
                [np.frombuffer(image.tobytes()      ,							dtype="""uint8"""    ).reshape((1, image.height, image.width)    ) for image in images]    )
            a_    :     Union[str, Any]										=    (sample / 2_5_5) * 2 - 1
            a_    :     Optional[Any]										=    torch.Tensor(lowerCamelCase_    ).to(self.device    )

            for t in self.progress_bar(torch.flip(self.scheduler.timesteps      ,							(0,)    )    ):
                 a_    :     str										=    t - self.scheduler.config.num_train_timesteps // self.scheduler.num_inference_steps
                 a_    :     Any										=    self.scheduler.alphas_cumprod[t]
                 a_    :     Dict										=    (
                     self.scheduler.alphas_cumprod[prev_timestep]
                     if prev_timestep >= 0
                     else self.scheduler.final_alpha_cumprod
                 )
                 a_    :     Optional[Any]										=    1 - alpha_prod_t
                 a_    :     Optional[int]										=    self.unet(lowerCamelCase_      ,							lowerCamelCase_    )["""sample"""]
                 a_    :     Tuple										=    (1 - alpha_prod_t_prev) ** 0.5 * model_output
                 a_    :     Tuple										=    (sample - pred_sample_direction) * alpha_prod_t_prev ** (-0.5)
                 a_    :     Tuple										=    sample * alpha_prod_t ** 0.5 + beta_prod_t ** 0.5 * model_output

            return sample

       @staticmethod
       def 						UpperCAmelCase(   lowerCamelCase_      :     torch.Tensor      ,							lowerCamelCase_      :     torch.Tensor      ,							lowerCamelCase_      :     float    ):
            a_    :     Dict										=    acos(torch.dot(torch.flatten(lowerCamelCase_    )      ,							torch.flatten(lowerCamelCase_    )    ) / torch.norm(lowerCamelCase_    ) / torch.norm(lowerCamelCase_    )    )
            return sin((1 - alpha) * theta    ) * xa / sin(lowerCamelCase_    ) + sin(alpha * theta    ) * xa / sin(lowerCamelCase_    )



