
import math
from typing import Any, Callable, List, Optional, Tuple, Union

import numpy as np
import torch

from ...models import TaFilmDecoder
from ...schedulers import DDPMScheduler
from ...utils import is_onnx_available, logging, randn_tensor


if is_onnx_available():
			from ..onnx_utils import OnnxRuntimeModel

from ..pipeline_utils import AudioPipelineOutput, DiffusionPipeline
from .continous_encoder import SpectrogramContEncoder
from .notes_encoder import SpectrogramNotesEncoder


_lowerCamelCase       :				List[Any]       =			logging.get_logger(__name__)  # pylint: disable=invalid-name

_lowerCamelCase       :				Dict       =			256




class    SCREAMING_SNAKE_CASE__     (	UpperCAmelCase					):





	'''simple docstring'''



	_UpperCAmelCase       :		List[Any]       =					["melgan"]



	def __init__(  self	:    Dict			,  lowercase	:    SpectrogramNotesEncoder			,  lowercase	:    SpectrogramContEncoder			,  lowercase	:    TaFilmDecoder			,  lowercase	:    DDPMScheduler			,  lowercase	:    OnnxRuntimeModel if is_onnx_available() else Any			,  ):


				'''simple docstring'''


				super().__init__()

				# From MELGAN
				_snake_case		=			math.log(1E-5	)  # Matches MelGAN training.
				_snake_case		=			4.0  # Largest value for most examples
				_snake_case		=			128

				self.register_modules(
				    notes_encoder=lowercase			,  continuous_encoder=lowercase			,  decoder=lowercase			,  scheduler=lowercase			,  melgan=lowercase			,  )



	def  A							(  self	:    Any			,  lowercase	:    int			,  lowercase	:    Dict=(-1.0, 1.0)			,  lowercase	:    Dict=False	):


				'''simple docstring'''


				_snake_case	,					_snake_case		=			output_range
				if clip:
							_snake_case		=			torch.clip(lowercase			,  self.min_value			,  self.max_value	)
				# Scale to [0, 1].
				_snake_case		=			(features - self.min_value) / (self.max_value - self.min_value)
				# Scale to [min_out, max_out].
				return zero_one * (max_out - min_out) + min_out



	def  A							(  self	:    Optional[int]			,  lowercase	:    Optional[int]			,  lowercase	:    str=(-1.0, 1.0)			,  lowercase	:    List[str]=False	):


				'''simple docstring'''


				_snake_case	,					_snake_case		=			input_range
				_snake_case		=			torch.clip(lowercase			,  lowercase			,  lowercase	) if clip else outputs
				# Scale to [0, 1].
				_snake_case		=			(outputs - min_out) / (max_out - min_out)
				# Scale to [self.min_value, self.max_value].
				return zero_one * (self.max_value - self.min_value) + self.min_value



	def  A							(  self	:    Optional[int]			,  lowercase	:    Optional[int]			,  lowercase	:    List[Any]			,  lowercase	:    Optional[int]	):


				'''simple docstring'''


				_snake_case		=			input_tokens > 0
				_snake_case	,					_snake_case		=			self.notes_encoder(
				    encoder_input_tokens=lowercase			,  encoder_inputs_mask=lowercase	)

				_snake_case	,					_snake_case		=			self.continuous_encoder(
				    encoder_inputs=lowercase			,  encoder_inputs_mask=lowercase	)

				return [(tokens_encoded, tokens_mask), (continuous_encoded, continuous_mask)]



	def  A							(  self	:    Optional[int]			,  lowercase	:    List[str]			,  lowercase	:    List[Any]			,  lowercase	:    List[str]	):


				'''simple docstring'''


				_snake_case		=			noise_time
				if not torch.is_tensor(lowercase	):
							_snake_case		=			torch.tensor([timesteps]			,  dtype=torch.long			,  device=input_tokens.device	)
				elif torch.is_tensor(lowercase	) and len(timesteps.shape	) == 0:
							_snake_case		=			timesteps[None].to(input_tokens.device	)

				# broadcast to batch dimension in a way that's compatible with ONNX/Core ML
				_snake_case		=			timesteps * torch.ones(input_tokens.shape[0]			,  dtype=timesteps.dtype			,  device=timesteps.device	)

				_snake_case		=			self.decoder(
				    encodings_and_masks=lowercase			,  decoder_input_tokens=lowercase			,  decoder_noise_time=lowercase	)
				return logits





	@torch.no_grad()
	def __call__(  self	:    List[Any]			,  lowercase	:    List[List[int]]			,  lowercase	:    Optional[torch.Generator] = None			,  lowercase	:    int = 100			,  lowercase	:    bool = True			,  lowercase	:    str = "numpy"			,  lowercase	:    Optional[Callable[[int, int, torch.FloatTensor], None]] = None			,  lowercase	:    int = 1			,  ):


				'''simple docstring'''


				if (callback_steps is None) or (
				    callback_steps is not None and (not isinstance(lowercase			,  lowercase	) or callback_steps <= 0)
				):
							raise ValueError(
							    f'''`callback_steps` has to be a positive integer but is {callback_steps} of type'''
							    f''' {type(lowercase	)}.'''	)

				_snake_case		=			np.zeros([1, TARGET_FEATURE_LENGTH, self.n_dims]			,  dtype=np.floataa	)
				_snake_case		=			np.zeros([1, 0, self.n_dims]			,  np.floataa	)
				_snake_case		=			torch.ones((1, TARGET_FEATURE_LENGTH)			,  dtype=lowercase			,  device=self.device	)

				for i, encoder_input_tokens in enumerate(lowercase	):
							if i == 0:
										_snake_case		=			torch.from_numpy(pred_mel[:1].copy()	).to(
										    device=self.device			,  dtype=self.decoder.dtype	)
										# The first chunk has no previous context.
										_snake_case		=			torch.zeros((1, TARGET_FEATURE_LENGTH)			,  dtype=lowercase			,  device=self.device	)
							else:
										# The full song pipeline does not feed in a context feature, so the mask
										# will be all 0s after the feature converter. Because we know we're
										# feeding in a full context chunk from the previous prediction, set it
										# to all 1s.
										_snake_case		=			ones

							_snake_case		=			self.scale_features(
							    lowercase			,  output_range=[-1.0, 1.0]			,  clip=lowercase	)

							_snake_case		=			self.encode(
							    input_tokens=torch.IntTensor([encoder_input_tokens]	).to(device=self.device	)			,  continuous_inputs=lowercase			,  continuous_mask=lowercase			,  )

							# Sample encoder_continuous_inputs shaped gaussian noise to begin loop
							_snake_case		=			randn_tensor(
							    shape=encoder_continuous_inputs.shape			,  generator=lowercase			,  device=self.device			,  dtype=self.decoder.dtype			,  )

							# set step values
							self.scheduler.set_timesteps(lowercase	)

							# Denoising diffusion loop
							for j, t in enumerate(self.progress_bar(self.scheduler.timesteps	)	):
										_snake_case		=			self.decode(
										    encodings_and_masks=lowercase			,  input_tokens=lowercase			,  noise_time=t / self.scheduler.config.num_train_timesteps			,  )

										# Compute previous output: x_t -> x_t-1
										_snake_case		=			self.scheduler.step(lowercase			,  lowercase			,  lowercase			,  generator=lowercase	).prev_sample

							_snake_case		=			self.scale_to_features(lowercase			,  input_range=[-1.0, 1.0]	)
							_snake_case		=			mel[:1]
							_snake_case		=			mel.cpu().float().numpy()

							_snake_case		=			np.concatenate([full_pred_mel, pred_mel[:1]]			,  axis=1	)

							# call the callback, if provided
							if callback is not None and i % callback_steps == 0:
										callback(lowercase			,  lowercase	)

							logger.info('Generated segment'			,  lowercase	)

				if output_type == "numpy" and not is_onnx_available():
							raise ValueError(
							    'Cannot return output in \'np\' format if ONNX is not available. Make sure to have ONNX installed or set \'output_type\' to \'mel\'.'	)
				elif output_type == "numpy" and self.melgan is None:
							raise ValueError(
							    'Cannot return output in \'np\' format if melgan component is not defined. Make sure to define `self.melgan` or set \'output_type\' to \'mel\'.'	)

				if output_type == "numpy":
							_snake_case		=			self.melgan(input_features=full_pred_mel.astype(np.floataa	)	)
				else:
							_snake_case		=			full_pred_mel

				if not return_dict:
							return (output,)

				return AudioPipelineOutput(audios=lowercase	)

