



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

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, XLMRobertaTokenizer

from diffusers import AltDiffusionPipeline, AutoencoderKL, DDIMScheduler, PNDMScheduler, UNetaDConditionModel
from diffusers.pipelines.alt_diffusion.modeling_roberta_series import (
    RobertaSeriesConfig,
    RobertaSeriesModelWithTransformation,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()




class       snake_case_    (    __lowercase  ,__lowercase  ,__lowercase  ,unittest.TestCase					):
							__lowerCAmelCase            =       AltDiffusionPipeline
							__lowerCAmelCase            =       TEXT_TO_IMAGE_PARAMS
							__lowerCAmelCase            =       TEXT_TO_IMAGE_BATCH_PARAMS
							__lowerCAmelCase            =       TEXT_TO_IMAGE_IMAGE_PARAMS
							__lowerCAmelCase            =       TEXT_TO_IMAGE_IMAGE_PARAMS






							def snake_case_	( self   ):
									torch.manual_seed(0   )
									a_  :   Dict			    =  UNetaDConditionModel(
									    block_out_channels=(3_2, 6_4)      ,  layers_per_block=2      ,  sample_size=3_2      ,  in_channels=4      ,  out_channels=4      ,  down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")      ,  up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")      ,  cross_attention_dim=3_2      ,  )
									a_  :   Any			    =  DDIMScheduler(
									    beta_start=0.00_085      ,  beta_end=0.012      ,  beta_schedule="scaled_linear"      ,  clip_sample=a_      ,  set_alpha_to_one=a_      ,  )
									torch.manual_seed(0   )
									a_  :   Any			    =  AutoencoderKL(
									    block_out_channels=[3_2, 6_4]      ,  in_channels=3      ,  out_channels=3      ,  down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]      ,  up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]      ,  latent_channels=4      ,  )

									# TODO: address the non-deterministic text encoder (fails for save-load tests)
									# torch.manual_seed(0)
									# text_encoder_config = RobertaSeriesConfig(
									#     hidden_size=32,
									#     project_dim=32,
									#     intermediate_size=37,
									#     layer_norm_eps=1e-05,
									#     num_attention_heads=4,
									#     num_hidden_layers=5,
									#     vocab_size=5002,
									# )
									# text_encoder = RobertaSeriesModelWithTransformation(text_encoder_config)

									torch.manual_seed(0   )
									a_  :   int			    =  CLIPTextConfig(
									    bos_token_id=0      ,  eos_token_id=2      ,  hidden_size=3_2      ,  projection_dim=3_2      ,  intermediate_size=3_7      ,  layer_norm_eps=1e-05      ,  num_attention_heads=4      ,  num_hidden_layers=5      ,  pad_token_id=1      ,  vocab_size=5_0_0_2      ,  )
									a_  :   Dict			    =  CLIPTextModel(a_   )

									a_  :   Optional[int]			    =  XLMRobertaTokenizer.from_pretrained("hf-internal-testing/tiny-xlm-roberta"   )
									a_  :   str			    =  7_7

									a_  :   List[Any]			    =  {
									    """unet""": unet,
									    """scheduler""": scheduler,
									    """vae""": vae,
									    """text_encoder""": text_encoder,
									    """tokenizer""": tokenizer,
									    """safety_checker""": None,
									    """feature_extractor""": None,
									}
									return components






							def snake_case_	( self      ,  a_      ,  a_=0   ):

									if str(a_   ).startswith("mps"   ):
											a_  :   int			    =  torch.manual_seed(a_   )
									else:
											a_  :   Optional[Any]			    =  torch.Generator(device=a_   ).manual_seed(a_   )
									a_  :   List[Any]			    =  {
									    """prompt""": """A painting of a squirrel eating a burger""",
									    """generator""": generator,
									    """num_inference_steps""": 2,
									    """guidance_scale""": 6.0,
									    """output_type""": """numpy""",
									}
									return inputs






							def snake_case_	( self   ):
									super().test_attention_slicing_forward_pass(expected_max_diff=3e-3   )






							def snake_case_	( self   ):
									super().test_inference_batch_single_identical(expected_max_diff=3e-3   )






							def snake_case_	( self   ):
									a_  :   Dict			    =  """cpu"""  # ensure determinism for the device-dependent torch.Generator

									a_  :   Optional[Any]			    =  self.get_dummy_components()
									torch.manual_seed(0   )
									a_  :   str			    =  RobertaSeriesConfig(
									    hidden_size=3_2      ,  project_dim=3_2      ,  intermediate_size=3_7      ,  layer_norm_eps=1e-05      ,  num_attention_heads=4      ,  num_hidden_layers=5      ,  vocab_size=5_0_0_2      ,  )
									# TODO: remove after fixing the non-deterministic text encoder
									a_  :   Optional[int]			    =  RobertaSeriesModelWithTransformation(a_   )
									a_  :   Optional[Any]			    =  text_encoder

									a_  :   Any			    =  AltDiffusionPipeline(**a_   )
									a_  :   Any			    =  alt_pipe.to(a_   )
									alt_pipe.set_progress_bar_config(disable=a_   )

									a_  :   int			    =  self.get_dummy_inputs(a_   )
									a_  :   Union[str, Any]			    =  """A photo of an astronaut"""
									a_  :   Any			    =  alt_pipe(**a_   )
									a_  :   List[Any]			    =  output.images
									a_  :   Optional[Any]			    =  image[0, -3:, -3:, -1]

									assert image.shape == (1, 6_4, 6_4, 3)
									a_  :   int			    =  np.array(
									    [0.5_748_162, 0.60_447_145, 0.48_821_217, 0.50_100_636, 0.5_431_185, 0.45_763_683, 0.49_657_696, 0.48_132_733, 0.47_573_093]   )

									assert np.abs(image_slice.flatten() - expected_slice   ).max() < 1e-2






							def snake_case_	( self   ):
									a_  :   Any			    =  """cpu"""  # ensure determinism for the device-dependent torch.Generator

									a_  :   Any			    =  self.get_dummy_components()
									a_  :   List[Any]			    =  PNDMScheduler(skip_prk_steps=a_   )
									torch.manual_seed(0   )
									a_  :   int			    =  RobertaSeriesConfig(
									    hidden_size=3_2      ,  project_dim=3_2      ,  intermediate_size=3_7      ,  layer_norm_eps=1e-05      ,  num_attention_heads=4      ,  num_hidden_layers=5      ,  vocab_size=5_0_0_2      ,  )
									# TODO: remove after fixing the non-deterministic text encoder
									a_  :   Dict			    =  RobertaSeriesModelWithTransformation(a_   )
									a_  :   Tuple			    =  text_encoder
									a_  :   Optional[Any]			    =  AltDiffusionPipeline(**a_   )
									a_  :   Optional[int]			    =  alt_pipe.to(a_   )
									alt_pipe.set_progress_bar_config(disable=a_   )

									a_  :   Dict			    =  self.get_dummy_inputs(a_   )
									a_  :   List[Any]			    =  alt_pipe(**a_   )
									a_  :   Dict			    =  output.images
									a_  :   Any			    =  image[0, -3:, -3:, -1]

									assert image.shape == (1, 6_4, 6_4, 3)
									a_  :   str			    =  np.array(
									    [0.51_605_093, 0.5_707_241, 0.47_365_507, 0.50_578_886, 0.5_633_877, 0.4_642_503, 0.5_182_081, 0.48_763_484, 0.49_084_237]   )

									assert np.abs(image_slice.flatten() - expected_slice   ).max() < 1e-2




