'''simple docstring'''




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

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, apply_forward_hook
from .attention_processor import AttentionProcessor, AttnProcessor
from .modeling_utils import ModelMixin
from .vae import Decoder, DecoderOutput, DiagonalGaussianDistribution, Encoder






@dataclass
class 				a__(					lowerCAmelCase__	):


			'''simple docstring'''






			UpperCAmelCase_			:     "DiagonalGaussianDistribution"







class 				a__(					lowerCAmelCase__ ,						lowerCAmelCase__	):


			'''simple docstring'''






			UpperCAmelCase_			:     str      =       True



			@register_to_config
			def __init__(   self       ,    __lowerCAmelCase = 3       ,    __lowerCAmelCase = 3       ,    __lowerCAmelCase = ("DownEncoderBlock2D",)       ,    __lowerCAmelCase = ("UpDecoderBlock2D",)       ,    __lowerCAmelCase = (64,)       ,    __lowerCAmelCase = 1       ,    __lowerCAmelCase = "silu"       ,    __lowerCAmelCase = 4       ,    __lowerCAmelCase = 32       ,    __lowerCAmelCase = 32       ,    __lowerCAmelCase = 0.18215       ,    ):




					"""simple docstring"""


					super().__init__()

					# pass init params to Encoder
					lowerCAmelCase   					=  Encoder(
					    in_channels=__lowerCAmelCase       ,    out_channels=__lowerCAmelCase       ,    down_block_types=__lowerCAmelCase       ,    block_out_channels=__lowerCAmelCase       ,    layers_per_block=__lowerCAmelCase       ,    act_fn=__lowerCAmelCase       ,    norm_num_groups=__lowerCAmelCase       ,    double_z=__lowerCAmelCase       ,    )

					# pass init params to Decoder
					lowerCAmelCase   					=  Decoder(
					    in_channels=__lowerCAmelCase       ,    out_channels=__lowerCAmelCase       ,    up_block_types=__lowerCAmelCase       ,    block_out_channels=__lowerCAmelCase       ,    layers_per_block=__lowerCAmelCase       ,    norm_num_groups=__lowerCAmelCase       ,    act_fn=__lowerCAmelCase       ,    )

					lowerCAmelCase   					=  nn.Convad(2 * latent_channels       ,    2 * latent_channels       ,    1)
					lowerCAmelCase   					=  nn.Convad(__lowerCAmelCase       ,    __lowerCAmelCase       ,    1)

					lowerCAmelCase   					=  False
					lowerCAmelCase   					=  False

					# only relevant if vae tiling is enabled
					lowerCAmelCase   					=  self.config.sample_size
					lowerCAmelCase   					=  (
					    self.config.sample_size[0]
					    if isinstance(self.config.sample_size       ,    (list, tuple))
					    else self.config.sample_size
					)
					lowerCAmelCase   					=  int(sample_size / (2 ** (len(self.config.block_out_channels) - 1)))
					lowerCAmelCase   					=  0.25



			def 							a_       (   self       ,    __lowerCAmelCase       ,    __lowerCAmelCase=False):




					"""simple docstring"""


					if isinstance(__lowerCAmelCase       ,    (Encoder, Decoder)):
							lowerCAmelCase   					=  value



			def 							a_       (   self       ,    __lowerCAmelCase = True):




					"""simple docstring"""


					lowerCAmelCase   					=  use_tiling



			def 							a_       (   self):




					"""simple docstring"""


					self.enable_tiling(__lowerCAmelCase)



			def 							a_       (   self):




					"""simple docstring"""


					lowerCAmelCase   					=  True



			def 							a_       (   self):




					"""simple docstring"""


					lowerCAmelCase   					=  False



			@property
			# Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
			def 							a_       (   self):




					"""simple docstring"""


					lowerCAmelCase   					=  {}

					def fn_recursive_add_processors(__lowerCAmelCase       ,    __lowerCAmelCase       ,    __lowerCAmelCase):
							if hasattr(__lowerCAmelCase       ,    """set_processor"""):
									lowerCAmelCase   					=  module.processor

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

							return processors

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

					return processors



			def 							a_       (   self       ,    __lowerCAmelCase):




					"""simple docstring"""


					lowerCAmelCase   					=  len(self.attn_processors.keys())

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

					def fn_recursive_attn_processor(__lowerCAmelCase       ,    __lowerCAmelCase       ,    __lowerCAmelCase):
							if hasattr(__lowerCAmelCase       ,    """set_processor"""):
									if not isinstance(__lowerCAmelCase       ,    __lowerCAmelCase):
											module.set_processor(__lowerCAmelCase)
									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}"       ,    __lowerCAmelCase       ,    __lowerCAmelCase)

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



			def 							a_       (   self):




					"""simple docstring"""


					self.set_attn_processor(AttnProcessor())



			@apply_forward_hook
			def 							a_       (   self       ,    __lowerCAmelCase       ,    __lowerCAmelCase = True):




					"""simple docstring"""


					if self.use_tiling and (x.shape[-1] > self.tile_sample_min_size or x.shape[-2] > self.tile_sample_min_size):
							return self.tiled_encode(__lowerCAmelCase       ,    return_dict=__lowerCAmelCase)

					if self.use_slicing and x.shape[0] > 1:
							lowerCAmelCase   					=  [self.encoder(__lowerCAmelCase) for x_slice in x.split(1)]
							lowerCAmelCase   					=  torch.cat(__lowerCAmelCase)
					else:
							lowerCAmelCase   					=  self.encoder(__lowerCAmelCase)

					lowerCAmelCase   					=  self.quant_conv(__lowerCAmelCase)
					lowerCAmelCase   					=  DiagonalGaussianDistribution(__lowerCAmelCase)

					if not return_dict:
							return (posterior,)

					return AutoencoderKLOutput(latent_dist=__lowerCAmelCase)



			def 							a_       (   self       ,    __lowerCAmelCase       ,    __lowerCAmelCase = True):




					"""simple docstring"""


					if self.use_tiling and (z.shape[-1] > self.tile_latent_min_size or z.shape[-2] > self.tile_latent_min_size):
							return self.tiled_decode(__lowerCAmelCase       ,    return_dict=__lowerCAmelCase)

					lowerCAmelCase   					=  self.post_quant_conv(__lowerCAmelCase)
					lowerCAmelCase   					=  self.decoder(__lowerCAmelCase)

					if not return_dict:
							return (dec,)

					return DecoderOutput(sample=__lowerCAmelCase)



			@apply_forward_hook
			def 							a_       (   self       ,    __lowerCAmelCase       ,    __lowerCAmelCase = True):




					"""simple docstring"""


					if self.use_slicing and z.shape[0] > 1:
							lowerCAmelCase   					=  [self._decode(__lowerCAmelCase).sample for z_slice in z.split(1)]
							lowerCAmelCase   					=  torch.cat(__lowerCAmelCase)
					else:
							lowerCAmelCase   					=  self._decode(__lowerCAmelCase).sample

					if not return_dict:
							return (decoded,)

					return DecoderOutput(sample=__lowerCAmelCase)



			def 							a_       (   self       ,    __lowerCAmelCase       ,    __lowerCAmelCase       ,    __lowerCAmelCase):




					"""simple docstring"""


					lowerCAmelCase   					=  min(a.shape[2]       ,    b.shape[2]       ,    __lowerCAmelCase)
					for y in range(__lowerCAmelCase):
							lowerCAmelCase   					=  a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
					return b



			def 							a_       (   self       ,    __lowerCAmelCase       ,    __lowerCAmelCase       ,    __lowerCAmelCase):




					"""simple docstring"""


					lowerCAmelCase   					=  min(a.shape[3]       ,    b.shape[3]       ,    __lowerCAmelCase)
					for x in range(__lowerCAmelCase):
							lowerCAmelCase   					=  a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
					return b



			def 							a_       (   self       ,    __lowerCAmelCase       ,    __lowerCAmelCase = True):




					"""simple docstring"""


					lowerCAmelCase   					=  int(self.tile_sample_min_size * (1 - self.tile_overlap_factor))
					lowerCAmelCase   					=  int(self.tile_latent_min_size * self.tile_overlap_factor)
					lowerCAmelCase   					=  self.tile_latent_min_size - blend_extent

					# Split the image into 512x512 tiles and encode them separately.
					lowerCAmelCase   					=  []
					for i in range(0       ,    x.shape[2]       ,    __lowerCAmelCase):
							lowerCAmelCase   					=  []
							for j in range(0       ,    x.shape[3]       ,    __lowerCAmelCase):
									lowerCAmelCase   					=  x[:, :, i : i + self.tile_sample_min_size, j : j + self.tile_sample_min_size]
									lowerCAmelCase   					=  self.encoder(__lowerCAmelCase)
									lowerCAmelCase   					=  self.quant_conv(__lowerCAmelCase)
									row.append(__lowerCAmelCase)
							rows.append(__lowerCAmelCase)
					lowerCAmelCase   					=  []
					for i, row in enumerate(__lowerCAmelCase):
							lowerCAmelCase   					=  []
							for j, tile in enumerate(__lowerCAmelCase):
									# blend the above tile and the left tile
									# to the current tile and add the current tile to the result row
									if i > 0:
											lowerCAmelCase   					=  self.blend_v(rows[i - 1][j]       ,    __lowerCAmelCase       ,    __lowerCAmelCase)
									if j > 0:
											lowerCAmelCase   					=  self.blend_h(row[j - 1]       ,    __lowerCAmelCase       ,    __lowerCAmelCase)
									result_row.append(tile[:, :, :row_limit, :row_limit])
							result_rows.append(torch.cat(__lowerCAmelCase       ,    dim=3))

					lowerCAmelCase   					=  torch.cat(__lowerCAmelCase       ,    dim=2)
					lowerCAmelCase   					=  DiagonalGaussianDistribution(__lowerCAmelCase)

					if not return_dict:
							return (posterior,)

					return AutoencoderKLOutput(latent_dist=__lowerCAmelCase)



			def 							a_       (   self       ,    __lowerCAmelCase       ,    __lowerCAmelCase = True):




					"""simple docstring"""


					lowerCAmelCase   					=  int(self.tile_latent_min_size * (1 - self.tile_overlap_factor))
					lowerCAmelCase   					=  int(self.tile_sample_min_size * self.tile_overlap_factor)
					lowerCAmelCase   					=  self.tile_sample_min_size - blend_extent

					# Split z into overlapping 64x64 tiles and decode them separately.
					# The tiles have an overlap to avoid seams between tiles.
					lowerCAmelCase   					=  []
					for i in range(0       ,    z.shape[2]       ,    __lowerCAmelCase):
							lowerCAmelCase   					=  []
							for j in range(0       ,    z.shape[3]       ,    __lowerCAmelCase):
									lowerCAmelCase   					=  z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
									lowerCAmelCase   					=  self.post_quant_conv(__lowerCAmelCase)
									lowerCAmelCase   					=  self.decoder(__lowerCAmelCase)
									row.append(__lowerCAmelCase)
							rows.append(__lowerCAmelCase)
					lowerCAmelCase   					=  []
					for i, row in enumerate(__lowerCAmelCase):
							lowerCAmelCase   					=  []
							for j, tile in enumerate(__lowerCAmelCase):
									# blend the above tile and the left tile
									# to the current tile and add the current tile to the result row
									if i > 0:
											lowerCAmelCase   					=  self.blend_v(rows[i - 1][j]       ,    __lowerCAmelCase       ,    __lowerCAmelCase)
									if j > 0:
											lowerCAmelCase   					=  self.blend_h(row[j - 1]       ,    __lowerCAmelCase       ,    __lowerCAmelCase)
									result_row.append(tile[:, :, :row_limit, :row_limit])
							result_rows.append(torch.cat(__lowerCAmelCase       ,    dim=3))

					lowerCAmelCase   					=  torch.cat(__lowerCAmelCase       ,    dim=2)
					if not return_dict:
							return (dec,)

					return DecoderOutput(sample=__lowerCAmelCase)







			def 							a_       (   self       ,    __lowerCAmelCase       ,    __lowerCAmelCase = False       ,    __lowerCAmelCase = True       ,    __lowerCAmelCase = None       ,    ):




					"""simple docstring"""


					lowerCAmelCase   					=  sample
					lowerCAmelCase   					=  self.encode(__lowerCAmelCase).latent_dist
					if sample_posterior:
							lowerCAmelCase   					=  posterior.sample(generator=__lowerCAmelCase)
					else:
							lowerCAmelCase   					=  posterior.mode()
					lowerCAmelCase   					=  self.decode(__lowerCAmelCase).sample

					if not return_dict:
							return (dec,)

					return DecoderOutput(sample=__lowerCAmelCase)