import os
from typing import Any, Callable, Dict, List, Optional, Tuple, Union

import torch
from torch import nn

from ...models.controlnet import ControlNetModel, ControlNetOutput
from ...models.modeling_utils import ModelMixin
from ...utils import logging


_lowerCamelCase       :				List[str]       =			logging.get_logger(__name__)




class    SCREAMING_SNAKE_CASE__     (	UpperCAmelCase					):





	'''simple docstring'''



	def __init__(  self	:    Dict			,  lowercase	:    Union[List[ControlNetModel], Tuple[ControlNetModel]]	):


				'''simple docstring'''


				super().__init__()
				_snake_case		=			nn.ModuleList(lowercase	)



	def  A							(  self	:    Optional[int]			,  lowercase	:    torch.FloatTensor			,  lowercase	:    Union[torch.Tensor, float, int]			,  lowercase	:    torch.Tensor			,  lowercase	:    List[torch.tensor]			,  lowercase	:    List[float]			,  lowercase	:    Optional[torch.Tensor] = None			,  lowercase	:    Optional[torch.Tensor] = None			,  lowercase	:    Optional[torch.Tensor] = None			,  lowercase	:    Optional[Dict[str, Any]] = None			,  lowercase	:    bool = False			,  lowercase	:    bool = True			,  ):


				'''simple docstring'''


				for i, (image, scale, controlnet) in enumerate(zip(lowercase			,  lowercase			,  self.nets	)	):
							_snake_case	,					_snake_case		=			controlnet(
							    lowercase			,  lowercase			,  lowercase			,  lowercase			,  lowercase			,  lowercase			,  lowercase			,  lowercase			,  lowercase			,  lowercase			,  lowercase			,  )

							# merge samples
							if i == 0:
										_snake_case	,					_snake_case		=			down_samples, mid_sample
							else:
										_snake_case		=			[
										    samples_prev + samples_curr
										    for samples_prev, samples_curr in zip(lowercase			,  lowercase	)
										]
										mid_block_res_sample += mid_sample

				return down_block_res_samples, mid_block_res_sample



	def  A							(  self	:    Dict			,  lowercase	:    Union[str, os.PathLike]			,  lowercase	:    bool = True			,  lowercase	:    Callable = None			,  lowercase	:    bool = False			,  lowercase	:    Optional[str] = None			,  ):


				'''simple docstring'''


				_snake_case		=			0
				_snake_case		=			save_directory
				for controlnet in self.nets:
							controlnet.save_pretrained(
							    lowercase			,  is_main_process=lowercase			,  save_function=lowercase			,  safe_serialization=lowercase			,  variant=lowercase			,  )

							idx += 1
							_snake_case		=			model_path_to_save + f'''_{idx}'''





	@classmethod
	def  A							(  cls	:    Any			,  lowercase	:    Optional[Union[str, os.PathLike]]			,  **lowercase	:    List[str]	):


				'''simple docstring'''


				_snake_case		=			0
				_snake_case		=			[]

				# load controlnet and append to list until no controlnet directory exists anymore
				# first controlnet has to be saved under `./mydirectory/controlnet` to be compliant with `DiffusionPipeline.from_prertained`
				# second, third, ... controlnets have to be saved under `./mydirectory/controlnet_1`, `./mydirectory/controlnet_2`, ...
				_snake_case		=			pretrained_model_path
				while os.path.isdir(lowercase	):
							_snake_case		=			ControlNetModel.from_pretrained(lowercase			,  **lowercase	)
							controlnets.append(lowercase	)

							idx += 1
							_snake_case		=			pretrained_model_path + f'''_{idx}'''

				logger.info(f'''{len(lowercase	)} controlnets loaded from {pretrained_model_path}.'''	)

				if len(lowercase	) == 0:
							raise ValueError(
							    f'''No ControlNets found under {os.path.dirname(lowercase	)}. Expected at least {pretrained_model_path + '_0'}.'''	)

				return cls(lowercase	)