






import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    UNetaDConditionModel,
)
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 PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()





class        a__      (	lowerCAmelCase__	,       lowerCAmelCase__	,       unittest.TestCase		):
		UpperCAmelCase_							:       Optional[int] 					= StableDiffusionParadigmsPipeline
		UpperCAmelCase_							:       Tuple 					= TEXT_TO_IMAGE_PARAMS
		UpperCAmelCase_							:       Dict 					= TEXT_TO_IMAGE_BATCH_PARAMS
		UpperCAmelCase_							:       List[Any] 					= TEXT_TO_IMAGE_IMAGE_PARAMS
		UpperCAmelCase_							:       Tuple 					= TEXT_TO_IMAGE_IMAGE_PARAMS






		def 					_lowerCamelCase					(      self						)				->				Optional[int]:
				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							, attention_head_dim=(2, 4)							, use_linear_projection=lowercase__							, )
				__A    =							DDIMParallelScheduler(
				    beta_start=0.0_0085							, beta_end=0.012							, beta_schedule="scaled_linear"							, clip_sample=lowercase__							, set_alpha_to_one=lowercase__							, )
				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							, sample_size=128							, )
				torch.manual_seed(0						)
				__A    =							CLIPTextConfig(
				    bos_token_id=0							, eos_token_id=2							, hidden_size=32							, intermediate_size=37							, layer_norm_eps=1e-05							, num_attention_heads=4							, num_hidden_layers=5							, pad_token_id=1							, vocab_size=1000							, hidden_act="gelu"							, projection_dim=512							, )
				__A    =							CLIPTextModel(lowercase__						)
				__A    =							CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"						)

				__A    =							{
				    "unet": unet,
				    "scheduler": scheduler,
				    "vae": vae,
				    "text_encoder": text_encoder,
				    "tokenizer": tokenizer,
				    "safety_checker": None,
				    "feature_extractor": None,
				}
				return components






		def 					_lowerCamelCase					(      self							, lowercase__							, lowercase__=0						)				->				Union[str, Any]:
				if str(lowercase__						).startswith("mps"						):
						__A    =							torch.manual_seed(lowercase__						)
				else:
						__A    =							torch.Generator(device=lowercase__						).manual_seed(lowercase__						)
				__A    =							{
				    "prompt": "a photograph of an astronaut riding a horse",
				    "generator": generator,
				    "num_inference_steps": 10,
				    "guidance_scale": 6.0,
				    "output_type": "numpy",
				    "parallel": 3,
				    "debug": True,
				}
				return inputs






		def 					_lowerCamelCase					(      self						)				->				Tuple:
				__A    =							"cpu"  # ensure determinism for the device-dependent torch.Generator
				__A    =							self.get_dummy_components()
				__A    =							StableDiffusionParadigmsPipeline(**lowercase__						)
				__A    =							sd_pipe.to(lowercase__						)
				sd_pipe.set_progress_bar_config(disable=lowercase__						)

				__A    =							self.get_dummy_inputs(lowercase__						)
				__A    =							sd_pipe(**lowercase__						).images
				__A    =							image[0, -3:, -3:, -1]
				assert image.shape == (1, 64, 64, 3)

				__A    =							np.array([0.4773, 0.5417, 0.4723, 0.4925, 0.5631, 0.4752, 0.5240, 0.4935, 0.5023]						)

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






		def 					_lowerCamelCase					(      self						)				->				Optional[Any]:
				__A    =							"cpu"  # ensure determinism for the device-dependent torch.Generator
				__A    =							self.get_dummy_components()
				torch.manual_seed(0						)
				__A    =							DDPMParallelScheduler()
				torch.manual_seed(0						)
				__A    =							StableDiffusionParadigmsPipeline(**lowercase__						)
				__A    =							sd_pipe.to(lowercase__						)
				sd_pipe.set_progress_bar_config(disable=lowercase__						)

				__A    =							self.get_dummy_inputs(lowercase__						)
				__A    =							sd_pipe(**lowercase__						).images
				__A    =							image[0, -3:, -3:, -1]
				assert image.shape == (1, 64, 64, 3)

				__A    =							np.array([0.3573, 0.4420, 0.4960, 0.4799, 0.3796, 0.3879, 0.4819, 0.4365, 0.4468]						)

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






		def 					_lowerCamelCase					(      self						)				->				Union[str, Any]:
				super().test_inference_batch_consistent(batch_sizes=[1, 2]						)






		def 					_lowerCamelCase					(      self						)				->				List[str]:
				super().test_inference_batch_single_identical(batch_size=2							, expected_max_diff=3e-3						)






		def 					_lowerCamelCase					(      self						)				->				Optional[Any]:
				__A    =							"cpu"  # ensure determinism for the device-dependent torch.Generator
				__A    =							self.get_dummy_components()
				__A    =							StableDiffusionParadigmsPipeline(**lowercase__						)
				__A    =							sd_pipe.to(lowercase__						)
				sd_pipe.set_progress_bar_config(disable=lowercase__						)

				__A    =							self.get_dummy_inputs(lowercase__						)
				__A    =							"french fries"
				__A    =							sd_pipe(**lowercase__							, negative_prompt=lowercase__						)
				__A    =							output.images
				__A    =							image[0, -3:, -3:, -1]

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

				__A    =							np.array([0.4771, 0.5420, 0.4683, 0.4918, 0.5636, 0.4725, 0.5230, 0.4923, 0.5015]						)

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





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






		def 					_lowerCamelCase					(      self						)				->				Optional[int]:
				super().tearDown()
				gc.collect()
				torch.cuda.empty_cache()






		def 					_lowerCamelCase					(      self							, lowercase__=0						)				->				Optional[Any]:
				__A    =							torch.Generator(device=lowercase__						).manual_seed(lowercase__						)
				__A    =							{
				    "prompt": "a photograph of an astronaut riding a horse",
				    "generator": generator,
				    "num_inference_steps": 10,
				    "guidance_scale": 7.5,
				    "output_type": "numpy",
				    "parallel": 3,
				    "debug": True,
				}
				return inputs






		def 					_lowerCamelCase					(      self						)				->				Dict:
				__A    =							"stabilityai/stable-diffusion-2-base"
				__A    =							DDIMParallelScheduler.from_pretrained(lowercase__							, subfolder="scheduler"						)
				__A    =							StableDiffusionParadigmsPipeline.from_pretrained(lowercase__							, scheduler=lowercase__							, safety_checker=lowercase__						)
				pipe.to(lowercase__						)
				pipe.set_progress_bar_config(disable=lowercase__						)
				pipe.enable_attention_slicing()

				__A    =							self.get_inputs()
				__A    =							pipe(**lowercase__						).images
				__A    =							image[0, -3:, -3:, -1].flatten()

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

				__A    =							np.array([0.9622, 0.9602, 0.9748, 0.9591, 0.9630, 0.9691, 0.9661, 0.9631, 0.9741]						)

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




