




"""simple docstring"""

import logging
import os
import time
from argparse import Namespace
from pathlib import Path

import datasets
import torch
from accelerate import Accelerator, DistributedType
from accelerate.utils import ProjectConfiguration
from arguments import TrainingArguments
from datasets import load_dataset
from huggingface_hub import Repository
from torch.optim import AdamW
from torch.utils.data import IterableDataset
from torch.utils.data.dataloader import DataLoader
from torch.utils.data.datapipes.iter.combinatorics import ShufflerIterDataPipe

import transformers
from transformers import AutoModelForCausalLM, AutoTokenizer, HfArgumentParser, get_scheduler, set_seed

class     lowercase (      snake_case_):







				def __init__(							self				:   Any ,       _lowerCamelCase				:   Dict ,       _lowerCamelCase				:   Union[str, Any] ,       _lowerCamelCase				:   Optional[Any]=False ,       _lowerCamelCase				:   Dict=10_24 ,       _lowerCamelCase				:   Optional[Any]=10_24 ,       _lowerCamelCase				:   Tuple=3.6 ,       _lowerCamelCase				:   Union[str, Any]=False ,       ):



					"""simple docstring"""

					A_					:			Optional[Any]						=						tokenizer
					A_					:			int						=						tokenizer.bos_token_id
					A_					:			Tuple						=						dataset
					A_					:			Any						=						seq_length
					A_					:			List[str]						=						0
					A_					:			List[str]						=						infinite
					A_					:			str						=						0
					A_					:			List[Any]						=						tokenized

					if self.tokenized:
						A_					:			List[str]						=						seq_length * num_of_sequences
						A_					:			Any						=						"input_ids"
					else:
						A_					:			int						=						seq_length * chars_per_token * num_of_sequences
						A_					:			Optional[Any]						=						"content"







				def __iter__(							self				:   Union[str, Any]    ):



					"""simple docstring"""

					A_					:			Optional[int]						=						iter(self.dataset    )
					A_					:			Union[str, Any]						=						True
					while more_examples:
						A_					:			int						=						[], 0
						while True:
							if buffer_len >= self.max_buffer_size:
								break
							try:
								buffer.append(next(_lowerCamelCase    )[self.content_field]    )
								buffer_len += len(buffer[-1]    )
							except StopIteration:
								if self.infinite:
									A_					:			List[str]						=						iter(self.dataset    )
									self.epoch += 1
									logger.info(F"""Dataset epoch: {self.epoch}"""    )
								else:
									A_					:			List[Any]						=						False
									break
						if self.tokenized:
							A_					:			List[Any]						=						buffer
						else:
							A_					:			Dict						=						self.tokenizer(_lowerCamelCase ,       truncation=_lowerCamelCase    )["input_ids"]
						A_					:			int						=						[]
						for tokenized_input in tokenized_inputs:
							all_token_ids.extend(tokenized_input + [self.concat_token_id]    )
						for i in range(0 ,       len(_lowerCamelCase    ) ,       self.seq_length    ):
							A_					:			str						=						all_token_ids[i : i + self.seq_length]
							if len(_lowerCamelCase    ) == self.seq_length:
								self.current_size += 1
								yield torch.tensor(_lowerCamelCase    )







				def    a_					(							self				:   int ,       _lowerCamelCase				:   Optional[Any]=10_00    ):



					"""simple docstring"""

					return ShufflerIterDataPipe(self ,       buffer_size=_lowerCamelCase    )

