
from dataclasses import dataclass
from typing import Optional

import numpy as np
import torch
import torch.nn as nn

from ..utils import BaseOutput, is_torch_version, randn_tensor
from .attention_processor import SpatialNorm
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block





@dataclass
class      lowercase__    (							UpperCamelCase__  ):
  __UpperCamelCase	     =  42





class      lowercase__    (							nn.Module  ):







  def __init__(	self    ,						_lowercase=3    ,						_lowercase=3    ,						_lowercase=("DownEncoderBlock2D",)    ,						_lowercase=(64,)    ,						_lowercase=2    ,						_lowercase=32    ,						_lowercase="silu"    ,						_lowercase=True    ,						):
       super().__init__()
       lowerCAmelCase_					:       Optional[int]    =		layers_per_block

       lowerCAmelCase_					:       Tuple    =		torch.nn.Convad(
           _a    ,						block_out_channels[0]    ,						kernel_size=3    ,						stride=1    ,						padding=1    ,						)

       lowerCAmelCase_					:       str    =		None
       lowerCAmelCase_					:       int    =		nn.ModuleList([]						)

       # down
       lowerCAmelCase_					:       Tuple    =		block_out_channels[0]
       for i, down_block_type in enumerate(_a						):
            lowerCAmelCase_					:       Optional[int]    =		output_channel
            lowerCAmelCase_					:       List[Any]    =		block_out_channels[i]
            lowerCAmelCase_					:       List[Any]    =		i == len(_a						) - 1

            lowerCAmelCase_					:       Union[str, Any]    =		get_down_block(
                _a    ,						num_layers=self.layers_per_block    ,						in_channels=_a    ,						out_channels=_a    ,						add_downsample=not is_final_block    ,						resnet_eps=1e-6    ,						downsample_padding=0    ,						resnet_act_fn=_a    ,						resnet_groups=_a    ,						attention_head_dim=_a    ,						temb_channels=_a    ,						)
            self.down_blocks.append(_a						)

       # mid
       lowerCAmelCase_					:       List[str]    =		UNetMidBlockaD(
           in_channels=block_out_channels[-1]    ,						resnet_eps=1e-6    ,						resnet_act_fn=_a    ,						output_scale_factor=1    ,						resnet_time_scale_shift="""default"""    ,						attention_head_dim=block_out_channels[-1]    ,						resnet_groups=_a    ,						temb_channels=_a    ,						)

       # out
       lowerCAmelCase_					:       Dict    =		nn.GroupNorm(num_channels=block_out_channels[-1]    ,						num_groups=_a    ,						eps=1e-6						)
       lowerCAmelCase_					:       Optional[int]    =		nn.SiLU()

       lowerCAmelCase_					:       List[str]    =		2 * out_channels if double_z else out_channels
       lowerCAmelCase_					:       List[str]    =		nn.Convad(block_out_channels[-1]    ,						_a    ,						3    ,						padding=1						)

       lowerCAmelCase_					:       int    =		False







  def 	UpperCAmelCase__		(	self    ,						_lowercase						):
       lowerCAmelCase_					:       Dict    =		x
       lowerCAmelCase_					:       Tuple    =		self.conv_in(_a						)

       if self.training and self.gradient_checkpointing:

            def create_custom_forward(_lowercase						):
                 def custom_forward(*_lowercase						):
                      return module(*_a						)

                 return custom_forward

            # down
            if is_torch_version(""">="""    ,						"""1.11.0"""						):
                 for down_block in self.down_blocks:
                      lowerCAmelCase_					:       Tuple    =		torch.utils.checkpoint.checkpoint(
                          create_custom_forward(_a						)    ,						_a    ,						use_reentrant=_a						)
                 # middle
                 lowerCAmelCase_					:       Optional[int]    =		torch.utils.checkpoint.checkpoint(
                     create_custom_forward(self.mid_block						)    ,						_a    ,						use_reentrant=_a						)
            else:
                 for down_block in self.down_blocks:
                      lowerCAmelCase_					:       Optional[Any]    =		torch.utils.checkpoint.checkpoint(create_custom_forward(_a						)    ,						_a						)
                 # middle
                 lowerCAmelCase_					:       Union[str, Any]    =		torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block						)    ,						_a						)

       else:
            # down
            for down_block in self.down_blocks:
                 lowerCAmelCase_					:       Optional[int]    =		down_block(_a						)

            # middle
            lowerCAmelCase_					:       int    =		self.mid_block(_a						)

       # post-process
       lowerCAmelCase_					:       Tuple    =		self.conv_norm_out(_a						)
       lowerCAmelCase_					:       Optional[int]    =		self.conv_act(_a						)
       lowerCAmelCase_					:       Optional[int]    =		self.conv_out(_a						)

       return sample





