




'''simple docstring'''





from __future__ import annotations

import unittest

import numpy as np

from transformers import OPTConfig, is_tf_available
from transformers.testing_utils import require_sentencepiece, require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
     import tensorflow as tf

     from transformers import GPTaTokenizer, TFOPTForCausalLM, TFOPTModel



def      __lowerCAmelCase			(					UpperCamelCase	,      UpperCamelCase	,      UpperCamelCase=None	,      UpperCamelCase=None      )     ->	List[str]:
  if attention_mask is None:
    lowerCAmelCase__				:  Optional[Any]		       =     tf.cast(tf.math.not_equal(UpperCamelCase	,      config.pad_token_id      )	,      tf.inta      )
  return {"input_ids": input_ids, "attention_mask": attention_mask}






@require_tf
class  _lowerCAmelCase     :
      A__		     =		OPTConfig
      A__		     =		{}
      A__		     =		'gelu'

      def __init__(   self      ,				__UpperCAmelCase      ,				__UpperCAmelCase=13      ,				__UpperCAmelCase=7      ,				__UpperCAmelCase=True      ,				__UpperCAmelCase=False      ,				__UpperCAmelCase=99      ,				__UpperCAmelCase=16      ,				__UpperCAmelCase=2      ,				__UpperCAmelCase=4      ,				__UpperCAmelCase=4      ,				__UpperCAmelCase="gelu"      ,				__UpperCAmelCase=0.1      ,				__UpperCAmelCase=0.1      ,				__UpperCAmelCase=20      ,				__UpperCAmelCase=2      ,				__UpperCAmelCase=1      ,				__UpperCAmelCase=0      ,				__UpperCAmelCase=16      ,				__UpperCAmelCase=16      ,				):
        lowerCAmelCase__				:  Dict		       =     parent
        lowerCAmelCase__				:  List[Any]		       =     batch_size
        lowerCAmelCase__				:  str		       =     seq_length
        lowerCAmelCase__				:  int		       =     is_training
        lowerCAmelCase__				:  Optional[int]		       =     use_labels
        lowerCAmelCase__				:  int		       =     vocab_size
        lowerCAmelCase__				:  Optional[Any]		       =     hidden_size
        lowerCAmelCase__				:  str		       =     num_hidden_layers
        lowerCAmelCase__				:  Optional[Any]		       =     num_attention_heads
        lowerCAmelCase__				:  List[str]		       =     intermediate_size
        lowerCAmelCase__				:  List[Any]		       =     hidden_act
        lowerCAmelCase__				:  Any		       =     hidden_dropout_prob
        lowerCAmelCase__				:  Any		       =     attention_probs_dropout_prob
        lowerCAmelCase__				:  Optional[int]		       =     max_position_embeddings
        lowerCAmelCase__				:  Tuple		       =     eos_token_id
        lowerCAmelCase__				:  Tuple		       =     pad_token_id
        lowerCAmelCase__				:  Tuple		       =     bos_token_id
        lowerCAmelCase__				:  Union[str, Any]		       =     embed_dim
        lowerCAmelCase__				:  Optional[int]		       =     word_embed_proj_dim
        lowerCAmelCase__				:  int		       =     False

      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  Union[str, Any]		       =     ids_tensor([self.batch_size, self.seq_length - 1]      ,				self.vocab_size		)
        lowerCAmelCase__				:  int		       =     tf.expand_dims(tf.constant([self.eos_token_id] * self.batch_size		)      ,				1		)
        lowerCAmelCase__				:  Tuple		       =     tf.concat([input_ids, eos_tensor]      ,				axis=1		)

        lowerCAmelCase__				:  str		       =     self.config_cls(
            vocab_size=self.vocab_size      ,				hidden_size=self.hidden_size      ,				num_hidden_layers=self.num_hidden_layers      ,				num_attention_heads=self.num_attention_heads      ,				ffn_dim=self.intermediate_size      ,				dropout=self.hidden_dropout_prob      ,				attention_dropout=self.attention_probs_dropout_prob      ,				max_position_embeddings=self.max_position_embeddings      ,				eos_token_id=self.eos_token_id      ,				bos_token_id=self.bos_token_id      ,				pad_token_id=self.pad_token_id      ,				embed_dim=self.embed_dim      ,				word_embed_proj_dim=self.word_embed_proj_dim      ,				is_encoder_decoder=__UpperCAmelCase      ,				**self.config_updates      ,				)
        lowerCAmelCase__				:  Optional[Any]		       =     prepare_opt_inputs_dict(__UpperCAmelCase      ,				__UpperCAmelCase		)
        return config, inputs_dict







      def 	__magic_name__(   self      ,				__UpperCAmelCase      ,				__UpperCAmelCase		):
        lowerCAmelCase__				:  Any		       =     TFOPTModel(config=__UpperCAmelCase		)
        lowerCAmelCase__				:  Tuple		       =     inputs_dict['''input_ids''']

        lowerCAmelCase__				:  str		       =     input_ids[:1, :]
        lowerCAmelCase__				:  str		       =     inputs_dict['''attention_mask'''][:1, :]
        lowerCAmelCase__				:  List[str]		       =     1

        # first forward pass
        lowerCAmelCase__				:  str		       =     model(__UpperCAmelCase      ,				attention_mask=__UpperCAmelCase      ,				use_cache=__UpperCAmelCase		)

        lowerCAmelCase__				:  List[Any]		       =     outputs.to_tuple()

        # create hypothetical next token and extent to next_input_ids
        lowerCAmelCase__				:  Optional[int]		       =     ids_tensor((self.batch_size, 3)      ,				config.vocab_size		)
        lowerCAmelCase__				:  List[Any]		       =     tf.cast(ids_tensor((self.batch_size, 3)      ,				2		)      ,				tf.inta		)

        # append to next input_ids and
        lowerCAmelCase__				:  Tuple		       =     tf.concat([input_ids, next_tokens]      ,				axis=-1		)
        lowerCAmelCase__				:  Any		       =     tf.concat([attention_mask, next_attn_mask]      ,				axis=-1		)

        lowerCAmelCase__				:  Dict		       =     model(__UpperCAmelCase      ,				attention_mask=__UpperCAmelCase		)[0]
        lowerCAmelCase__				:  Any		       =     model(__UpperCAmelCase      ,				attention_mask=__UpperCAmelCase      ,				past_key_values=__UpperCAmelCase		)[0]

        self.parent.assertEqual(next_tokens.shape[1]      ,				output_from_past.shape[1]		)

        # select random slice
        lowerCAmelCase__				:  Union[str, Any]		       =     int(ids_tensor((1,)      ,				output_from_past.shape[-1]		)		)
        lowerCAmelCase__				:  Tuple		       =     output_from_no_past[:, -3:, random_slice_idx]
        lowerCAmelCase__				:  Tuple		       =     output_from_past[:, :, random_slice_idx]

        # test that outputs are equal for slice
        tf.debugging.assert_near(__UpperCAmelCase      ,				__UpperCAmelCase      ,				rtol=1e-3		)






