


'''simple docstring'''






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 	a       (       _SCREAMING_SNAKE_CASE    ):
 _lowerCAmelCase	   =		42





class 	a       (       nn.Module    ):
 def __init__(	self ,			__magic_name__=3 ,			__magic_name__=3 ,			__magic_name__=("DownEncoderBlock2D",) ,			__magic_name__=(64,) ,			__magic_name__=2 ,			__magic_name__=32 ,			__magic_name__="silu" ,			__magic_name__=True ,			)   ->							str:
      super().__init__()
      _a      =							layers_per_block

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

      _a      =							None
      _a      =							nn.ModuleList([]  )

      # down
      _a      =							block_out_channels[0]
      for i, down_block_type in enumerate(__magic_name__  ):
           _a      =							output_channel
           _a      =							block_out_channels[i]
           _a      =							i == len(__magic_name__  ) - 1

           _a      =							get_down_block(
               __magic_name__ ,			num_layers=self.layers_per_block ,			in_channels=__magic_name__ ,			out_channels=__magic_name__ ,			add_downsample=not is_final_block ,			resnet_eps=1e-6 ,			downsample_padding=0 ,			resnet_act_fn=__magic_name__ ,			resnet_groups=__magic_name__ ,			attention_head_dim=__magic_name__ ,			temb_channels=__magic_name__ ,			)
           self.down_blocks.append(__magic_name__  )

      # mid
      _a      =							UNetMidBlockaD(
          in_channels=block_out_channels[-1] ,			resnet_eps=1e-6 ,			resnet_act_fn=__magic_name__ ,			output_scale_factor=1 ,			resnet_time_scale_shift='default' ,			attention_head_dim=block_out_channels[-1] ,			resnet_groups=__magic_name__ ,			temb_channels=__magic_name__ ,			)

      # out
      _a      =							nn.GroupNorm(num_channels=block_out_channels[-1] ,			num_groups=__magic_name__ ,			eps=1e-6  )
      _a      =							nn.SiLU()

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

      _a      =							False




 def 							__UpperCAmelCase						(	self ,			__magic_name__  )   ->							Optional[Any]:
      _a      =							x
      _a      =							self.conv_in(__magic_name__  )

      if self.training and self.gradient_checkpointing:

           def create_custom_forward(__magic_name__  ):
                def custom_forward(*__magic_name__  ):
                     return module(*__magic_name__  )

                return custom_forward

           # down
           if is_torch_version('>=' ,			'1.11.0'  ):
                for down_block in self.down_blocks:
                     _a      =							torch.utils.checkpoint.checkpoint(
                         create_custom_forward(__magic_name__  ) ,			__magic_name__ ,			use_reentrant=__magic_name__  )
                # middle
                _a      =							torch.utils.checkpoint.checkpoint(
                    create_custom_forward(self.mid_block  ) ,			__magic_name__ ,			use_reentrant=__magic_name__  )
           else:
                for down_block in self.down_blocks:
                     _a      =							torch.utils.checkpoint.checkpoint(create_custom_forward(__magic_name__  ) ,			__magic_name__  )
                # middle
                _a      =							torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block  ) ,			__magic_name__  )

      else:
           # down
           for down_block in self.down_blocks:
                _a      =							down_block(__magic_name__  )

           # middle
           _a      =							self.mid_block(__magic_name__  )

      # post-process
      _a      =							self.conv_norm_out(__magic_name__  )
      _a      =							self.conv_act(__magic_name__  )
      _a      =							self.conv_out(__magic_name__  )

      return sample