def 						lowercase_ (						_UpperCAmelCase		):




	"""simple docstring"""






	A_					:			List[Any]						=						args.model_ckpt.split('''/'''		)[-1]
	A_					:			List[Any]						=						logging.getLogger(__name__		)
	A_					:			Tuple						=						Path(args.save_dir		) / "log/"
	log_dir.mkdir(exist_ok=SCREAMING_SNAKE_CASE_		)
	A_					:			Optional[Any]						=						f"""debug_{accelerator.process_index}.log"""
	logging.basicConfig(
	    format='''%(asctime)s - %(levelname)s - %(name)s - %(message)s'''					, datefmt='''%m/%d/%Y %H:%M:%S'''					, level=logging.INFO					, handlers=[logging.FileHandler(log_dir / filename		), logging.StreamHandler()]					, )
	if accelerator.is_main_process:  # we only want to setup logging once
		accelerator.init_trackers(SCREAMING_SNAKE_CASE_					, vars(SCREAMING_SNAKE_CASE_		)		)
		A_					:			Any						=						accelerator.trackers[0].run.name
		logger.setLevel(logging.INFO		)
		datasets.utils.logging.set_verbosity_info()
		transformers.utils.logging.set_verbosity_info()
	else:
		A_					:			Dict						=						""
		logger.setLevel(logging.ERROR		)
		datasets.utils.logging.set_verbosity_error()
		transformers.utils.logging.set_verbosity_error()
	return logger, run_name

def 						lowercase_ (						_UpperCAmelCase		):




	"""simple docstring"""






	A_					:			Optional[int]						=						{"streaming": True}
	A_					:			List[str]						=						load_dataset(args.dataset_name_train					, split='''train'''					, **SCREAMING_SNAKE_CASE_		)
	A_					:			Dict						=						train_data.shuffle(buffer_size=args.shuffle_buffer					, seed=args.seed		)
	A_					:			Optional[Any]						=						load_dataset(args.dataset_name_valid					, split='''train'''					, **SCREAMING_SNAKE_CASE_		)
	A_					:			Optional[Any]						=						ConstantLengthDataset(
	    SCREAMING_SNAKE_CASE_					, SCREAMING_SNAKE_CASE_					, infinite=SCREAMING_SNAKE_CASE_					, seq_length=args.seq_length					, tokenized=args.tokenized		)
	A_					:			Tuple						=						ConstantLengthDataset(
	    SCREAMING_SNAKE_CASE_					, SCREAMING_SNAKE_CASE_					, infinite=SCREAMING_SNAKE_CASE_					, seq_length=args.seq_length					, tokenized=args.tokenized		)
	A_					:			str						=						train_dataset.shuffle(buffer_size=args.shuffle_buffer		)
	A_					:			List[Any]						=						DataLoader(SCREAMING_SNAKE_CASE_					, batch_size=args.train_batch_size					, shuffle=SCREAMING_SNAKE_CASE_		)
	A_					:			List[str]						=						DataLoader(SCREAMING_SNAKE_CASE_					, batch_size=args.valid_batch_size		)
	return train_dataloader, eval_dataloader

def 						lowercase_ (						_UpperCAmelCase					, _UpperCAmelCase					, _UpperCAmelCase=["bias", "ln_1.weight", "ln_2.weight", "ln_f.weight"]		):




	"""simple docstring"""






	A_					:			Any						=						[], []
	for n, p in model.named_parameters():
		if any(nd in n for nd in no_decay		):
			params_without_wd.append(SCREAMING_SNAKE_CASE_		)
		else:
			params_with_wd.append(SCREAMING_SNAKE_CASE_		)
	return [
	    {"params": params_with_wd, "weight_decay": args.weight_decay},
	    {"params": params_without_wd, "weight_decay": 0.0},
	]

def 						lowercase_ (						_UpperCAmelCase					, _UpperCAmelCase		):




	"""simple docstring"""






	logger.info(f"""Step {step}: {metrics}"""		)
	if accelerator.is_main_process:
		accelerator.log(SCREAMING_SNAKE_CASE_					, SCREAMING_SNAKE_CASE_		)

def 						lowercase_ (						_UpperCAmelCase					, _UpperCAmelCase					, _UpperCAmelCase		):




	"""simple docstring"""






	A_					:			str						=						accelerator.unwrap_model(SCREAMING_SNAKE_CASE_		).config
	A_					:			Optional[int]						=						4 if args.gradient_checkpointing else 3
	A_					:			int						=						args.train_batch_size * accelerator.state.num_processes * args.gradient_accumulation_steps
	A_					:			Optional[int]						=						24 * checkpoint_factor * batch_size * args.seq_length * config_model.n_layer * (config_model.n_embd**2)
	A_					:			List[str]						=						factor * (
	    1.0
	    + (args.seq_length / (6.0 * config_model.n_embd))
	    + (tokenizer.vocab_size / (16.0 * config_model.n_layer * config_model.n_embd))
	)
	A_					:			List[Any]						=						flops_per_iteration / (elapsed_time * accelerator.state.num_processes * (10**12))
	return tflops

