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

      __UpperCamelCase:  "DiagonalGaussianDistribution"

class 	lowerCamelCase_	(snake_case__							, snake_case__    ):
      '''simple docstring'''

      __UpperCamelCase:  Dict						=						True




      @register_to_config
      def __init__(       self						:				Dict						,					A						:				int = 3						,					A						:				int = 3						,					A						:				Tuple[str] = ("DownEncoderBlock2D",)						,					A						:				Tuple[str] = ("UpDecoderBlock2D",)						,					A						:				Tuple[int] = (64,)						,					A						:				int = 1						,					A						:				str = "silu"						,					A						:				int = 4						,					A						:				int = 32						,					A						:				int = 32						,					A						:				float = 0.18_215						,					):
             super().__init__()

             # pass init params to Encoder
             _UpperCAmelCase     :  Union[str, Any]							      =    Encoder(
                 in_channels=A						,					out_channels=A						,					down_block_types=A						,					block_out_channels=A						,					layers_per_block=A						,					act_fn=A						,					norm_num_groups=A						,					double_z=A						,					)

             # pass init params to Decoder
             _UpperCAmelCase     :  List[str]							      =    Decoder(
                 in_channels=A						,					out_channels=A						,					up_block_types=A						,					block_out_channels=A						,					layers_per_block=A						,					norm_num_groups=A						,					act_fn=A						,					)

             _UpperCAmelCase     :  Union[str, Any]							      =    nn.Convad(2 * latent_channels						,					2 * latent_channels						,					1       )
             _UpperCAmelCase     :  Optional[Any]							      =    nn.Convad(A						,					A						,					1       )

             _UpperCAmelCase     :  List[Any]							      =    False
             _UpperCAmelCase     :  List[Any]							      =    False

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




      def      _A	(       self						:				Optional[Any]						,					A						:				str						,					A						:				Union[str, Any]=False       ):
             if isinstance(A						,					(Encoder, Decoder)       ):
                    _UpperCAmelCase     :  Union[str, Any]							      =    value




      def      _A	(       self						:				int						,					A						:				bool = True       ):
             _UpperCAmelCase     :  Any							      =    use_tiling




      def      _A	(       self						:				Dict       ):
             self.enable_tiling(A       )




      def      _A	(       self						:				str       ):
             _UpperCAmelCase     :  Optional[Any]							      =    True




      def      _A	(       self						:				Optional[int]       ):
             _UpperCAmelCase     :  Tuple							      =    False




      @property
      # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
      def      _A	(       self						:				Union[str, Any]       ):
             _UpperCAmelCase     :  List[str]							      =    {}

             def fn_recursive_add_processors(A						:				str						,					A						:				torch.nn.Module						,					A						:				Dict[str, AttentionProcessor]       ):
                    if hasattr(A						,					"set_processor"       ):
                           _UpperCAmelCase     :  Any							      =    module.processor

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

                    return processors

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

             return processors




      def      _A	(       self						:				Optional[int]						,					A						:				Union[AttentionProcessor, Dict[str, AttentionProcessor]]       ):
             _UpperCAmelCase     :  str							      =    len(self.attn_processors.keys()       )

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

             def fn_recursive_attn_processor(A						:				str						,					A						:				torch.nn.Module						,					A						:				str       ):
                    if hasattr(A						,					"set_processor"       ):
                           if not isinstance(A						,					A       ):
                                  module.set_processor(A       )
                           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}"""						,					A						,					A       )

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




      def      _A	(       self						:				Optional[int]       ):
             self.set_attn_processor(AttnProcessor()       )




      @apply_forward_hook
      def      _A	(       self						:				Optional[Any]						,					A						:				torch.FloatTensor						,					A						:				bool = True       ):
             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(A						,					return_dict=A       )

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

             _UpperCAmelCase     :  Union[str, Any]							      =    self.quant_conv(A       )
             _UpperCAmelCase     :  str							      =    DiagonalGaussianDistribution(A       )

             if not return_dict:
                    return (posterior,)

             return AutoencoderKLOutput(latent_dist=A       )




      def      _A	(       self						:				List[str]						,					A						:				torch.FloatTensor						,					A						:				bool = True       ):
             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(A						,					return_dict=A       )

             _UpperCAmelCase     :  Dict							      =    self.post_quant_conv(A       )
             _UpperCAmelCase     :  Any							      =    self.decoder(A       )

             if not return_dict:
                    return (dec,)

             return DecoderOutput(sample=A       )




      @apply_forward_hook
      def      _A	(       self						:				List[str]						,					A						:				torch.FloatTensor						,					A						:				bool = True       ):
             if self.use_slicing and z.shape[0] > 1:
                    _UpperCAmelCase     :  List[Any]							      =    [self._decode(A       ).sample for z_slice in z.split(1       )]
                    _UpperCAmelCase     :  str							      =    torch.cat(A       )
             else:
                    _UpperCAmelCase     :  Optional[int]							      =    self._decode(A       ).sample

             if not return_dict:
                    return (decoded,)

             return DecoderOutput(sample=A       )




      def      _A	(       self						:				Tuple						,					A						:				Any						,					A						:				Tuple						,					A						:				Optional[int]       ):
             _UpperCAmelCase     :  int							      =    min(a.shape[2]						,					b.shape[2]						,					A       )
             for y in range(A       ):
                    _UpperCAmelCase     :  Tuple							      =    a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
             return b




      def      _A	(       self						:				Optional[Any]						,					A						:				Dict						,					A						:				Union[str, Any]						,					A						:				List[str]       ):
             _UpperCAmelCase     :  Optional[int]							      =    min(a.shape[3]						,					b.shape[3]						,					A       )
             for x in range(A       ):
                    _UpperCAmelCase     :  Optional[int]							      =    a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
             return b




      def      _A	(       self						:				Any						,					A						:				torch.FloatTensor						,					A						:				bool = True       ):
             _UpperCAmelCase     :  str							      =    int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)       )
             _UpperCAmelCase     :  Optional[Any]							      =    int(self.tile_latent_min_size * self.tile_overlap_factor       )
             _UpperCAmelCase     :  List[Any]							      =    self.tile_latent_min_size - blend_extent

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

             _UpperCAmelCase     :  Optional[int]							      =    torch.cat(A						,					dim=2       )
             _UpperCAmelCase     :  List[Any]							      =    DiagonalGaussianDistribution(A       )

             if not return_dict:
                    return (posterior,)

             return AutoencoderKLOutput(latent_dist=A       )




      def      _A	(       self						:				str						,					A						:				torch.FloatTensor						,					A						:				bool = True       ):
             _UpperCAmelCase     :  Any							      =    int(self.tile_latent_min_size * (1 - self.tile_overlap_factor)       )
             _UpperCAmelCase     :  List[Any]							      =    int(self.tile_sample_min_size * self.tile_overlap_factor       )
             _UpperCAmelCase     :  Optional[Any]							      =    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.
             _UpperCAmelCase     :  str							      =    []
             for i in range(0						,					z.shape[2]						,					A       ):
                    _UpperCAmelCase     :  List[str]							      =    []
                    for j in range(0						,					z.shape[3]						,					A       ):
                           _UpperCAmelCase     :  Tuple							      =    z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
                           _UpperCAmelCase     :  List[str]							      =    self.post_quant_conv(A       )
                           _UpperCAmelCase     :  Dict							      =    self.decoder(A       )
                           row.append(A       )
                    rows.append(A       )
             _UpperCAmelCase     :  List[Any]							      =    []
             for i, row in enumerate(A       ):
                    _UpperCAmelCase     :  List[str]							      =    []
                    for j, tile in enumerate(A       ):
                           # blend the above tile and the left tile
                           # to the current tile and add the current tile to the result row
                           if i > 0:
                                  _UpperCAmelCase     :  List[Any]							      =    self.blend_v(rows[i - 1][j]						,					A						,					A       )
                           if j > 0:
                                  _UpperCAmelCase     :  Optional[int]							      =    self.blend_h(row[j - 1]						,					A						,					A       )
                           result_row.append(tile[:, :, :row_limit, :row_limit]       )
                    result_rows.append(torch.cat(A						,					dim=3       )       )

             _UpperCAmelCase     :  Any							      =    torch.cat(A						,					dim=2       )
             if not return_dict:
                    return (dec,)

             return DecoderOutput(sample=A       )





      def      _A	(       self						:				List[Any]						,					A						:				torch.FloatTensor						,					A						:				bool = False						,					A						:				bool = True						,					A						:				Optional[torch.Generator] = None						,					):
             _UpperCAmelCase     :  Union[str, Any]							      =    sample
             _UpperCAmelCase     :  Tuple							      =    self.encode(A       ).latent_dist
             if sample_posterior:
                    _UpperCAmelCase     :  Any							      =    posterior.sample(generator=A       )
             else:
                    _UpperCAmelCase     :  Tuple							      =    posterior.mode()
             _UpperCAmelCase     :  List[Any]							      =    self.decode(A       ).sample

             if not return_dict:
                    return (dec,)

             return DecoderOutput(sample=A       )


'''simple docstring'''




import math
from typing import Optional

import numpy as np

from ...configuration_utils import PretrainedConfig
from ...utils import logging


__SCREAMING_SNAKE_CASE       :     Dict     =  logging.get_logger(__name__)

__SCREAMING_SNAKE_CASE       :     List[Any]     =  {
    """facebook/encodec_24khz""": """https://huggingface.co/facebook/encodec_24khz/resolve/main/config.json""",
    """facebook/encodec_48khz""": """https://huggingface.co/facebook/encodec_48khz/resolve/main/config.json""",
}