'''simple docstring'''




import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np

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


__lowercase          =  logging.get_logger(__name__)






class 				a__(					lowerCAmelCase__	):


			'''simple docstring'''






			UpperCAmelCase_			:     Union[str, Any]      =       ['''input_features''']



			def __init__(   self       ,    __lowerCAmelCase=80       ,    __lowerCAmelCase=16000       ,    __lowerCAmelCase=160       ,    __lowerCAmelCase=30       ,    __lowerCAmelCase=400       ,    __lowerCAmelCase=0.0       ,    __lowerCAmelCase=False       ,    **__lowerCAmelCase       ,    ):




					"""simple docstring"""


					super().__init__(
					    feature_size=__lowerCAmelCase       ,    sampling_rate=__lowerCAmelCase       ,    padding_value=__lowerCAmelCase       ,    return_attention_mask=__lowerCAmelCase       ,    **__lowerCAmelCase       ,    )
					lowerCAmelCase   					=  n_fft
					lowerCAmelCase   					=  hop_length
					lowerCAmelCase   					=  chunk_length
					lowerCAmelCase   					=  chunk_length * sampling_rate
					lowerCAmelCase   					=  self.n_samples // hop_length
					lowerCAmelCase   					=  sampling_rate
					lowerCAmelCase   					=  mel_filter_bank(
					    num_frequency_bins=1 + n_fft // 2       ,    num_mel_filters=__lowerCAmelCase       ,    min_frequency=0.0       ,    max_frequency=8000.0       ,    sampling_rate=__lowerCAmelCase       ,    norm="""slaney"""       ,    mel_scale="""slaney"""       ,    )



			def 							a_       (   self       ,    __lowerCAmelCase):




					"""simple docstring"""


					lowerCAmelCase   					=  spectrogram(
					    __lowerCAmelCase       ,    window_function(self.n_fft       ,    """hann""")       ,    frame_length=self.n_fft       ,    hop_length=self.hop_length       ,    power=2.0       ,    mel_filters=self.mel_filters       ,    log_mel="""log10"""       ,    )
					lowerCAmelCase   					=  log_spec[:, :-1]
					lowerCAmelCase   					=  np.maximum(__lowerCAmelCase       ,    log_spec.max() - 8.0)
					lowerCAmelCase   					=  (log_spec + 4.0) / 4.0
					return log_spec



			@staticmethod
			# Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
			def 							a_       (   __lowerCAmelCase       ,    __lowerCAmelCase       ,    __lowerCAmelCase = 0.0):




					"""simple docstring"""


					if attention_mask is not None:
							lowerCAmelCase   					=  np.array(__lowerCAmelCase       ,    np.intaa)
							lowerCAmelCase   					=  []

							for vector, length in zip(__lowerCAmelCase       ,    attention_mask.sum(-1)):
									lowerCAmelCase   					=  (vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7)
									if length < normed_slice.shape[0]:
											lowerCAmelCase   					=  padding_value

									normed_input_values.append(__lowerCAmelCase)
					else:
							lowerCAmelCase   					=  [(x - x.mean()) / np.sqrt(x.var() + 1E-7) for x in input_values]

					return normed_input_values



			def __call__(   self       ,    __lowerCAmelCase       ,    __lowerCAmelCase = True       ,    __lowerCAmelCase = None       ,    __lowerCAmelCase = None       ,    __lowerCAmelCase = None       ,    __lowerCAmelCase = "max_length"       ,    __lowerCAmelCase = None       ,    __lowerCAmelCase = None       ,    __lowerCAmelCase = None       ,    **__lowerCAmelCase       ,    ):




					"""simple docstring"""


					if sampling_rate is not None:
							if sampling_rate != self.sampling_rate:
									raise ValueError(
									    f"The model corresponding to this feature extractor: {self.__class__.__name__} was trained using a"
									    f" sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input"
									    f" was sampled 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.""")

					lowerCAmelCase   					=  isinstance(__lowerCAmelCase       ,    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}")
					lowerCAmelCase   					=  is_batched_numpy or (
					    isinstance(__lowerCAmelCase       ,    (list, tuple)) and (isinstance(raw_speech[0]       ,    (np.ndarray, tuple, list)))
					)

					if is_batched:
							lowerCAmelCase   					=  [np.asarray([speech]       ,    dtype=np.floataa).T for speech in raw_speech]
					elif not is_batched and not isinstance(__lowerCAmelCase       ,    np.ndarray):
							lowerCAmelCase   					=  np.asarray(__lowerCAmelCase       ,    dtype=np.floataa)
					elif isinstance(__lowerCAmelCase       ,    np.ndarray) and raw_speech.dtype is np.dtype(np.floataa):
							lowerCAmelCase   					=  raw_speech.astype(np.floataa)

					# always return batch
					if not is_batched:
							lowerCAmelCase   					=  [np.asarray([raw_speech]).T]

					lowerCAmelCase   					=  BatchFeature({"""input_features""": raw_speech})

					# convert into correct format for padding

					lowerCAmelCase   					=  self.pad(
					    __lowerCAmelCase       ,    padding=__lowerCAmelCase       ,    max_length=max_length if max_length else self.n_samples       ,    truncation=__lowerCAmelCase       ,    pad_to_multiple_of=__lowerCAmelCase       ,    return_attention_mask=return_attention_mask or do_normalize       ,    )

					# zero-mean and unit-variance normalization
					if do_normalize:
							lowerCAmelCase   					=  self.zero_mean_unit_var_norm(
							    padded_inputs["""input_features"""]       ,    attention_mask=padded_inputs["""attention_mask"""]       ,    padding_value=self.padding_value       ,    )
							lowerCAmelCase   					=  np.stack(padded_inputs["""input_features"""]       ,    axis=0)

					# make sure list is in array format
					lowerCAmelCase   					=  padded_inputs.get("""input_features""").transpose(2       ,    0       ,    1)

					lowerCAmelCase   					=  [self._np_extract_fbank_features(__lowerCAmelCase) for waveform in input_features[0]]

					if isinstance(input_features[0]       ,    __lowerCAmelCase):
							lowerCAmelCase   					=  [np.asarray(__lowerCAmelCase       ,    dtype=np.floataa) for feature in input_features]
					else:
							lowerCAmelCase   					=  input_features

					if return_attention_mask:
							# rescale from sample (48000) to feature (3000)
							lowerCAmelCase   					=  padded_inputs["""attention_mask"""][:, :: self.hop_length]

					if return_tensors is not None:
							lowerCAmelCase   					=  padded_inputs.convert_to_tensors(__lowerCAmelCase)

					return padded_inputs







			def 							a_       (   self):




					"""simple docstring"""


					lowerCAmelCase   					=  copy.deepcopy(self.__dict__)
					lowerCAmelCase   					=  self.__class__.__name__
					if "mel_filters" in output:
							del output["mel_filters"]
					return output