@require_tf
class  _lowerCAmelCase     (  _lowercase    ,			_lowercase    ,			unittest.TestCase	):
      A__		     =		(TFOPTModel, TFOPTForCausalLM) if is_tf_available() else ()
      A__		     =		(TFOPTForCausalLM,) if is_tf_available() else ()
      A__		     =		(
          {'feature-extraction': TFOPTModel, 'text-generation': TFOPTForCausalLM} if is_tf_available() else {}
      )
      A__		     =		False
      A__		     =		False
      A__		     =		False
      A__		     =		10

      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  Union[str, Any]		       =     TFOPTModelTester(self		)
        lowerCAmelCase__				:  int		       =     ConfigTester(self      ,				config_class=__UpperCAmelCase		)

      def 	__magic_name__(   self		):
        self.config_tester.run_common_tests()

      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  str		       =     self.model_tester.prepare_config_and_inputs_for_common()
        self.model_tester.check_decoder_model_past_large_inputs(*__UpperCAmelCase		)







      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  Tuple		       =     self.model_tester.prepare_config_and_inputs_for_common()

        def _get_word_embedding_weight(__UpperCAmelCase      ,				__UpperCAmelCase		):
          if hasattr(__UpperCAmelCase      ,				'''weight'''		):
            return embedding_layer.weight
          else:
            # Here we build the word embeddings weights if not exists.
            # And then we retry to get the attribute once built.
            model.build()
            if hasattr(__UpperCAmelCase      ,				'''weight'''		):
              return embedding_layer.weight
            else:
              return None

        for model_class in self.all_model_classes:
          for size in [config.vocab_size - 10, config.vocab_size + 10]:
            # build the embeddings
            lowerCAmelCase__				:  Dict		       =     model_class(config=__UpperCAmelCase		)
            lowerCAmelCase__				:  Optional[Any]		       =     _get_word_embedding_weight(__UpperCAmelCase      ,				model.get_input_embeddings()		)
            lowerCAmelCase__				:  Optional[int]		       =     _get_word_embedding_weight(__UpperCAmelCase      ,				model.get_output_embeddings()		)

            # reshape the embeddings
            model.resize_token_embeddings(__UpperCAmelCase		)
            lowerCAmelCase__				:  Tuple		       =     _get_word_embedding_weight(__UpperCAmelCase      ,				model.get_input_embeddings()		)
            lowerCAmelCase__				:  List[Any]		       =     _get_word_embedding_weight(__UpperCAmelCase      ,				model.get_output_embeddings()		)

            # check that the resized embeddings size matches the desired size.
            lowerCAmelCase__				:  Any		       =     size if size is not None else config.vocab_size

            self.assertEqual(new_input_embeddings.shape[0]      ,				__UpperCAmelCase		)

            # check that weights remain the same after resizing
            lowerCAmelCase__				:  Tuple		       =     True
            for pa, pa in zip(old_input_embeddings.value()      ,				new_input_embeddings.value()		):
              if tf.math.reduce_sum(tf.math.abs(pa - pa		)		) > 0:
                lowerCAmelCase__				:  List[Any]		       =     False
            self.assertTrue(__UpperCAmelCase		)

            if old_output_embeddings is not None and new_output_embeddings is not None:
              self.assertEqual(new_output_embeddings.shape[0]      ,				__UpperCAmelCase		)

              lowerCAmelCase__				:  int		       =     True
              for pa, pa in zip(old_output_embeddings.value()      ,				new_output_embeddings.value()		):
                if tf.math.reduce_sum(tf.math.abs(pa - pa		)		) > 0:
                  lowerCAmelCase__				:  int		       =     False
              self.assertTrue(__UpperCAmelCase		)