class      lowercase__    (							nn.Module  ):







  def __init__(	self    ,						_lowercase=3    ,						_lowercase=3    ,						_lowercase=("UpDecoderBlock2D",)    ,						_lowercase=(64,)    ,						_lowercase=2    ,						_lowercase=32    ,						_lowercase="silu"    ,						_lowercase="group"    ,						):
       super().__init__()
       lowerCAmelCase_					:       str    =		layers_per_block

       lowerCAmelCase_					:       str    =		nn.Convad(
           _a    ,						block_out_channels[-1]    ,						kernel_size=3    ,						stride=1    ,						padding=1    ,						)

       lowerCAmelCase_					:       Optional[int]    =		None
       lowerCAmelCase_					:       int    =		nn.ModuleList([]						)

       lowerCAmelCase_					:       Dict    =		in_channels if norm_type == """spatial""" else None

       # mid
       lowerCAmelCase_					:       Tuple    =		UNetMidBlockaD(
           in_channels=block_out_channels[-1]    ,						resnet_eps=1e-6    ,						resnet_act_fn=_a    ,						output_scale_factor=1    ,						resnet_time_scale_shift="""default""" if norm_type == """group""" else norm_type    ,						attention_head_dim=block_out_channels[-1]    ,						resnet_groups=_a    ,						temb_channels=_a    ,						)

       # up
       lowerCAmelCase_					:       Union[str, Any]    =		list(reversed(_a						)						)
       lowerCAmelCase_					:       int    =		reversed_block_out_channels[0]
       for i, up_block_type in enumerate(_a						):
            lowerCAmelCase_					:       List[Any]    =		output_channel
            lowerCAmelCase_					:       Tuple    =		reversed_block_out_channels[i]

            lowerCAmelCase_					:       Optional[Any]    =		i == len(_a						) - 1

            lowerCAmelCase_					:       Optional[int]    =		get_up_block(
                _a    ,						num_layers=self.layers_per_block + 1    ,						in_channels=_a    ,						out_channels=_a    ,						prev_output_channel=_a    ,						add_upsample=not is_final_block    ,						resnet_eps=1e-6    ,						resnet_act_fn=_a    ,						resnet_groups=_a    ,						attention_head_dim=_a    ,						temb_channels=_a    ,						resnet_time_scale_shift=_a    ,						)
            self.up_blocks.append(_a						)
            lowerCAmelCase_					:       str    =		output_channel

       # out
       if norm_type == "spatial":
            lowerCAmelCase_					:       str    =		SpatialNorm(block_out_channels[0]    ,						_a						)
       else:
            lowerCAmelCase_					:       int    =		nn.GroupNorm(num_channels=block_out_channels[0]    ,						num_groups=_a    ,						eps=1e-6						)
       lowerCAmelCase_					:       Dict    =		nn.SiLU()
       lowerCAmelCase_					:       str    =		nn.Convad(block_out_channels[0]    ,						_a    ,						3    ,						padding=1						)

       lowerCAmelCase_					:       int    =		False







  def 	UpperCAmelCase__		(	self    ,						_lowercase    ,						_lowercase=None						):
       lowerCAmelCase_					:       Union[str, Any]    =		z
       lowerCAmelCase_					:       int    =		self.conv_in(_a						)

       lowerCAmelCase_					:       Any    =		next(iter(self.up_blocks.parameters()						)						).dtype
       if self.training and self.gradient_checkpointing:

            def create_custom_forward(_lowercase						):
                 def custom_forward(*_lowercase						):
                      return module(*_a						)

                 return custom_forward

            if is_torch_version(""">="""    ,						"""1.11.0"""						):
                 # middle
                 lowerCAmelCase_					:       List[str]    =		torch.utils.checkpoint.checkpoint(
                     create_custom_forward(self.mid_block						)    ,						_a    ,						_a    ,						use_reentrant=_a						)
                 lowerCAmelCase_					:       List[Any]    =		sample.to(_a						)

                 # up
                 for up_block in self.up_blocks:
                      lowerCAmelCase_					:       Optional[int]    =		torch.utils.checkpoint.checkpoint(
                          create_custom_forward(_a						)    ,						_a    ,						_a    ,						use_reentrant=_a						)
            else:
                 # middle
                 lowerCAmelCase_					:       int    =		torch.utils.checkpoint.checkpoint(
                     create_custom_forward(self.mid_block						)    ,						_a    ,						_a						)
                 lowerCAmelCase_					:       int    =		sample.to(_a						)

                 # up
                 for up_block in self.up_blocks:
                      lowerCAmelCase_					:       int    =		torch.utils.checkpoint.checkpoint(create_custom_forward(_a						)    ,						_a    ,						_a						)
       else:
            # middle
            lowerCAmelCase_					:       Any    =		self.mid_block(_a    ,						_a						)
            lowerCAmelCase_					:       Union[str, Any]    =		sample.to(_a						)

            # up
            for up_block in self.up_blocks:
                 lowerCAmelCase_					:       Union[str, Any]    =		up_block(_a    ,						_a						)

        # post-process
       if latent_embeds is None:
            lowerCAmelCase_					:       Any    =		self.conv_norm_out(_a						)
       else:
            lowerCAmelCase_					:       Union[str, Any]    =		self.conv_norm_out(_a    ,						_a						)
       lowerCAmelCase_					:       str    =		self.conv_act(_a						)
       lowerCAmelCase_					:       Tuple    =		self.conv_out(_a						)

       return sample