import unittest

from transformers import MarkupLMConfig, is_torch_available
from transformers.testing_utils import require_torch, slow, torch_device
from transformers.utils import cached_property

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
       import torch

       from transformers import (
           MarkupLMForQuestionAnswering,
           MarkupLMForSequenceClassification,
           MarkupLMForTokenClassification,
           MarkupLMModel,
       )

# TODO check dependencies
from transformers import MarkupLMFeatureExtractor, MarkupLMProcessor, MarkupLMTokenizer







class 		__magic_name__     :

   '''simple docstring'''




   def __init__(							self,       lowercase_,       lowercase_=13,       lowercase_=7,       lowercase_=True,       lowercase_=True,       lowercase_=True,       lowercase_=True,       lowercase_=99,       lowercase_=32,       lowercase_=5,       lowercase_=4,       lowercase_=37,       lowercase_="gelu",       lowercase_=0.1,       lowercase_=0.1,       lowercase_=512,       lowercase_=16,       lowercase_=2,       lowercase_=0.02,       lowercase_=3,       lowercase_=None,       lowercase_=20,       lowercase_=30,       lowercase_=2,       lowercase_=2,       lowercase_=10,       )				->       Dict:






     """simple docstring"""






     a__    =parent
     a__    =batch_size
     a__    =seq_length
     a__    =is_training
     a__    =use_input_mask
     a__    =use_token_type_ids
     a__    =use_labels
     a__    =vocab_size
     a__    =hidden_size
     a__    =num_hidden_layers
     a__    =num_attention_heads
     a__    =intermediate_size
     a__    =hidden_act
     a__    =hidden_dropout_prob
     a__    =attention_probs_dropout_prob
     a__    =max_position_embeddings
     a__    =type_vocab_size
     a__    =type_sequence_label_size
     a__    =initializer_range
     a__    =num_labels
     a__    =scope
     a__    =max_xpath_tag_unit_embeddings
     a__    =max_xpath_subs_unit_embeddings
     a__    =tag_pad_id
     a__    =subs_pad_id
     a__    =max_depth



   def 					_UpperCAmelCase					(							self			)				->       Optional[Any]:






     """simple docstring"""






     a__    =ids_tensor([self.batch_size, self.seq_length],       self.vocab_size			)

     a__    =ids_tensor(
         [self.batch_size, self.seq_length, self.max_depth],       self.max_xpath_tag_unit_embeddings			)

     a__    =ids_tensor(
         [self.batch_size, self.seq_length, self.max_depth],       self.max_xpath_subs_unit_embeddings			)

     a__    =None
     if self.use_input_mask:
       a__    =ids_tensor([self.batch_size, self.seq_length],       vocab_size=2			)

     a__    =None
     if self.use_token_type_ids:
       a__    =ids_tensor([self.batch_size, self.seq_length],       self.type_vocab_size			)

     a__    =None
     a__    =None
     if self.use_labels:
       a__    =ids_tensor([self.batch_size],       self.type_sequence_label_size			)
       a__    =ids_tensor([self.batch_size, self.seq_length],       self.num_labels			)

     a__    =self.get_config()

     return (
         config,
         input_ids,
         xpath_tags_seq,
         xpath_subs_seq,
         token_type_ids,
         input_mask,
         sequence_labels,
         token_labels,
     )



   def 					_UpperCAmelCase					(							self			)				->       List[str]:






     """simple docstring"""






     return MarkupLMConfig(
         vocab_size=self.vocab_size,       hidden_size=self.hidden_size,       num_hidden_layers=self.num_hidden_layers,       num_attention_heads=self.num_attention_heads,       intermediate_size=self.intermediate_size,       hidden_act=self.hidden_act,       hidden_dropout_prob=self.hidden_dropout_prob,       attention_probs_dropout_prob=self.attention_probs_dropout_prob,       max_position_embeddings=self.max_position_embeddings,       type_vocab_size=self.type_vocab_size,       initializer_range=self.initializer_range,       max_xpath_tag_unit_embeddings=self.max_xpath_tag_unit_embeddings,       max_xpath_subs_unit_embeddings=self.max_xpath_subs_unit_embeddings,       tag_pad_id=self.tag_pad_id,       subs_pad_id=self.subs_pad_id,       max_depth=self.max_depth,       )



   def 					_UpperCAmelCase					(							self,       lowercase_,       lowercase_,       lowercase_,       lowercase_,       lowercase_,       lowercase_,       lowercase_,       lowercase_,       )				->       Tuple:






     """simple docstring"""






     a__    =MarkupLMModel(config=lowercase_			)
     model.to(lowercase_			)
     model.eval()
     print('''Configs:''',       model.config.tag_pad_id,       model.config.subs_pad_id			)
     a__    =model(lowercase_,       attention_mask=lowercase_,       token_type_ids=lowercase_			)
     a__    =model(lowercase_,       token_type_ids=lowercase_			)
     a__    =model(lowercase_			)
     self.parent.assertEqual(result.last_hidden_state.shape,       (self.batch_size, self.seq_length, self.hidden_size)			)
     self.parent.assertEqual(result.pooler_output.shape,       (self.batch_size, self.hidden_size)			)



   def 					_UpperCAmelCase					(							self,       lowercase_,       lowercase_,       lowercase_,       lowercase_,       lowercase_,       lowercase_,       lowercase_,       lowercase_,       )				->       Tuple:






     """simple docstring"""






     a__    =self.num_labels
     a__    =MarkupLMForSequenceClassification(lowercase_			)
     model.to(lowercase_			)
     model.eval()
     a__    =model(
         lowercase_,       xpath_tags_seq=lowercase_,       xpath_subs_seq=lowercase_,       attention_mask=lowercase_,       token_type_ids=lowercase_,       labels=lowercase_,       )
     self.parent.assertEqual(result.logits.shape,       (self.batch_size, self.num_labels)			)



   def 					_UpperCAmelCase					(							self,       lowercase_,       lowercase_,       lowercase_,       lowercase_,       lowercase_,       lowercase_,       lowercase_,       lowercase_,       )				->       Union[str, Any]:






     """simple docstring"""






     a__    =self.num_labels
     a__    =MarkupLMForTokenClassification(config=lowercase_			)
     model.to(lowercase_			)
     model.eval()
     a__    =model(
         lowercase_,       xpath_tags_seq=lowercase_,       xpath_subs_seq=lowercase_,       attention_mask=lowercase_,       token_type_ids=lowercase_,       labels=lowercase_,       )
     self.parent.assertEqual(result.logits.shape,       (self.batch_size, self.seq_length, self.num_labels)			)



   def 					_UpperCAmelCase					(							self,       lowercase_,       lowercase_,       lowercase_,       lowercase_,       lowercase_,       lowercase_,       lowercase_,       lowercase_,       )				->       Any:






     """simple docstring"""






     a__    =MarkupLMForQuestionAnswering(config=lowercase_			)
     model.to(lowercase_			)
     model.eval()
     a__    =model(
         lowercase_,       xpath_tags_seq=lowercase_,       xpath_subs_seq=lowercase_,       attention_mask=lowercase_,       token_type_ids=lowercase_,       start_positions=lowercase_,       end_positions=lowercase_,       )
     self.parent.assertEqual(result.start_logits.shape,       (self.batch_size, self.seq_length)			)
     self.parent.assertEqual(result.end_logits.shape,       (self.batch_size, self.seq_length)			)



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






     """simple docstring"""






     a__    =self.prepare_config_and_inputs()
     (
         (
         a__
     ), (
         a__
     ), (
         a__
     ), (
         a__
     ), (
         a__
     ), (
         a__
     ), (
         a__
     ), (
         a__
     ), 
     )    =config_and_inputs
     a__    ={
         '''input_ids''': input_ids,
         '''xpath_tags_seq''': xpath_tags_seq,
         '''xpath_subs_seq''': xpath_subs_seq,
         '''token_type_ids''': token_type_ids,
         '''attention_mask''': input_mask,
     }
     return config, inputs_dict







