




'''simple docstring'''





import gc
import importlib.metadata
import tempfile
import unittest

from packaging import version

from transformers import (
    AutoModel,
    AutoModelForCausalLM,
    AutoModelForSeqaSeqLM,
    AutoModelForSequenceClassification,
    AutoTokenizer,
    BitsAndBytesConfig,
    pipeline,
)
from transformers.testing_utils import (
    is_torch_available,
    require_accelerate,
    require_bitsandbytes,
    require_torch,
    require_torch_gpu,
    require_torch_multi_gpu,
    slow,
)



def      __lowerCAmelCase			(					UpperCamelCase      )     ->	Any:
  if model.config.model_type == "gpt2":
    return model.transformer.h[0].mlp.c_fc
  return model.transformer.h[0].mlp.dense_ah_to_h


if is_torch_available():
     import torch
     import torch.nn as nn






     class  _lowerCAmelCase     (  nn.Module	):

           def __init__(   self      ,				__UpperCAmelCase      ,				__UpperCAmelCase		):
             super().__init__()
             lowerCAmelCase__				:  List[Any]		       =     module
             lowerCAmelCase__				:  str		       =     nn.Sequential(
                 nn.Linear(module.in_features      ,				__UpperCAmelCase      ,				bias=__UpperCAmelCase		)      ,				nn.Linear(__UpperCAmelCase      ,				module.out_features      ,				bias=__UpperCAmelCase		)      ,				)
             lowerCAmelCase__				:  Optional[Any]		       =     (2.0 / (5 * min(module.in_features      ,				module.out_features		))) ** 0.5
             nn.init.normal_(self.adapter[0].weight      ,				std=__UpperCAmelCase		)
             nn.init.zeros_(self.adapter[1].weight		)
             self.adapter.to(module.weight.device		)







           def 	__magic_name__(   self      ,				__UpperCAmelCase      ,				*__UpperCAmelCase      ,				**__UpperCAmelCase		):
             return self.module(__UpperCAmelCase      ,				*__UpperCAmelCase      ,				**__UpperCAmelCase		) + self.adapter(__UpperCAmelCase		)






@require_bitsandbytes
@require_accelerate
@require_torch
@require_torch_gpu
@slow
class  _lowerCAmelCase     (  unittest.TestCase	):
      A__		     =		'bigscience/bloom-1b7'

      # Constant values
      A__		     =		2.109659552692574

      A__		     =		'Hello my name is'
      A__		     =		set()
      EXPECTED_OUTPUTS.add('Hello my name is John and I am a professional photographer. I'	)
      EXPECTED_OUTPUTS.add('Hello my name is John.\nI am a friend of your father.\n'	)
      EXPECTED_OUTPUTS.add('Hello my name is John Doe, I am a student at the University'	)
      A__		     =		10

      def 	__magic_name__(   self		):
        # Models and tokenizer
        lowerCAmelCase__				:  Optional[Any]		       =     AutoTokenizer.from_pretrained(self.model_name		)






