





import itertools
import random
import unittest

import numpy as np

from transformers import ASTFeatureExtractor
from transformers.testing_utils import require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


_SCREAMING_SNAKE_CASE 		=						random.Random()

if is_torch_available():
						import torch


def 							SCREAMING_SNAKE_CASE__  (      __a				,	__a=1.0				,	__a=None				,	__a=None				):
	if rng is None:
		snake_case_						:     str     =		global_rng

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

	return values
class SCREAMING_SNAKE_CASE_  (   unittest.TestCase   ):






						def __init__( self : Dict						,				_A : Optional[int]						,				_A : Dict=7						,				_A : Dict=400						,				_A : str=2000						,				_A : Optional[Any]=1						,				_A : int=0.0						,				_A : List[Any]=16000						,				_A : str=True						,				_A : Optional[Any]=True						,				)							->      Any:







							"""simple docstring"""


							snake_case_						:     Tuple     =		parent
							snake_case_						:     Optional[int]     =		batch_size
							snake_case_						:     Dict     =		min_seq_length
							snake_case_						:     Optional[Any]     =		max_seq_length
							snake_case_						:     List[Any]     =		(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
							snake_case_						:     Optional[int]     =		feature_size
							snake_case_						:     Optional[Any]     =		padding_value
							snake_case_						:     Any     =		sampling_rate
							snake_case_						:     Dict     =		return_attention_mask
							snake_case_						:     Any     =		do_normalize






						def UpperCAmelCase_			( self : Dict		)							->      str:







							"""simple docstring"""


							return {
							    "feature_size": self.feature_size,
							    "padding_value": self.padding_value,
							    "sampling_rate": self.sampling_rate,
							    "return_attention_mask": self.return_attention_mask,
							    "do_normalize": self.do_normalize,
							}






						def UpperCAmelCase_			( self : int						,				_A : Optional[Any]=False						,				_A : str=False		)							->      Dict:







							"""simple docstring"""


							def _flatten(_A : Optional[int]		):
								return list(itertools.chain(*_A		)		)

							if equal_length:
								snake_case_						:     str     =		floats_list((self.batch_size, self.max_seq_length)		)
							else:
								# make sure that inputs increase in size
								snake_case_						:     List[str]     =		[
								    _flatten(floats_list((x, self.feature_size)		)		)
								    for x in range(self.min_seq_length						,				self.max_seq_length						,				self.seq_length_diff		)
								]

							if numpify:
								snake_case_						:     Optional[Any]     =		[np.asarray(_A		) for x in speech_inputs]

							return speech_inputs
@require_torch
@require_torchaudio
class SCREAMING_SNAKE_CASE_  (   snake_case_    ,			unittest.TestCase   ):
						__magic_name__:     Dict	 =    ASTFeatureExtractor






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







							"""simple docstring"""


							snake_case_						:     str     =		ASTFeatureExtractionTester(self		)






						def UpperCAmelCase_			( self : Optional[Any]		)							->      Optional[Any]:







							"""simple docstring"""


							snake_case_						:     int     =		self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()		)
							# create three inputs of length 800, 1000, and 1200
							snake_case_						:     Any     =		[floats_list((1, x)		)[0] for x in range(800						,				1400						,				200		)]
							snake_case_						:     List[Any]     =		[np.asarray(_A		) for speech_input in speech_inputs]

							# Test not batched input
							snake_case_						:     int     =		feat_extract(speech_inputs[0]						,				return_tensors='np'		).input_values
							snake_case_						:     Dict     =		feat_extract(np_speech_inputs[0]						,				return_tensors='np'		).input_values
							self.assertTrue(np.allclose(_A						,				_A						,				atol=1E-3		)		)

							# Test batched
							snake_case_						:     Tuple     =		feat_extract(_A						,				padding=_A						,				return_tensors='np'		).input_values
							snake_case_						:     Union[str, Any]     =		feat_extract(_A						,				padding=_A						,				return_tensors='np'		).input_values
							for enc_seq_a, enc_seq_a in zip(_A						,				_A		):
								self.assertTrue(np.allclose(_A						,				_A						,				atol=1E-3		)		)

							# Test 2-D numpy arrays are batched.
							snake_case_						:     Dict     =		[floats_list((1, x)		)[0] for x in (800, 800, 800)]
							snake_case_						:     Tuple     =		np.asarray(_A		)
							snake_case_						:     Tuple     =		feat_extract(_A						,				return_tensors='np'		).input_values
							snake_case_						:     Dict     =		feat_extract(_A						,				return_tensors='np'		).input_values
							for enc_seq_a, enc_seq_a in zip(_A						,				_A		):
								self.assertTrue(np.allclose(_A						,				_A						,				atol=1E-3		)		)






						@require_torch
						def UpperCAmelCase_			( self : Tuple		)							->      Union[str, Any]:







							"""simple docstring"""


							import torch

							snake_case_						:     str     =		self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()		)
							snake_case_						:     Tuple     =		np.random.rand(100		).astype(np.floataa		)
							snake_case_						:     List[str]     =		np_speech_inputs.tolist()

							for inputs in [py_speech_inputs, np_speech_inputs]:
								snake_case_						:     Dict     =		feature_extractor.pad([{'input_values': inputs}]						,				return_tensors='np'		)
								self.assertTrue(np_processed.input_values.dtype == np.floataa		)
								snake_case_						:     Any     =		feature_extractor.pad([{'input_values': inputs}]						,				return_tensors='pt'		)
								self.assertTrue(pt_processed.input_values.dtype == torch.floataa		)






						def UpperCAmelCase_			( self : Union[str, Any]						,				_A : Union[str, Any]		)							->      Optional[int]:







							"""simple docstring"""


							from datasets import load_dataset

							snake_case_						:     Optional[Any]     =		load_dataset('hf-internal-testing/librispeech_asr_dummy'						,				'clean'						,				split='validation'		)
							# automatic decoding with librispeech
							snake_case_						:     Optional[Any]     =		ds.sort('id'		).select(range(_A		)		)[:num_samples]['audio']

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






						@require_torch
						def UpperCAmelCase_			( self : List[str]		)							->      int:







							"""simple docstring"""


							snake_case_						:     Optional[int]     =		torch.tensor(
							    [-0.9_8_9_4, -1.2_7_7_6, -0.9_0_6_6, -1.2_7_7_6, -0.9_3_4_9, -1.2_6_0_9, -1.0_3_8_6, -1.2_7_7_6,
							     -1.1_5_6_1, -1.2_7_7_6, -1.2_0_5_2, -1.2_7_2_3, -1.2_1_9_0, -1.2_1_3_2, -1.2_7_7_6, -1.1_1_3_3,
							     -1.1_9_5_3, -1.1_3_4_3, -1.1_5_8_4, -1.2_2_0_3, -1.1_7_7_0, -1.2_4_7_4, -1.2_3_8_1, -1.1_9_3_6,
							     -0.9_2_7_0, -0.8_3_1_7, -0.8_0_4_9, -0.7_7_0_6, -0.7_5_6_5, -0.7_8_6_9]		)
							# fmt: on

							snake_case_						:     int     =		self._load_datasamples(1		)
							snake_case_						:     Dict     =		ASTFeatureExtractor()
							snake_case_						:     Any     =		feature_extractor(_A						,				return_tensors='pt'		).input_values
							self.assertEquals(input_values.shape						,				(1, 1024, 128)		)
							self.assertTrue(torch.allclose(input_values[0, 0, :30]						,				_A						,				atol=1E-4		)		)








