




"""simple docstring"""

import gc
import random
import unittest

import numpy as np
import torch
from transformers import XLMRobertaTokenizer

from diffusers import (
    AltDiffusionImgaImgPipeline,
    AutoencoderKL,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.image_processor import VaeImageProcessor
from diffusers.pipelines.alt_diffusion.modeling_roberta_series import (
    RobertaSeriesConfig,
    RobertaSeriesModelWithTransformation,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu


enable_full_determinism()





class 							lowercase_			(	unittest.TestCase				):



 '''simple docstring'''

 def      lowerCAmelCase_				(					self			:			List[Any]						):
       # clean up the VRAM after each test
       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()

 @property
 def      lowerCAmelCase_				(					self			:			int						):
       _A     =  1
       _A     =  3
       _A     =  (32, 32)

       _A     =  floats_tensor((batch_size, num_channels) + sizes       ,				rng=random.Random(0						)						).to(__a						)
       return image

 @property
 def      lowerCAmelCase_				(					self			:			Optional[Any]						):
       torch.manual_seed(0						)
       _A     =  UNetaDConditionModel(
           block_out_channels=(32, 64)       ,				layers_per_block=2       ,				sample_size=32       ,				in_channels=4       ,				out_channels=4       ,				down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')       ,				up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')       ,				cross_attention_dim=32       ,				)
       return model

 @property
 def      lowerCAmelCase_				(					self			:			Optional[Any]						):
       torch.manual_seed(0						)
       _A     =  AutoencoderKL(
           block_out_channels=[32, 64]       ,				in_channels=3       ,				out_channels=3       ,				down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']       ,				up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']       ,				latent_channels=4       ,				)
       return model

 @property
 def      lowerCAmelCase_				(					self			:			List[Any]						):
       torch.manual_seed(0						)
       _A     =  RobertaSeriesConfig(
           hidden_size=32       ,				project_dim=32       ,				intermediate_size=37       ,				layer_norm_eps=1E-0_5       ,				num_attention_heads=4       ,				num_hidden_layers=5       ,				pad_token_id=1       ,				vocab_size=5_006       ,				)
       return RobertaSeriesModelWithTransformation(__a						)

 @property
 def      lowerCAmelCase_				(					self			:			str						):
       def extract(*_UpperCAmelCase			:			Optional[Any]       ,				**_UpperCAmelCase			:			Any						):





             class 							lowercase_			:



              '''simple docstring'''

              def __init__(					self			:			Optional[Any]						):
                    _A     =  torch.ones([0]						)



              def      lowerCAmelCase_				(					self			:			Any       ,				_UpperCAmelCase			:			List[Any]						):
                    self.pixel_values.to(__a						)
                    return self



             return Out()

       return extract

 def      lowerCAmelCase_				(					self			:			int						):
       _A     =  'cpu'  # ensure determinism for the device-dependent torch.Generator
       _A     =  self.dummy_cond_unet
       _A     =  PNDMScheduler(skip_prk_steps=__a						)
       _A     =  self.dummy_vae
       _A     =  self.dummy_text_encoder
       _A     =  XLMRobertaTokenizer.from_pretrained('hf-internal-testing/tiny-xlm-roberta'						)
       _A     =  77

       _A     =  self.dummy_image.to(__a						)
       _A     =  init_image / 2 + 0.5

       # make sure here that pndm scheduler skips prk
       _A     =  AltDiffusionImgaImgPipeline(
           unet=__a       ,				scheduler=__a       ,				vae=__a       ,				text_encoder=__a       ,				tokenizer=__a       ,				safety_checker=__a       ,				feature_extractor=self.dummy_extractor       ,				)
       _A     =  VaeImageProcessor(vae_scale_factor=alt_pipe.vae_scale_factor       ,				do_normalize=__a						)
       _A     =  alt_pipe.to(__a						)
       alt_pipe.set_progress_bar_config(disable=__a						)

       _A     =  'A painting of a squirrel eating a burger'
       _A     =  torch.Generator(device=__a						).manual_seed(0						)
       _A     =  alt_pipe(
           [prompt]       ,				generator=__a       ,				guidance_scale=6.0       ,				num_inference_steps=2       ,				output_type='np'       ,				image=__a       ,				)

       _A     =  output.images

       _A     =  torch.Generator(device=__a						).manual_seed(0						)
       _A     =  alt_pipe(
           [prompt]       ,				generator=__a       ,				guidance_scale=6.0       ,				num_inference_steps=2       ,				output_type='np'       ,				image=__a       ,				return_dict=__a       ,				)[0]

       _A     =  image[0, -3:, -3:, -1]
       _A     =  image_from_tuple[0, -3:, -3:, -1]

       assert image.shape == (1, 32, 32, 3)
       _A     =  np.array([0.4427, 0.3731, 0.4249, 0.4941, 0.4546, 0.4148, 0.4193, 0.4666, 0.4499]						)

       assert np.abs(image_slice.flatten() - expected_slice						).max() < 5E-3
       assert np.abs(image_from_tuple_slice.flatten() - expected_slice						).max() < 5E-3

 @unittest.skipIf(torch_device != 'cuda'       ,				'This test requires a GPU'						)
 def      lowerCAmelCase_				(					self			:			Union[str, Any]						):
       _A     =  self.dummy_cond_unet
       _A     =  PNDMScheduler(skip_prk_steps=__a						)
       _A     =  self.dummy_vae
       _A     =  self.dummy_text_encoder
       _A     =  XLMRobertaTokenizer.from_pretrained('hf-internal-testing/tiny-xlm-roberta'						)
       _A     =  77

       _A     =  self.dummy_image.to(__a						)

       # put models in fp16
       _A     =  unet.half()
       _A     =  vae.half()
       _A     =  bert.half()

       # make sure here that pndm scheduler skips prk
       _A     =  AltDiffusionImgaImgPipeline(
           unet=__a       ,				scheduler=__a       ,				vae=__a       ,				text_encoder=__a       ,				tokenizer=__a       ,				safety_checker=__a       ,				feature_extractor=self.dummy_extractor       ,				)
       _A     =  VaeImageProcessor(vae_scale_factor=alt_pipe.vae_scale_factor       ,				do_normalize=__a						)
       _A     =  alt_pipe.to(__a						)
       alt_pipe.set_progress_bar_config(disable=__a						)

       _A     =  'A painting of a squirrel eating a burger'
       _A     =  torch.manual_seed(0						)
       _A     =  alt_pipe(
           [prompt]       ,				generator=__a       ,				num_inference_steps=2       ,				output_type='np'       ,				image=__a       ,				).images

       assert image.shape == (1, 32, 32, 3)



 @unittest.skipIf(torch_device != 'cuda'       ,				'This test requires a GPU'						)
 def      lowerCAmelCase_				(					self			:			str						):
       _A     =  load_image(
           'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
           '/img2img/sketch-mountains-input.jpg'						)
       # resize to resolution that is divisible by 8 but not 16 or 32
       _A     =  init_image.resize((760, 504)						)

       _A     =  'BAAI/AltDiffusion'
       _A     =  AltDiffusionImgaImgPipeline.from_pretrained(
           __a       ,				safety_checker=__a       ,				)
       pipe.to(__a						)
       pipe.set_progress_bar_config(disable=__a						)
       pipe.enable_attention_slicing()

       _A     =  'A fantasy landscape, trending on artstation'

       _A     =  torch.manual_seed(0						)
       _A     =  pipe(
           prompt=__a       ,				image=__a       ,				strength=0.75       ,				guidance_scale=7.5       ,				generator=__a       ,				output_type='np'       ,				)
       _A     =  output.images[0]

       _A     =  image[255:258, 383:386, -1]

       assert image.shape == (504, 760, 3)
       _A     =  np.array([0.9358, 0.9397, 0.9599, 0.9901, 1.0000, 1.0000, 0.9882, 1.0000, 1.0000]						)

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





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



 '''simple docstring'''

 def      lowerCAmelCase_				(					self			:			List[Any]						):
       # clean up the VRAM after each test
       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()



 def      lowerCAmelCase_				(					self			:			str						):
       _A     =  load_image(
           'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
           '/img2img/sketch-mountains-input.jpg'						)
       _A     =  init_image.resize((768, 512)						)
       _A     =  load_numpy(
           'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/img2img/fantasy_landscape_alt.npy'						)

       _A     =  'BAAI/AltDiffusion'
       _A     =  AltDiffusionImgaImgPipeline.from_pretrained(
           __a       ,				safety_checker=__a       ,				)
       pipe.to(__a						)
       pipe.set_progress_bar_config(disable=__a						)
       pipe.enable_attention_slicing()

       _A     =  'A fantasy landscape, trending on artstation'

       _A     =  torch.manual_seed(0						)
       _A     =  pipe(
           prompt=__a       ,				image=__a       ,				strength=0.75       ,				guidance_scale=7.5       ,				generator=__a       ,				output_type='np'       ,				)
       _A     =  output.images[0]

       assert image.shape == (512, 768, 3)
       # img2img is flaky across GPUs even in fp32, so using MAE here
       assert np.abs(expected_image - image						).max() < 1E-2