def      __lowerCAmelCase			(					UpperCamelCase      )     ->	Tuple:
  return tf.constant(UpperCamelCase	,      dtype=tf.intaa      )






@require_tf
class  _lowerCAmelCase     (  unittest.TestCase	):
      A__		     =		99

      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  Optional[Any]		       =     tf.ones((4, 1)      ,				dtype=tf.intaa		) * 2
        lowerCAmelCase__				:  Optional[Any]		       =     tf.concat([ids_tensor((4, 6)      ,				self.vocab_size - 3		) + 3, eos_column_vector]      ,				axis=1		)
        lowerCAmelCase__				:  Tuple		       =     input_ids.shape[0]
        lowerCAmelCase__				:  Optional[Any]		       =     OPTConfig(
            vocab_size=self.vocab_size      ,				hidden_size=24      ,				num_hidden_layers=2      ,				num_attention_heads=2      ,				ffn_dim=32      ,				max_position_embeddings=48      ,				eos_token_id=2      ,				pad_token_id=1      ,				bos_token_id=0      ,				)
        return config, input_ids, batch_size






@require_sentencepiece
@require_tf
class  _lowerCAmelCase     (  unittest.TestCase	):

      @slow
      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  Optional[int]		       =     TFOPTModel.from_pretrained('''facebook/opt-350m'''		)
        lowerCAmelCase__				:  List[str]		       =     _long_tensor([[0, 3_1414, 232, 328, 740, 1140, 1_2695, 69, 4_6078, 1588, 2]]		)
        lowerCAmelCase__				:  str		       =     tf.not_equal(__UpperCAmelCase      ,				model.config.pad_token_id		)
        with tf.GradientTape():
          lowerCAmelCase__				:  Dict		       =     model(input_ids=__UpperCAmelCase      ,				attention_mask=__UpperCAmelCase		).last_hidden_state
        lowerCAmelCase__				:  int		       =     (1, 11, 512)
        self.assertEqual(output.shape      ,				__UpperCAmelCase		)
        lowerCAmelCase__				:  Dict		       =     tf.constant(
            [[-0.2873, -1.9218, -0.3033], [-1.2710, -0.1338, -0.1902], [0.4095, 0.1214, -1.3121]]		)
        self.assertTrue(np.allclose(output[:, :3, :3]      ,				__UpperCAmelCase      ,				atol=4e-3		)		)

        lowerCAmelCase__				:  Any		       =     tf.function(__UpperCAmelCase      ,				jit_compile=__UpperCAmelCase		)
        lowerCAmelCase__				:  Tuple		       =     xla_generate(__UpperCAmelCase      ,				__UpperCAmelCase		)[0]
        self.assertTrue(np.allclose(output[:, :3, :3]      ,				__UpperCAmelCase      ,				atol=4e-2		)		)






