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       lowercase__		(     UpperCamelCase_):
						UpperCamelCase_  =	42






class       lowercase__		(     nn.Module):


						def __init__( self			:       List[Any]					,	UpperCamelCase__			:       int=3					,	UpperCamelCase__			:       Optional[int]=3					,	UpperCamelCase__			:       Union[str, Any]=("DownEncoderBlock2D",)					,	UpperCamelCase__			:       Tuple=(64,)					,	UpperCamelCase__			:       List[str]=2					,	UpperCamelCase__			:       Union[str, Any]=32					,	UpperCamelCase__			:       str="silu"					,	UpperCamelCase__			:       int=True					,	):





												'''simple docstring'''





												super().__init__()
												SCREAMING_SNAKE_CASE							:					int           =							layers_per_block

												SCREAMING_SNAKE_CASE							:					Dict           =							torch.nn.Convad(
												    UpperCamelCase__					,	block_out_channels[0]					,	kernel_size=3					,	stride=1					,	padding=1					,	)

												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							None
												SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							nn.ModuleList([]		)

												# down
												SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							block_out_channels[0]
												for i, down_block_type in enumerate(UpperCamelCase__		):
																		SCREAMING_SNAKE_CASE							:					Optional[int]           =							output_channel
																		SCREAMING_SNAKE_CASE							:					List[str]           =							block_out_channels[i]
																		SCREAMING_SNAKE_CASE							:					Tuple           =							i == len(UpperCamelCase__		) - 1

																		SCREAMING_SNAKE_CASE							:					Any           =							get_down_block(
																		    UpperCamelCase__					,	num_layers=self.layers_per_block					,	in_channels=UpperCamelCase__					,	out_channels=UpperCamelCase__					,	add_downsample=not is_final_block					,	resnet_eps=1E-6					,	downsample_padding=0					,	resnet_act_fn=UpperCamelCase__					,	resnet_groups=UpperCamelCase__					,	attention_head_dim=UpperCamelCase__					,	temb_channels=UpperCamelCase__					,	)
																		self.down_blocks.append(UpperCamelCase__		)

												# mid
												SCREAMING_SNAKE_CASE							:					Optional[int]           =							UNetMidBlockaD(
												    in_channels=block_out_channels[-1]					,	resnet_eps=1E-6					,	resnet_act_fn=UpperCamelCase__					,	output_scale_factor=1					,	resnet_time_scale_shift='''default'''					,	attention_head_dim=block_out_channels[-1]					,	resnet_groups=UpperCamelCase__					,	temb_channels=UpperCamelCase__					,	)

												# out
												SCREAMING_SNAKE_CASE							:					Any           =							nn.GroupNorm(num_channels=block_out_channels[-1]					,	num_groups=UpperCamelCase__					,	eps=1E-6		)
												SCREAMING_SNAKE_CASE							:					Optional[int]           =							nn.SiLU()

												SCREAMING_SNAKE_CASE							:					int           =							2 * out_channels if double_z else out_channels
												SCREAMING_SNAKE_CASE							:					int           =							nn.Convad(block_out_channels[-1]					,	UpperCamelCase__					,	3					,	padding=1		)

												SCREAMING_SNAKE_CASE							:					List[Any]           =							False



						def  __A ( self			:       Tuple					,	UpperCamelCase__			:       List[str]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Any           =							x
												SCREAMING_SNAKE_CASE							:					Tuple           =							self.conv_in(UpperCamelCase__		)

												if self.training and self.gradient_checkpointing:

																		def create_custom_forward(UpperCamelCase__			:       Union[str, Any]		):
																								def custom_forward(*UpperCamelCase__			:       List[str]		):
																														return module(*UpperCamelCase__		)

																								return custom_forward

																		# down
																		if is_torch_version('''>='''					,	'''1.11.0'''		):
																								for down_block in self.down_blocks:
																														SCREAMING_SNAKE_CASE							:					Optional[int]           =							torch.utils.checkpoint.checkpoint(
																														    create_custom_forward(UpperCamelCase__		)					,	UpperCamelCase__					,	use_reentrant=UpperCamelCase__		)
																								# middle
																								SCREAMING_SNAKE_CASE							:					Any           =							torch.utils.checkpoint.checkpoint(
																								    create_custom_forward(self.mid_block		)					,	UpperCamelCase__					,	use_reentrant=UpperCamelCase__		)
																		else:
																								for down_block in self.down_blocks:
																														SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							torch.utils.checkpoint.checkpoint(create_custom_forward(UpperCamelCase__		)					,	UpperCamelCase__		)
																								# middle
																								SCREAMING_SNAKE_CASE							:					Tuple           =							torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block		)					,	UpperCamelCase__		)

												else:
																		# down
																		for down_block in self.down_blocks:
																								SCREAMING_SNAKE_CASE							:					str           =							down_block(UpperCamelCase__		)

																		# middle
																		SCREAMING_SNAKE_CASE							:					Optional[Any]           =							self.mid_block(UpperCamelCase__		)

												# post-process
												SCREAMING_SNAKE_CASE							:					Dict           =							self.conv_norm_out(UpperCamelCase__		)
												SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							self.conv_act(UpperCamelCase__		)
												SCREAMING_SNAKE_CASE							:					Any           =							self.conv_out(UpperCamelCase__		)

												return sample