'''simple docstring'''


import copy
import inspect
import unittest

from transformers import AutoBackbone
from transformers.configuration_utils import PretrainedConfig
from transformers.testing_utils import require_timm, require_torch, torch_device
from transformers.utils.import_utils import is_torch_available

from ...test_backbone_common import BackboneTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor


if is_torch_available():
					import torch

					from transformers import TimmBackbone, TimmBackboneConfig

from ...test_pipeline_mixin import PipelineTesterMixin






class     A :





							def __init__(   self					:				List[str]  ,		__a					:				List[Any]  ,		__a					:				Dict=None  ,		__a					:				str=None  ,		__a					:				List[Any]=None  ,		__a					:				Union[str, Any]="resnet50"  ,		__a					:				List[Any]=3  ,		__a					:				List[Any]=3_2  ,		__a					:				List[Any]=3  ,		__a					:				Optional[int]=True  ,		__a					:				str=True  ,		)				->						Any:
								__UpperCAmelCase      					=      parent
								__UpperCAmelCase      					=      out_indices if out_indices is not None else [4]
								__UpperCAmelCase      					=      stage_names
								__UpperCAmelCase      					=      out_features
								__UpperCAmelCase      					=      backbone
								__UpperCAmelCase      					=      batch_size
								__UpperCAmelCase      					=      image_size
								__UpperCAmelCase      					=      num_channels
								__UpperCAmelCase      					=      use_pretrained_backbone
								__UpperCAmelCase      					=      is_training





							def    snake_case__    (   self					:				Optional[Any]						)				->						Optional[int]:
								__UpperCAmelCase      					=      floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]						)
								__UpperCAmelCase      					=      self.get_config()

								return config, pixel_values





							def    snake_case__    (   self					:				Union[str, Any]						)				->						str:
								return TimmBackboneConfig(
								    image_size=self.image_size  ,		num_channels=self.num_channels  ,		out_features=self.out_features  ,		out_indices=self.out_indices  ,		stage_names=self.stage_names  ,		use_pretrained_backbone=self.use_pretrained_backbone  ,		backbone=self.backbone  ,		)





							def    snake_case__    (   self					:				int  ,		__a					:				Union[str, Any]  ,		__a					:				List[Any]						)				->						Tuple:
								__UpperCAmelCase      					=      TimmBackbone(config=__a						)
								model.to(__a						)
								model.eval()
								with torch.no_grad():
									__UpperCAmelCase      					=      model(__a						)
								self.parent.assertEqual(
								    result.feature_map[-1].shape  ,		(self.batch_size, model.channels[-1], 1_4, 1_4)  ,		)






							def    snake_case__    (   self					:				Dict						)				->						Tuple:
								__UpperCAmelCase      					=      self.prepare_config_and_inputs()
								__UpperCAmelCase					,      __UpperCAmelCase      					=      config_and_inputs
								__UpperCAmelCase      					=      {'''pixel_values''': pixel_values}
								return config, inputs_dict