class 	a       (       nn.Module    ):
 def __init__(	self ,			__magic_name__=3 ,			__magic_name__=3 ,			__magic_name__=("UpDecoderBlock2D",) ,			__magic_name__=(64,) ,			__magic_name__=2 ,			__magic_name__=32 ,			__magic_name__="silu" ,			__magic_name__="group" ,			)   ->							List[Any]:
      super().__init__()
      _a      =							layers_per_block

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

      _a      =							None
      _a      =							nn.ModuleList([]  )

      _a      =							in_channels if norm_type == 'spatial' else None

      # mid
      _a      =							UNetMidBlockaD(
          in_channels=block_out_channels[-1] ,			resnet_eps=1e-6 ,			resnet_act_fn=__magic_name__ ,			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=__magic_name__ ,			temb_channels=__magic_name__ ,			)

      # up
      _a      =							list(reversed(__magic_name__  )  )
      _a      =							reversed_block_out_channels[0]
      for i, up_block_type in enumerate(__magic_name__  ):
           _a      =							output_channel
           _a      =							reversed_block_out_channels[i]

           _a      =							i == len(__magic_name__  ) - 1

           _a      =							get_up_block(
               __magic_name__ ,			num_layers=self.layers_per_block + 1 ,			in_channels=__magic_name__ ,			out_channels=__magic_name__ ,			prev_output_channel=__magic_name__ ,			add_upsample=not is_final_block ,			resnet_eps=1e-6 ,			resnet_act_fn=__magic_name__ ,			resnet_groups=__magic_name__ ,			attention_head_dim=__magic_name__ ,			temb_channels=__magic_name__ ,			resnet_time_scale_shift=__magic_name__ ,			)
           self.up_blocks.append(__magic_name__  )
           _a      =							output_channel

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

      _a      =							False




 def 							__UpperCAmelCase						(	self ,			__magic_name__ ,			__magic_name__=None  )   ->							str:
      _a      =							z
      _a      =							self.conv_in(__magic_name__  )

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

           def create_custom_forward(__magic_name__  ):
                def custom_forward(*__magic_name__  ):
                     return module(*__magic_name__  )

                return custom_forward

           if is_torch_version('>=' ,			'1.11.0'  ):
                # middle
                _a      =							torch.utils.checkpoint.checkpoint(
                    create_custom_forward(self.mid_block  ) ,			__magic_name__ ,			__magic_name__ ,			use_reentrant=__magic_name__  )
                _a      =							sample.to(__magic_name__  )

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

                # up
                for up_block in self.up_blocks:
                     _a      =							torch.utils.checkpoint.checkpoint(create_custom_forward(__magic_name__  ) ,			__magic_name__ ,			__magic_name__  )
      else:
           # middle
           _a      =							self.mid_block(__magic_name__ ,			__magic_name__  )
           _a      =							sample.to(__magic_name__  )

           # up
           for up_block in self.up_blocks:
                _a      =							up_block(__magic_name__ ,			__magic_name__  )

        # post-process
      if latent_embeds is None:
           _a      =							self.conv_norm_out(__magic_name__  )
      else:
           _a      =							self.conv_norm_out(__magic_name__ ,			__magic_name__  )
      _a      =							self.conv_act(__magic_name__  )
      _a      =							self.conv_out(__magic_name__  )

      return sample