def 						lowercase_ (						_UpperCAmelCase		):




	"""simple docstring"""






	model.eval()
	A_					:			Any						=						[]
	for step, batch in enumerate(SCREAMING_SNAKE_CASE_		):
		with torch.no_grad():
			A_					:			str						=						model(SCREAMING_SNAKE_CASE_					, labels=SCREAMING_SNAKE_CASE_		)
		A_					:			Union[str, Any]						=						outputs.loss.repeat(args.valid_batch_size		)
		losses.append(accelerator.gather(SCREAMING_SNAKE_CASE_		)		)
		if args.max_eval_steps > 0 and step >= args.max_eval_steps:
			break
	A_					:			Any						=						torch.cat(SCREAMING_SNAKE_CASE_		)
	A_					:			Union[str, Any]						=						losses[: eval_dataloader.dataset.current_size].mean()
	try:
		A_					:			Tuple						=						torch.exp(SCREAMING_SNAKE_CASE_		)
	except OverflowError:
		A_					:			Any						=						float('''inf'''		)
	return loss.item(), perplexity.item()


# Settings
_lowerCamelCase					:     List[str]				       =    HfArgumentParser(TrainingArguments)
_lowerCamelCase					:     Optional[Any]				       =    parser.parse_args()

# Accelerator
_lowerCamelCase					:     List[Any]				       =    ProjectConfiguration(project_dir=args.save_dir, logging_dir='log')
_lowerCamelCase					:     Union[str, Any]				       =    Accelerator(log_with=['wandb', 'tensorboard'], project_config=config)
_lowerCamelCase					:     Tuple				       =    {str(k): str(v) for k, v in accelerator.state.__dict__.items()}

_lowerCamelCase					:     int				       =    Namespace(**vars(args), **acc_state)
_lowerCamelCase					:     Optional[int]				       =    accelerator.state.num_processes * args.train_batch_size
set_seed(args.seed)

# Clone model repository
if accelerator.is_main_process:
		_lowerCamelCase					:     List[str]				       =    Repository(args.save_dir, clone_from=args.model_ckpt)

# Logging
_lowerCamelCase					:     Optional[Any]				       =    setup_logging(args)
logger.info(accelerator.state)

# Checkout new branch on repo
if accelerator.is_main_process:
		hf_repo.git_checkout(run_name, create_branch_ok=True)

# Load model and tokenizer
_lowerCamelCase					:     int				       =    AutoModelForCausalLM.from_pretrained(args.save_dir)
if args.gradient_checkpointing:
		model.gradient_checkpointing_enable()
_lowerCamelCase					:     Optional[Any]				       =    AutoTokenizer.from_pretrained(args.save_dir)

# Load dataset and dataloader
_lowerCamelCase					:     Union[str, Any]				       =    create_dataloaders(args)

# Prepare the optimizer and learning rate scheduler
_lowerCamelCase					:     Any				       =    AdamW(get_grouped_params(model, args), lr=args.learning_rate)
_lowerCamelCase					:     Tuple				       =    get_scheduler(
    name=args.lr_scheduler_type,
    optimizer=optimizer,
    num_warmup_steps=args.num_warmup_steps,
    num_training_steps=args.max_train_steps,
)
accelerator.register_for_checkpointing(lr_scheduler)

def 						lowercase_ (						):




	"""simple docstring"""






	return optimizer.param_groups[0]["lr"]


# Prepare everything with our `accelerator`.
_lowerCamelCase					:     List[str]				       =    accelerator.prepare(
    model, optimizer, train_dataloader, eval_dataloader
)

