




import datetime
import platform
import subprocess
from typing import Optional, Tuple, Union

import numpy as np







def lowerCamelCase__       (      _lowercase			,						_lowercase				):

     '''simple docstring'''


     UpperCAmelCase_      :							List[str]       =      f'''{sampling_rate}'''
     UpperCAmelCase_      :							List[str]       =      '''1'''
     UpperCAmelCase_      :							List[str]       =      '''f32le'''
     UpperCAmelCase_      :							int       =      [
         '''ffmpeg''',
         '''-i''',
         '''pipe:0''',
         '''-ac''',
         ac,
         '''-ar''',
         ar,
         '''-f''',
         format_for_conversion,
         '''-hide_banner''',
         '''-loglevel''',
         '''quiet''',
         '''pipe:1''',
     ]

     try:
          with subprocess.Popen(_lowercase			,						stdin=subprocess.PIPE			,						stdout=subprocess.PIPE				) as ffmpeg_process:
               UpperCAmelCase_      :							int       =      ffmpeg_process.communicate(_lowercase				)
     except FileNotFoundError as error:
          raise ValueError('''ffmpeg was not found but is required to load audio files from filename'''				) from error
     UpperCAmelCase_      :							int       =      output_stream[0]
     UpperCAmelCase_      :							Union[str, Any]       =      np.frombuffer(_lowercase			,						np.floataa				)
     if audio.shape[0] == 0:
          raise ValueError('''Malformed soundfile'''				)
     return audio







def lowerCamelCase__       (      _lowercase			,						_lowercase			,						_lowercase = "f32le"			,						):

     '''simple docstring'''


     UpperCAmelCase_      :							Optional[int]       =      f'''{sampling_rate}'''
     UpperCAmelCase_      :							Union[str, Any]       =      '''1'''
     if format_for_conversion == "s16le":
          UpperCAmelCase_      :							Optional[Any]       =      2
     elif format_for_conversion == "f32le":
          UpperCAmelCase_      :							Dict       =      4
     else:
          raise ValueError(f'''Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`'''				)

     UpperCAmelCase_      :							int       =      platform.system()
     if system == "Linux":
          UpperCAmelCase_      :							Optional[Any]       =      '''alsa'''
          UpperCAmelCase_      :							List[str]       =      '''default'''
     elif system == "Darwin":
          UpperCAmelCase_      :							Tuple       =      '''avfoundation'''
          UpperCAmelCase_      :							int       =      ''':0'''
     elif system == "Windows":
          UpperCAmelCase_      :							Union[str, Any]       =      '''dshow'''
          UpperCAmelCase_      :							List[Any]       =      '''default'''

     UpperCAmelCase_      :							List[Any]       =      [
         '''ffmpeg''',
         '''-f''',
         format_,
         '''-i''',
         input_,
         '''-ac''',
         ac,
         '''-ar''',
         ar,
         '''-f''',
         format_for_conversion,
         '''-fflags''',
         '''nobuffer''',
         '''-hide_banner''',
         '''-loglevel''',
         '''quiet''',
         '''pipe:1''',
     ]
     UpperCAmelCase_      :							Optional[Any]       =      int(round(sampling_rate * chunk_length_s				)				) * size_of_sample
     UpperCAmelCase_      :							int       =      _ffmpeg_stream(_lowercase			,						_lowercase				)
     for item in iterator:
          yield item







