



from dataclasses import dataclass
from typing import Dict, Optional, Union

import torch
import torch.nn.functional as F
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .attention import BasicTransformerBlock
from .attention_processor import AttentionProcessor, AttnProcessor
from .embeddings import TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin

@dataclass
class       _SCREAMING_SNAKE_CASE	(UpperCamelCase ):
     lowerCAmelCase					=			42






class       _SCREAMING_SNAKE_CASE	(UpperCamelCase			,    UpperCamelCase ):





     @register_to_config
     def __init__(       self :							Tuple     ,  UpperCamelCase :							int = 3_2     ,  UpperCamelCase :							int = 6_4     ,  UpperCamelCase :							int = 2_0     ,  UpperCamelCase :							int = 7_6_8     ,  UpperCamelCase :							Dict=7_7     ,  UpperCamelCase :							Any=4     ,  UpperCamelCase :							float = 0.0     ,  UpperCamelCase :							str = "silu"     ,  UpperCamelCase :							Optional[str] = None     ,  UpperCamelCase :							Optional[str] = None     ,  UpperCamelCase :							Optional[str] = "linear"     ,  UpperCamelCase :							Optional[str] = "prd"     ,  UpperCamelCase :							Optional[int] = None     ,  UpperCamelCase :							Optional[int] = None     ,  UpperCamelCase :							Optional[int] = None     ,  )->Any:
      super().__init__()
      __SCREAMING_SNAKE_CASE       :    Any       =	num_attention_heads
      __SCREAMING_SNAKE_CASE       :    Any       =	attention_head_dim
      __SCREAMING_SNAKE_CASE       :    Dict       =	num_attention_heads * attention_head_dim
      __SCREAMING_SNAKE_CASE       :    Optional[int]       =	additional_embeddings

      __SCREAMING_SNAKE_CASE       :    str       =	time_embed_dim or inner_dim
      __SCREAMING_SNAKE_CASE       :    Any       =	embedding_proj_dim or embedding_dim
      __SCREAMING_SNAKE_CASE       :    str       =	clip_embed_dim or embedding_dim

      __SCREAMING_SNAKE_CASE       :    List[Any]       =	Timesteps(UpperCamelCase     ,  UpperCamelCase     ,  0				)
      __SCREAMING_SNAKE_CASE       :    Dict       =	TimestepEmbedding(UpperCamelCase     ,  UpperCamelCase     ,  out_dim=UpperCamelCase     ,  act_fn=UpperCamelCase				)

      __SCREAMING_SNAKE_CASE       :    Optional[Any]       =	nn.Linear(UpperCamelCase     ,  UpperCamelCase				)

      if embedding_proj_norm_type is None:
       __SCREAMING_SNAKE_CASE       :    Dict       =	None
      elif embedding_proj_norm_type == "layer":
       __SCREAMING_SNAKE_CASE       :    Dict       =	nn.LayerNorm(UpperCamelCase				)
      else:
       raise ValueError(F"""unsupported embedding_proj_norm_type: {embedding_proj_norm_type}"""				)

      __SCREAMING_SNAKE_CASE       :    Any       =	nn.Linear(UpperCamelCase     ,  UpperCamelCase				)

      if encoder_hid_proj_type is None:
       __SCREAMING_SNAKE_CASE       :    List[Any]       =	None
      elif encoder_hid_proj_type == "linear":
       __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	nn.Linear(UpperCamelCase     ,  UpperCamelCase				)
      else:
       raise ValueError(F"""unsupported encoder_hid_proj_type: {encoder_hid_proj_type}"""				)

      __SCREAMING_SNAKE_CASE       :    List[str]       =	nn.Parameter(torch.zeros(1     ,  num_embeddings + additional_embeddings     ,  UpperCamelCase				)				)

      if added_emb_type == "prd":
       __SCREAMING_SNAKE_CASE       :    str       =	nn.Parameter(torch.zeros(1     ,  1     ,  UpperCamelCase				)				)
      elif added_emb_type is None:
       __SCREAMING_SNAKE_CASE       :    Tuple       =	None
      else:
       raise ValueError(
           F"""`added_emb_type`: {added_emb_type} is not supported. Make sure to choose one of `'prd'` or `None`."""				)

      __SCREAMING_SNAKE_CASE       :    List[str]       =	nn.ModuleList(
          [
              BasicTransformerBlock(
                  UpperCamelCase     ,  UpperCamelCase     ,  UpperCamelCase     ,  dropout=UpperCamelCase     ,  activation_fn="gelu"     ,  attention_bias=UpperCamelCase     ,  )
              for d in range(UpperCamelCase				)
          ]				)

      if norm_in_type == "layer":
       __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	nn.LayerNorm(UpperCamelCase				)
      elif norm_in_type is None:
       __SCREAMING_SNAKE_CASE       :    List[Any]       =	None
      else:
       raise ValueError(F"""Unsupported norm_in_type: {norm_in_type}."""				)

      __SCREAMING_SNAKE_CASE       :    Optional[Any]       =	nn.LayerNorm(UpperCamelCase				)

      __SCREAMING_SNAKE_CASE       :    Dict       =	nn.Linear(UpperCamelCase     ,  UpperCamelCase				)

      __SCREAMING_SNAKE_CASE       :    int       =	torch.full(
          [num_embeddings + additional_embeddings, num_embeddings + additional_embeddings]     ,  -1_0_0_0_0.0				)
      causal_attention_mask.triu_(1				)
      __SCREAMING_SNAKE_CASE       :    int       =	causal_attention_mask[None, ...]
      self.register_buffer("causal_attention_mask"     ,  UpperCamelCase     ,  persistent=UpperCamelCase				)

      __SCREAMING_SNAKE_CASE       :    List[str]       =	nn.Parameter(torch.zeros(1     ,  UpperCamelCase				)				)
      __SCREAMING_SNAKE_CASE       :    str       =	nn.Parameter(torch.zeros(1     ,  UpperCamelCase				)				)





     @property
     # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
     def 	__snake_case    (       self :							Optional[Any]				)->Dict[str, AttentionProcessor]:
      __SCREAMING_SNAKE_CASE       :    str       =	{}

      def fn_recursive_add_processors(UpperCamelCase :							str     ,  UpperCamelCase :							torch.nn.Module     ,  UpperCamelCase :							Dict[str, AttentionProcessor]				):
       if hasattr(UpperCamelCase     ,  "set_processor"				):
        __SCREAMING_SNAKE_CASE       :    Any       =	module.processor

       for sub_name, child in module.named_children():
        fn_recursive_add_processors(F"""{name}.{sub_name}"""     ,  UpperCamelCase     ,  UpperCamelCase				)

       return processors

      for name, module in self.named_children():
       fn_recursive_add_processors(UpperCamelCase     ,  UpperCamelCase     ,  UpperCamelCase				)

      return processors





     def 	__snake_case    (       self :							Union[str, Any]     ,  UpperCamelCase :							Union[AttentionProcessor, Dict[str, AttentionProcessor]]				)->str:
      __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	len(self.attn_processors.keys()				)

      if isinstance(UpperCamelCase     ,  UpperCamelCase				) and len(UpperCamelCase				) != count:
       raise ValueError(
           F"""A dict of processors was passed, but the number of processors {len(UpperCamelCase				)} does not match the"""
           F""" number of attention layers: {count}. Please make sure to pass {count} processor classes."""				)

      def fn_recursive_attn_processor(UpperCamelCase :							str     ,  UpperCamelCase :							torch.nn.Module     ,  UpperCamelCase :							Optional[int]				):
       if hasattr(UpperCamelCase     ,  "set_processor"				):
        if not isinstance(UpperCamelCase     ,  UpperCamelCase				):
         module.set_processor(UpperCamelCase				)
        else:
         module.set_processor(processor.pop(F"""{name}.processor"""				)				)

       for sub_name, child in module.named_children():
        fn_recursive_attn_processor(F"""{name}.{sub_name}"""     ,  UpperCamelCase     ,  UpperCamelCase				)

      for name, module in self.named_children():
       fn_recursive_attn_processor(UpperCamelCase     ,  UpperCamelCase     ,  UpperCamelCase				)





     def 	__snake_case    (       self :							Any				)->List[Any]:
      self.set_attn_processor(AttnProcessor()				)





     def 	__snake_case    (       self :							Optional[Any]     ,  UpperCamelCase :							Optional[Any]     ,  UpperCamelCase :							Union[torch.Tensor, float, int]     ,  UpperCamelCase :							torch.FloatTensor     ,  UpperCamelCase :							Optional[torch.FloatTensor] = None     ,  UpperCamelCase :							Optional[torch.BoolTensor] = None     ,  UpperCamelCase :							bool = True     ,  )->Optional[Any]:
      __SCREAMING_SNAKE_CASE       :    Tuple       =	hidden_states.shape[0]

      __SCREAMING_SNAKE_CASE       :    str       =	timestep
      if not torch.is_tensor(UpperCamelCase				):
       __SCREAMING_SNAKE_CASE       :    str       =	torch.tensor([timesteps]     ,  dtype=torch.long     ,  device=hidden_states.device				)
      elif torch.is_tensor(UpperCamelCase				) and len(timesteps.shape				) == 0:
       __SCREAMING_SNAKE_CASE       :    List[str]       =	timesteps[None].to(hidden_states.device				)

      # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
      __SCREAMING_SNAKE_CASE       :    Dict       =	timesteps * torch.ones(UpperCamelCase     ,  dtype=timesteps.dtype     ,  device=timesteps.device				)

      __SCREAMING_SNAKE_CASE       :    Dict       =	self.time_proj(UpperCamelCase				)

      # timesteps does not contain any weights and will always return f32 tensors
      # but time_embedding might be fp16, so we need to cast here.
      __SCREAMING_SNAKE_CASE       :    List[str]       =	timesteps_projected.to(dtype=self.dtype				)
      __SCREAMING_SNAKE_CASE       :    int       =	self.time_embedding(UpperCamelCase				)

      if self.embedding_proj_norm is not None:
       __SCREAMING_SNAKE_CASE       :    Tuple       =	self.embedding_proj_norm(UpperCamelCase				)

      __SCREAMING_SNAKE_CASE       :    List[str]       =	self.embedding_proj(UpperCamelCase				)
      if self.encoder_hidden_states_proj is not None and encoder_hidden_states is not None:
       __SCREAMING_SNAKE_CASE       :    int       =	self.encoder_hidden_states_proj(UpperCamelCase				)
      elif self.encoder_hidden_states_proj is not None and encoder_hidden_states is None:
       raise ValueError("`encoder_hidden_states_proj` requires `encoder_hidden_states` to be set"				)

      __SCREAMING_SNAKE_CASE       :    Any       =	self.proj_in(UpperCamelCase				)

      __SCREAMING_SNAKE_CASE       :    int       =	self.positional_embedding.to(hidden_states.dtype				)

      __SCREAMING_SNAKE_CASE       :    List[str]       =	[]
      __SCREAMING_SNAKE_CASE       :    Any       =	0

      if encoder_hidden_states is not None:
       additional_embeds.append(UpperCamelCase				)
       additional_embeddings_len += encoder_hidden_states.shape[1]

      if len(proj_embeddings.shape				) == 2:
       __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	proj_embeddings[:, None, :]

      if len(hidden_states.shape				) == 2:
       __SCREAMING_SNAKE_CASE       :    Optional[Any]       =	hidden_states[:, None, :]

      __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	additional_embeds + [
          proj_embeddings,
          time_embeddings[:, None, :],
          hidden_states,
      ]

      if self.prd_embedding is not None:
       __SCREAMING_SNAKE_CASE       :    Any       =	self.prd_embedding.to(hidden_states.dtype				).expand(UpperCamelCase     ,  -1     ,  -1				)
       additional_embeds.append(UpperCamelCase				)

      __SCREAMING_SNAKE_CASE       :    str       =	torch.cat(
          UpperCamelCase     ,  dim=1     ,  )

      # Allow positional_embedding to not include the `addtional_embeddings` and instead pad it with zeros for these additional tokens
      __SCREAMING_SNAKE_CASE       :    List[Any]       =	additional_embeddings_len + proj_embeddings.shape[1] + 1
      if positional_embeddings.shape[1] < hidden_states.shape[1]:
       __SCREAMING_SNAKE_CASE       :    Tuple       =	F.pad(
           UpperCamelCase     ,  (
               0,
               0,
               additional_embeddings_len,
               self.prd_embedding.shape[1] if self.prd_embedding is not None else 0,
           )     ,  value=0.0     ,  )

      __SCREAMING_SNAKE_CASE       :    Any       =	hidden_states + positional_embeddings

      if attention_mask is not None:
       __SCREAMING_SNAKE_CASE       :    Dict       =	(1 - attention_mask.to(hidden_states.dtype				)) * -1_0_0_0_0.0
       __SCREAMING_SNAKE_CASE       :    List[Any]       =	F.pad(UpperCamelCase     ,  (0, self.additional_embeddings)     ,  value=0.0				)
       __SCREAMING_SNAKE_CASE       :    str       =	(attention_mask[:, None, :] + self.causal_attention_mask).to(hidden_states.dtype				)
       __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	attention_mask.repeat_interleave(self.config.num_attention_heads     ,  dim=0				)

      if self.norm_in is not None:
       __SCREAMING_SNAKE_CASE       :    Any       =	self.norm_in(UpperCamelCase				)

      for block in self.transformer_blocks:
       __SCREAMING_SNAKE_CASE       :    List[str]       =	block(UpperCamelCase     ,  attention_mask=UpperCamelCase				)

      __SCREAMING_SNAKE_CASE       :    Dict       =	self.norm_out(UpperCamelCase				)

      if self.prd_embedding is not None:
       __SCREAMING_SNAKE_CASE       :    Optional[int]       =	hidden_states[:, -1]
      else:
       __SCREAMING_SNAKE_CASE       :    str       =	hidden_states[:, additional_embeddings_len:]

      __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	self.proj_to_clip_embeddings(UpperCamelCase				)

      if not return_dict:
       return (predicted_image_embedding,)

      return PriorTransformerOutput(predicted_image_embedding=UpperCamelCase				)






     def 	__snake_case    (       self :							List[str]     ,  UpperCamelCase :							str				)->Union[str, Any]:
      __SCREAMING_SNAKE_CASE       :    Optional[int]       =	(prior_latents * self.clip_std) + self.clip_mean
      return prior_latents











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       _SCREAMING_SNAKE_CASE	(UpperCamelCase ):
     lowerCAmelCase					=			["""vqvae"""]





     def __init__(       self :							Tuple     ,  UpperCamelCase :							AutoencoderKL     ,  UpperCamelCase :							UNetaDConditionModel     ,  UpperCamelCase :							Mel     ,  UpperCamelCase :							Union[DDIMScheduler, DDPMScheduler]     ,  )->Tuple:
      super().__init__()
      self.register_modules(unet=UpperCamelCase     ,  scheduler=UpperCamelCase     ,  mel=UpperCamelCase     ,  vqvae=UpperCamelCase				)





     def 	__snake_case    (       self :							List[Any]				)->int:
      return 5_0 if isinstance(self.scheduler     ,  UpperCamelCase				) else 1_0_0_0





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

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

       if self.vqvae is not None:
        __SCREAMING_SNAKE_CASE       :    Optional[Any]       =	self.vqvae.encode(torch.unsqueeze(UpperCamelCase     ,  0				)				).latent_dist.sample(
            generator=UpperCamelCase				)[0]
        __SCREAMING_SNAKE_CASE       :    int       =	self.vqvae.config.scaling_factor * input_images

       if start_step > 0:
        __SCREAMING_SNAKE_CASE       :    List[str]       =	self.scheduler.add_noise(UpperCamelCase     ,  UpperCamelCase     ,  self.scheduler.timesteps[start_step - 1]				)

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

      for step, t in enumerate(self.progress_bar(self.scheduler.timesteps[start_step:]				)				):
       if isinstance(self.unet     ,  UpperCamelCase				):
        __SCREAMING_SNAKE_CASE       :    str       =	self.unet(UpperCamelCase     ,  UpperCamelCase     ,  UpperCamelCase				)["sample"]
       else:
        __SCREAMING_SNAKE_CASE       :    int       =	self.unet(UpperCamelCase     ,  UpperCamelCase				)["sample"]

       if isinstance(self.scheduler     ,  UpperCamelCase				):
        __SCREAMING_SNAKE_CASE       :    int       =	self.scheduler.step(
            model_output=UpperCamelCase     ,  timestep=UpperCamelCase     ,  sample=UpperCamelCase     ,  eta=UpperCamelCase     ,  generator=UpperCamelCase     ,  )["prev_sample"]
       else:
        __SCREAMING_SNAKE_CASE       :    Tuple       =	self.scheduler.step(
            model_output=UpperCamelCase     ,  timestep=UpperCamelCase     ,  sample=UpperCamelCase     ,  generator=UpperCamelCase     ,  )["prev_sample"]

       if mask is not None:
        if mask_start > 0:
         __SCREAMING_SNAKE_CASE       :    int       =	mask[:, step, :, :mask_start]
        if mask_end > 0:
         __SCREAMING_SNAKE_CASE       :    Optional[Any]       =	mask[:, step, :, -mask_end:]

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

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

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

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





     @torch.no_grad()
     def 	__snake_case    (       self :							Dict     ,  UpperCamelCase :							List[Image.Image]     ,  UpperCamelCase :							int = 5_0				)->np.ndarray:
      assert isinstance(self.scheduler     ,  UpperCamelCase				)
      self.scheduler.set_timesteps(UpperCamelCase				)
      __SCREAMING_SNAKE_CASE       :    int       =	np.array(
          [np.frombuffer(image.tobytes()     ,  dtype="uint8"				).reshape((1, image.height, image.width)				) for image in images]				)
      __SCREAMING_SNAKE_CASE       :    Dict       =	(sample / 2_5_5) * 2 - 1
      __SCREAMING_SNAKE_CASE       :    Optional[int]       =	torch.Tensor(UpperCamelCase				).to(self.device				)

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

      return sample






     @staticmethod
     def 	__snake_case    (       UpperCamelCase :							torch.Tensor     ,  UpperCamelCase :							torch.Tensor     ,  UpperCamelCase :							float				)->torch.Tensor:
      __SCREAMING_SNAKE_CASE       :    List[str]       =	acos(torch.dot(torch.flatten(UpperCamelCase				)     ,  torch.flatten(UpperCamelCase				)				) / torch.norm(UpperCamelCase				) / torch.norm(UpperCamelCase				)				)
      return sin((1 - alpha) * theta				) * xa / sin(UpperCamelCase				) + sin(alpha * theta				) * xa / sin(UpperCamelCase				)






