




"""simple docstring"""

import os
from typing import Optional, Union

import numpy as np
from huggingface_hub import hf_hub_download

from ... import AutoTokenizer
from ...utils import logging


_lowerCamelCase					:     Tuple				       =    'block_records.npy'


_lowerCamelCase					:     Any				       =    logging.get_logger(__name__)

def 						lowercase_ (						_UpperCAmelCase					, _UpperCAmelCase		):




	"""simple docstring"""






	import tensorflow.compat.va as tf

	A_					:			Any						=						tf.data.TFRecordDataset(_UpperCAmelCase					, buffer_size=512 * 1024 * 1024		)
	A_					:			Any						=						blocks_dataset.batch(_UpperCAmelCase					, drop_remainder=_UpperCAmelCase		)
	A_					:			Tuple						=						next(blocks_dataset.take(1		).as_numpy_iterator()		)

	return np_record

class     lowercase :







				def __init__(							self				:   List[str] ,       _lowerCamelCase				:   Union[str, Any] ,       _lowerCamelCase				:   Any ,       _lowerCamelCase				:   Tuple=2 ,       _lowerCamelCase				:   str=10_00 ,       _lowerCamelCase				:   Optional[int]=1_00 ,       _lowerCamelCase				:   List[str]=10_00_00 ,       ):



					"""simple docstring"""

					from scann.scann_ops.py.scann_ops_pybind import builder as Builder

					A_					:			List[str]						=						Builder(db=_lowerCamelCase ,       num_neighbors=_lowerCamelCase ,       distance_measure='''dot_product'''    )
					A_					:			Tuple						=						builder.tree(
					    num_leaves=_lowerCamelCase ,       num_leaves_to_search=_lowerCamelCase ,       training_sample_size=_lowerCamelCase    )
					A_					:			Dict						=						builder.score_ah(dimensions_per_block=_lowerCamelCase    )

					A_					:			int						=						builder.build()







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



					"""simple docstring"""

					A_				,	A_					:			List[Any]						=						self.searcher.search_batched(question_projection.detach().cpu()    )
					return retrieved_block_ids.astype('''int64'''    )

class     lowercase :







				def __init__(							self				:   Union[str, Any] ,       _lowerCamelCase				:   Any ,       _lowerCamelCase				:   List[Any]    ):



					"""simple docstring"""

					super().__init__()
					A_					:			str						=						block_records
					A_					:			List[Any]						=						tokenizer







				def __call__(							self				:   List[str] ,       _lowerCamelCase				:   Optional[int] ,       _lowerCamelCase				:   Union[str, Any] ,       _lowerCamelCase				:   Tuple ,       _lowerCamelCase				:   List[Any]=None ,       _lowerCamelCase				:   Optional[int]="pt"    ):



					"""simple docstring"""

					A_					:			Optional[int]						=						np.take(self.block_records ,       indices=_lowerCamelCase ,       axis=0    )

					A_					:			Any						=						self.tokenizer.decode(question_input_ids[0] ,       skip_special_tokens=_lowerCamelCase    )

					A_					:			Optional[Any]						=						[]
					A_					:			Optional[Any]						=						[]
					for retrieved_block in retrieved_blocks:
						text.append(_lowerCamelCase    )
						text_pair.append(retrieved_block.decode()    )

					A_					:			Optional[Any]						=						self.tokenizer(
					    _lowerCamelCase ,       _lowerCamelCase ,       padding=_lowerCamelCase ,       truncation=_lowerCamelCase ,       return_special_tokens_mask=_lowerCamelCase ,       max_length=_lowerCamelCase    )
					A_					:			Any						=						concat_inputs.convert_to_tensors(_lowerCamelCase    )

					if answer_ids is not None:
						return self.block_has_answer(_lowerCamelCase ,       _lowerCamelCase    ) + (concat_inputs_tensors,)
					else:
						return (None, None, None, concat_inputs_tensors)







				@classmethod
				def    a_					(							cls				:   int ,       _lowerCamelCase				:   Optional[Union[str, os.PathLike]] ,       *_lowerCamelCase				:   Any ,       **_lowerCamelCase				:   Optional[int]    ):



					"""simple docstring"""

					if os.path.isdir(_lowerCamelCase    ):
						A_					:			Any						=						os.path.join(_lowerCamelCase ,       _REALM_BLOCK_RECORDS_FILENAME    )
					else:
						A_					:			List[str]						=						hf_hub_download(
						    repo_id=_lowerCamelCase ,       filename=_REALM_BLOCK_RECORDS_FILENAME ,       **_lowerCamelCase    )
					A_					:			Optional[int]						=						np.load(_lowerCamelCase ,       allow_pickle=_lowerCamelCase    )

					A_					:			str						=						AutoTokenizer.from_pretrained(_lowerCamelCase ,       *_lowerCamelCase ,       **_lowerCamelCase    )

					return cls(_lowerCamelCase ,       _lowerCamelCase    )







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



					"""simple docstring"""

					np.save(os.path.join(_lowerCamelCase ,       _REALM_BLOCK_RECORDS_FILENAME    ) ,       self.block_records    )
					# save tokenizer
					self.tokenizer.save_pretrained(_lowerCamelCase    )







				def    a_					(							self				:   int ,       _lowerCamelCase				:   str ,       _lowerCamelCase				:   Any    ):



					"""simple docstring"""

					A_					:			Tuple						=						[]
					A_					:			Tuple						=						[]
					A_					:			List[str]						=						[]
					A_					:			int						=						0

					for input_id in concat_inputs.input_ids:
						A_					:			Optional[Any]						=						input_id.tolist()
						# Check answers between two [SEP] tokens
						A_					:			Optional[Any]						=						input_id_list.index(self.tokenizer.sep_token_id    )
						A_					:			List[Any]						=						first_sep_idx + 1 + input_id_list[first_sep_idx + 1 :].index(self.tokenizer.sep_token_id    )

						start_pos.append([]    )
						end_pos.append([]    )
						for answer in answer_ids:
							for idx in range(first_sep_idx + 1 ,       _lowerCamelCase    ):
								if answer[0] == input_id_list[idx]:
									if input_id_list[idx : idx + len(_lowerCamelCase    )] == answer:
										start_pos[-1].append(_lowerCamelCase    )
										end_pos[-1].append(idx + len(_lowerCamelCase    ) - 1    )

						if len(start_pos[-1]    ) == 0:
							has_answers.append(_lowerCamelCase    )
						else:
							has_answers.append(_lowerCamelCase    )
							if len(start_pos[-1]    ) > max_answers:
								A_					:			int						=						len(start_pos[-1]    )

        # Pad -1 to max_answers
					for start_pos_, end_pos_ in zip(_lowerCamelCase ,       _lowerCamelCase    ):
						if len(start_pos_    ) < max_answers:
							A_					:			Tuple						=						[-1] * (max_answers - len(start_pos_    ))
							start_pos_ += padded
							end_pos_ += padded
					return has_answers, start_pos, end_pos











"""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    )    )