@require_torch
@require_timm
class     A ( UpperCAmelCase							,						UpperCAmelCase							,						UpperCAmelCase							,						unittest.TestCase ):
							a_										=   (TimmBackbone,) if is_torch_available() else ()
							a_										=   {'''feature-extraction''': TimmBackbone} if is_torch_available() else {}
							a_										=   False
							a_										=   False
							a_										=   False
							a_										=   False





							def    snake_case__    (   self					:				Optional[int]						)				->						List[str]:
								__UpperCAmelCase      					=      TimmBackboneModelTester(self						)
								__UpperCAmelCase      					=      ConfigTester(self  ,		config_class=__a  ,		has_text_modality=__a						)





							def    snake_case__    (   self					:				List[str]						)				->						int:
								self.config_tester.create_and_test_config_to_json_string()
								self.config_tester.create_and_test_config_to_json_file()
								self.config_tester.create_and_test_config_from_and_save_pretrained()
								self.config_tester.create_and_test_config_with_num_labels()
								self.config_tester.check_config_can_be_init_without_params()
								self.config_tester.check_config_arguments_init()





							def    snake_case__    (   self					:				List[str]						)				->						str:
								__UpperCAmelCase      					=      '''resnet18'''
								__UpperCAmelCase      					=      '''microsoft/resnet-18'''

								__UpperCAmelCase      					=      AutoBackbone.from_pretrained(__a  ,		use_timm_backbone=__a						)
								__UpperCAmelCase      					=      AutoBackbone.from_pretrained(__a						)

								self.assertEqual(len(timm_model.out_features						)  ,		len(transformers_model.out_features						)						)
								self.assertEqual(len(timm_model.stage_names						)  ,		len(transformers_model.stage_names						)						)
								self.assertEqual(timm_model.channels  ,		transformers_model.channels						)
								# Out indices are set to the last layer by default. For timm models, we don't know
								# the number of layers in advance, so we set it to (-1,), whereas for transformers
								# models, we set it to [len(stage_names) - 1] (kept for backward compatibility).
								self.assertEqual(timm_model.out_indices  ,		(-1,)						)
								self.assertEqual(transformers_model.out_indices  ,		[len(timm_model.stage_names						) - 1]						)

								__UpperCAmelCase      					=      AutoBackbone.from_pretrained(__a  ,		use_timm_backbone=__a  ,		out_indices=[1, 2, 3]						)
								__UpperCAmelCase      					=      AutoBackbone.from_pretrained(__a  ,		out_indices=[1, 2, 3]						)

								self.assertEqual(timm_model.out_indices  ,		transformers_model.out_indices						)
								self.assertEqual(len(timm_model.out_features						)  ,		len(transformers_model.out_features						)						)
								self.assertEqual(timm_model.channels  ,		transformers_model.channels						)





							@unittest.skip('''TimmBackbone doesn\'t support feed forward chunking'''						)
							def    snake_case__    (   self					:				str						)				->						Optional[Any]:
								pass





							@unittest.skip('''TimmBackbone doesn\'t have num_hidden_layers attribute'''						)
							def    snake_case__    (   self					:				str						)				->						List[Any]:
								pass





							@unittest.skip('''TimmBackbone initialization is managed on the timm side'''						)
							def    snake_case__    (   self					:				str						)				->						int:
								pass





							@unittest.skip('''TimmBackbone models doesn\'t have inputs_embeds'''						)
							def    snake_case__    (   self					:				Optional[Any]						)				->						Union[str, Any]:
								pass





							@unittest.skip('''TimmBackbone models doesn\'t have inputs_embeds'''						)
							def    snake_case__    (   self					:				Optional[int]						)				->						Dict:
								pass





							@unittest.skip('''TimmBackbone model cannot be created without specifying a backbone checkpoint'''						)
							def    snake_case__    (   self					:				str						)				->						Dict:
								pass





							@unittest.skip('''Only checkpoints on timm can be loaded into TimmBackbone'''						)
							def    snake_case__    (   self					:				Optional[Any]						)				->						Union[str, Any]:
								pass





							@unittest.skip('''model weights aren\'t tied in TimmBackbone.'''						)
							def    snake_case__    (   self					:				Union[str, Any]						)				->						Dict:
								pass





							@unittest.skip('''model weights aren\'t tied in TimmBackbone.'''						)
							def    snake_case__    (   self					:				List[str]						)				->						Union[str, Any]:
								pass





							@unittest.skip('''Only checkpoints on timm can be loaded into TimmBackbone'''						)
							def    snake_case__    (   self					:				str						)				->						List[Any]:
								pass





							@unittest.skip('''Only checkpoints on timm can be loaded into TimmBackbone'''						)
							def    snake_case__    (   self					:				Tuple						)				->						int:
								pass





							@unittest.skip('''TimmBackbone doesn\'t have hidden size info in its configuration.'''						)
							def    snake_case__    (   self					:				Union[str, Any]						)				->						Union[str, Any]:
								pass





							@unittest.skip('''TimmBackbone doesn\'t support output_attentions.'''						)
							def    snake_case__    (   self					:				List[str]						)				->						Dict:
								pass





							@unittest.skip('''Safetensors is not supported by timm.'''						)
							def    snake_case__    (   self					:				int						)				->						int:
								pass





							@unittest.skip('''Will be fixed soon by reducing the size of the model used for common tests.'''						)
							def    snake_case__    (   self					:				int						)				->						int:
								pass





							def    snake_case__    (   self					:				Optional[Any]						)				->						Optional[int]:
								__UpperCAmelCase					,      __UpperCAmelCase      					=      self.model_tester.prepare_config_and_inputs_for_common()

								for model_class in self.all_model_classes:
									__UpperCAmelCase      					=      model_class(__a						)
									__UpperCAmelCase      					=      inspect.signature(model.forward						)
									# signature.parameters is an OrderedDict => so arg_names order is deterministic
									__UpperCAmelCase      					=      [*signature.parameters.keys()]

									__UpperCAmelCase      					=      ['''pixel_values''']
									self.assertListEqual(arg_names[:1]  ,		__a						)





							def    snake_case__    (   self					:				Any						)				->						int:
								__UpperCAmelCase					,      __UpperCAmelCase      					=      self.model_tester.prepare_config_and_inputs_for_common()
								__UpperCAmelCase      					=      True
								__UpperCAmelCase      					=      self.has_attentions

								# no need to test all models as different heads yield the same functionality
								__UpperCAmelCase      					=      self.all_model_classes[0]
								__UpperCAmelCase      					=      model_class(__a						)
								model.to(__a						)

								__UpperCAmelCase      					=      self._prepare_for_class(__a  ,		__a						)
								__UpperCAmelCase      					=      model(**__a						)
								__UpperCAmelCase      					=      outputs[0][-1]

								# Encoder-/Decoder-only models
								__UpperCAmelCase      					=      outputs.hidden_states[0]
								hidden_states.retain_grad()

								if self.has_attentions:
									__UpperCAmelCase      					=      outputs.attentions[0]
									attentions.retain_grad()

								output.flatten()[0].backward(retain_graph=__a						)

								self.assertIsNotNone(hidden_states.grad						)

								if self.has_attentions:
									self.assertIsNotNone(attentions.grad						)






							def    snake_case__    (   self					:				Any						)				->						Union[str, Any]:
								__UpperCAmelCase					,      __UpperCAmelCase      					=      self.model_tester.prepare_config_and_inputs_for_common()

								for model_class in self.all_model_classes:
									__UpperCAmelCase      					=      model_class(__a						)
									model.to(__a						)
									model.eval()
									__UpperCAmelCase      					=      model(**__a						)

									self.assertEqual(len(result.feature_maps						)  ,		len(config.out_indices						)						)
									self.assertEqual(len(model.channels						)  ,		len(config.out_indices						)						)

									# Check output of last stage is taken if out_features=None, out_indices=None
									__UpperCAmelCase      					=      copy.deepcopy(__a						)
									__UpperCAmelCase      					=      None
									__UpperCAmelCase      					=      model_class(__a						)
									model.to(__a						)
									model.eval()
									__UpperCAmelCase      					=      model(**__a						)

									self.assertEqual(len(result.feature_maps						)  ,		1						)
									self.assertEqual(len(model.channels						)  ,		1						)

									# Check backbone can be initialized with fresh weights
									__UpperCAmelCase      					=      copy.deepcopy(__a						)
									__UpperCAmelCase      					=      False
									__UpperCAmelCase      					=      model_class(__a						)
									model.to(__a						)
									model.eval()
									__UpperCAmelCase      					=      model(**__a						)