def lowerCamelCase__       (      _lowercase			,						_lowercase			,						_lowercase = None			,						_lowercase = None			,						_lowercase = "f32le"			,						):

     '''simple docstring'''


     if stream_chunk_s is not None:
          UpperCAmelCase_      :							str       =      stream_chunk_s
     else:
          UpperCAmelCase_      :							List[str]       =      chunk_length_s

     UpperCAmelCase_      :							Tuple       =      ffmpeg_microphone(_lowercase			,						_lowercase			,						format_for_conversion=_lowercase				)
     if format_for_conversion == "s16le":
          UpperCAmelCase_      :							Optional[int]       =      np.intaa
          UpperCAmelCase_      :							int       =      2
     elif format_for_conversion == "f32le":
          UpperCAmelCase_      :							Tuple       =      np.floataa
          UpperCAmelCase_      :							Optional[int]       =      4
     else:
          raise ValueError(f'''Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`'''				)

     if stride_length_s is None:
          UpperCAmelCase_      :							Any       =      chunk_length_s / 6
     UpperCAmelCase_      :							Union[str, Any]       =      int(round(sampling_rate * chunk_length_s				)				) * size_of_sample
     if isinstance(_lowercase			,						(int, float)				):
          UpperCAmelCase_      :							Optional[int]       =      [stride_length_s, stride_length_s]

     UpperCAmelCase_      :							List[str]       =      int(round(sampling_rate * stride_length_s[0]				)				) * size_of_sample
     UpperCAmelCase_      :							List[Any]       =      int(round(sampling_rate * stride_length_s[1]				)				) * size_of_sample
     UpperCAmelCase_      :							List[str]       =      datetime.datetime.now()
     UpperCAmelCase_      :							List[str]       =      datetime.timedelta(seconds=_lowercase				)
     for item in chunk_bytes_iter(_lowercase			,						_lowercase			,						stride=(stride_left, stride_right)			,						stream=_lowercase				):
          # Put everything back in numpy scale
          UpperCAmelCase_      :							Union[str, Any]       =      np.frombuffer(item['''raw''']			,						dtype=_lowercase				)
          UpperCAmelCase_      :							Optional[int]       =      (
              item['''stride'''][0] // size_of_sample,
              item['''stride'''][1] // size_of_sample,
          )
          UpperCAmelCase_      :							str       =      sampling_rate
          audio_time += delta
          if datetime.datetime.now() > audio_time + 10 * delta:
               # We're late !! SKIP
               continue
          yield item







def lowerCamelCase__       (      _lowercase			,						_lowercase			,						_lowercase			,						_lowercase = False				):

     '''simple docstring'''


     UpperCAmelCase_      :							Dict       =      B''''''
     UpperCAmelCase_,   UpperCAmelCase_      :							Dict       =      stride
     if stride_left + stride_right >= chunk_len:
          raise ValueError(
              f'''Stride needs to be strictly smaller than chunk_len: ({stride_left}, {stride_right}) vs {chunk_len}'''				)
     UpperCAmelCase_      :							Optional[Any]       =      0
     for raw in iterator:
          acc += raw
          if stream and len(_lowercase				) < chunk_len:
               UpperCAmelCase_      :							Dict       =      (_stride_left, 0)
               yield {"raw": acc[:chunk_len], "stride": stride, "partial": True}
          else:
               while len(_lowercase				) >= chunk_len:
                    # We are flushing the accumulator
                    UpperCAmelCase_      :							int       =      (_stride_left, stride_right)
                    UpperCAmelCase_      :							List[str]       =      {'''raw''': acc[:chunk_len], '''stride''': stride}
                    if stream:
                         UpperCAmelCase_      :							List[str]       =      False
                    yield item
                    UpperCAmelCase_      :							Dict       =      stride_left
                    UpperCAmelCase_      :							List[str]       =      acc[chunk_len - stride_left - stride_right :]
    # Last chunk
     if len(_lowercase				) > stride_left:
          UpperCAmelCase_      :							int       =      {'''raw''': acc, '''stride''': (_stride_left, 0)}
          if stream:
               UpperCAmelCase_      :							Optional[int]       =      False
          yield item







def lowerCamelCase__       (      _lowercase			,						_lowercase				):

     '''simple docstring'''


     UpperCAmelCase_      :							Optional[Any]       =      2**24  # 16Mo
     try:
          with subprocess.Popen(_lowercase			,						stdout=subprocess.PIPE			,						bufsize=_lowercase				) as ffmpeg_process:
               while True:
                    UpperCAmelCase_      :							Union[str, Any]       =      ffmpeg_process.stdout.read(_lowercase				)
                    if raw == b"":
                         break
                    yield raw
     except FileNotFoundError as error:
          raise ValueError('''ffmpeg was not found but is required to stream audio files from filename'''				) from error





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(					_a       ):






    """simple docstring"""





    lowerCAmelCase        =				42


