




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 		lowercase							(     UpperCamelCase__	):
     _a											=		42







class 		lowercase							(     UpperCamelCase__,UpperCamelCase__	):
     _a											=		True

     @register_to_config
     def __init__(						self	,  _a = 3	,  _a = 3	,  _a = ("DownEncoderBlock2D",)	,  _a = ("UpDecoderBlock2D",)	,  _a = (64,)	,  _a = 1	,  _a = "silu"	,  _a = 4	,  _a = 32	,  _a = 32	,  _a = 0.18215	,  )		->							Any:
           super().__init__()

           # pass init params to Encoder
           _A							:  Dict    =		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
           _A							:  Dict    =		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	,  )

           _A							:  Union[str, Any]    =		nn.Convad(2 * latent_channels	,  2 * latent_channels	,  1       )
           _A							:  Tuple    =		nn.Convad(_a	,  _a	,  1       )

           _A							:  int    =		False
           _A							:  Union[str, Any]    =		False

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

     def       a__			(						self	,  _a	,  _a=False       )		->							List[str]:
           if isinstance(_a	,  (Encoder, Decoder)       ):
                 _A							:  Dict    =		value

     def       a__			(						self	,  _a = True       )		->							Any:
           _A							:  Tuple    =		use_tiling

     def       a__			(						self       )		->							List[Any]:
           self.enable_tiling(_a       )

     def       a__			(						self       )		->							Dict:
           _A							:  Dict    =		True

     def       a__			(						self       )		->							List[str]:
           _A							:  int    =		False

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

           def fn_recursive_add_processors(_a	,  _a	,  _a       ):
                 if hasattr(_a	,  """set_processor"""       ):
                       _A							:  List[str]    =		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	,  _a       )		->							Union[str, Any]:
           _A							:  List[Any]    =		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	,  _a	,  _a       ):
                 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       )		->							Union[str, Any]:
           self.set_attn_processor(AttnProcessor()       )

     @apply_forward_hook
     def       a__			(						self	,  _a	,  _a = 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(_a	,  return_dict=_a       )

           if self.use_slicing and x.shape[0] > 1:
                 _A							:  List[Any]    =		[self.encoder(_a       ) for x_slice in x.split(1       )]
                 _A							:  List[str]    =		torch.cat(_a       )
           else:
                 _A							:  Union[str, Any]    =		self.encoder(_a       )

           _A							:  List[Any]    =		self.quant_conv(_a       )
           _A							:  str    =		DiagonalGaussianDistribution(_a       )

           if not return_dict:
                 return (posterior,)

           return AutoencoderKLOutput(latent_dist=_a       )

     def       a__			(						self	,  _a	,  _a = 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(_a	,  return_dict=_a       )

           _A							:  Optional[Any]    =		self.post_quant_conv(_a       )
           _A							:  Tuple    =		self.decoder(_a       )

           if not return_dict:
                 return (dec,)

           return DecoderOutput(sample=_a       )

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

           if not return_dict:
                 return (decoded,)

           return DecoderOutput(sample=_a       )

     def       a__			(						self	,  _a	,  _a	,  _a       )		->							List[str]:
           _A							:  str    =		min(a.shape[2]	,  b.shape[2]	,  _a       )
           for y in range(_a       ):
                 _A							:  Dict    =		a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
           return b

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

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

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

           _A							:  int    =		torch.cat(_a	,  dim=2       )
           _A							:  Optional[int]    =		DiagonalGaussianDistribution(_a       )

           if not return_dict:
                 return (posterior,)

           return AutoencoderKLOutput(latent_dist=_a       )

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

           _A							:  Optional[Any]    =		torch.cat(_a	,  dim=2       )
           if not return_dict:
                 return (dec,)

           return DecoderOutput(sample=_a       )



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

           if not return_dict:
                 return (dec,)

           return DecoderOutput(sample=_a       )







import itertools
import os
import random
import tempfile
import unittest

import numpy as np

from transformers import TvltFeatureExtractor, is_datasets_available
from transformers.testing_utils import check_json_file_has_correct_format, require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_torch_available():
  import torch

if is_datasets_available():
  from datasets import load_dataset

_snake_case            = random.Random()







def        lowerCAmelCase_			(  snake_case_,snake_case_=1.0,snake_case_=None,snake_case_=None  ):
      if rng is None:
            _A							:  str    =		global_rng

      _A							:  List[Any]    =		[]
      for batch_idx in range(shape[0]  ):
            values.append([]  )
            for _ in range(shape[1]  ):
                  values[-1].append(rng.random() * scale  )

      return values







class 		lowercase							(     unittest.TestCase	):

     def __init__(						self	,  _a	,  _a=7	,  _a=400	,  _a=2000	,  _a=2048	,  _a=128	,  _a=1	,  _a=512	,  _a=30	,  _a=4_4100	,  )		->							Tuple:
           _A							:  Any    =		parent
           _A							:  str    =		batch_size
           _A							:  Union[str, Any]    =		min_seq_length
           _A							:  int    =		max_seq_length
           _A							:  List[str]    =		(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
           _A							:  Tuple    =		spectrogram_length
           _A							:  int    =		feature_size
           _A							:  str    =		num_audio_channels
           _A							:  Tuple    =		hop_length
           _A							:  List[str]    =		chunk_length
           _A							:  Union[str, Any]    =		sampling_rate

     def       a__			(						self       )		->							Tuple:
           return {
               "spectrogram_length": self.spectrogram_length,
               "feature_size": self.feature_size,
               "num_audio_channels": self.num_audio_channels,
               "hop_length": self.hop_length,
               "chunk_length": self.chunk_length,
               "sampling_rate": self.sampling_rate,
           }



     def       a__			(						self	,  _a=False	,  _a=False       )		->							Optional[int]:
           def _flatten(_a       ):
                 return list(itertools.chain(*_a       )       )

           if equal_length:
                 _A							:  List[str]    =		[floats_list((self.max_seq_length, self.feature_size)       ) for _ in range(self.batch_size       )]
           else:
                 # make sure that inputs increase in size
                 _A							:  Union[str, Any]    =		[
                     floats_list((x, self.feature_size)       )
                     for x in range(self.min_seq_length	,  self.max_seq_length	,  self.seq_length_diff       )
                 ]
           if numpify:
                 _A							:  List[Any]    =		[np.asarray(_a       ) for x in speech_inputs]
           return speech_inputs







@require_torch
@require_torchaudio
class 		lowercase							(     UpperCamelCase__,unittest.TestCase	):
     _a											=		TvltFeatureExtractor

     def       a__			(						self       )		->							Any:
           _A							:  int    =		TvltFeatureExtractionTester(self       )

     def       a__			(						self       )		->							List[Any]:
           _A							:  Tuple    =		self.feature_extraction_class(**self.feat_extract_dict       )
           self.assertTrue(hasattr(_a	,  """spectrogram_length"""       )       )
           self.assertTrue(hasattr(_a	,  """feature_size"""       )       )
           self.assertTrue(hasattr(_a	,  """num_audio_channels"""       )       )
           self.assertTrue(hasattr(_a	,  """hop_length"""       )       )
           self.assertTrue(hasattr(_a	,  """chunk_length"""       )       )
           self.assertTrue(hasattr(_a	,  """sampling_rate"""       )       )

     def       a__			(						self       )		->							Optional[int]:
           _A							:  str    =		self.feature_extraction_class(**self.feat_extract_dict       )

           with tempfile.TemporaryDirectory() as tmpdirname:
                 _A							:  Tuple    =		feat_extract_first.save_pretrained(_a       )[0]
                 check_json_file_has_correct_format(_a       )
                 _A							:  Optional[int]    =		self.feature_extraction_class.from_pretrained(_a       )

           _A							:  Optional[Any]    =		feat_extract_first.to_dict()
           _A							:  int    =		feat_extract_second.to_dict()
           _A							:  int    =		dict_first.pop("""mel_filters"""       )
           _A							:  Optional[int]    =		dict_second.pop("""mel_filters"""       )
           self.assertTrue(np.allclose(_a	,  _a       )       )
           self.assertEqual(_a	,  _a       )

     def       a__			(						self       )		->							int:
           _A							:  str    =		self.feature_extraction_class(**self.feat_extract_dict       )

           with tempfile.TemporaryDirectory() as tmpdirname:
                 _A							:  List[str]    =		os.path.join(_a	,  """feat_extract.json"""       )
                 feat_extract_first.to_json_file(_a       )
                 _A							:  Union[str, Any]    =		self.feature_extraction_class.from_json_file(_a       )

           _A							:  Optional[Any]    =		feat_extract_first.to_dict()
           _A							:  Union[str, Any]    =		feat_extract_second.to_dict()
           _A							:  List[Any]    =		dict_first.pop("""mel_filters"""       )
           _A							:  Optional[Any]    =		dict_second.pop("""mel_filters"""       )
           self.assertTrue(np.allclose(_a	,  _a       )       )
           self.assertEqual(_a	,  _a       )

     def       a__			(						self       )		->							Optional[Any]:
           # Initialize feature_extractor
           _A							:  Tuple    =		self.feature_extraction_class(**self.feat_extract_dict       )

           # create three inputs of length 800, 1000, and 1200
           _A							:  Optional[int]    =		[floats_list((1, x)       )[0] for x in range(800	,  1400	,  200       )]
           _A							:  List[str]    =		[np.asarray(_a       ) for speech_input in speech_inputs]

           # Test not batched input
           _A							:  Optional[int]    =		feature_extractor(np_speech_inputs[0]	,  return_tensors="""np"""	,  sampling_rate=4_4100       ).audio_values

           self.assertTrue(encoded_audios.ndim == 4       )
           self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size       )
           self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length       )
           self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels       )

           # Test batched
           _A							:  str    =		feature_extractor(_a	,  return_tensors="""np"""	,  sampling_rate=4_4100       ).audio_values

           self.assertTrue(encoded_audios.ndim == 4       )
           self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size       )
           self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length       )
           self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels       )

           # Test audio masking
           _A							:  Optional[Any]    =		feature_extractor(
               _a	,  return_tensors="""np"""	,  sampling_rate=4_4100	,  mask_audio=_a       ).audio_values

           self.assertTrue(encoded_audios.ndim == 4       )
           self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size       )
           self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length       )
           self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels       )

           # Test 2-D numpy arrays are batched.
           _A							:  Dict    =		[floats_list((1, x)       )[0] for x in (800, 800, 800)]
           _A							:  Optional[int]    =		np.asarray(_a       )
           _A							:  Optional[Any]    =		feature_extractor(_a	,  return_tensors="""np"""	,  sampling_rate=4_4100       ).audio_values
           self.assertTrue(encoded_audios.ndim == 4       )
           self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size       )
           self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length       )
           self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels       )

     def       a__			(						self	,  _a       )		->							str:
           _A							:  Optional[Any]    =		load_dataset("""hf-internal-testing/librispeech_asr_dummy"""	,  """clean"""	,  split="""validation"""       )
           # automatic decoding with librispeech
           _A							:  Dict    =		ds.sort("""id"""       ).select(range(_a       )       )[:num_samples]["""audio"""]

           return [x["array"] for x in speech_samples]



     def       a__			(						self       )		->							Optional[Any]:
           _A							:  List[str]    =		self._load_datasamples(1       )
           _A							:  List[str]    =		TvltFeatureExtractor()
           _A							:  List[Any]    =		feature_extractor(_a	,  return_tensors="""pt"""       ).audio_values

           self.assertEquals(audio_values.shape	,  (1, 1, 192, 128)       )

           _A							:  int    =		torch.tensor([[-0.3032, -0.2708], [-0.4434, -0.4007]]       )
           self.assertTrue(torch.allclose(audio_values[0, 0, :2, :2]	,  _a	,  atol=1e-4       )       )

