





'''simple docstring'''




from math import ceil
from typing import List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import BatchFeature, SequenceFeatureExtractor
from ...utils import TensorType, logging


a__					:			str          =     logging.get_logger(__name__)






class 				__snake_case     (					__magic_name__							):
	__lowerCAmelCase       						=  ['''audio_values''', '''audio_mask''']





	def __init__(		self      ,       UpperCamelCase_=2048      ,       UpperCamelCase_=1      ,       UpperCamelCase_=[16, 16]      ,       UpperCamelCase_=128      ,       UpperCamelCase_=4_4100      ,       UpperCamelCase_=86      ,       UpperCamelCase_=2048      ,       UpperCamelCase_=0.0      ,       **UpperCamelCase_      ,       )      ->		Any:
		super().__init__(
		    feature_size=UpperCamelCase_      ,       sampling_rate=UpperCamelCase_      ,       padding_value=UpperCamelCase_      ,       **UpperCamelCase_      ,       )

		snake_case__	       =   spectrogram_length
		snake_case__	       =   num_channels
		snake_case__	       =   patch_size
		snake_case__	       =   feature_size // self.patch_size[1]
		snake_case__	       =   n_fft
		snake_case__	       =   sampling_rate // hop_length_to_sampling_rate
		snake_case__	       =   sampling_rate
		snake_case__	       =   padding_value
		snake_case__	       =   mel_filter_bank(
		    num_frequency_bins=1 + n_fft // 2      ,       num_mel_filters=UpperCamelCase_      ,       min_frequency=0.0      ,       max_frequency=2_2_0_5_0.0      ,       sampling_rate=UpperCamelCase_      ,       norm='slaney'      ,       mel_scale='slaney'      ,       ).T





	def   _snake_case   (		self      ,       UpperCamelCase_		)      ->		np.ndarray:
		snake_case__	       =   spectrogram(
		    UpperCamelCase_      ,       window_function(self.n_fft      ,       'hann'		)      ,       frame_length=self.n_fft      ,       hop_length=self.hop_length      ,       power=2.0      ,       mel_filters=self.mel_filters.T      ,       log_mel='dB'      ,       db_range=8_0.0      ,       )
		snake_case__	       =   log_spec[:, :-1]
		snake_case__	       =   log_spec - 2_0.0
		snake_case__	       =   np.clip(log_spec / 4_0.0      ,       -2.0      ,       0.0		) + 1.0
		return log_spec





	def __call__(		self      ,       UpperCamelCase_      ,       UpperCamelCase_ = None      ,       UpperCamelCase_ = True      ,       UpperCamelCase_ = None      ,       UpperCamelCase_ = False      ,       UpperCamelCase_ = False      ,       **UpperCamelCase_      ,       )      ->		BatchFeature:

		if sampling_rate is not None:
			if sampling_rate != self.sampling_rate:
				raise ValueError(
				    'This feature extractor is set to support sampling rate'
				    F''' of {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled'''
				    F''' with {self.sampling_rate} and not {sampling_rate}.'''		)
		else:
			logger.warning(
			    'It is strongly recommended to pass the `sampling_rate` argument to this function. '
			    'Failing to do so can result in silent errors that might be hard to debug.'		)

		snake_case__	       =   isinstance(UpperCamelCase_      ,       np.ndarray		) and len(raw_speech.shape		) > 1
		if is_batched_numpy and len(raw_speech.shape		) > 2:
			raise ValueError(F'''Only mono-channel audio is supported for input to {self}'''		)
		snake_case__	       =   is_batched_numpy or (
		    isinstance(UpperCamelCase_      ,       (list, tuple)		) and (isinstance(raw_speech[0]      ,       (np.ndarray, tuple, list)		))
		)
		if is_batched:
			snake_case__	       =   [np.asarray([speech]      ,       dtype=np.floataa		).T for speech in raw_speech]
		elif not is_batched and not isinstance(UpperCamelCase_      ,       np.ndarray		):
			snake_case__	       =   np.asarray(UpperCamelCase_      ,       dtype=np.floataa		)
		elif isinstance(UpperCamelCase_      ,       np.ndarray		) and raw_speech.dtype is np.dtype(np.floataa		):
			snake_case__	       =   raw_speech.astype(np.floataa		)
		# always return batch
		if not is_batched:
			snake_case__	       =   [np.asarray([raw_speech]		).T]

		# Convert audio signals to log mel spectrograms, truncate by time axis
		snake_case__	       =   [
		    self._np_extract_fbank_features(waveform.squeeze()		).T[: self.spectrogram_length] for waveform in raw_speech
		]
		if isinstance(audio_features[0]      ,       UpperCamelCase_		):
			snake_case__	       =   [np.asarray(UpperCamelCase_      ,       dtype=np.floataa		) for feature in audio_features]

		# Create audio attention mask
		snake_case__	       =   max(
		    [ceil(feature.shape[0] / self.patch_size[0]		) * self.freq_len for feature in audio_features]		)  # The maximum number of audio patches in a batch
		if return_attention_mask:
			snake_case__	       =   [
			    (ceil(feature.shape[0] / self.patch_size[0]		) * self.freq_len) * [1]
			    + (max_patch_len - ceil(feature.shape[0] / self.patch_size[0]		) * self.freq_len) * [0]
			    for feature in audio_features
			]
			snake_case__	       =   np.array(UpperCamelCase_		).astype(np.floataa		)

		# convert into correct format for padding
		snake_case__	       =   max_patch_len // self.freq_len * self.patch_size[0]  # The maximum audio size in a batch
		snake_case__	       =   np.ones([len(UpperCamelCase_		), 1, max_time_len, self.feature_size]		).astype(np.floataa		)
		snake_case__	       =   padded_audio_features * self.padding_value
		for i in range(len(UpperCamelCase_		)		):
			snake_case__	       =   audio_features[i]
			snake_case__	       =   feature

		# return as BatchFeature
		if return_attention_mask:
			snake_case__	       =   {'audio_values': padded_audio_features, 'audio_mask': audio_mask}
		else:
			snake_case__	       =   {'audio_values': padded_audio_features}

		snake_case__	       =   BatchFeature(data=UpperCamelCase_      ,       tensor_type=UpperCamelCase_		)
		return encoded_inputs










'''simple docstring'''




from typing import List, Optional, Tuple, Union

import torch

from ...schedulers import DDIMScheduler
from ...utils import randn_tensor
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput






class 				__snake_case     (					__magic_name__							):





	def __init__(		self      ,       UpperCamelCase_      ,       UpperCamelCase_		)      ->		int:
		super().__init__()

		# make sure scheduler can always be converted to DDIM
		snake_case__	       =   DDIMScheduler.from_config(scheduler.config		)

		self.register_modules(unet=UpperCamelCase_      ,       scheduler=UpperCamelCase_		)





	@torch.no_grad()
	def __call__(		self      ,       UpperCamelCase_ = 1      ,       UpperCamelCase_ = None      ,       UpperCamelCase_ = 0.0      ,       UpperCamelCase_ = 50      ,       UpperCamelCase_ = None      ,       UpperCamelCase_ = "pil"      ,       UpperCamelCase_ = True      ,       )      ->		Union[ImagePipelineOutput, Tuple]:

		# Sample gaussian noise to begin loop
		if isinstance(self.unet.config.sample_size      ,       UpperCamelCase_		):
			snake_case__	       =   (
			    batch_size,
			    self.unet.config.in_channels,
			    self.unet.config.sample_size,
			    self.unet.config.sample_size,
			)
		else:
			snake_case__	       =   (batch_size, self.unet.config.in_channels, *self.unet.config.sample_size)

		if isinstance(UpperCamelCase_      ,       UpperCamelCase_		) and len(UpperCamelCase_		) != batch_size:
			raise ValueError(
			    F'''You have passed a list of generators of length {len(UpperCamelCase_		)}, but requested an effective batch'''
			    F''' size of {batch_size}. Make sure the batch size matches the length of the generators.'''		)

		snake_case__	       =   randn_tensor(UpperCamelCase_      ,       generator=UpperCamelCase_      ,       device=self.device      ,       dtype=self.unet.dtype		)

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

		for t in self.progress_bar(self.scheduler.timesteps		):
			# 1. predict noise model_output
			snake_case__	       =   self.unet(UpperCamelCase_      ,       UpperCamelCase_		).sample

			# 2. predict previous mean of image x_t-1 and add variance depending on eta
			# eta corresponds to η in paper and should be between [0, 1]
			# do x_t -> x_t-1
			snake_case__	       =   self.scheduler.step(
			    UpperCamelCase_      ,       UpperCamelCase_      ,       UpperCamelCase_      ,       eta=UpperCamelCase_      ,       use_clipped_model_output=UpperCamelCase_      ,       generator=UpperCamelCase_		).prev_sample

		snake_case__	       =   (image / 2 + 0.5).clamp(0      ,       1		)
		snake_case__	       =   image.cpu().permute(0      ,       2      ,       3      ,       1		).numpy()
		if output_type == "pil":
			snake_case__	       =   self.numpy_to_pil(UpperCamelCase_		)

		if not return_dict:
			return (image,)

		return ImagePipelineOutput(images=UpperCamelCase_		)