class       lowercase__		(     nn.Module):


						def __init__( self			:       List[str]					,	UpperCamelCase__			:       int=3					,	UpperCamelCase__			:       Optional[Any]=3					,	UpperCamelCase__			:       Any=("UpDecoderBlock2D",)					,	UpperCamelCase__			:       Union[str, Any]=(64,)					,	UpperCamelCase__			:       List[Any]=2					,	UpperCamelCase__			:       Optional[int]=32					,	UpperCamelCase__			:       List[Any]="silu"					,	UpperCamelCase__			:       Optional[Any]="group"					,	):





												'''simple docstring'''





												super().__init__()
												SCREAMING_SNAKE_CASE							:					Dict           =							layers_per_block

												SCREAMING_SNAKE_CASE							:					List[Any]           =							nn.Convad(
												    UpperCamelCase__					,	block_out_channels[-1]					,	kernel_size=3					,	stride=1					,	padding=1					,	)

												SCREAMING_SNAKE_CASE							:					List[str]           =							None
												SCREAMING_SNAKE_CASE							:					Tuple           =							nn.ModuleList([]		)

												SCREAMING_SNAKE_CASE							:					int           =							in_channels if norm_type == '''spatial''' else None

												# mid
												SCREAMING_SNAKE_CASE							:					Any           =							UNetMidBlockaD(
												    in_channels=block_out_channels[-1]					,	resnet_eps=1E-6					,	resnet_act_fn=UpperCamelCase__					,	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=UpperCamelCase__					,	temb_channels=UpperCamelCase__					,	)

												# up
												SCREAMING_SNAKE_CASE							:					int           =							list(reversed(UpperCamelCase__		)		)
												SCREAMING_SNAKE_CASE							:					List[Any]           =							reversed_block_out_channels[0]
												for i, up_block_type in enumerate(UpperCamelCase__		):
																		SCREAMING_SNAKE_CASE							:					List[str]           =							output_channel
																		SCREAMING_SNAKE_CASE							:					Optional[int]           =							reversed_block_out_channels[i]

																		SCREAMING_SNAKE_CASE							:					int           =							i == len(UpperCamelCase__		) - 1

																		SCREAMING_SNAKE_CASE							:					Dict           =							get_up_block(
																		    UpperCamelCase__					,	num_layers=self.layers_per_block + 1					,	in_channels=UpperCamelCase__					,	out_channels=UpperCamelCase__					,	prev_output_channel=UpperCamelCase__					,	add_upsample=not is_final_block					,	resnet_eps=1E-6					,	resnet_act_fn=UpperCamelCase__					,	resnet_groups=UpperCamelCase__					,	attention_head_dim=UpperCamelCase__					,	temb_channels=UpperCamelCase__					,	resnet_time_scale_shift=UpperCamelCase__					,	)
																		self.up_blocks.append(UpperCamelCase__		)
																		SCREAMING_SNAKE_CASE							:					Optional[int]           =							output_channel

												# out
												if norm_type == "spatial":
																		SCREAMING_SNAKE_CASE							:					List[str]           =							SpatialNorm(block_out_channels[0]					,	UpperCamelCase__		)
												else:
																		SCREAMING_SNAKE_CASE							:					List[Any]           =							nn.GroupNorm(num_channels=block_out_channels[0]					,	num_groups=UpperCamelCase__					,	eps=1E-6		)
												SCREAMING_SNAKE_CASE							:					List[Any]           =							nn.SiLU()
												SCREAMING_SNAKE_CASE							:					str           =							nn.Convad(block_out_channels[0]					,	UpperCamelCase__					,	3					,	padding=1		)

												SCREAMING_SNAKE_CASE							:					Any           =							False



						def  __A ( self			:       Optional[int]					,	UpperCamelCase__			:       Any					,	UpperCamelCase__			:       Dict=None		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					int           =							z
												SCREAMING_SNAKE_CASE							:					Any           =							self.conv_in(UpperCamelCase__		)

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

																		def create_custom_forward(UpperCamelCase__			:       Optional[int]		):
																								def custom_forward(*UpperCamelCase__			:       Dict		):
																														return module(*UpperCamelCase__		)

																								return custom_forward

																		if is_torch_version('''>='''					,	'''1.11.0'''		):
																								# middle
																								SCREAMING_SNAKE_CASE							:					List[str]           =							torch.utils.checkpoint.checkpoint(
																								    create_custom_forward(self.mid_block		)					,	UpperCamelCase__					,	UpperCamelCase__					,	use_reentrant=UpperCamelCase__		)
																								SCREAMING_SNAKE_CASE							:					Optional[int]           =							sample.to(UpperCamelCase__		)

																								# up
																								for up_block in self.up_blocks:
																														SCREAMING_SNAKE_CASE							:					List[str]           =							torch.utils.checkpoint.checkpoint(
																														    create_custom_forward(UpperCamelCase__		)					,	UpperCamelCase__					,	UpperCamelCase__					,	use_reentrant=UpperCamelCase__		)
																		else:
																								# middle
																								SCREAMING_SNAKE_CASE							:					Dict           =							torch.utils.checkpoint.checkpoint(
																								    create_custom_forward(self.mid_block		)					,	UpperCamelCase__					,	UpperCamelCase__		)
																								SCREAMING_SNAKE_CASE							:					List[str]           =							sample.to(UpperCamelCase__		)

																								# up
																								for up_block in self.up_blocks:
																														SCREAMING_SNAKE_CASE							:					Tuple           =							torch.utils.checkpoint.checkpoint(create_custom_forward(UpperCamelCase__		)					,	UpperCamelCase__					,	UpperCamelCase__		)
												else:
																		# middle
																		SCREAMING_SNAKE_CASE							:					Optional[Any]           =							self.mid_block(UpperCamelCase__					,	UpperCamelCase__		)
																		SCREAMING_SNAKE_CASE							:					Optional[int]           =							sample.to(UpperCamelCase__		)

																		# up
																		for up_block in self.up_blocks:
																								SCREAMING_SNAKE_CASE							:					Any           =							up_block(UpperCamelCase__					,	UpperCamelCase__		)

        # post-process
												if latent_embeds is None:
																		SCREAMING_SNAKE_CASE							:					str           =							self.conv_norm_out(UpperCamelCase__		)
												else:
																		SCREAMING_SNAKE_CASE							:					Dict           =							self.conv_norm_out(UpperCamelCase__					,	UpperCamelCase__		)
												SCREAMING_SNAKE_CASE							:					str           =							self.conv_act(UpperCamelCase__		)
												SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							self.conv_out(UpperCamelCase__		)

												return sample






class       lowercase__		(     nn.Module):


						def __init__( self			:       int					,	UpperCamelCase__			:       Union[str, Any]					,	UpperCamelCase__			:       Dict					,	UpperCamelCase__			:       List[Any]					,	UpperCamelCase__			:       Optional[Any]=None					,	UpperCamelCase__			:       Tuple="random"					,	UpperCamelCase__			:       Dict=False					,	UpperCamelCase__			:       Optional[Any]=True		):





												'''simple docstring'''





												super().__init__()
												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							n_e
												SCREAMING_SNAKE_CASE							:					Tuple           =							vq_embed_dim
												SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							beta
												SCREAMING_SNAKE_CASE							:					List[str]           =							legacy

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

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

												SCREAMING_SNAKE_CASE							:					int           =							sane_index_shape


						def  __A ( self			:       Tuple					,	UpperCamelCase__			:       Union[str, Any]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							inds.shape
												assert len(UpperCamelCase__		) > 1
												SCREAMING_SNAKE_CASE							:					Optional[int]           =							inds.reshape(ishape[0]					,	-1		)
												SCREAMING_SNAKE_CASE							:					int           =							self.used.to(UpperCamelCase__		)
												SCREAMING_SNAKE_CASE							:					List[str]           =							(inds[:, :, None] == used[None, None, ...]).long()
												SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							match.argmax(-1		)
												SCREAMING_SNAKE_CASE							:					List[Any]           =							match.sum(2		) < 1
												if self.unknown_index == "random":
																		SCREAMING_SNAKE_CASE							:					Optional[Any]           =							torch.randint(0					,	self.re_embed					,	size=new[unknown].shape		).to(device=new.device		)
												else:
																		SCREAMING_SNAKE_CASE							:					str           =							self.unknown_index
												return new.reshape(UpperCamelCase__		)


						def  __A ( self			:       List[str]					,	UpperCamelCase__			:       Dict		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							inds.shape
												assert len(UpperCamelCase__		) > 1
												SCREAMING_SNAKE_CASE							:					List[Any]           =							inds.reshape(ishape[0]					,	-1		)
												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							self.used.to(UpperCamelCase__		)
												if self.re_embed > self.used.shape[0]:  # extra token
																		SCREAMING_SNAKE_CASE							:					Any           =							0  # simply set to zero
												SCREAMING_SNAKE_CASE							:					Any           =							torch.gather(used[None, :][inds.shape[0] * [0], :]					,	1					,	UpperCamelCase__		)
												return back.reshape(UpperCamelCase__		)


						def  __A ( self			:       int					,	UpperCamelCase__			:       Dict		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					List[Any]           =							z.permute(0					,	2					,	3					,	1		).contiguous()
												SCREAMING_SNAKE_CASE							:					Any           =							z.view(-1					,	self.vq_embed_dim		)

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

												SCREAMING_SNAKE_CASE							:					List[Any]           =							self.embedding(UpperCamelCase__		).view(z.shape		)
												SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							None
												SCREAMING_SNAKE_CASE							:					Any           =							None

												# compute loss for embedding
												if not self.legacy:
																		SCREAMING_SNAKE_CASE							:					Optional[int]           =							self.beta * torch.mean((z_q.detach() - z) ** 2		) + torch.mean((z_q - z.detach()) ** 2		)
												else:
																		SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							torch.mean((z_q.detach() - z) ** 2		) + self.beta * torch.mean((z_q - z.detach()) ** 2		)

												# preserve gradients
												SCREAMING_SNAKE_CASE							:					List[Any]           =							z + (z_q - z).detach()

												# reshape back to match original input shape
												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							z_q.permute(0					,	3					,	1					,	2		).contiguous()

												if self.remap is not None:
																		SCREAMING_SNAKE_CASE							:					Optional[Any]           =							min_encoding_indices.reshape(z.shape[0]					,	-1		)  # add batch axis
																		SCREAMING_SNAKE_CASE							:					Optional[Any]           =							self.remap_to_used(UpperCamelCase__		)
																		SCREAMING_SNAKE_CASE							:					str           =							min_encoding_indices.reshape(-1					,	1		)  # flatten

												if self.sane_index_shape:
																		SCREAMING_SNAKE_CASE							:					List[Any]           =							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  __A ( self			:       Union[str, Any]					,	UpperCamelCase__			:       Optional[Any]					,	UpperCamelCase__			:       Any		):





												'''simple docstring'''





												if self.remap is not None:
																		SCREAMING_SNAKE_CASE							:					Any           =							indices.reshape(shape[0]					,	-1		)  # add batch axis
																		SCREAMING_SNAKE_CASE							:					List[str]           =							self.unmap_to_all(UpperCamelCase__		)
																		SCREAMING_SNAKE_CASE							:					int           =							indices.reshape(-1		)  # flatten again

												# get quantized latent vectors
												SCREAMING_SNAKE_CASE							:					Any           =							self.embedding(UpperCamelCase__		)

												if shape is not None:
																		SCREAMING_SNAKE_CASE							:					List[Any]           =							z_q.view(UpperCamelCase__		)
																		# reshape back to match original input shape
																		SCREAMING_SNAKE_CASE							:					Dict           =							z_q.permute(0					,	3					,	1					,	2		).contiguous()

												return z_q






class       lowercase__		(     UpperCamelCase_):


						def __init__( self			:       Union[str, Any]					,	UpperCamelCase__			:       Any					,	UpperCamelCase__			:       List[str]=False		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					List[Any]           =							parameters
												SCREAMING_SNAKE_CASE       ,						SCREAMING_SNAKE_CASE							:					Dict           =							torch.chunk(UpperCamelCase__					,	2					,	dim=1		)
												SCREAMING_SNAKE_CASE							:					Any           =							torch.clamp(self.logvar					,	-30.0					,	20.0		)
												SCREAMING_SNAKE_CASE							:					Optional[int]           =							deterministic
												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							torch.exp(0.5 * self.logvar		)
												SCREAMING_SNAKE_CASE							:					Tuple           =							torch.exp(self.logvar		)
												if self.deterministic:
																		SCREAMING_SNAKE_CASE							:					int           =							torch.zeros_like(
																		    self.mean					,	device=self.parameters.device					,	dtype=self.parameters.dtype		)


						def  __A ( self			:       Any					,	UpperCamelCase__			:       Optional[torch.Generator] = None		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					int           =							randn_tensor(
												    self.mean.shape					,	generator=UpperCamelCase__					,	device=self.parameters.device					,	dtype=self.parameters.dtype		)
												SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							self.mean + self.std * sample
												return x


						def  __A ( self			:       str					,	UpperCamelCase__			:       str=None		):





												'''simple docstring'''





												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  __A ( self			:       List[Any]					,	UpperCamelCase__			:       Tuple					,	UpperCamelCase__			:       str=[1, 2, 3]		):





												'''simple docstring'''





												if self.deterministic:
																		return torch.Tensor([0.0]		)
												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							np.log(2.0 * np.pi		)
												return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean					,	2		) / self.var					,	dim=UpperCamelCase__		)



						def  __A ( self			:       List[str]		):





												'''simple docstring'''





												return self.mean






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

__UpperCamelCase	: Dict    		=  random.Random()
def    A			(					_lowercase	,						_lowercase=1.0	,						_lowercase=None	,						_lowercase=None		):
						if rng is None:
												SCREAMING_SNAKE_CASE							:					Any           =							global_rng

						SCREAMING_SNAKE_CASE							:					int           =							[]
						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			:       List[Any]					,	UpperCamelCase__			:       Optional[int]					,	UpperCamelCase__			:       str=7					,	UpperCamelCase__			:       Any=400					,	UpperCamelCase__			:       List[str]=2000					,	UpperCamelCase__			:       List[Any]=2048					,	UpperCamelCase__			:       Any=128					,	UpperCamelCase__			:       Union[str, Any]=1					,	UpperCamelCase__			:       List[Any]=512					,	UpperCamelCase__			:       str=30					,	UpperCamelCase__			:       Tuple=4_4100					,	):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					str           =							parent
												SCREAMING_SNAKE_CASE							:					str           =							batch_size
												SCREAMING_SNAKE_CASE							:					str           =							min_seq_length
												SCREAMING_SNAKE_CASE							:					Dict           =							max_seq_length
												SCREAMING_SNAKE_CASE							:					Optional[int]           =							(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							spectrogram_length
												SCREAMING_SNAKE_CASE							:					Optional[int]           =							feature_size
												SCREAMING_SNAKE_CASE							:					Tuple           =							num_audio_channels
												SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							hop_length
												SCREAMING_SNAKE_CASE							:					List[Any]           =							chunk_length
												SCREAMING_SNAKE_CASE							:					str           =							sampling_rate


						def  __A ( self			:       Optional[Any]		):





												'''simple docstring'''





												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			:       Tuple					,	UpperCamelCase__			:       List[str]=False					,	UpperCamelCase__			:       Optional[int]=False		):





												'''simple docstring'''





												def _flatten(UpperCamelCase__			:       str		):
																		return list(itertools.chain(*UpperCamelCase__		)		)

												if equal_length:
																		SCREAMING_SNAKE_CASE							:					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
																		SCREAMING_SNAKE_CASE							:					int           =							[
																		    floats_list((x, self.feature_size)		)
																		    for x in range(self.min_seq_length					,	self.max_seq_length					,	self.seq_length_diff		)
																		]
												if numpify:
																		SCREAMING_SNAKE_CASE							:					Optional[Any]           =							[np.asarray(UpperCamelCase__		) for x in speech_inputs]
												return speech_inputs






@require_torch
@require_torchaudio
class       lowercase__		(     UpperCamelCase_			,							unittest.TestCase):
						UpperCamelCase_  =	TvltFeatureExtractor


						def  __A ( self			:       List[str]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Dict           =							TvltFeatureExtractionTester(self		)


						def  __A ( self			:       Union[str, Any]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Any           =							self.feature_extraction_class(**self.feat_extract_dict		)
												self.assertTrue(hasattr(UpperCamelCase__					,	'''spectrogram_length'''		)		)
												self.assertTrue(hasattr(UpperCamelCase__					,	'''feature_size'''		)		)
												self.assertTrue(hasattr(UpperCamelCase__					,	'''num_audio_channels'''		)		)
												self.assertTrue(hasattr(UpperCamelCase__					,	'''hop_length'''		)		)
												self.assertTrue(hasattr(UpperCamelCase__					,	'''chunk_length'''		)		)
												self.assertTrue(hasattr(UpperCamelCase__					,	'''sampling_rate'''		)		)


						def  __A ( self			:       Union[str, Any]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							self.feature_extraction_class(**self.feat_extract_dict		)

												with tempfile.TemporaryDirectory() as tmpdirname:
																		SCREAMING_SNAKE_CASE							:					Optional[Any]           =							feat_extract_first.save_pretrained(UpperCamelCase__		)[0]
																		check_json_file_has_correct_format(UpperCamelCase__		)
																		SCREAMING_SNAKE_CASE							:					List[Any]           =							self.feature_extraction_class.from_pretrained(UpperCamelCase__		)

												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							feat_extract_first.to_dict()
												SCREAMING_SNAKE_CASE							:					str           =							feat_extract_second.to_dict()
												SCREAMING_SNAKE_CASE							:					Optional[int]           =							dict_first.pop('''mel_filters'''		)
												SCREAMING_SNAKE_CASE							:					Optional[int]           =							dict_second.pop('''mel_filters'''		)
												self.assertTrue(np.allclose(UpperCamelCase__					,	UpperCamelCase__		)		)
												self.assertEqual(UpperCamelCase__					,	UpperCamelCase__		)


						def  __A ( self			:       int		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Dict           =							self.feature_extraction_class(**self.feat_extract_dict		)

												with tempfile.TemporaryDirectory() as tmpdirname:
																		SCREAMING_SNAKE_CASE							:					Tuple           =							os.path.join(UpperCamelCase__					,	'''feat_extract.json'''		)
																		feat_extract_first.to_json_file(UpperCamelCase__		)
																		SCREAMING_SNAKE_CASE							:					List[str]           =							self.feature_extraction_class.from_json_file(UpperCamelCase__		)

												SCREAMING_SNAKE_CASE							:					Any           =							feat_extract_first.to_dict()
												SCREAMING_SNAKE_CASE							:					str           =							feat_extract_second.to_dict()
												SCREAMING_SNAKE_CASE							:					int           =							dict_first.pop('''mel_filters'''		)
												SCREAMING_SNAKE_CASE							:					Any           =							dict_second.pop('''mel_filters'''		)
												self.assertTrue(np.allclose(UpperCamelCase__					,	UpperCamelCase__		)		)
												self.assertEqual(UpperCamelCase__					,	UpperCamelCase__		)


						def  __A ( self			:       Any		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					int           =							self.feature_extraction_class(**self.feat_extract_dict		)

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

												# Test not batched input
												SCREAMING_SNAKE_CASE							:					List[Any]           =							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
												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							feature_extractor(UpperCamelCase__					,	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
												SCREAMING_SNAKE_CASE							:					List[str]           =							feature_extractor(
												    UpperCamelCase__					,	return_tensors='''np'''					,	sampling_rate=4_4100					,	mask_audio=UpperCamelCase__		).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.
												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							[floats_list((1, x)		)[0] for x in (800, 800, 800)]
												SCREAMING_SNAKE_CASE							:					Dict           =							np.asarray(UpperCamelCase__		)
												SCREAMING_SNAKE_CASE							:					str           =							feature_extractor(UpperCamelCase__					,	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			:       Optional[int]					,	UpperCamelCase__			:       int		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					List[str]           =							load_dataset('''hf-internal-testing/librispeech_asr_dummy'''					,	'''clean'''					,	split='''validation'''		)
												# automatic decoding with librispeech
												SCREAMING_SNAKE_CASE							:					Dict           =							ds.sort('''id'''		).select(range(UpperCamelCase__		)		)[:num_samples]['''audio''']

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



						def  __A ( self			:       List[Any]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							self._load_datasamples(1		)
												SCREAMING_SNAKE_CASE							:					int           =							TvltFeatureExtractor()
												SCREAMING_SNAKE_CASE							:					List[Any]           =							feature_extractor(UpperCamelCase__					,	return_tensors='''pt'''		).audio_values

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

												SCREAMING_SNAKE_CASE							:					str           =							torch.tensor([[-0.3032, -0.2708], [-0.4434, -0.4007]]		)
												self.assertTrue(torch.allclose(audio_values[0, 0, :2, :2]					,	UpperCamelCase__					,	atol=1E-4		)		)