class      lowercase__    (							nn.Module  ):







  def __init__(	self    ,						_lowercase    ,						_lowercase    ,						_lowercase    ,						_lowercase=None    ,						_lowercase="random"    ,						_lowercase=False    ,						_lowercase=True						):
       super().__init__()
       lowerCAmelCase_					:       Tuple    =		n_e
       lowerCAmelCase_					:       Optional[Any]    =		vq_embed_dim
       lowerCAmelCase_					:       Tuple    =		beta
       lowerCAmelCase_					:       Dict    =		legacy

       lowerCAmelCase_					:       List[str]    =		nn.Embedding(self.n_e    ,						self.vq_embed_dim						)
       self.embedding.weight.data.uniform_(-1.0 / self.n_e    ,						1.0 / self.n_e						)

       lowerCAmelCase_					:       Tuple    =		remap
       if self.remap is not None:
            self.register_buffer("""used"""    ,						torch.tensor(np.load(self.remap						)						)						)
            lowerCAmelCase_					:       int    =		self.used.shape[0]
            lowerCAmelCase_					:       Optional[int]    =		unknown_index  # "random" or "extra" or integer
            if self.unknown_index == "extra":
                 lowerCAmelCase_					:       Optional[int]    =		self.re_embed
                 lowerCAmelCase_					:       Optional[Any]    =		self.re_embed + 1
            print(
                F'Remapping {self.n_e} indices to {self.re_embed} indices. '
                F'Using {self.unknown_index} for unknown indices.'						)
       else:
            lowerCAmelCase_					:       Dict    =		n_e

       lowerCAmelCase_					:       Dict    =		sane_index_shape







  def 	UpperCAmelCase__		(	self    ,						_lowercase						):
       lowerCAmelCase_					:       Tuple    =		inds.shape
       assert len(_a						) > 1
       lowerCAmelCase_					:       Union[str, Any]    =		inds.reshape(ishape[0]    ,						-1						)
       lowerCAmelCase_					:       Union[str, Any]    =		self.used.to(_a						)
       lowerCAmelCase_					:       Optional[Any]    =		(inds[:, :, None] == used[None, None, ...]).long()
       lowerCAmelCase_					:       int    =		match.argmax(-1						)
       lowerCAmelCase_					:       Any    =		match.sum(2						) < 1
       if self.unknown_index == "random":
            lowerCAmelCase_					:       List[str]    =		torch.randint(0    ,						self.re_embed    ,						size=new[unknown].shape						).to(device=new.device						)
       else:
            lowerCAmelCase_					:       int    =		self.unknown_index
       return new.reshape(_a						)







  def 	UpperCAmelCase__		(	self    ,						_lowercase						):
       lowerCAmelCase_					:       Tuple    =		inds.shape
       assert len(_a						) > 1
       lowerCAmelCase_					:       List[Any]    =		inds.reshape(ishape[0]    ,						-1						)
       lowerCAmelCase_					:       Union[str, Any]    =		self.used.to(_a						)
       if self.re_embed > self.used.shape[0]:  # extra token
            lowerCAmelCase_					:       str    =		0  # simply set to zero
       lowerCAmelCase_					:       Any    =		torch.gather(used[None, :][inds.shape[0] * [0], :]    ,						1    ,						_a						)
       return back.reshape(_a						)







  def 	UpperCAmelCase__		(	self    ,						_lowercase						):
       # reshape z -> (batch, height, width, channel) and flatten
       lowerCAmelCase_					:       Any    =		z.permute(0    ,						2    ,						3    ,						1						).contiguous()
       lowerCAmelCase_					:       int    =		z.view(-1    ,						self.vq_embed_dim						)

       # distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
       lowerCAmelCase_					:       Tuple    =		torch.argmin(torch.cdist(_a    ,						self.embedding.weight						)    ,						dim=1						)

       lowerCAmelCase_					:       Any    =		self.embedding(_a						).view(z.shape						)
       lowerCAmelCase_					:       Optional[Any]    =		None
       lowerCAmelCase_					:       Any    =		None

       # compute loss for embedding
       if not self.legacy:
            lowerCAmelCase_					:       List[str]    =		self.beta * torch.mean((z_q.detach() - z) ** 2						) + torch.mean((z_q - z.detach()) ** 2						)
       else:
            lowerCAmelCase_					:       Any    =		torch.mean((z_q.detach() - z) ** 2						) + self.beta * torch.mean((z_q - z.detach()) ** 2						)

       # preserve gradients
       lowerCAmelCase_					:       List[str]    =		z + (z_q - z).detach()

       # reshape back to match original input shape
       lowerCAmelCase_					:       Dict    =		z_q.permute(0    ,						3    ,						1    ,						2						).contiguous()

       if self.remap is not None:
            lowerCAmelCase_					:       str    =		min_encoding_indices.reshape(z.shape[0]    ,						-1						)  # add batch axis
            lowerCAmelCase_					:       int    =		self.remap_to_used(_a						)
            lowerCAmelCase_					:       List[Any]    =		min_encoding_indices.reshape(-1    ,						1						)  # flatten

       if self.sane_index_shape:
            lowerCAmelCase_					:       int    =		min_encoding_indices.reshape(z_q.shape[0]    ,						z_q.shape[2]    ,						z_q.shape[3]						)

       return z_q, loss, (perplexity, min_encodings, min_encoding_indices)







  def 	UpperCAmelCase__		(	self    ,						_lowercase    ,						_lowercase						):
       # shape specifying (batch, height, width, channel)
       if self.remap is not None:
            lowerCAmelCase_					:       List[str]    =		indices.reshape(shape[0]    ,						-1						)  # add batch axis
            lowerCAmelCase_					:       Optional[int]    =		self.unmap_to_all(_a						)
            lowerCAmelCase_					:       Any    =		indices.reshape(-1						)  # flatten again

       # get quantized latent vectors
       lowerCAmelCase_					:       str    =		self.embedding(_a						)

       if shape is not None:
            lowerCAmelCase_					:       int    =		z_q.view(_a						)
            # reshape back to match original input shape
            lowerCAmelCase_					:       Optional[Any]    =		z_q.permute(0    ,						3    ,						1    ,						2						).contiguous()

       return z_q