class 	lowerCamelCase_	(snake_case__    ):
      '''simple docstring'''

      __UpperCamelCase:  str						=						"encodec"




      def __init__(       self						:				Optional[int]						,					A						:				Union[str, Any]=[1.5, 3.0, 6.0, 12.0, 24.0]						,					A						:				List[Any]=24000						,					A						:				Union[str, Any]=1						,					A						:				List[Any]=False						,					A						:				Optional[int]=None						,					A						:				int=None						,					A						:				str=128						,					A						:				List[Any]=32						,					A						:				List[Any]=1						,					A						:				int=[8, 5, 4, 2]						,					A						:				Optional[int]="weight_norm"						,					A						:				List[Any]=7						,					A						:				Any=7						,					A						:				Dict=3						,					A						:				Optional[int]=2						,					A						:				Dict=True						,					A						:				Dict="reflect"						,					A						:				Any=2						,					A						:				Dict=2						,					A						:				str=1.0						,					A						:				Optional[int]=1024						,					A						:				Any=None						,					A						:				Any=True						,					**A						:				str						,					):
             _UpperCAmelCase     :  Optional[int]							      =    target_bandwidths
             _UpperCAmelCase     :  List[str]							      =    sampling_rate
             _UpperCAmelCase     :  Optional[int]							      =    audio_channels
             _UpperCAmelCase     :  str							      =    normalize
             _UpperCAmelCase     :  int							      =    chunk_length_s
             _UpperCAmelCase     :  str							      =    overlap
             _UpperCAmelCase     :  Optional[Any]							      =    hidden_size
             _UpperCAmelCase     :  int							      =    num_filters
             _UpperCAmelCase     :  Optional[Any]							      =    num_residual_layers
             _UpperCAmelCase     :  Optional[int]							      =    upsampling_ratios
             _UpperCAmelCase     :  int							      =    norm_type
             _UpperCAmelCase     :  List[Any]							      =    kernel_size
             _UpperCAmelCase     :  List[Any]							      =    last_kernel_size
             _UpperCAmelCase     :  List[Any]							      =    residual_kernel_size
             _UpperCAmelCase     :  List[str]							      =    dilation_growth_rate
             _UpperCAmelCase     :  Dict							      =    use_causal_conv
             _UpperCAmelCase     :  Tuple							      =    pad_mode
             _UpperCAmelCase     :  Tuple							      =    compress
             _UpperCAmelCase     :  List[str]							      =    num_lstm_layers
             _UpperCAmelCase     :  List[Any]							      =    trim_right_ratio
             _UpperCAmelCase     :  int							      =    codebook_size
             _UpperCAmelCase     :  Optional[Any]							      =    codebook_dim if codebook_dim is not None else hidden_size
             _UpperCAmelCase     :  Optional[int]							      =    use_conv_shortcut

             if self.norm_type not in ["weight_norm", "time_group_norm"]:
                    raise ValueError(
                        F"""self.norm_type must be one of `\"weight_norm\"`, `\"time_group_norm\"`), got {self.norm_type}"""       )

             super().__init__(**A       )




      @property
      def      _A	(       self						:				Any       ):
             if self.chunk_length_s is None:
                    return None
             else:
                    return int(self.chunk_length_s * self.sampling_rate       )




      @property
      def      _A	(       self						:				Union[str, Any]       ):
             if self.chunk_length_s is None or self.overlap is None:
                    return None
             else:
                    return max(1						,					int((1.0 - self.overlap) * self.chunk_length       )       )




      @property
      def      _A	(       self						:				Union[str, Any]       ):
             _UpperCAmelCase     :  Dict							      =    np.prod(self.upsampling_ratios       )
             return math.ceil(self.sampling_rate / hop_length       )





      @property
      def      _A	(       self						:				str       ):
             return int(1000 * self.target_bandwidths[-1] // (self.frame_rate * 10)       )