class 	a       (       nn.Module    ):
 def __init__(	self ,			__magic_name__ ,			__magic_name__ ,			__magic_name__ ,			__magic_name__=None ,			__magic_name__="random" ,			__magic_name__=False ,			__magic_name__=True  )   ->							Union[str, Any]:
      super().__init__()
      _a      =							n_e
      _a      =							vq_embed_dim
      _a      =							beta
      _a      =							legacy

      _a      =							nn.Embedding(self.n_e ,			self.vq_embed_dim  )
      self.embedding.weight.data.uniform_(-1.0 / self.n_e ,			1.0 / self.n_e  )

      _a      =							remap
      if self.remap is not None:
           self.register_buffer('used' ,			torch.tensor(np.load(self.remap  )  )  )
           _a      =							self.used.shape[0]
           _a      =							unknown_index  # "random" or "extra" or integer
           if self.unknown_index == "extra":
                _a      =							self.re_embed
                _a      =							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:
           _a      =							n_e

      _a      =							sane_index_shape
 def 							__UpperCAmelCase						(	self ,			__magic_name__  )   ->							Union[str, Any]:
      _a      =							inds.shape
      assert len(__magic_name__  ) > 1
      _a      =							inds.reshape(ishape[0] ,			-1  )
      _a      =							self.used.to(__magic_name__  )
      _a      =							(inds[:, :, None] == used[None, None, ...]).long()
      _a      =							match.argmax(-1  )
      _a      =							match.sum(2  ) < 1
      if self.unknown_index == "random":
           _a      =							torch.randint(0 ,			self.re_embed ,			size=new[unknown].shape  ).to(device=new.device  )
      else:
           _a      =							self.unknown_index
      return new.reshape(__magic_name__  )
 def 							__UpperCAmelCase						(	self ,			__magic_name__  )   ->							Dict:
      _a      =							inds.shape
      assert len(__magic_name__  ) > 1
      _a      =							inds.reshape(ishape[0] ,			-1  )
      _a      =							self.used.to(__magic_name__  )
      if self.re_embed > self.used.shape[0]:  # extra token
           _a      =							0  # simply set to zero
      _a      =							torch.gather(used[None, :][inds.shape[0] * [0], :] ,			1 ,			__magic_name__  )
      return back.reshape(__magic_name__  )
 def 							__UpperCAmelCase						(	self ,			__magic_name__  )   ->							Optional[Any]:
      # reshape z -> (batch, height, width, channel) and flatten
      _a      =							z.permute(0 ,			2 ,			3 ,			1  ).contiguous()
      _a      =							z.view(-1 ,			self.vq_embed_dim  )

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

      _a      =							self.embedding(__magic_name__  ).view(z.shape  )
      _a      =							None
      _a      =							None

      # compute loss for embedding
      if not self.legacy:
           _a      =							self.beta * torch.mean((z_q.detach() - z) ** 2  ) + torch.mean((z_q - z.detach()) ** 2  )
      else:
           _a      =							torch.mean((z_q.detach() - z) ** 2  ) + self.beta * torch.mean((z_q - z.detach()) ** 2  )

      # preserve gradients
      _a      =							z + (z_q - z).detach()

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

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

      if self.sane_index_shape:
           _a      =							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 ,			__magic_name__ ,			__magic_name__  )   ->							List[str]:
      # shape specifying (batch, height, width, channel)
      if self.remap is not None:
           _a      =							indices.reshape(shape[0] ,			-1  )  # add batch axis
           _a      =							self.unmap_to_all(__magic_name__  )
           _a      =							indices.reshape(-1  )  # flatten again

      # get quantized latent vectors
      _a      =							self.embedding(__magic_name__  )

      if shape is not None:
           _a      =							z_q.view(__magic_name__  )
           # reshape back to match original input shape
           _a      =							z_q.permute(0 ,			3 ,			1 ,			2  ).contiguous()

      return z_q





class 	a       (       _SCREAMING_SNAKE_CASE    ):
 def __init__(	self ,			__magic_name__ ,			__magic_name__=False  )   ->							Dict:
      _a      =							parameters
      _a ,				_a      =							torch.chunk(__magic_name__ ,			2 ,			dim=1  )
      _a      =							torch.clamp(self.logvar ,			-3_0.0 ,			2_0.0  )
      _a      =							deterministic
      _a      =							torch.exp(0.5 * self.logvar  )
      _a      =							torch.exp(self.logvar  )
      if self.deterministic:
           _a      =							_a      =							torch.zeros_like(
               self.mean ,			device=self.parameters.device ,			dtype=self.parameters.dtype  )
 def 							__UpperCAmelCase						(	self ,			__magic_name__ = None  )   ->							torch.FloatTensor:
      # make sure sample is on the same device as the parameters and has same dtype
      _a      =							randn_tensor(
          self.mean.shape ,			generator=__magic_name__ ,			device=self.parameters.device ,			dtype=self.parameters.dtype  )
      _a      =							self.mean + self.std * sample
      return x
 def 							__UpperCAmelCase						(	self ,			__magic_name__=None  )   ->							Dict:
      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 ,			__magic_name__ ,			__magic_name__=[1, 2, 3]  )   ->							int:
      if self.deterministic:
           return torch.Tensor([0.0]  )
      _a      =							np.log(2.0 * np.pi  )
      return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean ,			2  ) / self.var ,			dim=__magic_name__  )




 def 							__UpperCAmelCase						(	self  )   ->							Optional[int]:
      return self.mean





