


'''simple docstring'''
from dataclasses import dataclass
from typing import Any, Dict, Optional

import torch
import torch.nn.functional as F
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..models.embeddings import ImagePositionalEmbeddings
from ..utils import BaseOutput, deprecate
from .attention import BasicTransformerBlock
from .embeddings import PatchEmbed
from .modeling_utils import ModelMixin






@dataclass
class 				__lowerCAmelCase      (      __UpperCAmelCase	):

						"""simple docstring"""






						_UpperCAmelCase		:					Optional[Any]      	=42






class 				__lowerCAmelCase      (      __UpperCAmelCase							, __UpperCAmelCase	):

						"""simple docstring"""






						@register_to_config
						def __init__(     self       :			Optional[Any]   ,     lowerCAmelCase       :			Optional[Any] = 16   ,     lowerCAmelCase       :			Tuple = 88   ,     lowerCAmelCase       :			int = None   ,     lowerCAmelCase       :			List[str] = None   ,     lowerCAmelCase       :			Dict = 1   ,     lowerCAmelCase       :			str = 0.0   ,     lowerCAmelCase       :			Union[str, Any] = 32   ,     lowerCAmelCase       :			Optional[int] = None   ,     lowerCAmelCase       :			Union[str, Any] = False   ,     lowerCAmelCase       :			List[Any] = None   ,     lowerCAmelCase       :			Any = None   ,     lowerCAmelCase       :			Optional[Any] = None   ,     lowerCAmelCase       :			List[str] = "geglu"   ,     lowerCAmelCase       :			Union[str, Any] = None   ,     lowerCAmelCase       :			List[Any] = False   ,     lowerCAmelCase       :			Dict = False   ,     lowerCAmelCase       :			Any = False   ,     lowerCAmelCase       :			str = "layer_norm"   ,     lowerCAmelCase       :			Optional[Any] = True   ,     ):
													super().__init__()
													A_ 						=		use_linear_projection
													A_ 						=		num_attention_heads
													A_ 						=		attention_head_dim
													A_ 						=		num_attention_heads * attention_head_dim

													# 1. Transformer2DModel can process both standard continuous images of shape `(batch_size, num_channels, width, height)` as well as quantized image embeddings of shape `(batch_size, num_image_vectors)`
													# Define whether input is continuous or discrete depending on configuration
													A_ 						=		(in_channels is not None) and (patch_size is None)
													A_ 						=		num_vector_embeds is not None
													A_ 						=		in_channels is not None and patch_size is not None

													if norm_type == "layer_norm" and num_embeds_ada_norm is not None:
																				A_ 						=		(
																				    F"The configuration file of this model: {self.__class__} is outdated. `norm_type` is either not set or"
																				    ''' incorrectly set to `\'layer_norm\'`.Make sure to set `norm_type` to `\'ada_norm\'` in the config.'''
																				    ''' Please make sure to update the config accordingly as leaving `norm_type` might led to incorrect'''
																				    ''' results in future versions. If you have downloaded this checkpoint from the Hugging Face Hub, it'''
																				    ''' would be very nice if you could open a Pull request for the `transformer/config.json` file'''
																				)
																				deprecate("norm_type!=num_embeds_ada_norm"   ,     "1.0.0"   ,     lowercase_   ,     standard_warn=lowercase_  )
																				A_ 						=		'''ada_norm'''

													if self.is_input_continuous and self.is_input_vectorized:
																				raise ValueError(
																				    F"Cannot define both `in_channels`: {in_channels} and `num_vector_embeds`: {num_vector_embeds}. Make"
																				    " sure that either `in_channels` or `num_vector_embeds` is None."  )
													elif self.is_input_vectorized and self.is_input_patches:
																				raise ValueError(
																				    F"Cannot define both `num_vector_embeds`: {num_vector_embeds} and `patch_size`: {patch_size}. Make"
																				    " sure that either `num_vector_embeds` or `num_patches` is None."  )
													elif not self.is_input_continuous and not self.is_input_vectorized and not self.is_input_patches:
																				raise ValueError(
																				    F"Has to define `in_channels`: {in_channels}, `num_vector_embeds`: {num_vector_embeds}, or patch_size:"
																				    F" {patch_size}. Make sure that `in_channels`, `num_vector_embeds` or `num_patches` is not None."  )

													# 2. Define input layers
													if self.is_input_continuous:
																				A_ 						=		in_channels

																				A_ 						=		torch.nn.GroupNorm(num_groups=lowercase_   ,     num_channels=lowercase_   ,     eps=1e-6   ,     affine=lowercase_  )
																				if use_linear_projection:
																											A_ 						=		nn.Linear(lowercase_   ,     lowercase_  )
																				else:
																											A_ 						=		nn.Convad(lowercase_   ,     lowercase_   ,     kernel_size=1   ,     stride=1   ,     padding=0  )
													elif self.is_input_vectorized:
																				assert sample_size is not None, "Transformer2DModel over discrete input must provide sample_size"
																				assert num_vector_embeds is not None, "Transformer2DModel over discrete input must provide num_embed"

																				A_ 						=		sample_size
																				A_ 						=		sample_size
																				A_ 						=		num_vector_embeds
																				A_ 						=		self.height * self.width

																				A_ 						=		ImagePositionalEmbeddings(
																				    num_embed=lowercase_   ,     embed_dim=lowercase_   ,     height=self.height   ,     width=self.width  )
													elif self.is_input_patches:
																				assert sample_size is not None, "Transformer2DModel over patched input must provide sample_size"

																				A_ 						=		sample_size
																				A_ 						=		sample_size

																				A_ 						=		patch_size
																				A_ 						=		PatchEmbed(
																				    height=lowercase_   ,     width=lowercase_   ,     patch_size=lowercase_   ,     in_channels=lowercase_   ,     embed_dim=lowercase_   ,     )

													# 3. Define transformers blocks
													A_ 						=		nn.ModuleList(
													    [
													        BasicTransformerBlock(
													            lowercase_   ,     lowercase_   ,     lowercase_   ,     dropout=lowercase_   ,     cross_attention_dim=lowercase_   ,     activation_fn=lowercase_   ,     num_embeds_ada_norm=lowercase_   ,     attention_bias=lowercase_   ,     only_cross_attention=lowercase_   ,     upcast_attention=lowercase_   ,     norm_type=lowercase_   ,     norm_elementwise_affine=lowercase_   ,     )
													        for d in range(lowercase_  )
													    ]  )

													# 4. Define output layers
													A_ 						=		in_channels if out_channels is None else out_channels
													if self.is_input_continuous:
																				# TODO: should use out_channels for continuous projections
																				if use_linear_projection:
																											A_ 						=		nn.Linear(lowercase_   ,     lowercase_  )
																				else:
																											A_ 						=		nn.Convad(lowercase_   ,     lowercase_   ,     kernel_size=1   ,     stride=1   ,     padding=0  )
													elif self.is_input_vectorized:
																				A_ 						=		nn.LayerNorm(lowercase_  )
																				A_ 						=		nn.Linear(lowercase_   ,     self.num_vector_embeds - 1  )
													elif self.is_input_patches:
																				A_ 						=		nn.LayerNorm(lowercase_   ,     elementwise_affine=lowercase_   ,     eps=1e-6  )
																				A_ 						=		nn.Linear(lowercase_   ,     2 * inner_dim  )
																				A_ 						=		nn.Linear(lowercase_   ,     patch_size * patch_size * self.out_channels  )






						def 							_UpperCAmelCase					(     self       :			str   ,     lowerCAmelCase       :			int   ,     lowerCAmelCase       :			int = None   ,     lowerCAmelCase       :			List[Any] = None   ,     lowerCAmelCase       :			Dict = None   ,     lowerCAmelCase       :			Any = None   ,     lowerCAmelCase       :			Dict = None   ,     lowerCAmelCase       :			int = None   ,     lowerCAmelCase       :			List[str] = True   ,     ):

													if attention_mask is not None and attention_mask.ndim == 2:
																				# assume that mask is expressed as:
																				#   (1 = keep,      0 = discard)
																				# convert mask into a bias that can be added to attention scores:
																				#       (keep = +0,     discard = -10000.0)
																				A_ 						=		(1 - attention_mask.to(hidden_states.dtype  )) * -1_00_00.0
																				A_ 						=		attention_mask.unsqueeze(1  )

													# convert encoder_attention_mask to a bias the same way we do for attention_mask
													if encoder_attention_mask is not None and encoder_attention_mask.ndim == 2:
																				A_ 						=		(1 - encoder_attention_mask.to(hidden_states.dtype  )) * -1_00_00.0
																				A_ 						=		encoder_attention_mask.unsqueeze(1  )

													# 1. Input
													if self.is_input_continuous:
																				A_ 						=		hidden_states.shape
																				A_ 						=		hidden_states

																				A_ 						=		self.norm(lowercase_  )
																				if not self.use_linear_projection:
																											A_ 						=		self.proj_in(lowercase_  )
																											A_ 						=		hidden_states.shape[1]
																											A_ 						=		hidden_states.permute(0   ,     2   ,     3   ,     1  ).reshape(lowercase_   ,     height * width   ,     lowercase_  )
																				else:
																											A_ 						=		hidden_states.shape[1]
																											A_ 						=		hidden_states.permute(0   ,     2   ,     3   ,     1  ).reshape(lowercase_   ,     height * width   ,     lowercase_  )
																											A_ 						=		self.proj_in(lowercase_  )
													elif self.is_input_vectorized:
																				A_ 						=		self.latent_image_embedding(lowercase_  )
													elif self.is_input_patches:
																				A_ 						=		self.pos_embed(lowercase_  )

													# 2. Blocks
													for block in self.transformer_blocks:
																				A_ 						=		block(
																				    lowercase_   ,     attention_mask=lowercase_   ,     encoder_hidden_states=lowercase_   ,     encoder_attention_mask=lowercase_   ,     timestep=lowercase_   ,     cross_attention_kwargs=lowercase_   ,     class_labels=lowercase_   ,     )

													# 3. Output
													if self.is_input_continuous:
																				if not self.use_linear_projection:
																											A_ 						=		hidden_states.reshape(lowercase_   ,     lowercase_   ,     lowercase_   ,     lowercase_  ).permute(0   ,     3   ,     1   ,     2  ).contiguous()
																											A_ 						=		self.proj_out(lowercase_  )
																				else:
																											A_ 						=		self.proj_out(lowercase_  )
																											A_ 						=		hidden_states.reshape(lowercase_   ,     lowercase_   ,     lowercase_   ,     lowercase_  ).permute(0   ,     3   ,     1   ,     2  ).contiguous()

																				A_ 						=		hidden_states + residual
													elif self.is_input_vectorized:
																				A_ 						=		self.norm_out(lowercase_  )
																				A_ 						=		self.out(lowercase_  )
																				# (batch, self.num_vector_embeds - 1, self.num_latent_pixels)
																				A_ 						=		logits.permute(0   ,     2   ,     1  )

																				# log(p(x_0))
																				A_ 						=		F.log_softmax(logits.double()   ,     dim=1  ).float()
													elif self.is_input_patches:
																				# TODO: cleanup!
																				A_ 						=		self.transformer_blocks[0].norma.emb(
																				    lowercase_   ,     lowercase_   ,     hidden_dtype=hidden_states.dtype  )
																				A_ 						=		self.proj_out_a(F.silu(lowercase_  )  ).chunk(2   ,     dim=1  )
																				A_ 						=		self.norm_out(lowercase_  ) * (1 + scale[:, None]) + shift[:, None]
																				A_ 						=		self.proj_out_a(lowercase_  )

																				# unpatchify
																				A_ 						=		int(hidden_states.shape[1] ** 0.5  )
																				A_ 						=		hidden_states.reshape(
																				    shape=(-1, height, width, self.patch_size, self.patch_size, self.out_channels)  )
																				A_ 						=		torch.einsum("nhwpqc->nchpwq"   ,     lowercase_  )
																				A_ 						=		hidden_states.reshape(
																				    shape=(-1, self.out_channels, height * self.patch_size, width * self.patch_size)  )

													if not return_dict:
																				return (output,)

													return TransformeraDModelOutput(sample=lowercase_  )