@require_tf
@slow
class  _lowerCAmelCase     (  unittest.TestCase	):

      def 	__magic_name__(   self		):
        super().setUp()
        lowerCAmelCase__				:  str		       =     '''facebook/opt-350m'''







      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  List[str]		       =     TFOPTForCausalLM.from_pretrained(self.path_model		)
        lowerCAmelCase__				:  List[Any]		       =     GPTaTokenizer.from_pretrained(self.path_model		)

        lowerCAmelCase__				:  Any		       =     [
            '''Today is a beautiful day and I want to''',
            '''In the city of''',
            '''Paris is the capital of France and''',
            '''Computers and mobile phones have taken''',
        ]
        # verify that prompt without BOS token is identical to Metaseq -> add_special_tokens=False
        lowerCAmelCase__				:  List[str]		       =     tokenizer(__UpperCAmelCase      ,				return_tensors='''tf'''      ,				padding=__UpperCAmelCase      ,				add_special_tokens=__UpperCAmelCase		)
        lowerCAmelCase__				:  Optional[int]		       =     tf.math.reduce_mean(model(inputs.input_ids      ,				attention_mask=inputs.attention_mask		)[0]      ,				axis=-1		)
        lowerCAmelCase__				:  str		       =     tf.constant(
            [
                [1.3851, -13.8923, -10.5229, -10.7533, -0.2309, -10.2384, -0.5365, -9.0947, -5.1670],
                [-4.7073, -10.6276, -3.9415, -21.5242, -0.2822, -0.2822, -0.2822, -0.2822, -0.2822],
                [0.6247, -3.4229, -8.9179, -1.4297, -14.1650, 1.4146, -9.0218, -0.2703, -0.2703],
                [6.4783, -1.9913, -10.7926, -2.3336, 1.5092, -0.9974, -6.8213, 1.3477, 1.3477],
            ]		)
        self.assertTrue(np.allclose(__UpperCAmelCase      ,				__UpperCAmelCase      ,				atol=1e-4		)		)

        lowerCAmelCase__				:  Tuple		       =     tf.function(__UpperCAmelCase      ,				jit_compile=__UpperCAmelCase		)
        lowerCAmelCase__				:  Tuple		       =     tf.math.reduce_mean(xla_generate(inputs.input_ids      ,				attention_mask=inputs.attention_mask		)[0]      ,				axis=-1		)
        self.assertTrue(np.allclose(__UpperCAmelCase      ,				__UpperCAmelCase      ,				atol=1e-4		)		)