@require_torch
class 		__magic_name__     (				lowerCamelCase__						,			lowerCamelCase__						,			unittest.TestCase				):

   '''simple docstring'''




   lowerCamelCase__					:      Tuple        =  (
       (
           MarkupLMModel,
           MarkupLMForSequenceClassification,
           MarkupLMForTokenClassification,
           MarkupLMForQuestionAnswering,
       )
       if is_torch_available()
       else None
   )
   lowerCamelCase__					:      Union[str, Any]        =  (
       {
           'feature-extraction': MarkupLMModel,
           'question-answering': MarkupLMForQuestionAnswering,
           'text-classification': MarkupLMForSequenceClassification,
           'token-classification': MarkupLMForTokenClassification,
           'zero-shot': MarkupLMForSequenceClassification,
       }
       if is_torch_available()
       else {}
   )



   def 					_UpperCAmelCase					(							self,       lowercase_,       lowercase_,       lowercase_,       lowercase_,       lowercase_			)				->       Union[str, Any]:






     """simple docstring"""






     return True



   def 					_UpperCAmelCase					(							self			)				->       List[str]:






     """simple docstring"""






     a__    =MarkupLMModelTester(self			)
     a__    =ConfigTester(self,       config_class=lowercase_,       hidden_size=37			)



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






     """simple docstring"""






     self.config_tester.run_common_tests()



   def 					_UpperCAmelCase					(							self			)				->       List[str]:






     """simple docstring"""






     a__    =self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_model(*lowercase_			)



   def 					_UpperCAmelCase					(							self			)				->       Optional[int]:






     """simple docstring"""






     a__    =self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_for_sequence_classification(*lowercase_			)



   def 					_UpperCAmelCase					(							self			)				->       Tuple:






     """simple docstring"""






     a__    =self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_for_token_classification(*lowercase_			)



   def 					_UpperCAmelCase					(							self			)				->       int:






     """simple docstring"""






     a__    =self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_for_question_answering(*lowercase_			)