'''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


a_   :      Tuple				    =      logging.get_logger(__name__)




class 	a       (       _SCREAMING_SNAKE_CASE    ):
 _lowerCAmelCase	   =		["""input_features"""]
 def __init__(	self ,			__magic_name__=80 ,			__magic_name__=1_60_00 ,			__magic_name__=1_60 ,			__magic_name__=30 ,			__magic_name__=4_00 ,			__magic_name__=0.0 ,			__magic_name__=False ,			**__magic_name__ ,			)   ->							Optional[int]:
      super().__init__(
          feature_size=__magic_name__ ,			sampling_rate=__magic_name__ ,			padding_value=__magic_name__ ,			return_attention_mask=__magic_name__ ,			**__magic_name__ ,			)
      _a      =							n_fft
      _a      =							hop_length
      _a      =							chunk_length
      _a      =							chunk_length * sampling_rate
      _a      =							self.n_samples // hop_length
      _a      =							sampling_rate
      _a      =							mel_filter_bank(
          num_frequency_bins=1 + n_fft // 2 ,			num_mel_filters=__magic_name__ ,			min_frequency=0.0 ,			max_frequency=8_0_0_0.0 ,			sampling_rate=__magic_name__ ,			norm='slaney' ,			mel_scale='slaney' ,			)
 def 							__UpperCAmelCase						(	self ,			__magic_name__  )   ->							np.ndarray:
      _a      =							spectrogram(
          __magic_name__ ,			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' ,			)
      _a      =							log_spec[:, :-1]
      _a      =							np.maximum(__magic_name__ ,			log_spec.max() - 8.0  )
      _a      =							(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 							__UpperCAmelCase						(	__magic_name__ ,			__magic_name__ ,			__magic_name__ = 0.0  )   ->							List[np.ndarray]:
      if attention_mask is not None:
           _a      =							np.array(__magic_name__ ,			np.intaa  )
           _a      =							[]

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

                normed_input_values.append(__magic_name__  )
      else:
           _a      =							[(x - x.mean()) / np.sqrt(x.var() + 1e-7  ) for x in input_values]

      return normed_input_values
 def __call__(	self ,			__magic_name__ ,			__magic_name__ = True ,			__magic_name__ = None ,			__magic_name__ = None ,			__magic_name__ = None ,			__magic_name__ = "max_length" ,			__magic_name__ = None ,			__magic_name__ = None ,			__magic_name__ = None ,			**__magic_name__ ,			)   ->							BatchFeature:

      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.'  )

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

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

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

      _a      =							BatchFeature({'input_features': raw_speech}  )

      # convert into correct format for padding

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

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

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

      _a      =							[self._np_extract_fbank_features(__magic_name__  ) for waveform in input_features[0]]

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

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

      if return_tensors is not None:
           _a      =							padded_inputs.convert_to_tensors(__magic_name__  )

      return padded_inputs




 def 							__UpperCAmelCase						(	self  )   ->							Dict[str, Any]:
      _a      =							copy.deepcopy(self.__dict__  )
      _a      =							self.__class__.__name__
      if "mel_filters" in output:
           del output["mel_filters"]
      return output

