



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 		_lowercase		(							UpperCAmelCase__						):
   _UpperCAmelCase  						=      42



class 		_lowercase		(							UpperCAmelCase__, UpperCAmelCase__						):

   @register_to_config
   def __init__(   self      :							int				, __lowerCAmelCase      :							int = 16				, __lowerCAmelCase      :							int = 88				, __lowerCAmelCase      :							Optional[int] = None				, __lowerCAmelCase      :							Optional[int] = None				, __lowerCAmelCase      :							int = 1				, __lowerCAmelCase      :							float = 0.0				, __lowerCAmelCase      :							int = 32				, __lowerCAmelCase      :							Optional[int] = None				, __lowerCAmelCase      :							bool = False				, __lowerCAmelCase      :							Optional[int] = None				, __lowerCAmelCase      :							Optional[int] = None				, __lowerCAmelCase      :							Optional[int] = None				, __lowerCAmelCase      :							str = "geglu"				, __lowerCAmelCase      :							Optional[int] = None				, __lowerCAmelCase      :							bool = False				, __lowerCAmelCase      :							bool = False				, __lowerCAmelCase      :							bool = False				, __lowerCAmelCase      :							str = "layer_norm"				, __lowerCAmelCase      :							bool = True				, )							->					List[Any]:


    """simple docstring"""






    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"				, __lowerCAmelCase				, standard_warn=__lowerCAmelCase				)
     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=__lowerCAmelCase				, num_channels=__lowerCAmelCase				, eps=1E-6				, affine=__lowerCAmelCase				)
     if use_linear_projection:
      a							=		nn.Linear(__lowerCAmelCase				, __lowerCAmelCase				)
     else:
      a							=		nn.Convad(__lowerCAmelCase				, __lowerCAmelCase				, 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=__lowerCAmelCase				, embed_dim=__lowerCAmelCase				, 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=__lowerCAmelCase				, width=__lowerCAmelCase				, patch_size=__lowerCAmelCase				, in_channels=__lowerCAmelCase				, embed_dim=__lowerCAmelCase				, )

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

    # 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(__lowerCAmelCase				, __lowerCAmelCase				)
     else:
      a							=		nn.Convad(__lowerCAmelCase				, __lowerCAmelCase				, kernel_size=1				, stride=1				, padding=0				)
    elif self.is_input_vectorized:
     a							=		nn.LayerNorm(__lowerCAmelCase				)
     a							=		nn.Linear(__lowerCAmelCase				, self.num_vector_embeds - 1				)
    elif self.is_input_patches:
     a							=		nn.LayerNorm(__lowerCAmelCase				, elementwise_affine=__lowerCAmelCase				, eps=1E-6				)
     a							=		nn.Linear(__lowerCAmelCase				, 2 * inner_dim				)
     a							=		nn.Linear(__lowerCAmelCase				, patch_size * patch_size * self.out_channels				)




   def 		A				(   self      :							Any				, __lowerCAmelCase      :							torch.Tensor				, __lowerCAmelCase      :							Optional[torch.Tensor] = None				, __lowerCAmelCase      :							Optional[torch.LongTensor] = None				, __lowerCAmelCase      :							Optional[torch.LongTensor] = None				, __lowerCAmelCase      :							Dict[str, Any] = None				, __lowerCAmelCase      :							Optional[torch.Tensor] = None				, __lowerCAmelCase      :							Optional[torch.Tensor] = None				, __lowerCAmelCase      :							bool = True				, )							->					Dict:


    """simple docstring"""






    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_0_0_0_0.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_0_0_0_0.0
     a							=		encoder_attention_mask.unsqueeze(1				)

    # 1. Input
    if self.is_input_continuous:
     a		,	a		,	a		,	a							=		hidden_states.shape
     a							=		hidden_states

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

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

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

     a							=		hidden_states + residual
    elif self.is_input_vectorized:
     a							=		self.norm_out(__lowerCAmelCase				)
     a							=		self.out(__lowerCAmelCase				)
     # (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(
         __lowerCAmelCase				, __lowerCAmelCase				, hidden_dtype=hidden_states.dtype				)
     a		,	a							=		self.proj_out_a(F.silu(__lowerCAmelCase				)				).chunk(2				, dim=1				)
     a							=		self.norm_out(__lowerCAmelCase				) * (1 + scale[:, None]) + shift[:, None]
     a							=		self.proj_out_a(__lowerCAmelCase				)

     # unpatchify
     a							=		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"				, __lowerCAmelCase				)
     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=__lowerCAmelCase				)




import random
import shutil
import tempfile
import unittest

import numpy as np

from transformers import TaTokenizer, TaTokenizerFast
from transformers.testing_utils import require_sentencepiece, require_torch
from transformers.utils.import_utils import is_speech_available, is_torch_available


if is_torch_available():
    pass