class      lowercase__    (							UpperCamelCase__  ):







  def __init__(	self    ,						_lowercase    ,						_lowercase=False						):
       lowerCAmelCase_					:       Dict    =		parameters
       lowerCAmelCase_					:       Tuple    =		torch.chunk(_a    ,						2    ,						dim=1						)
       lowerCAmelCase_					:       Optional[Any]    =		torch.clamp(self.logvar    ,						-30.0    ,						20.0						)
       lowerCAmelCase_					:       Tuple    =		deterministic
       lowerCAmelCase_					:       Any    =		torch.exp(0.5 * self.logvar						)
       lowerCAmelCase_					:       Optional[int]    =		torch.exp(self.logvar						)
       if self.deterministic:
            lowerCAmelCase_					:       List[Any]    =		torch.zeros_like(
                self.mean    ,						device=self.parameters.device    ,						dtype=self.parameters.dtype						)







  def 	UpperCAmelCase__		(	self    ,						_lowercase = None						):
       # make sure sample is on the same device as the parameters and has same dtype
       lowerCAmelCase_					:       Union[str, Any]    =		randn_tensor(
           self.mean.shape    ,						generator=_a    ,						device=self.parameters.device    ,						dtype=self.parameters.dtype						)
       lowerCAmelCase_					:       int    =		self.mean + self.std * sample
       return x







  def 	UpperCAmelCase__		(	self    ,						_lowercase=None						):
       if self.deterministic:
            return torch.Tensor([0.0]						)
       else:
            if other is None:
                 return 0.5 * torch.sum(torch.pow(self.mean    ,						2						) + self.var - 1.0 - self.logvar    ,						dim=[1, 2, 3]						)
            else:
                 return 0.5 * torch.sum(
                     torch.pow(self.mean - other.mean    ,						2						) / other.var
                     + self.var / other.var
                     - 1.0
                     - self.logvar
                     + other.logvar    ,						dim=[1, 2, 3]    ,						)







  def 	UpperCAmelCase__		(	self    ,						_lowercase    ,						_lowercase=[1, 2, 3]						):
       if self.deterministic:
            return torch.Tensor([0.0]						)
       lowerCAmelCase_					:       List[str]    =		np.log(2.0 * np.pi						)
       return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean    ,						2						) / self.var    ,						dim=_a						)







  def 	UpperCAmelCase__		(	self						):
       return self.mean







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      lowercase__    (							__A  ):
  __UpperCamelCase	     =  ["""vqvae"""]







  def __init__(	self    ,						_lowercase    ,						_lowercase    ,						_lowercase    ,						_lowercase    ,						):
       super().__init__()
       self.register_modules(unet=_lowercase    ,						scheduler=_lowercase    ,						mel=_lowercase    ,						vqvae=_lowercase						)







  def 	UpperCAmelCase__		(	self						):
       return 50 if isinstance(self.scheduler    ,						_lowercase						) else 1_000







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

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

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

            if start_step > 0:
                 lowerCAmelCase_					:       str    =		self.scheduler.add_noise(_lowercase    ,						_lowercase    ,						self.scheduler.timesteps[start_step - 1]						)

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

       for step, t in enumerate(self.progress_bar(self.scheduler.timesteps[start_step:]						)						):
            if isinstance(self.unet    ,						_lowercase						):
                 lowerCAmelCase_					:       List[Any]    =		self.unet(_lowercase    ,						_lowercase    ,						_lowercase						)["""sample"""]
            else:
                 lowerCAmelCase_					:       Any    =		self.unet(_lowercase    ,						_lowercase						)["""sample"""]

            if isinstance(self.scheduler    ,						_lowercase						):
                 lowerCAmelCase_					:       str    =		self.scheduler.step(
                     model_output=_lowercase    ,						timestep=_lowercase    ,						sample=_lowercase    ,						eta=_lowercase    ,						generator=_lowercase    ,						)["""prev_sample"""]
            else:
                 lowerCAmelCase_					:       List[str]    =		self.scheduler.step(
                     model_output=_lowercase    ,						timestep=_lowercase    ,						sample=_lowercase    ,						generator=_lowercase    ,						)["""prev_sample"""]

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

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

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

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

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







  @torch.no_grad()
  def 	UpperCAmelCase__		(	self    ,						_lowercase    ,						_lowercase = 50						):
       assert isinstance(self.scheduler    ,						_lowercase						)
       self.scheduler.set_timesteps(_lowercase						)
       lowerCAmelCase_					:       List[Any]    =		np.array(
           [np.frombuffer(image.tobytes()    ,						dtype="""uint8"""						).reshape((1, image.height, image.width)						) for image in images]						)
       lowerCAmelCase_					:       List[str]    =		(sample / 255) * 2 - 1
       lowerCAmelCase_					:       Optional[Any]    =		torch.Tensor(_lowercase						).to(self.device						)

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

       return sample







  @staticmethod
  def 	UpperCAmelCase__		(	_lowercase    ,						_lowercase    ,						_lowercase						):
       lowerCAmelCase_					:       Optional[int]    =		acos(torch.dot(torch.flatten(_lowercase						)    ,						torch.flatten(_lowercase						)						) / torch.norm(_lowercase						) / torch.norm(_lowercase						)						)
       return sin((1 - alpha) * theta						) * xa / sin(_lowercase						) + sin(alpha * theta						) * xa / sin(_lowercase						)
