





"""simple docstring"""





from dataclasses import dataclass
from typing import Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .embeddings import GaussianFourierProjection, TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block





@dataclass
class       snake_case		(       lowerCAmelCase__       ):
      lowerCamelCase__      =			42







class       snake_case		(       lowerCAmelCase__			,					lowerCAmelCase__       ):

      @register_to_config
      def __init__(      self       :Tuple						,							_lowerCamelCase       :Union[str, Any] = None						,							_lowerCamelCase       :Any = 3						,							_lowerCamelCase       :Optional[Any] = 3						,							_lowerCamelCase       :List[Any] = False						,							_lowerCamelCase       :Any = "positional"						,							_lowerCamelCase       :Dict = 0						,							_lowerCamelCase       :int = True						,							_lowerCamelCase       :Tuple = ("DownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D")						,							_lowerCamelCase       :str = ("AttnUpBlock2D", "AttnUpBlock2D", "AttnUpBlock2D", "UpBlock2D")						,							_lowerCamelCase       :Any = (2_2_4, 4_4_8, 6_7_2, 8_9_6)						,							_lowerCamelCase       :Tuple = 2						,							_lowerCamelCase       :Any = 1						,							_lowerCamelCase       :Dict = 1						,							_lowerCamelCase       :Optional[int] = "conv"						,							_lowerCamelCase       :List[Any] = "conv"						,							_lowerCamelCase       :Union[str, Any] = "silu"						,							_lowerCamelCase       :Tuple = 8						,							_lowerCamelCase       :Any = 3_2						,							_lowerCamelCase       :Optional[Any] = 1e-5						,							_lowerCamelCase       :List[str] = "default"						,							_lowerCamelCase       :List[str] = True						,							_lowerCamelCase       :Tuple = None						,							_lowerCamelCase       :Tuple = None						,							):
       super().__init__()

       __SCREAMING_SNAKE_CASE			:      Optional[Any] 							=	sample_size
       __SCREAMING_SNAKE_CASE			:      List[Any] 							=	block_out_channels[0] * 4

       # Check inputs
       if len(UpperCamelCase__						) != len(UpperCamelCase__						):
        raise ValueError(
            f'''Must provide the same number of `down_block_types` as `up_block_types`. `down_block_types`: {down_block_types}. `up_block_types`: {up_block_types}.'''						)

       if len(UpperCamelCase__						) != len(UpperCamelCase__						):
        raise ValueError(
            f'''Must provide the same number of `block_out_channels` as `down_block_types`. `block_out_channels`: {block_out_channels}. `down_block_types`: {down_block_types}.'''						)

       # input
       __SCREAMING_SNAKE_CASE			:      int 							=	nn.Convad(UpperCamelCase__						,							block_out_channels[0]						,							kernel_size=3						,							padding=(1, 1)						)

       # time
       if time_embedding_type == "fourier":
        __SCREAMING_SNAKE_CASE			:      Dict 							=	GaussianFourierProjection(embedding_size=block_out_channels[0]						,							scale=1_6						)
        __SCREAMING_SNAKE_CASE			:      int 							=	2 * block_out_channels[0]
       elif time_embedding_type == "positional":
        __SCREAMING_SNAKE_CASE			:      List[str] 							=	Timesteps(block_out_channels[0]						,							UpperCamelCase__						,							UpperCamelCase__						)
        __SCREAMING_SNAKE_CASE			:      Optional[Any] 							=	block_out_channels[0]

       __SCREAMING_SNAKE_CASE			:      Optional[int] 							=	TimestepEmbedding(UpperCamelCase__						,							UpperCamelCase__						)

       # class embedding
       if class_embed_type is None and num_class_embeds is not None:
        __SCREAMING_SNAKE_CASE			:      List[str] 							=	nn.Embedding(UpperCamelCase__						,							UpperCamelCase__						)
       elif class_embed_type == "timestep":
        __SCREAMING_SNAKE_CASE			:      Dict 							=	TimestepEmbedding(UpperCamelCase__						,							UpperCamelCase__						)
       elif class_embed_type == "identity":
        __SCREAMING_SNAKE_CASE			:      str 							=	nn.Identity(UpperCamelCase__						,							UpperCamelCase__						)
       else:
        __SCREAMING_SNAKE_CASE			:      Dict 							=	None

       __SCREAMING_SNAKE_CASE			:      Optional[Any] 							=	nn.ModuleList([]						)
       __SCREAMING_SNAKE_CASE			:      List[str] 							=	None
       __SCREAMING_SNAKE_CASE			:      int 							=	nn.ModuleList([]						)

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

        __SCREAMING_SNAKE_CASE			:      List[Any] 							=	get_down_block(
            UpperCamelCase__						,							num_layers=UpperCamelCase__						,							in_channels=UpperCamelCase__						,							out_channels=UpperCamelCase__						,							temb_channels=UpperCamelCase__						,							add_downsample=not is_final_block						,							resnet_eps=UpperCamelCase__						,							resnet_act_fn=UpperCamelCase__						,							resnet_groups=UpperCamelCase__						,							attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel						,							downsample_padding=UpperCamelCase__						,							resnet_time_scale_shift=UpperCamelCase__						,							downsample_type=UpperCamelCase__						,							)
        self.down_blocks.append(UpperCamelCase__						)

       # mid
       __SCREAMING_SNAKE_CASE			:      Optional[Any] 							=	UNetMidBlockaD(
           in_channels=block_out_channels[-1]						,							temb_channels=UpperCamelCase__						,							resnet_eps=UpperCamelCase__						,							resnet_act_fn=UpperCamelCase__						,							output_scale_factor=UpperCamelCase__						,							resnet_time_scale_shift=UpperCamelCase__						,							attention_head_dim=attention_head_dim if attention_head_dim is not None else block_out_channels[-1]						,							resnet_groups=UpperCamelCase__						,							add_attention=UpperCamelCase__						,							)

       # up
       __SCREAMING_SNAKE_CASE			:      Dict 							=	list(reversed(UpperCamelCase__						)						)
       __SCREAMING_SNAKE_CASE			:      List[str] 							=	reversed_block_out_channels[0]
       for i, up_block_type in enumerate(UpperCamelCase__						):
        __SCREAMING_SNAKE_CASE			:      Any 							=	output_channel
        __SCREAMING_SNAKE_CASE			:      Optional[Any] 							=	reversed_block_out_channels[i]
        __SCREAMING_SNAKE_CASE			:      Optional[int] 							=	reversed_block_out_channels[min(i + 1						,							len(UpperCamelCase__						) - 1						)]

        __SCREAMING_SNAKE_CASE			:      Optional[Any] 							=	i == len(UpperCamelCase__						) - 1

        __SCREAMING_SNAKE_CASE			:      Tuple 							=	get_up_block(
            UpperCamelCase__						,							num_layers=layers_per_block + 1						,							in_channels=UpperCamelCase__						,							out_channels=UpperCamelCase__						,							prev_output_channel=UpperCamelCase__						,							temb_channels=UpperCamelCase__						,							add_upsample=not is_final_block						,							resnet_eps=UpperCamelCase__						,							resnet_act_fn=UpperCamelCase__						,							resnet_groups=UpperCamelCase__						,							attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel						,							resnet_time_scale_shift=UpperCamelCase__						,							upsample_type=UpperCamelCase__						,							)
        self.up_blocks.append(UpperCamelCase__						)
        __SCREAMING_SNAKE_CASE			:      Optional[Any] 							=	output_channel

       # out
       __SCREAMING_SNAKE_CASE			:      Optional[int] 							=	norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4						,							3_2						)
       __SCREAMING_SNAKE_CASE			:      Optional[Any] 							=	nn.GroupNorm(num_channels=block_out_channels[0]						,							num_groups=UpperCamelCase__						,							eps=UpperCamelCase__						)
       __SCREAMING_SNAKE_CASE			:      Optional[Any] 							=	nn.SiLU()
       __SCREAMING_SNAKE_CASE			:      Dict 							=	nn.Convad(block_out_channels[0]						,							UpperCamelCase__						,							kernel_size=3						,							padding=1						)



      def 						SCREAMING_SNAKE_CASE_      (      self       :Any						,							_lowerCamelCase       :Optional[int]						,							_lowerCamelCase       :Dict						,							_lowerCamelCase       :Optional[int] = None						,							_lowerCamelCase       :List[Any] = True						,							):
       # 0. center input if necessary
       if self.config.center_input_sample:
        __SCREAMING_SNAKE_CASE			:      Dict 							=	2 * sample - 1.0

       # 1. time
       __SCREAMING_SNAKE_CASE			:      int 							=	timestep
       if not torch.is_tensor(UpperCamelCase__						):
        __SCREAMING_SNAKE_CASE			:      Optional[int] 							=	torch.tensor([timesteps]						,							dtype=torch.long						,							device=sample.device						)
       elif torch.is_tensor(UpperCamelCase__						) and len(timesteps.shape						) == 0:
        __SCREAMING_SNAKE_CASE			:      Dict 							=	timesteps[None].to(sample.device						)

       # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
       __SCREAMING_SNAKE_CASE			:      Tuple 							=	timesteps * torch.ones(sample.shape[0]						,							dtype=timesteps.dtype						,							device=timesteps.device						)

       __SCREAMING_SNAKE_CASE			:      Dict 							=	self.time_proj(UpperCamelCase__						)

       # timesteps does not contain any weights and will always return f32 tensors
       # but time_embedding might actually be running in fp16. so we need to cast here.
       # there might be better ways to encapsulate this.
       __SCREAMING_SNAKE_CASE			:      Optional[int] 							=	t_emb.to(dtype=self.dtype						)
       __SCREAMING_SNAKE_CASE			:      Dict 							=	self.time_embedding(UpperCamelCase__						)

       if self.class_embedding is not None:
        if class_labels is None:
         raise ValueError('''class_labels should be provided when doing class conditioning'''						)

        if self.config.class_embed_type == "timestep":
         __SCREAMING_SNAKE_CASE			:      List[Any] 							=	self.time_proj(UpperCamelCase__						)

        __SCREAMING_SNAKE_CASE			:      Tuple 							=	self.class_embedding(UpperCamelCase__						).to(dtype=self.dtype						)
        __SCREAMING_SNAKE_CASE			:      Tuple 							=	emb + class_emb

       # 2. pre-process
       __SCREAMING_SNAKE_CASE			:      str 							=	sample
       __SCREAMING_SNAKE_CASE			:      Union[str, Any] 							=	self.conv_in(UpperCamelCase__						)

       # 3. down
       __SCREAMING_SNAKE_CASE			:      int 							=	(sample,)
       for downsample_block in self.down_blocks:
        if hasattr(UpperCamelCase__						,							'''skip_conv'''						):
         __SCREAMING_SNAKE_CASE			:      Optional[int] 							=	downsample_block(
             hidden_states=UpperCamelCase__						,							temb=UpperCamelCase__						,							skip_sample=UpperCamelCase__						)
        else:
         __SCREAMING_SNAKE_CASE			:      Union[str, Any] 							=	downsample_block(hidden_states=UpperCamelCase__						,							temb=UpperCamelCase__						)

        down_block_res_samples += res_samples

       # 4. mid
       __SCREAMING_SNAKE_CASE			:      Dict 							=	self.mid_block(UpperCamelCase__						,							UpperCamelCase__						)

       # 5. up
       __SCREAMING_SNAKE_CASE			:      Tuple 							=	None
       for upsample_block in self.up_blocks:
        __SCREAMING_SNAKE_CASE			:      List[Any] 							=	down_block_res_samples[-len(upsample_block.resnets						) :]
        __SCREAMING_SNAKE_CASE			:      int 							=	down_block_res_samples[: -len(upsample_block.resnets						)]

        if hasattr(UpperCamelCase__						,							'''skip_conv'''						):
         __SCREAMING_SNAKE_CASE			:      List[str] 							=	upsample_block(UpperCamelCase__						,							UpperCamelCase__						,							UpperCamelCase__						,							UpperCamelCase__						)
        else:
         __SCREAMING_SNAKE_CASE			:      Union[str, Any] 							=	upsample_block(UpperCamelCase__						,							UpperCamelCase__						,							UpperCamelCase__						)

        # 6. post-process
       __SCREAMING_SNAKE_CASE			:      Optional[int] 							=	self.conv_norm_out(UpperCamelCase__						)
       __SCREAMING_SNAKE_CASE			:      Tuple 							=	self.conv_act(UpperCamelCase__						)
       __SCREAMING_SNAKE_CASE			:      Optional[Any] 							=	self.conv_out(UpperCamelCase__						)

       if skip_sample is not None:
        sample += skip_sample

       if self.config.time_embedding_type == "fourier":
        __SCREAMING_SNAKE_CASE			:      Optional[int] 							=	timesteps.reshape((sample.shape[0], *([1] * len(sample.shape[1:]						)))						)
        __SCREAMING_SNAKE_CASE			:      int 							=	sample / timesteps

       if not return_dict:
        return (sample,)

       return UNetaDOutput(sample=UpperCamelCase__						)