# load in the weights and states from a previous save
if args.resume_from_checkpoint:
		if args.resume_from_checkpoint is not None or args.resume_from_checkpoint != "":
				accelerator.print(f'Resumed from checkpoint: {args.resume_from_checkpoint}')
				accelerator.load_state(args.resume_from_checkpoint)
				_lowerCamelCase					:     Dict				       =    os.path.basename(args.resume_from_checkpoint)
		else:
				# Get the most recent checkpoint
				_lowerCamelCase					:     Tuple				       =    [f.name for f in os.scandir(args.save_dir) if f.is_dir() and '''step''' in str(f)]
				dirs.sort(key=os.path.getctime)
				_lowerCamelCase					:     str				       =    dirs[-1]  # Sorts folders by date modified, most recent checkpoint is the last
		# Extract the step of the checkpoint to continue from there
		_lowerCamelCase					:     Dict				       =    os.path.splitext(path)[0]
		_lowerCamelCase					:     Dict				       =    int(training_difference.replace('step_', ''))

# Train model
model.train()
_lowerCamelCase					:     Tuple				       =    0
_lowerCamelCase					:     str				       =    time.time()
_lowerCamelCase					:     List[str]				       =    0
for step, batch in enumerate(train_dataloader, start=1):
		if args.resume_from_checkpoint and step < resume_step:
				continue  # we need to skip steps until we reach the resumed step
		_lowerCamelCase					:     Optional[int]				       =    model(batch, labels=batch, use_cache=False).loss
		_lowerCamelCase					:     Dict				       =    accelerator.gather(loss.repeat(args.train_batch_size)).mean()
		loss_tracking += avg_loss.item() / args.gradient_accumulation_steps
		log_metrics(step, {'samples': step * samples_per_step, 'loss_per_step/train': loss.item()})
		_lowerCamelCase					:     Optional[int]				       =    loss / args.gradient_accumulation_steps
		if step % args.gradient_accumulation_steps != 0:
				# Prevent backward from doing gradient all_reduce in every step
				if accelerator.distributed_type == DistributedType.MULTI_GPU:
						with model.no_sync():
								accelerator.backward(loss)
				else:
						accelerator.backward(loss)
		else:
				_lowerCamelCase					:     Dict				       =    get_lr()
				accelerator.backward(loss)
				accelerator.clip_grad_norm_(model.parameters(), 1.0)
				optimizer.step()
				lr_scheduler.step()
				optimizer.zero_grad()
				_lowerCamelCase					:     List[Any]				       =    time.time() - t_start
				_lowerCamelCase					:     Union[str, Any]				       =    compute_tflops(elapsed_time, accelerator, args)
				log_metrics(
				    step,
				    {
				        'steps': completed_steps,
				        'loss/train': loss_tracking,
				        'lr': lr,
				        'tflops': tflops,
				        'time_per_iteration': elapsed_time,
				    },
				)
				_lowerCamelCase					:     Optional[Any]				       =    time.time()
				_lowerCamelCase					:     List[Any]				       =    0
				completed_steps += 1
		if step % args.save_checkpoint_steps == 0:
				logger.info('Evaluating and saving model checkpoint')
				_lowerCamelCase					:     List[Any]				       =    evaluate(args)
				log_metrics(step, {'loss/eval': eval_loss, 'perplexity': perplexity})
				accelerator.wait_for_everyone()
				_lowerCamelCase					:     Union[str, Any]				       =    os.path.join(args.save_dir, f'step_{step}')
				accelerator.save_state(save_dir)
				if accelerator.is_main_process:
						hf_repo.push_to_hub(commit_message=f'step {step}')
				model.train()
		if completed_steps >= args.max_train_steps:
				break

# Evaluate and save the last checkpoint
logger.info('Evaluating and saving model after training')
_lowerCamelCase					:     Any				       =    evaluate(args)
log_metrics(step, {'loss/eval': eval_loss, 'perplexity': perplexity})
accelerator.wait_for_everyone()
_lowerCamelCase					:     Optional[Any]				       =    accelerator.unwrap_model(model)
unwrapped_model.save_pretrained(args.save_dir, save_function=accelerator.save)
_lowerCamelCase					:     List[str]				       =    os.path.join(args.save_dir, f'step_{step}')
accelerator.save_state(save_dir)
if accelerator.is_main_process:
		hf_repo.push_to_hub(commit_message='final model')