class  _lowerCAmelCase     (  lowercase__	):

      def 	__magic_name__(   self		):
        super().setUp()

        # Models and tokenizer
        lowerCAmelCase__				:  Optional[int]		       =     AutoModelForCausalLM.from_pretrained(
            self.model_name      ,				torch_dtype=torch.floataa      ,				device_map='''auto'''		)
        lowerCAmelCase__				:  Any		       =     AutoModelForCausalLM.from_pretrained(self.model_name      ,				load_in_abit=__UpperCAmelCase      ,				device_map='''auto'''		)

      def 	__magic_name__(   self		):
        del self.model_fpaa
        del self.model_abit

        gc.collect()
        torch.cuda.empty_cache()

      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  int		       =     self.model_abit.config

        self.assertTrue(hasattr(__UpperCAmelCase      ,				'''quantization_config'''		)		)

        lowerCAmelCase__				:  Any		       =     config.to_dict()
        lowerCAmelCase__				:  Union[str, Any]		       =     config.to_diff_dict()

        lowerCAmelCase__				:  str		       =     config.to_json_string()

      def 	__magic_name__(   self		):
        from bitsandbytes.nn import Paramsabit

        lowerCAmelCase__				:  List[Any]		       =     self.model_fpaa.get_memory_footprint()
        lowerCAmelCase__				:  List[Any]		       =     self.model_abit.get_memory_footprint()

        self.assertAlmostEqual(mem_fpaa / mem_abit      ,				self.EXPECTED_RELATIVE_DIFFERENCE		)
        lowerCAmelCase__				:  Dict		       =     get_some_linear_layer(self.model_abit		)
        self.assertTrue(linear.weight.__class__ == Paramsabit		)

      def 	__magic_name__(   self		):
        from transformers import TaPreTrainedModel

        self.model_fpaa.get_memory_footprint()
        self.model_abit.get_memory_footprint()

        for name, module in self.model_abit.named_modules():
          if isinstance(__UpperCAmelCase      ,				torch.nn.Linear		):
            if name not in ["lm_head"] + TaPreTrainedModel._keep_in_fpaa_modules:
              # 4-bit parameters are packed in uint8 variables
              self.assertTrue(module.weight.dtype == torch.uinta		)

      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  Optional[Any]		       =     self.tokenizer(self.input_text      ,				return_tensors='''pt'''		)
        lowerCAmelCase__				:  int		       =     self.model_abit.generate(input_ids=encoded_input['''input_ids'''].to(0		)      ,				max_new_tokens=10		)

        self.assertIn(self.tokenizer.decode(output_sequences[0]      ,				skip_special_tokens=__UpperCAmelCase		)      ,				self.EXPECTED_OUTPUTS		)

      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  Tuple		       =     BitsAndBytesConfig()
        lowerCAmelCase__				:  str		       =     True

        lowerCAmelCase__				:  int		       =     AutoModelForCausalLM.from_pretrained(
            self.model_name      ,				quantization_config=__UpperCAmelCase      ,				device_map='''auto'''		)

        lowerCAmelCase__				:  str		       =     self.tokenizer(self.input_text      ,				return_tensors='''pt'''		)
        lowerCAmelCase__				:  Optional[int]		       =     model_abit_from_config.generate(
            input_ids=encoded_input['''input_ids'''].to(0		)      ,				max_new_tokens=10		)

        self.assertIn(self.tokenizer.decode(output_sequences[0]      ,				skip_special_tokens=__UpperCAmelCase		)      ,				self.EXPECTED_OUTPUTS		)

      def 	__magic_name__(   self		):
        with self.assertRaises(__UpperCAmelCase		), tempfile.TemporaryDirectory() as tmpdirname:
          self.model_abit.save_pretrained(__UpperCAmelCase		)

      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  Union[str, Any]		       =     BitsAndBytesConfig()

        with self.assertRaises(__UpperCAmelCase		):
          lowerCAmelCase__				:  str		       =     AutoModelForCausalLM.from_pretrained(
              self.model_name      ,				quantization_config=__UpperCAmelCase      ,				load_in_abit=__UpperCAmelCase      ,				device_map='''auto'''      ,				bnb_abit_quant_type='''nf4'''      ,				)

      def 	__magic_name__(   self		):
        with self.assertRaises(__UpperCAmelCase		):
          # Tries with `str`
          self.model_abit.to('''cpu'''		)

        with self.assertRaises(__UpperCAmelCase		):
          # Tries with a `dtype``
          self.model_abit.to(torch.floataa		)

        with self.assertRaises(__UpperCAmelCase		):
          # Tries with a `device`
          self.model_abit.to(torch.device('''cuda:0'''		)		)

        with self.assertRaises(__UpperCAmelCase		):
          # Tries with a `device`
          self.model_abit.float()

        with self.assertRaises(__UpperCAmelCase		):
          # Tries with a `device`
          self.model_abit.half()

        # Test if we did not break anything
        lowerCAmelCase__				:  Any		       =     self.tokenizer(self.input_text      ,				return_tensors='''pt'''		)

        lowerCAmelCase__				:  str		       =     self.model_fpaa.to(torch.floataa		)
        lowerCAmelCase__				:  Optional[int]		       =     self.model_fpaa.generate(input_ids=encoded_input['''input_ids'''].to(0		)      ,				max_new_tokens=10		)

        # Check this does not throw an error
        lowerCAmelCase__				:  Union[str, Any]		       =     self.model_fpaa.to('''cpu'''		)

        # Check this does not throw an error
        lowerCAmelCase__				:  Optional[int]		       =     self.model_fpaa.half()

        # Check this does not throw an error
        lowerCAmelCase__				:  str		       =     self.model_fpaa.float()







      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  Any		       =     AutoModelForSeqaSeqLM.from_pretrained('''t5-small'''      ,				load_in_abit=__UpperCAmelCase      ,				device_map='''auto'''		)
        self.assertTrue(model.decoder.block[0].layer[2].DenseReluDense.wo.weight.dtype == torch.floataa		)