'''simple docstring'''
import shutil
import tempfile
import unittest
from pathlib import Path
from shutil import copyfile

from transformers import SpeechaTextTokenizer, is_speech_available
from transformers.models.speech_to_text.tokenization_speech_to_text import VOCAB_FILES_NAMES, save_json
from transformers.testing_utils import get_tests_dir, require_sentencepiece, require_torch, require_torchaudio
from transformers.utils import FEATURE_EXTRACTOR_NAME

from .test_feature_extraction_speech_to_text import floats_list


if is_speech_available():
							from transformers import SpeechaTextFeatureExtractor, SpeechaTextProcessor


__SCREAMING_SNAKE_CASE				:							int		       =		get_tests_dir('''fixtures/test_sentencepiece.model''')






@require_torch
@require_torchaudio
@require_sentencepiece
class 				__lowerCAmelCase      (      unittest.TestCase	):

						"""simple docstring"""






						def 							_UpperCAmelCase					(     self       :			List[str]  ):
													A_ 						=		tempfile.mkdtemp()

													A_ 						=		["<s>", "<pad>", "</s>", "<unk>", "▁This", "▁is", "▁a", "▁t", "est"]
													A_ 						=		dict(zip(lowerCAmelCase   ,     range(len(lowerCAmelCase  )  )  )  )
													A_ 						=		Path(self.tmpdirname  )
													save_json(lowerCAmelCase   ,     save_dir / VOCAB_FILES_NAMES["vocab_file"]  )
													if not (save_dir / VOCAB_FILES_NAMES["spm_file"]).exists():
																				copyfile(lowerCAmelCase   ,     save_dir / VOCAB_FILES_NAMES["spm_file"]  )

													A_ 						=		SpeechaTextTokenizer.from_pretrained(self.tmpdirname  )
													tokenizer.save_pretrained(self.tmpdirname  )

													A_ 						=		{
													    "feature_size": 24,
													    "num_mel_bins": 24,
													    "padding_value": 0.0,
													    "sampling_rate": 1_60_00,
													    "return_attention_mask": False,
													    "do_normalize": True,
													}
													save_json(lowerCAmelCase   ,     save_dir / FEATURE_EXTRACTOR_NAME  )






						def 							_UpperCAmelCase					(     self       :			Union[str, Any]   ,     **lowerCAmelCase       :			str  ):
													return SpeechaTextTokenizer.from_pretrained(self.tmpdirname   ,     **lowerCAmelCase  )






						def 							_UpperCAmelCase					(     self       :			List[Any]   ,     **lowerCAmelCase       :			Optional[Any]  ):
													return SpeechaTextFeatureExtractor.from_pretrained(self.tmpdirname   ,     **lowerCAmelCase  )






						def 							_UpperCAmelCase					(     self       :			Any  ):
													shutil.rmtree(self.tmpdirname  )






						def 							_UpperCAmelCase					(     self       :			Union[str, Any]  ):
													A_ 						=		self.get_tokenizer()
													A_ 						=		self.get_feature_extractor()

													A_ 						=		SpeechaTextProcessor(tokenizer=lowerCAmelCase   ,     feature_extractor=lowerCAmelCase  )

													processor.save_pretrained(self.tmpdirname  )
													A_ 						=		SpeechaTextProcessor.from_pretrained(self.tmpdirname  )

													self.assertEqual(processor.tokenizer.get_vocab()   ,     tokenizer.get_vocab()  )
													self.assertIsInstance(processor.tokenizer   ,     lowerCAmelCase  )

													self.assertEqual(processor.feature_extractor.to_json_string()   ,     feature_extractor.to_json_string()  )
													self.assertIsInstance(processor.feature_extractor   ,     lowerCAmelCase  )






						def 							_UpperCAmelCase					(     self       :			Dict  ):
													A_ 						=		SpeechaTextProcessor(
													    tokenizer=self.get_tokenizer()   ,     feature_extractor=self.get_feature_extractor()  )
													processor.save_pretrained(self.tmpdirname  )

													A_ 						=		self.get_tokenizer(bos_token="(BOS)"   ,     eos_token="(EOS)"  )
													A_ 						=		self.get_feature_extractor(do_normalize=lowerCAmelCase   ,     padding_value=1.0  )

													A_ 						=		SpeechaTextProcessor.from_pretrained(
													    self.tmpdirname   ,     bos_token="(BOS)"   ,     eos_token="(EOS)"   ,     do_normalize=lowerCAmelCase   ,     padding_value=1.0  )

													self.assertEqual(processor.tokenizer.get_vocab()   ,     tokenizer_add_kwargs.get_vocab()  )
													self.assertIsInstance(processor.tokenizer   ,     lowerCAmelCase  )

													self.assertEqual(processor.feature_extractor.to_json_string()   ,     feature_extractor_add_kwargs.to_json_string()  )
													self.assertIsInstance(processor.feature_extractor   ,     lowerCAmelCase  )






						def 							_UpperCAmelCase					(     self       :			str  ):
													A_ 						=		self.get_feature_extractor()
													A_ 						=		self.get_tokenizer()

													A_ 						=		SpeechaTextProcessor(tokenizer=lowerCAmelCase   ,     feature_extractor=lowerCAmelCase  )

													A_ 						=		floats_list((3, 10_00)  )

													A_ 						=		feature_extractor(lowerCAmelCase   ,     return_tensors="np"  )
													A_ 						=		processor(lowerCAmelCase   ,     return_tensors="np"  )

													for key in input_feat_extract.keys():
																				self.assertAlmostEqual(input_feat_extract[key].sum()   ,     input_processor[key].sum()   ,     delta=1e-2  )






						def 							_UpperCAmelCase					(     self       :			Optional[int]  ):
													A_ 						=		self.get_feature_extractor()
													A_ 						=		self.get_tokenizer()

													A_ 						=		SpeechaTextProcessor(tokenizer=lowerCAmelCase   ,     feature_extractor=lowerCAmelCase  )

													A_ 						=		"This is a test string"

													A_ 						=		processor(text=lowerCAmelCase  )

													A_ 						=		tokenizer(lowerCAmelCase  )

													for key in encoded_tok.keys():
																				self.assertListEqual(encoded_tok[key]   ,     encoded_processor[key]  )






						def 							_UpperCAmelCase					(     self       :			int  ):
													A_ 						=		self.get_feature_extractor()
													A_ 						=		self.get_tokenizer()

													A_ 						=		SpeechaTextProcessor(tokenizer=lowerCAmelCase   ,     feature_extractor=lowerCAmelCase  )

													A_ 						=		[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

													A_ 						=		processor.batch_decode(lowerCAmelCase  )
													A_ 						=		tokenizer.batch_decode(lowerCAmelCase  )

													self.assertListEqual(lowerCAmelCase   ,     lowerCAmelCase  )






						def 							_UpperCAmelCase					(     self       :			Tuple  ):
													A_ 						=		self.get_feature_extractor()
													A_ 						=		self.get_tokenizer()

													A_ 						=		SpeechaTextProcessor(tokenizer=lowerCAmelCase   ,     feature_extractor=lowerCAmelCase  )

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