if is_speech_available():
    from transformers import EncodecFeatureExtractor, MusicgenProcessor


lowerCamelCase_					  =   random.Random()



def   UpperCamelCase(   lowercase_     ,   lowercase_=1.0     ,   lowercase_=None     ,   lowercase_=None				)      ->							List[str]:





  '''simple docstring'''





  if rng is None:
    snake_case_					= global_rng

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

  return values







@require_torch
@require_sentencepiece
class 					__lowerCamelCase							(      unittest.TestCase      ):







      def       lowerCAmelCase_       (	self  )		->			Union[str, Any]:
        snake_case_					= """facebook/musicgen-small"""
        snake_case_					= tempfile.mkdtemp()







      def       lowerCAmelCase_       (	self				,    **lowerCamelCase  )		->			int:
        return TaTokenizer.from_pretrained(self.checkpoint				,    **lowerCamelCase  )







      def       lowerCAmelCase_       (	self				,    **lowerCamelCase  )		->			Dict:
        return EncodecFeatureExtractor.from_pretrained(self.checkpoint				,    **lowerCamelCase  )







      def       lowerCAmelCase_       (	self  )		->			Tuple:
        shutil.rmtree(self.tmpdirname  )







      def       lowerCAmelCase_       (	self  )		->			Union[str, Any]:
        snake_case_					= self.get_tokenizer()
        snake_case_					= self.get_feature_extractor()

        snake_case_					= MusicgenProcessor(tokenizer=lowerCamelCase				,    feature_extractor=lowerCamelCase  )

        processor.save_pretrained(self.tmpdirname  )
        snake_case_					= MusicgenProcessor.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       lowerCAmelCase_       (	self  )		->			List[str]:
        snake_case_					= MusicgenProcessor(tokenizer=self.get_tokenizer()				,    feature_extractor=self.get_feature_extractor()  )
        processor.save_pretrained(self.tmpdirname  )

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

        snake_case_					= MusicgenProcessor.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       lowerCAmelCase_       (	self  )		->			Tuple:
        snake_case_					= self.get_feature_extractor()
        snake_case_					= self.get_tokenizer()

        snake_case_					= MusicgenProcessor(tokenizer=lowerCamelCase				,    feature_extractor=lowerCamelCase  )

        snake_case_					= floats_list((3, 1000)  )

        snake_case_					= feature_extractor(lowerCamelCase				,    return_tensors="""np"""  )
        snake_case_					= 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       lowerCAmelCase_       (	self  )		->			List[str]:
        snake_case_					= self.get_feature_extractor()
        snake_case_					= self.get_tokenizer()

        snake_case_					= MusicgenProcessor(tokenizer=lowerCamelCase				,    feature_extractor=lowerCamelCase  )

        snake_case_					= """This is a test string"""

        snake_case_					= processor(text=lowerCamelCase  )

        snake_case_					= tokenizer(lowerCamelCase  )

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







      def       lowerCAmelCase_       (	self  )		->			int:
        snake_case_					= self.get_feature_extractor()
        snake_case_					= self.get_tokenizer()

        snake_case_					= MusicgenProcessor(tokenizer=lowerCamelCase				,    feature_extractor=lowerCamelCase  )

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

        snake_case_					= processor.batch_decode(sequences=lowerCamelCase  )
        snake_case_					= tokenizer.batch_decode(lowerCamelCase  )

        self.assertListEqual(lowerCamelCase				,    lowerCamelCase  )







      def       lowerCAmelCase_       (	self  )		->			Optional[int]:
        snake_case_					= self.get_feature_extractor()
        snake_case_					= self.get_tokenizer()

        snake_case_					= MusicgenProcessor(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"""				,    )







      def       lowerCAmelCase_       (	self  )		->			Optional[Any]:
        snake_case_					= self.get_feature_extractor(padding_side="""left"""  )
        snake_case_					= self.get_tokenizer()

        snake_case_					= MusicgenProcessor(tokenizer=lowerCamelCase				,    feature_extractor=lowerCamelCase  )

        snake_case_					= [floats_list((1, x)  )[0] for x in range(5				,    20				,    5  )]
        snake_case_					= processor(lowerCamelCase  ).padding_mask

        snake_case_					= np.asarray(floats_list((3, 20)  )  )[:, None, :]
        snake_case_					= processor.batch_decode(lowerCamelCase				,    padding_mask=lowerCamelCase  )

        self.assertIsInstance(lowerCamelCase				,    lowerCamelCase  )

        for audio in decoded_audios:
          self.assertIsInstance(lowerCamelCase				,    np.ndarray  )

        self.assertTrue(decoded_audios[0].shape == (1, 10)  )
        self.assertTrue(decoded_audios[1].shape == (1, 15)  )
        self.assertTrue(decoded_audios[2].shape == (1, 20)  )