






'''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  snake_case__      (					snake_case__):
					a_        =					42






class  snake_case__      (					snake_case__     ,				snake_case__):
					a_        =					True




					@register_to_config
					def __init__(  self  :		Dict				, _A  :		str = 3				, _A  :		Any = 3				, _A  :		int = ("DownEncoderBlock2D",)				, _A  :		Dict = ("UpDecoderBlock2D",)				, _A  :		Union[str, Any] = (64,)				, _A  :		List[Any] = 1				, _A  :		Optional[Any] = "silu"				, _A  :		Optional[Any] = 4				, _A  :		Optional[Any] = 32				, _A  :		str = 32				, _A  :		str = 0.18_215				, )    ->							str:
								super().__init__()

								# pass init params to Encoder
								UpperCAmelCase_      :       str          =       Encoder(
								    in_channels=__snake_case				, out_channels=__snake_case				, down_block_types=__snake_case				, block_out_channels=__snake_case				, layers_per_block=__snake_case				, act_fn=__snake_case				, norm_num_groups=__snake_case				, double_z=__snake_case				, )

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

								UpperCAmelCase_      :       str          =       nn.Convad(2 * latent_channels				, 2 * latent_channels				, 1      )
								UpperCAmelCase_      :       List[str]          =       nn.Convad(__snake_case				, __snake_case				, 1      )

								UpperCAmelCase_      :       List[str]          =       False
								UpperCAmelCase_      :       List[str]          =       False

								# only relevant if vae tiling is enabled
								UpperCAmelCase_      :       Dict          =       self.config.sample_size
								UpperCAmelCase_      :       Tuple          =       (
								    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_      :       List[Any]          =       0.25




					def 		A      (  self  :		int				, _A  :		Any				, _A  :		Tuple=False      )    ->							Union[str, Any]:
								if isinstance(__snake_case				, (Encoder, Decoder)      ):
											UpperCAmelCase_      :       int          =       value




					def 		A      (  self  :		List[str]				, _A  :		Any = True      )    ->							int:
								UpperCAmelCase_      :       List[Any]          =       use_tiling




					def 		A      (  self  :		int      )    ->							Dict:
								self.enable_tiling(__snake_case      )




					def 		A      (  self  :		List[str]      )    ->							Any:
								UpperCAmelCase_      :       Tuple          =       True




					def 		A      (  self  :		Optional[Any]      )    ->							Tuple:
								UpperCAmelCase_      :       Dict          =       False




					@property
					# Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
					def 		A      (  self  :		str      )    ->							str:
								UpperCAmelCase_      :       str          =       {}

								def fn_recursive_add_processors(_A  :		List[Any]				, _A  :		Optional[Any]				, _A  :		Union[str, Any]      ):
											if hasattr(__snake_case				, '''set_processor'''      ):
														UpperCAmelCase_      :       int          =       module.processor

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

											return processors

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

								return processors




					def 		A      (  self  :		List[Any]				, _A  :		Dict      )    ->							int:
								UpperCAmelCase_      :       Union[str, Any]          =       len(self.attn_processors.keys()      )

								if isinstance(__snake_case				, __snake_case      ) and len(__snake_case      ) != count:
											raise ValueError(
											    F"A dict of processors was passed, but the number of processors {len(__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(_A  :		Dict				, _A  :		Tuple				, _A  :		Tuple      ):
											if hasattr(__snake_case				, '''set_processor'''      ):
														if not isinstance(__snake_case				, __snake_case      ):
																	module.set_processor(__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}"				, __snake_case				, __snake_case      )

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




					def 		A      (  self  :		Optional[int]      )    ->							int:
								self.set_attn_processor(AttnProcessor()      )




					@apply_forward_hook
					def 		A      (  self  :		Dict				, _A  :		Any				, _A  :		List[str] = True      )    ->							Union[str, Any]:
								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(__snake_case				, return_dict=__snake_case      )

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

								UpperCAmelCase_      :       List[Any]          =       self.quant_conv(__snake_case      )
								UpperCAmelCase_      :       List[str]          =       DiagonalGaussianDistribution(__snake_case      )

								if not return_dict:
											return (posterior,)

								return AutoencoderKLOutput(latent_dist=__snake_case      )




					def 		A      (  self  :		Tuple				, _A  :		Any				, _A  :		List[Any] = True      )    ->							Dict:
								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(__snake_case				, return_dict=__snake_case      )

								UpperCAmelCase_      :       List[Any]          =       self.post_quant_conv(__snake_case      )
								UpperCAmelCase_      :       List[str]          =       self.decoder(__snake_case      )

								if not return_dict:
											return (dec,)

								return DecoderOutput(sample=__snake_case      )




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

								if not return_dict:
											return (decoded,)

								return DecoderOutput(sample=__snake_case      )




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




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




					def 		A      (  self  :		Union[str, Any]				, _A  :		Union[str, Any]				, _A  :		Dict = True      )    ->							Optional[Any]:
								UpperCAmelCase_      :       Union[str, Any]          =       int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)      )
								UpperCAmelCase_      :       Union[str, 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_      :       Union[str, Any]          =       []
								for i in range(0				, x.shape[2]				, __snake_case      ):
											UpperCAmelCase_      :       str          =       []
											for j in range(0				, x.shape[3]				, __snake_case      ):
														UpperCAmelCase_      :       Optional[int]          =       x[:, :, i : i + self.tile_sample_min_size, j : j + self.tile_sample_min_size]
														UpperCAmelCase_      :       Optional[Any]          =       self.encoder(__snake_case      )
														UpperCAmelCase_      :       List[Any]          =       self.quant_conv(__snake_case      )
														row.append(__snake_case      )
											rows.append(__snake_case      )
								UpperCAmelCase_      :       int          =       []
								for i, row in enumerate(__snake_case      ):
											UpperCAmelCase_      :       str          =       []
											for j, tile in enumerate(__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_      :       int          =       self.blend_v(rows[i - 1][j]				, __snake_case				, __snake_case      )
														if j > 0:
																	UpperCAmelCase_      :       Any          =       self.blend_h(row[j - 1]				, __snake_case				, __snake_case      )
														result_row.append(tile[:, :, :row_limit, :row_limit]      )
											result_rows.append(torch.cat(__snake_case				, dim=3      )      )

								UpperCAmelCase_      :       Any          =       torch.cat(__snake_case				, dim=2      )
								UpperCAmelCase_      :       Any          =       DiagonalGaussianDistribution(__snake_case      )

								if not return_dict:
											return (posterior,)

								return AutoencoderKLOutput(latent_dist=__snake_case      )




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

								UpperCAmelCase_      :       List[Any]          =       torch.cat(__snake_case				, dim=2      )
								if not return_dict:
											return (dec,)

								return DecoderOutput(sample=__snake_case      )





					def 		A      (  self  :		List[Any]				, _A  :		List[Any]				, _A  :		List[str] = False				, _A  :		List[Any] = True				, _A  :		Dict = None				, )    ->							List[str]:
								UpperCAmelCase_      :       List[Any]          =       sample
								UpperCAmelCase_      :       List[str]          =       self.encode(__snake_case      ).latent_dist
								if sample_posterior:
											UpperCAmelCase_      :       List[Any]          =       posterior.sample(generator=__snake_case      )
								else:
											UpperCAmelCase_      :       Dict          =       posterior.mode()
								UpperCAmelCase_      :       str          =       self.decode(__snake_case      ).sample

								if not return_dict:
											return (dec,)

								return DecoderOutput(sample=__snake_case      )






import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers import is_speech_available, is_vision_available
from transformers.testing_utils import require_torch


if is_vision_available():
 from transformers import TvltImageProcessor

if is_speech_available():
 from transformers import TvltFeatureExtractor

from transformers import TvltProcessor

@require_torch
class     A__   (       unittest.TestCase      ):







    """simple docstring"""

    def 						a_     (			self       ):
         snake_case											=							'''ZinengTang/tvlt-base'''
         snake_case											=							tempfile.mkdtemp()
    def 						a_     (			self     ,   **__snake_case       ):
         return TvltImageProcessor.from_pretrained(self.checkpoint     ,   **__snake_case       )
    def 						a_     (			self     ,   **__snake_case       ):
         return TvltFeatureExtractor.from_pretrained(self.checkpoint     ,   **__snake_case       )
    def 						a_     (			self       ):
         shutil.rmtree(self.tmpdirname       )
    def 						a_     (			self       ):
         snake_case											=							self.get_image_processor()
         snake_case											=							self.get_feature_extractor()

         snake_case											=							TvltProcessor(image_processor=__snake_case     ,   feature_extractor=__snake_case       )
         processor.save_pretrained(self.tmpdirname       )
         snake_case											=							TvltProcessor.from_pretrained(self.tmpdirname       )

         self.assertIsInstance(processor.feature_extractor     ,   __snake_case       )
         self.assertIsInstance(processor.image_processor     ,   __snake_case       )
    def 						a_     (			self       ):
         snake_case											=							self.get_image_processor()
         snake_case											=							self.get_feature_extractor()

         snake_case											=							TvltProcessor(image_processor=__snake_case     ,   feature_extractor=__snake_case       )

         snake_case											=							np.ones([1_2_0_0_0]       )

         snake_case											=							feature_extractor(__snake_case     ,   return_tensors='''np'''       )
         snake_case											=							processor(audio=__snake_case     ,   return_tensors='''np'''       )

         for key in audio_dict.keys():
              self.assertAlmostEqual(audio_dict[key].sum()     ,   input_processor[key].sum()     ,   delta=1E-2       )
    def 						a_     (			self       ):
         snake_case											=							self.get_image_processor()
         snake_case											=							self.get_feature_extractor()

         snake_case											=							TvltProcessor(image_processor=__snake_case     ,   feature_extractor=__snake_case       )

         snake_case											=							np.ones([3, 2_2_4, 2_2_4]       )

         snake_case											=							image_processor(__snake_case     ,   return_tensors='''np'''       )
         snake_case											=							processor(images=__snake_case     ,   return_tensors='''np'''       )

         for key in image_dict.keys():
              self.assertAlmostEqual(image_dict[key].sum()     ,   input_processor[key].sum()     ,   delta=1E-2       )
    def 						a_     (			self       ):
         snake_case											=							self.get_image_processor()
         snake_case											=							self.get_feature_extractor()

         snake_case											=							TvltProcessor(image_processor=__snake_case     ,   feature_extractor=__snake_case       )

         snake_case											=							np.ones([1_2_0_0_0]       )
         snake_case											=							np.ones([3, 2_2_4, 2_2_4]       )

         snake_case											=							processor(audio=__snake_case     ,   images=__snake_case       )

         self.assertListEqual(list(inputs.keys()       )     ,   ['''audio_values''', '''audio_mask''', '''pixel_values''', '''pixel_mask''']       )

         # test if it raises when no input is passed
         with pytest.raises(__snake_case       ):
              processor()




    def 						a_     (			self       ):
         snake_case											=							self.get_image_processor()
         snake_case											=							self.get_feature_extractor()

         snake_case											=							TvltProcessor(image_processor=__snake_case     ,   feature_extractor=__snake_case       )

         self.assertListEqual(
             processor.model_input_names     ,   image_processor.model_input_names + feature_extractor.model_input_names     ,   msg='''`processor` and `image_processor`+`feature_extractor` model input names do not match'''     ,   )