def      UpperCAmelCase__							(     ):





  '''simple docstring'''







  a__    ='''
    <!DOCTYPE html>
    <html>
    <head>
    <title>Page Title</title>
    </head>
    <body>

    <h1>This is a Heading</h1>
    <p>This is a paragraph.</p>

    </body>
    </html>
    '''

  return html_string







@require_torch
class 		__magic_name__     (				unittest.TestCase				):

   '''simple docstring'''




   @cached_property
   def 					_UpperCAmelCase					(							self			)				->       List[str]:






     """simple docstring"""






     a__    =MarkupLMFeatureExtractor()
     a__    =MarkupLMTokenizer.from_pretrained('''microsoft/markuplm-base'''			)

     return MarkupLMProcessor(lowercase_,       lowercase_			)



   @slow
   def 					_UpperCAmelCase					(							self			)				->       Optional[Any]:






     """simple docstring"""






     a__    =MarkupLMModel.from_pretrained('''microsoft/markuplm-base'''			).to(lowercase_			)

     a__    =self.default_processor

     a__    =processor(prepare_html_string(),       return_tensors='''pt'''			)
     a__    =inputs.to(lowercase_			)

     # forward pass
     with torch.no_grad():
       a__    =model(**lowercase_			)

     # verify the last hidden states
     a__    =torch.Size([1, 14, 768]			)
     self.assertEqual(outputs.last_hidden_state.shape,       lowercase_			)

     a__    =torch.tensor(
         [[0.0675, -0.0052, 0.5001], [-0.2281, 0.0802, 0.2192], [-0.0583, -0.3311, 0.1185]]			).to(lowercase_			)

     self.assertTrue(torch.allclose(outputs.last_hidden_state[0, :3, :3],       lowercase_,       atol=1E-4			)			)