"""simple docstring"""

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     lowercase :







				def __init__(							self				:   Union[str, Any] ,       _lowerCamelCase				:   int ,       _lowerCamelCase				:   Optional[int]=13 ,       _lowerCamelCase				:   Any=7 ,       _lowerCamelCase				:   int=True ,       _lowerCamelCase				:   List[Any]=True ,       _lowerCamelCase				:   str=True ,       _lowerCamelCase				:   Any=True ,       _lowerCamelCase				:   Optional[int]=99 ,       _lowerCamelCase				:   int=32 ,       _lowerCamelCase				:   List[Any]=5 ,       _lowerCamelCase				:   str=4 ,       _lowerCamelCase				:   List[Any]=37 ,       _lowerCamelCase				:   Optional[Any]="gelu" ,       _lowerCamelCase				:   Tuple=0.1 ,       _lowerCamelCase				:   Dict=0.1 ,       _lowerCamelCase				:   Tuple=5_12 ,       _lowerCamelCase				:   List[Any]=16 ,       _lowerCamelCase				:   Dict=2 ,       _lowerCamelCase				:   Optional[int]=0.02 ,       _lowerCamelCase				:   Any=3 ,       _lowerCamelCase				:   Union[str, Any]=None ,       _lowerCamelCase				:   List[str]=20 ,       _lowerCamelCase				:   List[str]=30 ,       _lowerCamelCase				:   int=2 ,       _lowerCamelCase				:   Union[str, Any]=2 ,       _lowerCamelCase				:   Optional[Any]=10 ,       ):



					"""simple docstring"""

					A_					:			Dict						=						parent
					A_					:			Tuple						=						batch_size
					A_					:			Dict						=						seq_length
					A_					:			Any						=						is_training
					A_					:			List[str]						=						use_input_mask
					A_					:			Union[str, Any]						=						use_token_type_ids
					A_					:			int						=						use_labels
					A_					:			List[Any]						=						vocab_size
					A_					:			Dict						=						hidden_size
					A_					:			List[Any]						=						num_hidden_layers
					A_					:			str						=						num_attention_heads
					A_					:			str						=						intermediate_size
					A_					:			Tuple						=						hidden_act
					A_					:			List[str]						=						hidden_dropout_prob
					A_					:			Any						=						attention_probs_dropout_prob
					A_					:			Union[str, Any]						=						max_position_embeddings
					A_					:			Any						=						type_vocab_size
					A_					:			List[Any]						=						type_sequence_label_size
					A_					:			str						=						initializer_range
					A_					:			str						=						num_labels
					A_					:			Tuple						=						scope
					A_					:			Optional[int]						=						max_xpath_tag_unit_embeddings
					A_					:			List[Any]						=						max_xpath_subs_unit_embeddings
					A_					:			List[str]						=						tag_pad_id
					A_					:			Tuple						=						subs_pad_id
					A_					:			int						=						max_depth







				def    a_					(							self				:   int    ):



					"""simple docstring"""

					A_					:			Optional[int]						=						ids_tensor([self.batch_size, self.seq_length] ,       self.vocab_size    )

					A_					:			Union[str, Any]						=						ids_tensor(
					    [self.batch_size, self.seq_length, self.max_depth] ,       self.max_xpath_tag_unit_embeddings    )

					A_					:			Optional[Any]						=						ids_tensor(
					    [self.batch_size, self.seq_length, self.max_depth] ,       self.max_xpath_subs_unit_embeddings    )

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

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

					A_					:			Optional[int]						=						None
					A_					:			Optional[Any]						=						None
					if self.use_labels:
						A_					:			str						=						ids_tensor([self.batch_size] ,       self.type_sequence_label_size    )
						A_					:			Optional[int]						=						ids_tensor([self.batch_size, self.seq_length] ,       self.num_labels    )

					A_					:			int						=						self.get_config()

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







				def    a_					(							self				:   int    ):



					"""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    a_					(							self				:   int ,       _lowerCamelCase				:   List[Any] ,       _lowerCamelCase				:   int ,       _lowerCamelCase				:   int ,       _lowerCamelCase				:   List[Any] ,       _lowerCamelCase				:   Optional[int] ,       _lowerCamelCase				:   Optional[Any] ,       _lowerCamelCase				:   int ,       _lowerCamelCase				:   Optional[int] ,       ):



					"""simple docstring"""

					A_					:			Tuple						=						MarkupLMModel(config=_lowerCamelCase    )
					model.to(_lowerCamelCase    )
					model.eval()
					print('''Configs:''' ,       model.config.tag_pad_id ,       model.config.subs_pad_id    )
					A_					:			List[str]						=						model(_lowerCamelCase ,       attention_mask=_lowerCamelCase ,       token_type_ids=_lowerCamelCase    )
					A_					:			List[str]						=						model(_lowerCamelCase ,       token_type_ids=_lowerCamelCase    )
					A_					:			Dict						=						model(_lowerCamelCase    )
					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    a_					(							self				:   Optional[Any] ,       _lowerCamelCase				:   Optional[Any] ,       _lowerCamelCase				:   Union[str, Any] ,       _lowerCamelCase				:   List[str] ,       _lowerCamelCase				:   Union[str, Any] ,       _lowerCamelCase				:   str ,       _lowerCamelCase				:   Dict ,       _lowerCamelCase				:   Dict ,       _lowerCamelCase				:   List[Any] ,       ):



					"""simple docstring"""

					A_					:			Any						=						self.num_labels
					A_					:			str						=						MarkupLMForSequenceClassification(_lowerCamelCase    )
					model.to(_lowerCamelCase    )
					model.eval()
					A_					:			str						=						model(
					    _lowerCamelCase ,       xpath_tags_seq=_lowerCamelCase ,       xpath_subs_seq=_lowerCamelCase ,       attention_mask=_lowerCamelCase ,       token_type_ids=_lowerCamelCase ,       labels=_lowerCamelCase ,       )
					self.parent.assertEqual(result.logits.shape ,       (self.batch_size, self.num_labels)    )







				def    a_					(							self				:   Union[str, Any] ,       _lowerCamelCase				:   Any ,       _lowerCamelCase				:   int ,       _lowerCamelCase				:   str ,       _lowerCamelCase				:   str ,       _lowerCamelCase				:   Tuple ,       _lowerCamelCase				:   Optional[int] ,       _lowerCamelCase				:   List[str] ,       _lowerCamelCase				:   List[Any] ,       ):



					"""simple docstring"""

					A_					:			str						=						self.num_labels
					A_					:			int						=						MarkupLMForTokenClassification(config=_lowerCamelCase    )
					model.to(_lowerCamelCase    )
					model.eval()
					A_					:			List[Any]						=						model(
					    _lowerCamelCase ,       xpath_tags_seq=_lowerCamelCase ,       xpath_subs_seq=_lowerCamelCase ,       attention_mask=_lowerCamelCase ,       token_type_ids=_lowerCamelCase ,       labels=_lowerCamelCase ,       )
					self.parent.assertEqual(result.logits.shape ,       (self.batch_size, self.seq_length, self.num_labels)    )







				def    a_					(							self				:   Any ,       _lowerCamelCase				:   Union[str, Any] ,       _lowerCamelCase				:   str ,       _lowerCamelCase				:   Union[str, Any] ,       _lowerCamelCase				:   int ,       _lowerCamelCase				:   int ,       _lowerCamelCase				:   Dict ,       _lowerCamelCase				:   Tuple ,       _lowerCamelCase				:   Optional[Any] ,       ):



					"""simple docstring"""

					A_					:			Dict						=						MarkupLMForQuestionAnswering(config=_lowerCamelCase    )
					model.to(_lowerCamelCase    )
					model.eval()
					A_					:			List[Any]						=						model(
					    _lowerCamelCase ,       xpath_tags_seq=_lowerCamelCase ,       xpath_subs_seq=_lowerCamelCase ,       attention_mask=_lowerCamelCase ,       token_type_ids=_lowerCamelCase ,       start_positions=_lowerCamelCase ,       end_positions=_lowerCamelCase ,       )
					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    a_					(							self				:   List[str]    ):



					"""simple docstring"""

					A_					:			str						=						self.prepare_config_and_inputs()
					(
					    (
					    A_
					)				,	(
					    A_
					)				,	(
					    A_
					)				,	(
					    A_
					)				,	(
					    A_
					)				,	(
					    A_
					)				,	(
					    A_
					)				,	(
					    A_
					)				,	
					)					:			List[Any]						=						config_and_inputs
					A_					:			Tuple						=						{
					    '''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     lowercase (      __UpperCAmelCase       ,		__UpperCAmelCase       ,		unittest.TestCase):
				__lowerCAmelCase      :      Union[str, Any]					    =				(
				    (
				        MarkupLMModel,
				        MarkupLMForSequenceClassification,
				        MarkupLMForTokenClassification,
				        MarkupLMForQuestionAnswering,
				    )
				    if is_torch_available()
				    else None
				)
				__lowerCAmelCase      :      Optional[int]					    =				(
				    {
				        """feature-extraction""": MarkupLMModel,
				        """question-answering""": MarkupLMForQuestionAnswering,
				        """text-classification""": MarkupLMForSequenceClassification,
				        """token-classification""": MarkupLMForTokenClassification,
				        """zero-shot""": MarkupLMForSequenceClassification,
				    }
				    if is_torch_available()
				    else {}
				)







				def    a_					(							self				:   List[str] ,       _lowerCamelCase				:   Optional[int] ,       _lowerCamelCase				:   Dict ,       _lowerCamelCase				:   Optional[int] ,       _lowerCamelCase				:   Optional[Any] ,       _lowerCamelCase				:   Any    ):



					"""simple docstring"""

					return True







				def    a_					(							self				:   int    ):



					"""simple docstring"""

					A_					:			Tuple						=						MarkupLMModelTester(self    )
					A_					:			List[str]						=						ConfigTester(self ,       config_class=_lowerCamelCase ,       hidden_size=37    )







				def    a_					(							self				:   List[str]    ):



					"""simple docstring"""

					self.config_tester.run_common_tests()







				def    a_					(							self				:   str    ):



					"""simple docstring"""

					A_					:			str						=						self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_model(*_lowerCamelCase    )







				def    a_					(							self				:   Tuple    ):



					"""simple docstring"""

					A_					:			Tuple						=						self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_for_sequence_classification(*_lowerCamelCase    )







				def    a_					(							self				:   Optional[Any]    ):



					"""simple docstring"""

					A_					:			Union[str, Any]						=						self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_for_token_classification(*_lowerCamelCase    )







				def    a_					(							self				:   Optional[int]    ):



					"""simple docstring"""

					A_					:			Tuple						=						self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_for_question_answering(*_lowerCamelCase    )

def 						lowercase_ (						):




	"""simple docstring"""






	A_					:			Dict						=						'''
    <!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     lowercase (      unittest.TestCase):







				@cached_property
				def    a_					(							self				:   Tuple    ):



					"""simple docstring"""

					A_					:			int						=						MarkupLMFeatureExtractor()
					A_					:			Optional[int]						=						MarkupLMTokenizer.from_pretrained('''microsoft/markuplm-base'''    )

					return MarkupLMProcessor(_lowerCamelCase ,       _lowerCamelCase    )







				@slow
				def    a_					(							self				:   Optional[Any]    ):



					"""simple docstring"""

					A_					:			List[Any]						=						MarkupLMModel.from_pretrained('''microsoft/markuplm-base'''    ).to(_lowerCamelCase    )

					A_					:			List[Any]						=						self.default_processor

					A_					:			List[Any]						=						processor(prepare_html_string() ,       return_tensors='''pt'''    )
					A_					:			Tuple						=						inputs.to(_lowerCamelCase    )

					# forward pass
					with torch.no_grad():
						A_					:			Tuple						=						model(**_lowerCamelCase    )

					# verify the last hidden states
					A_					:			Dict						=						torch.Size([1, 14, 7_68]    )
					self.assertEqual(outputs.last_hidden_state.shape ,       _lowerCamelCase    )

					A_					:			Dict						=						torch.tensor(
					    [[0.0675, -0.0052, 0.5001], [-0.2281, 0.0802, 0.2192], [-0.0583, -0.3311, 0.1185]]    ).to(_lowerCamelCase    )

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





