
"""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, apply_forward_hook
from .modeling_utils import ModelMixin
from .vae import Decoder, DecoderOutput, Encoder, VectorQuantizer
@dataclass
class 			UpperCamelCase      (							__SCREAMING_SNAKE_CASE  ):
     A__      =			42
class 			UpperCamelCase      (							__SCREAMING_SNAKE_CASE      ,       __SCREAMING_SNAKE_CASE  ):
     @register_to_config
     def __init__(     self  ,     snake_case__ = 3  ,     snake_case__ = 3  ,     snake_case__ = ("DownEncoderBlock2D",)  ,     snake_case__ = ("UpDecoderBlock2D",)  ,     snake_case__ = (64,)  ,     snake_case__ = 1  ,     snake_case__ = "silu"  ,     snake_case__ = 3  ,     snake_case__ = 32  ,     snake_case__ = 256  ,     snake_case__ = 32  ,     snake_case__ = None  ,     snake_case__ = 0.18_215  ,     snake_case__ = "group"  ,     ):




       """simple docstring"""


       super().__init__()

       # pass init params to Encoder
       _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = Encoder(
           in_channels=snake_case__  ,     out_channels=snake_case__  ,     down_block_types=snake_case__  ,     block_out_channels=snake_case__  ,     layers_per_block=snake_case__  ,     act_fn=snake_case__  ,     norm_num_groups=snake_case__  ,     double_z=snake_case__  ,     )

       _SCREAMING_SNAKE_CASE   :				Dict							    = vq_embed_dim if vq_embed_dim is not None else latent_channels

       _SCREAMING_SNAKE_CASE   :				Tuple							    = nn.Convad(snake_case__  ,     snake_case__  ,     1				)
       _SCREAMING_SNAKE_CASE   :				Optional[int]							    = VectorQuantizer(snake_case__  ,     snake_case__  ,     beta=0.25  ,     remap=snake_case__  ,     sane_index_shape=snake_case__				)
       _SCREAMING_SNAKE_CASE   :				List[str]							    = nn.Convad(snake_case__  ,     snake_case__  ,     1				)

       # pass init params to Decoder
       _SCREAMING_SNAKE_CASE   :				Optional[int]							    = Decoder(
           in_channels=snake_case__  ,     out_channels=snake_case__  ,     up_block_types=snake_case__  ,     block_out_channels=snake_case__  ,     layers_per_block=snake_case__  ,     act_fn=snake_case__  ,     norm_num_groups=snake_case__  ,     norm_type=snake_case__  ,     )
     @apply_forward_hook
     def 				__SCREAMING_SNAKE_CASE							(     self  ,     snake_case__  ,     snake_case__ = True				):




       """simple docstring"""


       _SCREAMING_SNAKE_CASE   :				List[str]							    = self.encoder(snake_case__				)
       _SCREAMING_SNAKE_CASE   :				Dict							    = self.quant_conv(snake_case__				)

       if not return_dict:
         return (h,)

       return VQEncoderOutput(latents=snake_case__				)
     @apply_forward_hook
     def 				__SCREAMING_SNAKE_CASE							(     self  ,     snake_case__  ,     snake_case__ = False  ,     snake_case__ = True				):




       """simple docstring"""


       if not force_not_quantize:
         _SCREAMING_SNAKE_CASE    ,		_SCREAMING_SNAKE_CASE    ,		_SCREAMING_SNAKE_CASE   :				int							    = self.quantize(snake_case__				)
       else:
         _SCREAMING_SNAKE_CASE   :				int							    = h
       _SCREAMING_SNAKE_CASE   :				Dict							    = self.post_quant_conv(snake_case__				)
       _SCREAMING_SNAKE_CASE   :				int							    = self.decoder(snake_case__  ,     quant if self.config.norm_type == "spatial" else None				)

       if not return_dict:
         return (dec,)

       return DecoderOutput(sample=snake_case__				)




     def 				__SCREAMING_SNAKE_CASE							(     self  ,     snake_case__  ,     snake_case__ = True				):




       """simple docstring"""


       _SCREAMING_SNAKE_CASE   :				Tuple							    = sample
       _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = self.encode(snake_case__				).latents
       _SCREAMING_SNAKE_CASE   :				Any							    = self.decode(snake_case__				).sample

       if not return_dict:
         return (dec,)

       return DecoderOutput(sample=snake_case__				)





"""simple docstring"""
import gc
import unittest

import numpy as np
import torch

from diffusers import AutoencoderKL, DDIMScheduler, DiTPipeline, DPMSolverMultistepScheduler, TransformeraDModel
from diffusers.utils import is_xformers_available, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    CLASS_CONDITIONED_IMAGE_GENERATION_BATCH_PARAMS,
    CLASS_CONDITIONED_IMAGE_GENERATION_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()
class 			UpperCamelCase      (							__SCREAMING_SNAKE_CASE      ,       unittest.TestCase  ):
     A__      =			DiTPipeline
     A__      =			CLASS_CONDITIONED_IMAGE_GENERATION_PARAMS
     A__      =			PipelineTesterMixin.required_optional_params - {
         """latents""",
         """num_images_per_prompt""",
         """callback""",
         """callback_steps""",
     }
     A__      =			CLASS_CONDITIONED_IMAGE_GENERATION_BATCH_PARAMS
     A__      =			False
     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       torch.manual_seed(0				)
       _SCREAMING_SNAKE_CASE   :				List[str]							    = TransformeraDModel(
           sample_size=16  ,     num_layers=2  ,     patch_size=4  ,     attention_head_dim=8  ,     num_attention_heads=2  ,     in_channels=4  ,     out_channels=8  ,     attention_bias=snake_case__  ,     activation_fn="gelu-approximate"  ,     num_embeds_ada_norm=1000  ,     norm_type="ada_norm_zero"  ,     norm_elementwise_affine=snake_case__  ,     )
       _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = AutoencoderKL()
       _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = DDIMScheduler()
       _SCREAMING_SNAKE_CASE   :				Dict							    = {"transformer": transformer.eval(), "vae": vae.eval(), "scheduler": scheduler}
       return components
     def 				__SCREAMING_SNAKE_CASE							(     self  ,     snake_case__  ,     snake_case__=0				):




       """simple docstring"""


       if str(snake_case__				).startswith("mps"				):
         _SCREAMING_SNAKE_CASE   :				Tuple							    = torch.manual_seed(snake_case__				)
       else:
         _SCREAMING_SNAKE_CASE   :				List[Any]							    = torch.Generator(device=snake_case__				).manual_seed(snake_case__				)
       _SCREAMING_SNAKE_CASE   :				List[str]							    = {
           "class_labels": [1],
           "generator": generator,
           "num_inference_steps": 2,
           "output_type": "numpy",
       }
       return inputs
     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       _SCREAMING_SNAKE_CASE   :				Dict							    = "cpu"

       _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = self.get_dummy_components()
       _SCREAMING_SNAKE_CASE   :				List[Any]							    = self.pipeline_class(**snake_case__				)
       pipe.to(snake_case__				)
       pipe.set_progress_bar_config(disable=snake_case__				)

       _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = self.get_dummy_inputs(snake_case__				)
       _SCREAMING_SNAKE_CASE   :				List[str]							    = pipe(**snake_case__				).images
       _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = image[0, -3:, -3:, -1]

       self.assertEqual(image.shape  ,     (1, 16, 16, 3)				)
       _SCREAMING_SNAKE_CASE   :				List[str]							    = np.array([0.2_946, 0.6_601, 0.4_329, 0.3_296, 0.4_144, 0.5_319, 0.7_273, 0.5_013, 0.4_457]				)
       _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = np.abs(image_slice.flatten() - expected_slice				).max()
       self.assertLessEqual(snake_case__  ,     1E-3				)
     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       self._test_inference_batch_single_identical(relax_max_difference=snake_case__  ,     expected_max_diff=1E-3				)




     @unittest.skipIf(
         torch_device != "cuda" or not is_xformers_available()  ,     reason="XFormers attention is only available with CUDA and `xformers` installed"  ,     )
     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1E-3				)
@require_torch_gpu
@slow
class 			UpperCamelCase      (							unittest.TestCase  ):
     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()
     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       _SCREAMING_SNAKE_CASE   :				str							    = torch.manual_seed(0				)

       _SCREAMING_SNAKE_CASE   :				str							    = DiTPipeline.from_pretrained("facebook/DiT-XL-2-256"				)
       pipe.to("cuda"				)

       _SCREAMING_SNAKE_CASE   :				Optional[int]							    = ["vase", "umbrella", "white shark", "white wolf"]
       _SCREAMING_SNAKE_CASE   :				Any							    = pipe.get_label_ids(snake_case__				)

       _SCREAMING_SNAKE_CASE   :				List[str]							    = pipe(snake_case__  ,     generator=snake_case__  ,     num_inference_steps=40  ,     output_type="np"				).images

       for word, image in zip(snake_case__  ,     snake_case__				):
         _SCREAMING_SNAKE_CASE   :				int							    = load_numpy(
             F'''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/dit/{word}.npy'''				)
         assert np.abs((expected_image - image).max()				) < 1E-2




     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       _SCREAMING_SNAKE_CASE   :				Any							    = DiTPipeline.from_pretrained("facebook/DiT-XL-2-512"				)
       _SCREAMING_SNAKE_CASE   :				Dict							    = DPMSolverMultistepScheduler.from_config(pipe.scheduler.config				)
       pipe.to("cuda"				)

       _SCREAMING_SNAKE_CASE   :				List[Any]							    = ["vase", "umbrella"]
       _SCREAMING_SNAKE_CASE   :				int							    = pipe.get_label_ids(snake_case__				)

       _SCREAMING_SNAKE_CASE   :				str							    = torch.manual_seed(0				)
       _SCREAMING_SNAKE_CASE   :				Tuple							    = pipe(snake_case__  ,     generator=snake_case__  ,     num_inference_steps=25  ,     output_type="np"				).images

       for word, image in zip(snake_case__  ,     snake_case__				):
         _SCREAMING_SNAKE_CASE   :				Tuple							    = load_numpy(
             "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
             F'''/dit/{word}_512.npy'''				)

         assert np.abs((expected_image - image).max()				) < 1E-1



