



"""simple docstring"""





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    ):
       '''simple docstring'''







       _UpperCAmelCase      				=					['''vqvae''']






       def __init__(  self	,       snake_case	,       snake_case	,       snake_case	,       snake_case	,       )			->							List[Any]:
         super().__init__()
         self.register_modules(unet=snake_case	,       scheduler=snake_case	,       mel=snake_case	,       vqvae=snake_case						)






       def        lowerCamelCase_   (  self						)			->							int:
         return 50 if isinstance(self.scheduler	,       snake_case						) else 1000






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

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

           if self.vqvae is not None:
             _UpperCAmelCase							 =		self.vqvae.encode(torch.unsqueeze(snake_case	,       0						)						).latent_dist.sample(
                 generator=snake_case						)[0]
             _UpperCAmelCase							 =		self.vqvae.config.scaling_factor * input_images

           if start_step > 0:
             _UpperCAmelCase							 =		self.scheduler.add_noise(snake_case	,       snake_case	,       self.scheduler.timesteps[start_step - 1]						)

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

         for step, t in enumerate(self.progress_bar(self.scheduler.timesteps[start_step:]						)						):
           if isinstance(self.unet	,       snake_case						):
             _UpperCAmelCase							 =		self.unet(snake_case	,       snake_case	,       snake_case						)['sample']
           else:
             _UpperCAmelCase							 =		self.unet(snake_case	,       snake_case						)['sample']

           if isinstance(self.scheduler	,       snake_case						):
             _UpperCAmelCase							 =		self.scheduler.step(
                 model_output=snake_case	,       timestep=snake_case	,       sample=snake_case	,       eta=snake_case	,       generator=snake_case	,       )['prev_sample']
           else:
             _UpperCAmelCase							 =		self.scheduler.step(
                 model_output=snake_case	,       timestep=snake_case	,       sample=snake_case	,       generator=snake_case	,       )['prev_sample']

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

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

         _UpperCAmelCase							 =		(images / 2 + 0.5).clamp(0	,       1						)
         _UpperCAmelCase							 =		images.cpu().permute(0	,       2	,       3	,       1						).numpy()
         _UpperCAmelCase							 =		(images * 255).round().astype('uint8'						)
         _UpperCAmelCase							 =		list(
             (Image.fromarray(_[:, :, 0]						) for _ in images)
             if images.shape[3] == 1
             else (Image.fromarray(snake_case	,       mode='RGB'						).convert('L'						) for _ in images)						)

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

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






       @torch.no_grad()
       def        lowerCamelCase_   (  self	,       snake_case	,       snake_case = 50						)			->							np.ndarray:
         assert isinstance(self.scheduler	,       snake_case						)
         self.scheduler.set_timesteps(snake_case						)
         _UpperCAmelCase							 =		np.array(
             [np.frombuffer(image.tobytes()	,       dtype='uint8'						).reshape((1, image.height, image.width)						) for image in images]						)
         _UpperCAmelCase							 =		(sample / 255) * 2 - 1
         _UpperCAmelCase							 =		torch.Tensor(snake_case						).to(self.device						)

         for t in self.progress_bar(torch.flip(self.scheduler.timesteps	,       (0,)						)						):
           _UpperCAmelCase							 =		t - self.scheduler.config.num_train_timesteps // self.scheduler.num_inference_steps
           _UpperCAmelCase							 =		self.scheduler.alphas_cumprod[t]
           _UpperCAmelCase							 =		(
               self.scheduler.alphas_cumprod[prev_timestep]
               if prev_timestep >= 0
               else self.scheduler.final_alpha_cumprod
           )
           _UpperCAmelCase							 =		1 - alpha_prod_t
           _UpperCAmelCase							 =		self.unet(snake_case	,       snake_case						)['sample']
           _UpperCAmelCase							 =		(1 - alpha_prod_t_prev) ** 0.5 * model_output
           _UpperCAmelCase							 =		(sample - pred_sample_direction) * alpha_prod_t_prev ** (-0.5)
           _UpperCAmelCase							 =		sample * alpha_prod_t ** 0.5 + beta_prod_t ** 0.5 * model_output

         return sample






       @staticmethod
       def        lowerCamelCase_   (  snake_case	,       snake_case	,       snake_case						)			->							torch.Tensor:
         _UpperCAmelCase							 =		acos(torch.dot(torch.flatten(snake_case						)	,       torch.flatten(snake_case						)						) / torch.norm(snake_case						) / torch.norm(snake_case						)						)
         return sin((1 - alpha) * theta						) * xa / sin(snake_case						) + sin(alpha * theta						) * xa / sin(snake_case						)







"""simple docstring"""





from typing import Optional

from torch import nn

from .transformer_ad import TransformeraDModel, TransformeraDModelOutput




class 			lowercase__			(      nn.Module    ):
							'''simple docstring'''







							def __init__(  self	,       snake_case = 16	,       snake_case = 88	,       snake_case = None	,       snake_case = 1	,       snake_case = 0.0	,       snake_case = 32	,       snake_case = None	,       snake_case = False	,       snake_case = None	,       snake_case = None	,       snake_case = "geglu"	,       snake_case = None	,       )			->							str:
									super().__init__()
									_UpperCAmelCase							 =		nn.ModuleList(
									    [
									        TransformeraDModel(
									            num_attention_heads=snake_case	,       attention_head_dim=snake_case	,       in_channels=snake_case	,       num_layers=snake_case	,       dropout=snake_case	,       norm_num_groups=snake_case	,       cross_attention_dim=snake_case	,       attention_bias=snake_case	,       sample_size=snake_case	,       num_vector_embeds=snake_case	,       activation_fn=snake_case	,       num_embeds_ada_norm=snake_case	,       )
									        for _ in range(2						)
									    ]						)

									# Variables that can be set by a pipeline:

									# The ratio of transformer1 to transformer2's output states to be combined during inference
									_UpperCAmelCase							 =		0.5

									# The shape of `encoder_hidden_states` is expected to be
									# `(batch_size, condition_lengths[0]+condition_lengths[1], num_features)`
									_UpperCAmelCase							 =		[77, 257]

									# Which transformer to use to encode which condition.
									# E.g. `(1, 0)` means that we'll use `transformers[1](conditions[0])` and `transformers[0](conditions[1])`
									_UpperCAmelCase							 =		[1, 0]






							def        lowerCamelCase_   (  self	,       snake_case	,       snake_case	,       snake_case=None	,       snake_case=None	,       snake_case=None	,       snake_case = True	,       )			->							Any:
									_UpperCAmelCase							 =		hidden_states

									_UpperCAmelCase							 =		[]
									_UpperCAmelCase							 =		0
									# attention_mask is not used yet
									for i in range(2						):
											# for each of the two transformers, pass the corresponding condition tokens
											_UpperCAmelCase							 =		encoder_hidden_states[:, tokens_start : tokens_start + self.condition_lengths[i]]
											_UpperCAmelCase							 =		self.transformer_index_for_condition[i]
											_UpperCAmelCase							 =		self.transformers[transformer_index](
											    snake_case	,       encoder_hidden_states=snake_case	,       timestep=snake_case	,       cross_attention_kwargs=snake_case	,       return_dict=snake_case	,       )[0]
											encoded_states.append(encoded_state - input_states						)
											tokens_start += self.condition_lengths[i]

									_UpperCAmelCase							 =		encoded_states[0] * self.mix_ratio + encoded_states[1] * (1 - self.mix_ratio)
									_UpperCAmelCase							 =		output_states + input_states

									if not return_dict:
											return (output_states,)

									return TransformeraDModelOutput(sample=snake_case						)