@require_tf
@slow
class  _lowerCAmelCase     (  unittest.TestCase	):

      @property
      def 	__magic_name__(   self		):
        return [
            "Today is a beautiful day and I want",
            "In the city of",
            "Paris is the capital of France and",
            "Computers and mobile phones have taken",
        ]

      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  Tuple		       =     '''facebook/opt-125m'''

        lowerCAmelCase__				:  str		       =     [
            '''Today is a beautiful day and I want to''',
            '''In the city of New York, the city''',
            '''Paris is the capital of France and the capital''',
            '''Computers and mobile phones have taken over the''',
        ]

        lowerCAmelCase__				:  Optional[Any]		       =     []
        lowerCAmelCase__				:  str		       =     GPTaTokenizer.from_pretrained(__UpperCAmelCase		)
        lowerCAmelCase__				:  List[str]		       =     TFOPTForCausalLM.from_pretrained(__UpperCAmelCase		)

        for prompt in self.prompts:
          lowerCAmelCase__				:  Optional[int]		       =     tokenizer(__UpperCAmelCase      ,				return_tensors='''tf'''		).input_ids

          lowerCAmelCase__				:  Dict		       =     model.generate(__UpperCAmelCase      ,				max_length=10		)

          lowerCAmelCase__				:  Tuple		       =     tokenizer.batch_decode(__UpperCAmelCase      ,				skip_special_tokens=__UpperCAmelCase		)
          predicted_outputs += generated_string

        self.assertListEqual(__UpperCAmelCase      ,				__UpperCAmelCase		)

      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  str		       =     '''facebook/opt-350m'''

        lowerCAmelCase__				:  Tuple		       =     GPTaTokenizer.from_pretrained(__UpperCAmelCase		)
        lowerCAmelCase__				:  str		       =     TFOPTForCausalLM.from_pretrained(__UpperCAmelCase		)

        lowerCAmelCase__				:  Dict		       =     '''left'''

        # use different length sentences to test batching
        lowerCAmelCase__				:  str		       =     [
            '''Hello, my dog is a little''',
            '''Today, I''',
        ]

        lowerCAmelCase__				:  Tuple		       =     tokenizer(__UpperCAmelCase      ,				return_tensors='''tf'''      ,				padding=__UpperCAmelCase		)
        lowerCAmelCase__				:  Dict		       =     inputs['''input_ids''']

        lowerCAmelCase__				:  Optional[int]		       =     model.generate(input_ids=__UpperCAmelCase      ,				attention_mask=inputs['''attention_mask''']		)

        lowerCAmelCase__				:  Union[str, Any]		       =     tokenizer(sentences[0]      ,				return_tensors='''tf'''		).input_ids
        lowerCAmelCase__				:  str		       =     model.generate(input_ids=__UpperCAmelCase		)

        lowerCAmelCase__				:  List[str]		       =     inputs_non_padded.shape[-1] - tf.math.reduce_sum(
            tf.cast(inputs['''attention_mask'''][-1]      ,				tf.intaa		)		)
        lowerCAmelCase__				:  Tuple		       =     tokenizer(sentences[1]      ,				return_tensors='''tf'''		).input_ids
        lowerCAmelCase__				:  int		       =     model.generate(input_ids=__UpperCAmelCase      ,				max_length=model.config.max_length - num_paddings		)

        lowerCAmelCase__				:  Optional[Any]		       =     tokenizer.batch_decode(__UpperCAmelCase      ,				skip_special_tokens=__UpperCAmelCase		)
        lowerCAmelCase__				:  int		       =     tokenizer.decode(output_non_padded[0]      ,				skip_special_tokens=__UpperCAmelCase		)
        lowerCAmelCase__				:  Dict		       =     tokenizer.decode(output_padded[0]      ,				skip_special_tokens=__UpperCAmelCase		)

        lowerCAmelCase__				:  Any		       =     [
            '''Hello, my dog is a little bit of a dork.\nI\'m a little bit''',
            '''Today, I was in the middle of a conversation with a friend about the''',
        ]
        self.assertListEqual(__UpperCAmelCase      ,				__UpperCAmelCase		)
        self.assertListEqual(__UpperCAmelCase      ,				[non_padded_sentence, padded_sentence]		)







      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  List[Any]		       =     '''facebook/opt-350m'''

        lowerCAmelCase__				:  Optional[int]		       =     [
            '''Today is a beautiful day and I want to''',
            '''In the city of San Francisco, the city''',
            '''Paris is the capital of France and the capital''',
            '''Computers and mobile phones have taken over the''',
        ]

        lowerCAmelCase__				:  List[str]		       =     []
        lowerCAmelCase__				:  int		       =     GPTaTokenizer.from_pretrained(__UpperCAmelCase		)
        lowerCAmelCase__				:  Optional[Any]		       =     TFOPTForCausalLM.from_pretrained(__UpperCAmelCase		)

        for prompt in self.prompts:
          lowerCAmelCase__				:  Optional[Any]		       =     tokenizer(__UpperCAmelCase      ,				return_tensors='''tf'''		).input_ids

          lowerCAmelCase__				:  Dict		       =     model.generate(__UpperCAmelCase      ,				max_length=10		)

          lowerCAmelCase__				:  Union[str, Any]		       =     tokenizer.batch_decode(__UpperCAmelCase      ,				skip_special_tokens=__UpperCAmelCase		)
          predicted_outputs += generated_string

        self.assertListEqual(__UpperCAmelCase      ,				__UpperCAmelCase		)







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 get_down_block, get_mid_block, get_out_block, get_up_block