class   __a(					_a    ,						_a       ):






    """simple docstring"""





    lowerCAmelCase        =				True





    @register_to_config
    def __init__(				self       ,_SCREAMING_SNAKE_CASE = 3       ,_SCREAMING_SNAKE_CASE = 3       ,_SCREAMING_SNAKE_CASE = ("DownEncoderBlock2D",)       ,_SCREAMING_SNAKE_CASE = ("UpDecoderBlock2D",)       ,_SCREAMING_SNAKE_CASE = (64,)       ,_SCREAMING_SNAKE_CASE = 1       ,_SCREAMING_SNAKE_CASE = "silu"       ,_SCREAMING_SNAKE_CASE = 4       ,_SCREAMING_SNAKE_CASE = 32       ,_SCREAMING_SNAKE_CASE = 32       ,_SCREAMING_SNAKE_CASE = 0.1_82_15       ,)			->		Optional[int]:
         super().__init__()

         # pass init params to Encoder
         UpperCAmelCase_      :							List[Any]       =      Encoder(
             in_channels=_SCREAMING_SNAKE_CASE       ,out_channels=_SCREAMING_SNAKE_CASE       ,down_block_types=_SCREAMING_SNAKE_CASE       ,block_out_channels=_SCREAMING_SNAKE_CASE       ,layers_per_block=_SCREAMING_SNAKE_CASE       ,act_fn=_SCREAMING_SNAKE_CASE       ,norm_num_groups=_SCREAMING_SNAKE_CASE       ,double_z=_SCREAMING_SNAKE_CASE       ,)

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

         UpperCAmelCase_      :							int       =      nn.Convad(2 * latent_channels       ,2 * latent_channels       ,1  )
         UpperCAmelCase_      :							Union[str, Any]       =      nn.Convad(_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE       ,1  )

         UpperCAmelCase_      :							Optional[Any]       =      False
         UpperCAmelCase_      :							Dict       =      False

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





    def 							a__				(				self       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE=False  )			->		int:
         if isinstance(_SCREAMING_SNAKE_CASE       ,(Encoder, Decoder)  ):
              UpperCAmelCase_      :							Union[str, Any]       =      value





    def 							a__				(				self       ,_SCREAMING_SNAKE_CASE = True  )			->		Optional[Any]:
         UpperCAmelCase_      :							Dict       =      use_tiling





    def 							a__				(				self  )			->		Optional[Any]:
         self.enable_tiling(_SCREAMING_SNAKE_CASE  )





    def 							a__				(				self  )			->		List[str]:
         UpperCAmelCase_      :							str       =      True





    def 							a__				(				self  )			->		Any:
         UpperCAmelCase_      :							Any       =      False





    @property
    # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
    def 							a__				(				self  )			->		Dict[str, AttentionProcessor]:
         UpperCAmelCase_      :							int       =      {}

         def fn_recursive_add_processors(_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE  ):
              if hasattr(_SCREAMING_SNAKE_CASE       ,'''set_processor'''  ):
                   UpperCAmelCase_      :							Tuple       =      module.processor

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

              return processors

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

         return processors





    def 							a__				(				self       ,_SCREAMING_SNAKE_CASE  )			->		Union[str, Any]:
         UpperCAmelCase_      :							Optional[int]       =      len(self.attn_processors.keys()  )

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

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

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





    def 							a__				(				self  )			->		Tuple:
         self.set_attn_processor(AttnProcessor()  )





    @apply_forward_hook
    def 							a__				(				self       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE = True  )			->		AutoencoderKLOutput:
         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(_SCREAMING_SNAKE_CASE       ,return_dict=_SCREAMING_SNAKE_CASE  )

         if self.use_slicing and x.shape[0] > 1:
              UpperCAmelCase_      :							int       =      [self.encoder(_SCREAMING_SNAKE_CASE  ) for x_slice in x.split(1  )]
              UpperCAmelCase_      :							List[str]       =      torch.cat(_SCREAMING_SNAKE_CASE  )
         else:
              UpperCAmelCase_      :							Optional[Any]       =      self.encoder(_SCREAMING_SNAKE_CASE  )

         UpperCAmelCase_      :							Any       =      self.quant_conv(_SCREAMING_SNAKE_CASE  )
         UpperCAmelCase_      :							Union[str, Any]       =      DiagonalGaussianDistribution(_SCREAMING_SNAKE_CASE  )

         if not return_dict:
              return (posterior,)

         return AutoencoderKLOutput(latent_dist=_SCREAMING_SNAKE_CASE  )





    def 							a__				(				self       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE = True  )			->		Union[DecoderOutput, torch.FloatTensor]:
         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(_SCREAMING_SNAKE_CASE       ,return_dict=_SCREAMING_SNAKE_CASE  )

         UpperCAmelCase_      :							Tuple       =      self.post_quant_conv(_SCREAMING_SNAKE_CASE  )
         UpperCAmelCase_      :							Tuple       =      self.decoder(_SCREAMING_SNAKE_CASE  )

         if not return_dict:
              return (dec,)

         return DecoderOutput(sample=_SCREAMING_SNAKE_CASE  )





    @apply_forward_hook
    def 							a__				(				self       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE = True  )			->		Union[DecoderOutput, torch.FloatTensor]:
         if self.use_slicing and z.shape[0] > 1:
              UpperCAmelCase_      :							Tuple       =      [self._decode(_SCREAMING_SNAKE_CASE  ).sample for z_slice in z.split(1  )]
              UpperCAmelCase_      :							str       =      torch.cat(_SCREAMING_SNAKE_CASE  )
         else:
              UpperCAmelCase_      :							str       =      self._decode(_SCREAMING_SNAKE_CASE  ).sample

         if not return_dict:
              return (decoded,)

         return DecoderOutput(sample=_SCREAMING_SNAKE_CASE  )





    def 							a__				(				self       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE  )			->		Union[str, Any]:
         UpperCAmelCase_      :							str       =      min(a.shape[2]       ,b.shape[2]       ,_SCREAMING_SNAKE_CASE  )
         for y in range(_SCREAMING_SNAKE_CASE  ):
              UpperCAmelCase_      :							str       =      a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
         return b





    def 							a__				(				self       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE  )			->		int:
         UpperCAmelCase_      :							Any       =      min(a.shape[3]       ,b.shape[3]       ,_SCREAMING_SNAKE_CASE  )
         for x in range(_SCREAMING_SNAKE_CASE  ):
              UpperCAmelCase_      :							List[str]       =      a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
         return b





    def 							a__				(				self       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE = True  )			->		AutoencoderKLOutput:
         UpperCAmelCase_      :							Optional[int]       =      int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)  )
         UpperCAmelCase_      :							Any       =      int(self.tile_latent_min_size * self.tile_overlap_factor  )
         UpperCAmelCase_      :							str       =      self.tile_latent_min_size - blend_extent

         # Split the image into 512x512 tiles and encode them separately.
         UpperCAmelCase_      :							Tuple       =      []
         for i in range(0       ,x.shape[2]       ,_SCREAMING_SNAKE_CASE  ):
              UpperCAmelCase_      :							List[str]       =      []
              for j in range(0       ,x.shape[3]       ,_SCREAMING_SNAKE_CASE  ):
                   UpperCAmelCase_      :							List[str]       =      x[:, :, i : i + self.tile_sample_min_size, j : j + self.tile_sample_min_size]
                   UpperCAmelCase_      :							int       =      self.encoder(_SCREAMING_SNAKE_CASE  )
                   UpperCAmelCase_      :							str       =      self.quant_conv(_SCREAMING_SNAKE_CASE  )
                   row.append(_SCREAMING_SNAKE_CASE  )
              rows.append(_SCREAMING_SNAKE_CASE  )
         UpperCAmelCase_      :							Union[str, Any]       =      []
         for i, row in enumerate(_SCREAMING_SNAKE_CASE  ):
              UpperCAmelCase_      :							str       =      []
              for j, tile in enumerate(_SCREAMING_SNAKE_CASE  ):
                   # 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_      :							Optional[Any]       =      self.blend_v(rows[i - 1][j]       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE  )
                   if j > 0:
                        UpperCAmelCase_      :							List[str]       =      self.blend_h(row[j - 1]       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE  )
                   result_row.append(tile[:, :, :row_limit, :row_limit]  )
              result_rows.append(torch.cat(_SCREAMING_SNAKE_CASE       ,dim=3  )  )

         UpperCAmelCase_      :							Optional[Any]       =      torch.cat(_SCREAMING_SNAKE_CASE       ,dim=2  )
         UpperCAmelCase_      :							Union[str, Any]       =      DiagonalGaussianDistribution(_SCREAMING_SNAKE_CASE  )

         if not return_dict:
              return (posterior,)

         return AutoencoderKLOutput(latent_dist=_SCREAMING_SNAKE_CASE  )





    def 							a__				(				self       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE = True  )			->		Union[DecoderOutput, torch.FloatTensor]:
         UpperCAmelCase_      :							Optional[Any]       =      int(self.tile_latent_min_size * (1 - self.tile_overlap_factor)  )
         UpperCAmelCase_      :							str       =      int(self.tile_sample_min_size * self.tile_overlap_factor  )
         UpperCAmelCase_      :							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_      :							Tuple       =      []
         for i in range(0       ,z.shape[2]       ,_SCREAMING_SNAKE_CASE  ):
              UpperCAmelCase_      :							Tuple       =      []
              for j in range(0       ,z.shape[3]       ,_SCREAMING_SNAKE_CASE  ):
                   UpperCAmelCase_      :							Optional[int]       =      z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
                   UpperCAmelCase_      :							List[str]       =      self.post_quant_conv(_SCREAMING_SNAKE_CASE  )
                   UpperCAmelCase_      :							Dict       =      self.decoder(_SCREAMING_SNAKE_CASE  )
                   row.append(_SCREAMING_SNAKE_CASE  )
              rows.append(_SCREAMING_SNAKE_CASE  )
         UpperCAmelCase_      :							Optional[Any]       =      []
         for i, row in enumerate(_SCREAMING_SNAKE_CASE  ):
              UpperCAmelCase_      :							int       =      []
              for j, tile in enumerate(_SCREAMING_SNAKE_CASE  ):
                   # 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_      :							Optional[int]       =      self.blend_v(rows[i - 1][j]       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE  )
                   if j > 0:
                        UpperCAmelCase_      :							Optional[int]       =      self.blend_h(row[j - 1]       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE  )
                   result_row.append(tile[:, :, :row_limit, :row_limit]  )
              result_rows.append(torch.cat(_SCREAMING_SNAKE_CASE       ,dim=3  )  )

         UpperCAmelCase_      :							Dict       =      torch.cat(_SCREAMING_SNAKE_CASE       ,dim=2  )
         if not return_dict:
              return (dec,)

         return DecoderOutput(sample=_SCREAMING_SNAKE_CASE  )





    def 							a__				(				self       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE = False       ,_SCREAMING_SNAKE_CASE = True       ,_SCREAMING_SNAKE_CASE = None       ,)			->		Union[DecoderOutput, torch.FloatTensor]:
         UpperCAmelCase_      :							str       =      sample
         UpperCAmelCase_      :							Optional[Any]       =      self.encode(_SCREAMING_SNAKE_CASE  ).latent_dist
         if sample_posterior:
              UpperCAmelCase_      :							List[Any]       =      posterior.sample(generator=_SCREAMING_SNAKE_CASE  )
         else:
              UpperCAmelCase_      :							Union[str, Any]       =      posterior.mode()
         UpperCAmelCase_      :							List[Any]       =      self.decode(_SCREAMING_SNAKE_CASE  ).sample

         if not return_dict:
              return (dec,)

         return DecoderOutput(sample=_SCREAMING_SNAKE_CASE  )