import torch
from torch import nn

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin
class SCREAMING_SNAKE_CASE_  (   snake_case_    ,			snake_case_   ):






						@register_to_config
						def __init__( self : Optional[Any]						,				*,
						    _A : int = 4						,				_A : int = 768						,				_A : int						,				_A : Tuple						,				)							->      Dict:







							"""simple docstring"""


							super().__init__()

							snake_case_						:     int     =		nn.Parameter(torch.zeros(_A		)		)

							# parameters for additional clip time embeddings
							snake_case_						:     Tuple     =		nn.Linear(_A						,				_A		)
							snake_case_						:     List[Any]     =		nn.Linear(_A						,				_A		)

							# parameters for encoder hidden states
							snake_case_						:     Union[str, Any]     =		clip_extra_context_tokens
							snake_case_						:     str     =		nn.Linear(
							    _A						,				self.clip_extra_context_tokens * cross_attention_dim		)
							snake_case_						:     Any     =		nn.Linear(_A						,				_A		)
							snake_case_						:     Tuple     =		nn.LayerNorm(_A		)






						def UpperCAmelCase_			( self : List[str]						,				*, _A : Tuple						,				_A : List[Any]						,				_A : str						,				_A : Optional[Any]		)							->      List[Any]:







							"""simple docstring"""


							if do_classifier_free_guidance:
								# Add the classifier free guidance embeddings to the image embeddings
								snake_case_						:     Optional[int]     =		image_embeddings.shape[0]
								snake_case_						:     Optional[Any]     =		self.learned_classifier_free_guidance_embeddings.unsqueeze(0		)
								snake_case_						:     Optional[Any]     =		classifier_free_guidance_embeddings.expand(
								    _A						,				-1		)
								snake_case_						:     Any     =		torch.cat([classifier_free_guidance_embeddings, image_embeddings]						,				dim=0		)

							# The image embeddings batch size and the text embeddings batch size are equal
							assert image_embeddings.shape[0] == prompt_embeds.shape[0]

							snake_case_						:     str     =		prompt_embeds.shape[0]

							# "Specifically, we modify the architecture described in Nichol et al. (2021) by projecting and
							# adding CLIP embeddings to the existing timestep embedding, ...
							snake_case_						:     str     =		self.embedding_proj(_A		)
							snake_case_						:     Dict     =		self.clip_image_embeddings_project_to_time_embeddings(_A		)
							snake_case_						:     Tuple     =		time_projected_image_embeddings + time_projected_prompt_embeds

							# ... and by projecting CLIP embeddings into four
							# extra tokens of context that are concatenated to the sequence of outputs from the GLIDE text encoder"
							snake_case_						:     List[str]     =		self.clip_extra_context_tokens_proj(_A		)
							snake_case_						:     Optional[Any]     =		clip_extra_context_tokens.reshape(_A						,				-1						,				self.clip_extra_context_tokens		)
							snake_case_						:     int     =		clip_extra_context_tokens.permute(0						,				2						,				1		)

							snake_case_						:     Optional[int]     =		self.encoder_hidden_states_proj(_A		)
							snake_case_						:     Any     =		self.text_encoder_hidden_states_norm(_A		)
							snake_case_						:     Dict     =		torch.cat([clip_extra_context_tokens, text_encoder_hidden_states]						,				dim=1		)

							return text_encoder_hidden_states, additive_clip_time_embeddings