"""simple docstring"""





import json
import os
import shutil
import tempfile
import unittest

from transformers import is_speech_available, is_torch_available
from transformers.models.speechta import SpeechTaTokenizer
from transformers.testing_utils import get_tests_dir, require_torch
from transformers.utils import FEATURE_EXTRACTOR_NAME


if is_speech_available() and is_torch_available():
    from transformers import SpeechTaFeatureExtractor, SpeechTaProcessor

    from .test_feature_extraction_speechta import floats_list


_lowerCamelCase  					=     get_tests_dir('''fixtures/test_sentencepiece_bpe_char.model''')





@require_torch
class       snake_case		(       unittest.TestCase       ):

      def 						SCREAMING_SNAKE_CASE_      (      self       :Union[str, Any]						):
       __SCREAMING_SNAKE_CASE			:      Union[str, Any] 							=	tempfile.mkdtemp()

       __SCREAMING_SNAKE_CASE			:      Optional[int] 							=	SpeechTaTokenizer(_lowerCamelCase						)
       tokenizer.save_pretrained(self.tmpdirname						)

       __SCREAMING_SNAKE_CASE			:      Optional[Any] 							=	{
           '''feature_size''': 1,
           '''padding_value''': 0.0,
           '''sampling_rate''': 1_6_0_0_0,
           '''do_normalize''': False,
           '''num_mel_bins''': 8_0,
           '''hop_length''': 1_6,
           '''win_length''': 6_4,
           '''win_function''': '''hann_window''',
           '''fmin''': 8_0,
           '''fmax''': 7_6_0_0,
           '''mel_floor''': 1e-10,
           '''reduction_factor''': 2,
           '''return_attention_mask''': True,
       }

       __SCREAMING_SNAKE_CASE			:      Any 							=	os.path.join(self.tmpdirname						,							_lowerCamelCase						)
       with open(self.feature_extraction_file						,							'''w'''						,							encoding='''utf-8'''						) as fp:
        fp.write(json.dumps(_lowerCamelCase						) + '''\n'''						)

      def 						SCREAMING_SNAKE_CASE_      (      self       :str						,							**_lowerCamelCase       :str						):
       return SpeechTaTokenizer.from_pretrained(self.tmpdirname						,							**_lowerCamelCase						)

      def 						SCREAMING_SNAKE_CASE_      (      self       :Dict						,							**_lowerCamelCase       :Dict						):
       return SpeechTaFeatureExtractor.from_pretrained(self.tmpdirname						,							**_lowerCamelCase						)

      def 						SCREAMING_SNAKE_CASE_      (      self       :List[str]						):
       shutil.rmtree(self.tmpdirname						)

      def 						SCREAMING_SNAKE_CASE_      (      self       :Union[str, Any]						):
       __SCREAMING_SNAKE_CASE			:      Any 							=	self.get_tokenizer()
       __SCREAMING_SNAKE_CASE			:      int 							=	self.get_feature_extractor()

       __SCREAMING_SNAKE_CASE			:      Any 							=	SpeechTaProcessor(tokenizer=_lowerCamelCase						,							feature_extractor=_lowerCamelCase						)

       processor.save_pretrained(self.tmpdirname						)
       __SCREAMING_SNAKE_CASE			:      Optional[Any] 							=	SpeechTaProcessor.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 						SCREAMING_SNAKE_CASE_      (      self       :Union[str, Any]						):
       __SCREAMING_SNAKE_CASE			:      Union[str, Any] 							=	SpeechTaProcessor(tokenizer=self.get_tokenizer()						,							feature_extractor=self.get_feature_extractor()						)
       processor.save_pretrained(self.tmpdirname						)

       __SCREAMING_SNAKE_CASE			:      Tuple 							=	self.get_tokenizer(bos_token='''(BOS)'''						,							eos_token='''(EOS)'''						)
       __SCREAMING_SNAKE_CASE			:      List[str] 							=	self.get_feature_extractor(do_normalize=_lowerCamelCase						,							padding_value=1.0						)

       __SCREAMING_SNAKE_CASE			:      Any 							=	SpeechTaProcessor.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 						SCREAMING_SNAKE_CASE_      (      self       :str						):
       __SCREAMING_SNAKE_CASE			:      Optional[int] 							=	self.get_feature_extractor()
       __SCREAMING_SNAKE_CASE			:      Tuple 							=	self.get_tokenizer()

       __SCREAMING_SNAKE_CASE			:      Union[str, Any] 							=	SpeechTaProcessor(tokenizer=_lowerCamelCase						,							feature_extractor=_lowerCamelCase						)

       __SCREAMING_SNAKE_CASE			:      int 							=	floats_list((3, 1_0_0_0)						)

       __SCREAMING_SNAKE_CASE			:      int 							=	feature_extractor(audio=_lowerCamelCase						,							return_tensors='''np'''						)
       __SCREAMING_SNAKE_CASE			:      List[str] 							=	processor(audio=_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 						SCREAMING_SNAKE_CASE_      (      self       :Tuple						):
       __SCREAMING_SNAKE_CASE			:      Any 							=	self.get_feature_extractor()
       __SCREAMING_SNAKE_CASE			:      int 							=	self.get_tokenizer()

       __SCREAMING_SNAKE_CASE			:      Any 							=	SpeechTaProcessor(tokenizer=_lowerCamelCase						,							feature_extractor=_lowerCamelCase						)

       __SCREAMING_SNAKE_CASE			:      Union[str, Any] 							=	floats_list((3, 1_0_0_0)						)

       __SCREAMING_SNAKE_CASE			:      Union[str, Any] 							=	feature_extractor(audio_target=_lowerCamelCase						,							return_tensors='''np'''						)
       __SCREAMING_SNAKE_CASE			:      List[Any] 							=	processor(audio_target=_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 						SCREAMING_SNAKE_CASE_      (      self       :List[Any]						):
       __SCREAMING_SNAKE_CASE			:      Optional[Any] 							=	self.get_feature_extractor()
       __SCREAMING_SNAKE_CASE			:      Union[str, Any] 							=	self.get_tokenizer()

       __SCREAMING_SNAKE_CASE			:      Union[str, Any] 							=	SpeechTaProcessor(tokenizer=_lowerCamelCase						,							feature_extractor=_lowerCamelCase						)

       __SCREAMING_SNAKE_CASE			:      int 							=	'''This is a test string'''

       __SCREAMING_SNAKE_CASE			:      str 							=	processor(text=_lowerCamelCase						)
       __SCREAMING_SNAKE_CASE			:      Union[str, Any] 							=	tokenizer(_lowerCamelCase						)

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

      def 						SCREAMING_SNAKE_CASE_      (      self       :Optional[int]						):
       __SCREAMING_SNAKE_CASE			:      int 							=	self.get_feature_extractor()
       __SCREAMING_SNAKE_CASE			:      str 							=	self.get_tokenizer()

       __SCREAMING_SNAKE_CASE			:      Tuple 							=	SpeechTaProcessor(tokenizer=_lowerCamelCase						,							feature_extractor=_lowerCamelCase						)

       __SCREAMING_SNAKE_CASE			:      Dict 							=	'''This is a test string'''

       __SCREAMING_SNAKE_CASE			:      str 							=	processor(text_target=_lowerCamelCase						)
       __SCREAMING_SNAKE_CASE			:      Dict 							=	tokenizer(_lowerCamelCase						)

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

      def 						SCREAMING_SNAKE_CASE_      (      self       :Optional[Any]						):
       __SCREAMING_SNAKE_CASE			:      Dict 							=	self.get_feature_extractor()
       __SCREAMING_SNAKE_CASE			:      Tuple 							=	self.get_tokenizer()

       __SCREAMING_SNAKE_CASE			:      Optional[Any] 							=	SpeechTaProcessor(tokenizer=_lowerCamelCase						,							feature_extractor=_lowerCamelCase						)

       __SCREAMING_SNAKE_CASE			:      Optional[Any] 							=	[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

       __SCREAMING_SNAKE_CASE			:      List[Any] 							=	processor.batch_decode(_lowerCamelCase						)
       __SCREAMING_SNAKE_CASE			:      List[Any] 							=	tokenizer.batch_decode(_lowerCamelCase						)

       self.assertListEqual(_lowerCamelCase						,							_lowerCamelCase						)



      def 						SCREAMING_SNAKE_CASE_      (      self       :Optional[Any]						):
       __SCREAMING_SNAKE_CASE			:      Optional[Any] 							=	self.get_feature_extractor()
       __SCREAMING_SNAKE_CASE			:      str 							=	self.get_tokenizer()

       __SCREAMING_SNAKE_CASE			:      str 							=	SpeechTaProcessor(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'''						,							)