@slow
@require_torch_gpu
class       snake_case_    (    unittest.TestCase					):






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






							def snake_case_	( self   ):
									a_  :   List[str]			    =  AltDiffusionPipeline.from_pretrained("BAAI/AltDiffusion"      ,  safety_checker=a_   )
									a_  :   List[str]			    =  alt_pipe.to(a_   )
									alt_pipe.set_progress_bar_config(disable=a_   )

									a_  :   int			    =  """A painting of a squirrel eating a burger"""
									a_  :   Tuple			    =  torch.manual_seed(0   )
									a_  :   Dict			    =  alt_pipe([prompt]      ,  generator=a_      ,  guidance_scale=6.0      ,  num_inference_steps=2_0      ,  output_type="np"   )

									a_  :   Optional[Any]			    =  output.images

									a_  :   Optional[Any]			    =  image[0, -3:, -3:, -1]

									assert image.shape == (1, 5_1_2, 5_1_2, 3)
									a_  :   str			    =  np.array([0.1_010, 0.0_800, 0.0_794, 0.0_885, 0.0_843, 0.0_762, 0.0_769, 0.0_729, 0.0_586]   )

									assert np.abs(image_slice.flatten() - expected_slice   ).max() < 1e-2






							def snake_case_	( self   ):
									a_  :   Union[str, Any]			    =  DDIMScheduler.from_pretrained("BAAI/AltDiffusion"      ,  subfolder="scheduler"   )

									a_  :   Dict			    =  AltDiffusionPipeline.from_pretrained("BAAI/AltDiffusion"      ,  scheduler=a_      ,  safety_checker=a_   )
									a_  :   List[str]			    =  alt_pipe.to(a_   )
									alt_pipe.set_progress_bar_config(disable=a_   )

									a_  :   Dict			    =  """A painting of a squirrel eating a burger"""
									a_  :   Dict			    =  torch.manual_seed(0   )

									a_  :   Any			    =  alt_pipe([prompt]      ,  generator=a_      ,  num_inference_steps=2      ,  output_type="numpy"   )
									a_  :   List[Any]			    =  output.images

									a_  :   Tuple			    =  image[0, -3:, -3:, -1]

									assert image.shape == (1, 5_1_2, 5_1_2, 3)
									a_  :   Union[str, Any]			    =  np.array([0.4_019, 0.4_052, 0.3_810, 0.4_119, 0.3_916, 0.3_982, 0.4_651, 0.4_195, 0.5_323]   )

									assert np.abs(image_slice.flatten() - expected_slice   ).max() < 1e-2