@dataclass
class  _lowerCAmelCase     (  _lowercase	):
      A__		     =		42






class  _lowerCAmelCase     (  _lowercase    ,			_lowercase	):

      @register_to_config
      def __init__(   self      ,				__UpperCAmelCase = 6_5536      ,				__UpperCAmelCase = None      ,				__UpperCAmelCase = 2      ,				__UpperCAmelCase = 2      ,				__UpperCAmelCase = 0      ,				__UpperCAmelCase = "fourier"      ,				__UpperCAmelCase = True      ,				__UpperCAmelCase = False      ,				__UpperCAmelCase = 0.0      ,				__UpperCAmelCase = ("DownBlock1DNoSkip", "DownBlock1D", "AttnDownBlock1D")      ,				__UpperCAmelCase = ("AttnUpBlock1D", "UpBlock1D", "UpBlock1DNoSkip")      ,				__UpperCAmelCase = "UNetMidBlock1D"      ,				__UpperCAmelCase = None      ,				__UpperCAmelCase = (32, 32, 64)      ,				__UpperCAmelCase = None      ,				__UpperCAmelCase = 8      ,				__UpperCAmelCase = 1      ,				__UpperCAmelCase = False      ,				):
        super().__init__()
        lowerCAmelCase__				:  Dict		       =     sample_size

        # time
        if time_embedding_type == "fourier":
          lowerCAmelCase__				:  str		       =     GaussianFourierProjection(
              embedding_size=8      ,				set_W_to_weight=__UpperCAmelCase      ,				log=__UpperCAmelCase      ,				flip_sin_to_cos=__UpperCAmelCase		)
          lowerCAmelCase__				:  int		       =     2 * block_out_channels[0]
        elif time_embedding_type == "positional":
          lowerCAmelCase__				:  int		       =     Timesteps(
              block_out_channels[0]      ,				flip_sin_to_cos=__UpperCAmelCase      ,				downscale_freq_shift=__UpperCAmelCase		)
          lowerCAmelCase__				:  str		       =     block_out_channels[0]

        if use_timestep_embedding:
          lowerCAmelCase__				:  str		       =     block_out_channels[0] * 4
          lowerCAmelCase__				:  Dict		       =     TimestepEmbedding(
              in_channels=__UpperCAmelCase      ,				time_embed_dim=__UpperCAmelCase      ,				act_fn=__UpperCAmelCase      ,				out_dim=block_out_channels[0]      ,				)

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

        # down
        lowerCAmelCase__				:  List[Any]		       =     in_channels
        for i, down_block_type in enumerate(__UpperCAmelCase		):
          lowerCAmelCase__				:  List[str]		       =     output_channel
          lowerCAmelCase__				:  Any		       =     block_out_channels[i]

          if i == 0:
            input_channel += extra_in_channels

          lowerCAmelCase__				:  Any		       =     i == len(__UpperCAmelCase		) - 1

          lowerCAmelCase__				:  Any		       =     get_down_block(
              __UpperCAmelCase      ,				num_layers=__UpperCAmelCase      ,				in_channels=__UpperCAmelCase      ,				out_channels=__UpperCAmelCase      ,				temb_channels=block_out_channels[0]      ,				add_downsample=not is_final_block or downsample_each_block      ,				)
          self.down_blocks.append(__UpperCAmelCase		)

        # mid
        lowerCAmelCase__				:  Optional[int]		       =     get_mid_block(
            __UpperCAmelCase      ,				in_channels=block_out_channels[-1]      ,				mid_channels=block_out_channels[-1]      ,				out_channels=block_out_channels[-1]      ,				embed_dim=block_out_channels[0]      ,				num_layers=__UpperCAmelCase      ,				add_downsample=__UpperCAmelCase      ,				)

        # up
        lowerCAmelCase__				:  Union[str, Any]		       =     list(reversed(__UpperCAmelCase		)		)
        lowerCAmelCase__				:  List[str]		       =     reversed_block_out_channels[0]
        if out_block_type is None:
          lowerCAmelCase__				:  Optional[int]		       =     out_channels
        else:
          lowerCAmelCase__				:  int		       =     block_out_channels[0]

        for i, up_block_type in enumerate(__UpperCAmelCase		):
          lowerCAmelCase__				:  Optional[int]		       =     output_channel
          lowerCAmelCase__				:  Any		       =     (
              reversed_block_out_channels[i + 1] if i < len(__UpperCAmelCase		) - 1 else final_upsample_channels
          )

          lowerCAmelCase__				:  Optional[int]		       =     i == len(__UpperCAmelCase		) - 1

          lowerCAmelCase__				:  Optional[int]		       =     get_up_block(
              __UpperCAmelCase      ,				num_layers=__UpperCAmelCase      ,				in_channels=__UpperCAmelCase      ,				out_channels=__UpperCAmelCase      ,				temb_channels=block_out_channels[0]      ,				add_upsample=not is_final_block      ,				)
          self.up_blocks.append(__UpperCAmelCase		)
          lowerCAmelCase__				:  int		       =     output_channel

        # out
        lowerCAmelCase__				:  Union[str, Any]		       =     norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4      ,				32		)
        lowerCAmelCase__				:  Dict		       =     get_out_block(
            out_block_type=__UpperCAmelCase      ,				num_groups_out=__UpperCAmelCase      ,				embed_dim=block_out_channels[0]      ,				out_channels=__UpperCAmelCase      ,				act_fn=__UpperCAmelCase      ,				fc_dim=block_out_channels[-1] // 4      ,				)







      def 	__magic_name__(   self      ,				__UpperCAmelCase      ,				__UpperCAmelCase      ,				__UpperCAmelCase = True      ,				):
        lowerCAmelCase__				:  int		       =     timestep
        if not torch.is_tensor(__UpperCAmelCase		):
          lowerCAmelCase__				:  Any		       =     torch.tensor([timesteps]      ,				dtype=torch.long      ,				device=sample.device		)
        elif torch.is_tensor(__UpperCAmelCase		) and len(timesteps.shape		) == 0:
          lowerCAmelCase__				:  Any		       =     timesteps[None].to(sample.device		)

        lowerCAmelCase__				:  int		       =     self.time_proj(__UpperCAmelCase		)
        if self.config.use_timestep_embedding:
          lowerCAmelCase__				:  str		       =     self.time_mlp(__UpperCAmelCase		)
        else:
          lowerCAmelCase__				:  Tuple		       =     timestep_embed[..., None]
          lowerCAmelCase__				:  Union[str, Any]		       =     timestep_embed.repeat([1, 1, sample.shape[2]]		).to(sample.dtype		)
          lowerCAmelCase__				:  Any		       =     timestep_embed.broadcast_to((sample.shape[:1] + timestep_embed.shape[1:])		)

        # 2. down
        lowerCAmelCase__				:  int		       =     ()
        for downsample_block in self.down_blocks:
          lowerCAmelCase__       ,			lowerCAmelCase__				:  int		       =     downsample_block(hidden_states=__UpperCAmelCase      ,				temb=__UpperCAmelCase		)
          down_block_res_samples += res_samples

        # 3. mid
        if self.mid_block:
          lowerCAmelCase__				:  Dict		       =     self.mid_block(__UpperCAmelCase      ,				__UpperCAmelCase		)

        # 4. up
        for i, upsample_block in enumerate(self.up_blocks		):
          lowerCAmelCase__				:  Tuple		       =     down_block_res_samples[-1:]
          lowerCAmelCase__				:  int		       =     down_block_res_samples[:-1]
          lowerCAmelCase__				:  Optional[int]		       =     upsample_block(__UpperCAmelCase      ,				res_hidden_states_tuple=__UpperCAmelCase      ,				temb=__UpperCAmelCase		)

        # 5. post-process
        if self.out_block:
          lowerCAmelCase__				:  Any		       =     self.out_block(__UpperCAmelCase      ,				__UpperCAmelCase		)

        if not return_dict:
          return (sample,)

        return UNetaDOutput(sample=__UpperCAmelCase		)