@require_bitsandbytes
@require_accelerate
@require_torch
@require_torch_gpu
@slow
class  _lowerCAmelCase     (  unittest.TestCase	):

      @classmethod
      def 	__magic_name__(   cls		):
        lowerCAmelCase__				:  Dict		       =     '''t5-small'''
        lowerCAmelCase__				:  int		       =     '''google/flan-t5-small'''  # flan-t5 uses dense-act instead of dense-relu-dense
        lowerCAmelCase__				:  int		       =     AutoTokenizer.from_pretrained(cls.model_name		)
        lowerCAmelCase__				:  Union[str, Any]		       =     '''Translate in German: Hello, my dog is cute'''

      def 	__magic_name__(   self		):
        gc.collect()
        torch.cuda.empty_cache()

      def 	__magic_name__(   self		):
        from transformers import TaForConditionalGeneration

        lowerCAmelCase__				:  str		       =     TaForConditionalGeneration._keep_in_fpaa_modules
        lowerCAmelCase__				:  List[Any]		       =     None

        # test with `t5-small`
        lowerCAmelCase__				:  Tuple		       =     TaForConditionalGeneration.from_pretrained(self.model_name      ,				load_in_abit=__UpperCAmelCase      ,				device_map='''auto'''		)
        lowerCAmelCase__				:  List[Any]		       =     self.tokenizer(self.input_text      ,				return_tensors='''pt'''		).to(0		)
        lowerCAmelCase__				:  Optional[Any]		       =     model.generate(**__UpperCAmelCase		)

        # test with `flan-t5-small`
        lowerCAmelCase__				:  int		       =     TaForConditionalGeneration.from_pretrained(
            self.dense_act_model_name      ,				load_in_abit=__UpperCAmelCase      ,				device_map='''auto'''		)
        lowerCAmelCase__				:  Optional[int]		       =     self.tokenizer(self.input_text      ,				return_tensors='''pt'''		).to(0		)
        lowerCAmelCase__				:  Any		       =     model.generate(**__UpperCAmelCase		)
        lowerCAmelCase__				:  int		       =     modules







      def 	__magic_name__(   self		):
        import bitsandbytes as bnb

        from transformers import TaForConditionalGeneration

        # test with `t5-small`
        lowerCAmelCase__				:  Any		       =     TaForConditionalGeneration.from_pretrained(self.model_name      ,				load_in_abit=__UpperCAmelCase      ,				device_map='''auto'''		)

        # there was a bug with decoders - this test checks that it is fixed
        self.assertTrue(isinstance(model.decoder.block[0].layer[0].SelfAttention.q      ,				bnb.nn.Linearabit		)		)

        lowerCAmelCase__				:  Union[str, Any]		       =     self.tokenizer(self.input_text      ,				return_tensors='''pt'''		).to(0		)
        lowerCAmelCase__				:  List[str]		       =     model.generate(**__UpperCAmelCase		)

        # test with `flan-t5-small`
        lowerCAmelCase__				:  str		       =     TaForConditionalGeneration.from_pretrained(
            self.dense_act_model_name      ,				load_in_abit=__UpperCAmelCase      ,				device_map='''auto'''		)
        lowerCAmelCase__				:  Dict		       =     self.tokenizer(self.input_text      ,				return_tensors='''pt'''		).to(0		)
        lowerCAmelCase__				:  Any		       =     model.generate(**__UpperCAmelCase		)






class  _lowerCAmelCase     (  lowercase__	):

      def 	__magic_name__(   self		):
        super().setUp()
        # model_name
        lowerCAmelCase__				:  Optional[Any]		       =     '''bigscience/bloom-560m'''
        lowerCAmelCase__				:  List[str]		       =     '''t5-small'''

        # Different types of model

        lowerCAmelCase__				:  List[str]		       =     AutoModel.from_pretrained(self.model_name      ,				load_in_abit=__UpperCAmelCase      ,				device_map='''auto'''		)
        # Sequence classification model
        lowerCAmelCase__				:  Dict		       =     AutoModelForSequenceClassification.from_pretrained(
            self.model_name      ,				load_in_abit=__UpperCAmelCase      ,				device_map='''auto'''		)
        # CausalLM model
        lowerCAmelCase__				:  Any		       =     AutoModelForCausalLM.from_pretrained(self.model_name      ,				load_in_abit=__UpperCAmelCase      ,				device_map='''auto'''		)
        # Seq2seq model
        lowerCAmelCase__				:  List[str]		       =     AutoModelForSeqaSeqLM.from_pretrained(
            self.seq_to_seq_name      ,				load_in_abit=__UpperCAmelCase      ,				device_map='''auto'''		)

      def 	__magic_name__(   self		):
        del self.base_model
        del self.sequence_model
        del self.model_abit
        del self.seq_to_seq_model

        gc.collect()
        torch.cuda.empty_cache()







      def 	__magic_name__(   self		):
        from bitsandbytes.nn import Paramsabit

        self.assertTrue(self.base_model.h[-1].mlp.dense_ah_to_h.weight.__class__ == Paramsabit		)

        # Other heads should be nn.Parameter
        self.assertTrue(self.model_abit.lm_head.weight.__class__ == torch.nn.Parameter		)
        self.assertTrue(self.sequence_model.score.weight.__class__ == torch.nn.Parameter		)
        self.assertTrue(self.seq_to_seq_model.lm_head.weight.__class__ == torch.nn.Parameter		)