'''simple docstring'''






from unittest import TestCase

from datasets import Sequence, Value
from datasets.arrow_dataset import Dataset




class  _A (					__lowercase		):

		def 	lowercase__     (   self					:   Any			)					->    str:


					"""simple docstring"""

					return [
					    {"col_1": 3, "col_2": "a"},
					    {"col_1": 2, "col_2": "b"},
					    {"col_1": 1, "col_2": "c"},
					    {"col_1": 0, "col_2": "d"},
					]

		def 	lowercase__     (   self					:   str			)					->    int:


					"""simple docstring"""

					__snake_case					:      Union[str, Any]							      =     {"""col_1""": [3, 2, 1, 0], """col_2""": ["""a""", """b""", """c""", """d"""]}
					return Dataset.from_dict(__magic_name__			)

		def 	lowercase__     (   self					:   str			)					->    List[Any]:


					"""simple docstring"""

					__snake_case					:      Any							      =     self._create_example_records()
					__snake_case					:      str							      =     Dataset.from_list(__magic_name__			)
					self.assertListEqual(dset.column_names  ,      ["""col_1""", """col_2"""]			)
					for i, r in enumerate(__magic_name__			):
								self.assertDictEqual(__magic_name__  ,      example_records[i]			)

		def 	lowercase__     (   self					:   Optional[Any]			)					->    List[Any]:


					"""simple docstring"""

					__snake_case					:      List[Any]							      =     self._create_example_records()
					__snake_case					:      Dict							      =     Dataset.from_list(__magic_name__			)
					__snake_case					:      List[Any]							      =     Dataset.from_dict({k: [r[k] for r in example_records] for k in example_records[0]}			)
					self.assertEqual(dset.info  ,      dset_from_dict.info			)

		def 	lowercase__     (   self					:   str			)					->    List[Any]:  # checks what happens with missing columns


					"""simple docstring"""

					__snake_case					:      Union[str, Any]							      =     [{"""col_1""": 1}, {"""col_2""": """x"""}]
					__snake_case					:      Optional[int]							      =     Dataset.from_list(__magic_name__			)
					self.assertDictEqual(dset[0]  ,      {"""col_1""": 1}			)
					self.assertDictEqual(dset[1]  ,      {"""col_1""": None}			)  # NB: first record is used for columns

		def 	lowercase__     (   self					:   List[str]			)					->    Optional[Any]:  # checks if the type can be inferred from the second record


					"""simple docstring"""

					__snake_case					:      List[Any]							      =     [{"""col_1""": []}, {"""col_1""": [1, 2]}]
					__snake_case					:      int							      =     Dataset.from_list(__magic_name__			)
					self.assertEqual(dset.info.features["""col_1"""]  ,      Sequence(Value("""int64"""			)			)			)

		def 	lowercase__     (   self					:   int			)					->    Union[str, Any]:


					"""simple docstring"""

					__snake_case					:      Tuple							      =     Dataset.from_list([]			)
					self.assertEqual(len(__magic_name__			)  ,      0			)
					self.assertListEqual(dset.column_names  ,      []			)