class  _lowerCAmelCase     (  lowercase__	):

      def 	__magic_name__(   self		):
        super().setUp()

      def 	__magic_name__(   self		):
        del self.pipe

        gc.collect()
        torch.cuda.empty_cache()







      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  Tuple		       =     pipeline(
            '''text-generation'''      ,				model=self.model_name      ,				model_kwargs={'''device_map''': '''auto''', '''load_in_4bit''': True, '''torch_dtype''': torch.floataa}      ,				max_new_tokens=self.MAX_NEW_TOKENS      ,				)

        # Real second forward pass
        lowerCAmelCase__				:  int		       =     self.pipe(self.input_text		)
        self.assertIn(pipeline_output[0]['''generated_text''']      ,				self.EXPECTED_OUTPUTS		)






@require_torch_multi_gpu
class  _lowerCAmelCase     (  lowercase__	):

      def 	__magic_name__(   self		):
        super().setUp()







      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  Optional[Any]		       =     AutoModelForCausalLM.from_pretrained(
            self.model_name      ,				load_in_abit=__UpperCAmelCase      ,				device_map='''balanced'''		)

        # Check correct device map
        self.assertEqual(set(model_parallel.hf_device_map.values()		)      ,				{0, 1}		)

        # Check that inference pass works on the model
        lowerCAmelCase__				:  int		       =     self.tokenizer(self.input_text      ,				return_tensors='''pt'''		)

        # Second real batch
        lowerCAmelCase__				:  Tuple		       =     model_parallel.generate(input_ids=encoded_input['''input_ids'''].to(0		)      ,				max_new_tokens=10		)
        self.assertIn(self.tokenizer.decode(output_parallel[0]      ,				skip_special_tokens=__UpperCAmelCase		)      ,				self.EXPECTED_OUTPUTS		)






class  _lowerCAmelCase     (  lowercase__	):

      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  Union[str, Any]		       =     '''facebook/opt-350m'''
        super().setUp()







      def 	__magic_name__(   self		):
        if version.parse(importlib.metadata.version('''bitsandbytes'''		)		) < version.parse('''0.37.0'''		):
          return

        # Step 1: freeze all parameters
        lowerCAmelCase__				:  int		       =     AutoModelForCausalLM.from_pretrained(self.model_name      ,				load_in_abit=__UpperCAmelCase		)

        self.assertEqual(set(model.hf_device_map.values()		)      ,				{torch.cuda.current_device()}		)

        for param in model.parameters():
          lowerCAmelCase__				:  str		       =     False  # freeze the model - train adapters later
          if param.ndim == 1:
            # cast the small parameters (e.g. layernorm) to fp32 for stability
            lowerCAmelCase__				:  str		       =     param.data.to(torch.floataa		)

        # Step 2: add adapters
        for _, module in model.named_modules():
          if "OPTAttention" in repr(type(__UpperCAmelCase		)		):
            lowerCAmelCase__				:  Union[str, Any]		       =     LoRALayer(module.q_proj      ,				rank=16		)
            lowerCAmelCase__				:  Dict		       =     LoRALayer(module.k_proj      ,				rank=16		)
            lowerCAmelCase__				:  Optional[int]		       =     LoRALayer(module.v_proj      ,				rank=16		)

        # Step 3: dummy batch
        lowerCAmelCase__				:  Dict		       =     self.tokenizer('''Test batch '''      ,				return_tensors='''pt'''		).to(0		)

        # Step 4: Check if the gradient is not None
        with torch.cuda.amp.autocast():
          lowerCAmelCase__				:  int		       =     model.forward(**__UpperCAmelCase		)
          out.logits.norm().backward()

        for module in model.modules():
          if isinstance(__UpperCAmelCase      ,				__UpperCAmelCase		):
            self.assertTrue(module.adapter[1].weight.grad is not None		)
            self.assertTrue(module.adapter[1].weight.grad.norm().item() > 0		)
          elif isinstance(__UpperCAmelCase      ,				nn.Embedding		):
            self.assertTrue(module.weight.grad is None		)






class  _lowerCAmelCase     (  lowercase__	):
      A__		     =		'gpt2-xl'
      A__		     =		3.3191854854152187







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		)

