



'''simple docstring'''





from ...configuration_utils import PretrainedConfig
from ...utils import is_detectrona_available, logging


A			:   Dict			=      logging.get_logger(__name__)

A			:   str			=      {
    """layoutlmv2-base-uncased""": """https://huggingface.co/microsoft/layoutlmv2-base-uncased/resolve/main/config.json""",
    """layoutlmv2-large-uncased""": """https://huggingface.co/microsoft/layoutlmv2-large-uncased/resolve/main/config.json""",
    # See all LayoutLMv2 models at https://huggingface.co/models?filter=layoutlmv2
}

# soft dependency
if is_detectrona_available():
      import detectrona





class 							lowerCAmelCase_		(	a_						):
    __UpperCAmelCase							=	'layoutlmv2'





    def __init__(    self    :    int,    _snake_case    :    List[Any]=30_522,    _snake_case    :    Union[str, Any]=768,    _snake_case    :    Union[str, Any]=12,    _snake_case    :    Union[str, Any]=12,    _snake_case    :    int=3_072,    _snake_case    :    List[str]="gelu",    _snake_case    :    Optional[Any]=0.1,    _snake_case    :    Optional[int]=0.1,    _snake_case    :    str=512,    _snake_case    :    List[Any]=2,    _snake_case    :    Dict=0.02,    _snake_case    :    Optional[int]=1E-12,    _snake_case    :    Dict=0,    _snake_case    :    List[Any]=1_024,    _snake_case    :    str=128,    _snake_case    :    List[Any]=32,    _snake_case    :    List[Any]=True,    _snake_case    :    int=256,    _snake_case    :    Union[str, Any]=64,    _snake_case    :    int=True,    _snake_case    :    Tuple=[7, 7, 256],    _snake_case    :    Optional[int]=128,    _snake_case    :    Any=128,    _snake_case    :    Optional[Any]=True,    _snake_case    :    Union[str, Any]=True,    _snake_case    :    Dict=False,    _snake_case    :    Union[str, Any]=None,    **_snake_case    :    Any,    ):



       '''simple docstring'''







       super().__init__(
           vocab_size=_snake_case,    hidden_size=_snake_case,    num_hidden_layers=_snake_case,    num_attention_heads=_snake_case,    intermediate_size=_snake_case,    hidden_act=_snake_case,    hidden_dropout_prob=_snake_case,    attention_probs_dropout_prob=_snake_case,    max_position_embeddings=_snake_case,    type_vocab_size=_snake_case,    initializer_range=_snake_case,    layer_norm_eps=_snake_case,    pad_token_id=_snake_case,    **_snake_case,    )
       snake_case   :							str						  =max_ad_position_embeddings
       snake_case   :							Dict						  =max_rel_pos
       snake_case   :							Any						  =rel_pos_bins
       snake_case   :							int						  =fast_qkv
       snake_case   :							Union[str, Any]						  =max_rel_ad_pos
       snake_case   :							Optional[int]						  =rel_ad_pos_bins
       snake_case   :							Optional[int]						  =convert_sync_batchnorm
       snake_case   :							Optional[int]						  =image_feature_pool_shape
       snake_case   :							Optional[int]						  =coordinate_size
       snake_case   :							Optional[int]						  =shape_size
       snake_case   :							Optional[Any]						  =has_relative_attention_bias
       snake_case   :							Optional[int]						  =has_spatial_attention_bias
       snake_case   :							List[str]						  =has_visual_segment_embedding
       snake_case   :							int						  =(
           detectrona_config_args if detectrona_config_args is not None else self.get_default_detectrona_config()
       )





    @classmethod
    def 					__snake_case       (    self    :    Any     ):



       '''simple docstring'''







       return {
           "MODEL.MASK_ON": True,
           "MODEL.PIXEL_STD": [57.375, 57.120, 58.395],
           "MODEL.BACKBONE.NAME": "build_resnet_fpn_backbone",
           "MODEL.FPN.IN_FEATURES": ["res2", "res3", "res4", "res5"],
           "MODEL.ANCHOR_GENERATOR.SIZES": [[32], [64], [128], [256], [512]],
           "MODEL.RPN.IN_FEATURES": ["p2", "p3", "p4", "p5", "p6"],
           "MODEL.RPN.PRE_NMS_TOPK_TRAIN": 2_000,
           "MODEL.RPN.PRE_NMS_TOPK_TEST": 1_000,
           "MODEL.RPN.POST_NMS_TOPK_TRAIN": 1_000,
           "MODEL.POST_NMS_TOPK_TEST": 1_000,
           "MODEL.ROI_HEADS.NAME": "StandardROIHeads",
           "MODEL.ROI_HEADS.NUM_CLASSES": 5,
           "MODEL.ROI_HEADS.IN_FEATURES": ["p2", "p3", "p4", "p5"],
           "MODEL.ROI_BOX_HEAD.NAME": "FastRCNNConvFCHead",
           "MODEL.ROI_BOX_HEAD.NUM_FC": 2,
           "MODEL.ROI_BOX_HEAD.POOLER_RESOLUTION": 14,
           "MODEL.ROI_MASK_HEAD.NAME": "MaskRCNNConvUpsampleHead",
           "MODEL.ROI_MASK_HEAD.NUM_CONV": 4,
           "MODEL.ROI_MASK_HEAD.POOLER_RESOLUTION": 7,
           "MODEL.RESNETS.DEPTH": 101,
           "MODEL.RESNETS.SIZES": [[32], [64], [128], [256], [512]],
           "MODEL.RESNETS.ASPECT_RATIOS": [[0.5, 1.0, 2.0]],
           "MODEL.RESNETS.OUT_FEATURES": ["res2", "res3", "res4", "res5"],
           "MODEL.RESNETS.NUM_GROUPS": 32,
           "MODEL.RESNETS.WIDTH_PER_GROUP": 8,
           "MODEL.RESNETS.STRIDE_IN_1X1": False,
       }






    def 					__snake_case       (    self    :    List[str]     ):



       '''simple docstring'''







       snake_case   :							List[str]						  =detectrona.config.get_cfg()
       for k, v in self.detectrona_config_args.items():
          snake_case   :							Any						  =k.split('''.'''     )
          snake_case   :							Optional[Any]						  =detectrona_config
          for attribute in attributes[:-1]:
             snake_case   :							Dict						  =getattr(_snake_case,    _snake_case     )
          setattr(_snake_case,    attributes[-1],    _snake_case     )

       return detectrona_config










'''simple docstring'''





import collections
import types

import numpy as np

from ..utils import (
    add_end_docstrings,
    is_tensorflow_probability_available,
    is_tf_available,
    is_torch_available,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, ArgumentHandler, Dataset, Pipeline, PipelineException


if is_torch_available():
      import torch

      from ..models.auto.modeling_auto import (
          MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING,
          MODEL_FOR_TABLE_QUESTION_ANSWERING_MAPPING,
      )

if is_tf_available() and is_tensorflow_probability_available():
      import tensorflow as tf
      import tensorflow_probability as tfp

      from ..models.auto.modeling_tf_auto import (
          TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING,
          TF_MODEL_FOR_TABLE_QUESTION_ANSWERING_MAPPING,
      )





class 							lowerCAmelCase_		(	a_						):





    def __call__(    self    :    str,    _snake_case    :    Dict=None,    _snake_case    :    str=None,    **_snake_case    :    Any     ):



       '''simple docstring'''







       requires_backends(self,    '''pandas'''     )
       import pandas as pd

       if table is None:
          raise ValueError('''Keyword argument `table` cannot be None.'''     )
       elif query is None:
          if isinstance(_snake_case,    _snake_case     ) and table.get('''query'''     ) is not None and table.get('''table'''     ) is not None:
             snake_case   :							Union[str, Any]						  =[table]
          elif isinstance(_snake_case,    _snake_case     ) and len(_snake_case     ) > 0:
             if not all(isinstance(_snake_case,    _snake_case     ) for d in table     ):
                raise ValueError(
                    f'''Keyword argument `table` should be a list of dict, but is {(type(_snake_case     ) for d in table)}'''     )

             if table[0].get('''query'''     ) is not None and table[0].get('''table'''     ) is not None:
                snake_case   :							Dict						  =table
             else:
                raise ValueError(
                    '''If keyword argument `table` is a list of dictionaries, each dictionary should have a `table`'''
                    f''' and `query` key, but only dictionary has keys {table[0].keys()} `table` and `query` keys.'''     )
          elif Dataset is not None and isinstance(_snake_case,    _snake_case     ) or isinstance(_snake_case,    types.GeneratorType     ):
             return table
          else:
             raise ValueError(
                 '''Invalid input. Keyword argument `table` should be either of type `dict` or `list`, but '''
                 f'''is {type(_snake_case     )})'''     )
       else:
          snake_case   :							Union[str, Any]						  =[{'''table''': table, '''query''': query}]

       for tqa_pipeline_input in tqa_pipeline_inputs:
          if not isinstance(tqa_pipeline_input['''table'''],    pd.DataFrame     ):
             if tqa_pipeline_input["table"] is None:
                raise ValueError('''Table cannot be None.'''     )

             snake_case   :							Any						  =pd.DataFrame(tqa_pipeline_input['''table''']     )

       return tqa_pipeline_inputs





@add_end_docstrings(a_						)
class 							lowerCAmelCase_		(	a_						):
    __UpperCAmelCase							=	'table,query'





    def __init__(    self    :    Any,    _snake_case    :    Any=TableQuestionAnsweringArgumentHandler(),    *_snake_case    :    List[Any],    **_snake_case    :    Dict     ):



       '''simple docstring'''







       super().__init__(*_snake_case,    **_snake_case     )
       snake_case   :							str						  =args_parser

       self.check_model_type(
           dict(
               TF_MODEL_FOR_TABLE_QUESTION_ANSWERING_MAPPING.items()
               + TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING.items()     )
           if self.framework == '''tf'''
           else dict(
               MODEL_FOR_TABLE_QUESTION_ANSWERING_MAPPING.items() + MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING.items()     )     )

       snake_case   :							Dict						  =bool(getattr(self.model.config,    '''aggregation_labels''',    _snake_case     )     ) and bool(
           getattr(self.model.config,    '''num_aggregation_labels''',    _snake_case     )     )
       snake_case   :							List[str]						  ='''tapas''' if hasattr(self.model.config,    '''aggregation_labels'''     ) else None





    def 					__snake_case       (    self    :    Dict,    **_snake_case    :    Tuple     ):



       '''simple docstring'''







       return self.model(**_snake_case     )





    def 					__snake_case       (    self    :    List[Any],    **_snake_case    :    List[str]     ):



       '''simple docstring'''







       if self.framework == "pt":
          snake_case   :							str						  =[]
          snake_case   :							str						  =[]
          snake_case   :							Optional[int]						  =None
          snake_case   :							List[str]						  =inputs['''input_ids'''].shape[0]

          snake_case   :							Union[str, Any]						  =inputs['''input_ids'''].to(self.device     )
          snake_case   :							List[str]						  =inputs['''attention_mask'''].to(self.device     )
          snake_case   :							Optional[int]						  =inputs['''token_type_ids'''].to(self.device     )
          snake_case   :							Any						  =None

          for index in range(_snake_case     ):
             # If sequences have already been processed, the token type IDs will be created according to the previous
             # answer.
             if prev_answers is not None:
                snake_case   :							Union[str, Any]						  =token_type_ids_example[:, 3]  # shape (seq_len,)
                snake_case   :							Any						  =np.zeros_like(prev_labels_example.cpu().numpy()     )  # shape (seq_len,)

                snake_case   :							Any						  =token_type_ids[index]  # shape (seq_len, 7)
                for i in range(model_labels.shape[0]     ):
                   snake_case   :							Tuple						  =token_type_ids_example[:, 0].tolist()[i]
                   snake_case   :							Tuple						  =token_type_ids_example[:, 1].tolist()[i] - 1
                   snake_case   :							Optional[int]						  =token_type_ids_example[:, 2].tolist()[i] - 1

                   if row_id >= 0 and col_id >= 0 and segment_id == 1:
                      snake_case   :							List[Any]						  =int(prev_answers[(col_id, row_id)]     )

                snake_case   :							List[Any]						  =torch.from_numpy(_snake_case     ).type(torch.long     ).to(self.device     )

             snake_case   :							Tuple						  =input_ids[index]
             snake_case   :							Optional[int]						  =attention_mask[index]  # shape (seq_len,)
             snake_case   :							List[Any]						  =token_type_ids[index]  # shape (seq_len, 7)
             snake_case   :							Union[str, Any]						  =self.model(
                 input_ids=input_ids_example.unsqueeze(0     ),    attention_mask=attention_mask_example.unsqueeze(0     ),    token_type_ids=token_type_ids_example.unsqueeze(0     ),    )
             snake_case   :							int						  =outputs.logits

             if self.aggregate:
                all_aggregations.append(outputs.logits_aggregation     )

             all_logits.append(_snake_case     )

             snake_case   :							List[Any]						  =torch.distributions.Bernoulli(logits=_snake_case     )
             snake_case   :							Optional[Any]						  =dist_per_token.probs * attention_mask_example.type(torch.floataa     ).to(
                 dist_per_token.probs.device     )

             snake_case   :							Dict						  =collections.defaultdict(_snake_case     )
             for i, p in enumerate(probabilities.squeeze().tolist()     ):
                snake_case   :							int						  =token_type_ids_example[:, 0].tolist()[i]
                snake_case   :							Optional[int]						  =token_type_ids_example[:, 1].tolist()[i] - 1
                snake_case   :							Dict						  =token_type_ids_example[:, 2].tolist()[i] - 1
                if col >= 0 and row >= 0 and segment_id == 1:
                   coords_to_probs[(col, row)].append(_snake_case     )

             snake_case   :							Union[str, Any]						  ={key: np.array(coords_to_probs[key]     ).mean() > 0.5 for key in coords_to_probs}

          snake_case   :							Tuple						  =torch.cat(tuple(_snake_case     ),    0     )

          return (logits_batch,) if not self.aggregate else (logits_batch, torch.cat(tuple(_snake_case     ),    0     ))
       else:
          snake_case   :							Tuple						  =[]
          snake_case   :							Optional[int]						  =[]
          snake_case   :							str						  =None
          snake_case   :							Dict						  =inputs['''input_ids'''].shape[0]

          snake_case   :							Optional[Any]						  =inputs['''input_ids''']
          snake_case   :							Optional[int]						  =inputs['''attention_mask''']
          snake_case   :							Union[str, Any]						  =inputs['''token_type_ids'''].numpy()
          snake_case   :							Tuple						  =None

          for index in range(_snake_case     ):
             # If sequences have already been processed, the token type IDs will be created according to the previous
             # answer.
             if prev_answers is not None:
                snake_case   :							Any						  =token_type_ids_example[:, 3]  # shape (seq_len,)
                snake_case   :							str						  =np.zeros_like(_snake_case,    dtype=np.intaa     )  # shape (seq_len,)

                snake_case   :							Any						  =token_type_ids[index]  # shape (seq_len, 7)
                for i in range(model_labels.shape[0]     ):
                   snake_case   :							Union[str, Any]						  =token_type_ids_example[:, 0].tolist()[i]
                   snake_case   :							Dict						  =token_type_ids_example[:, 1].tolist()[i] - 1
                   snake_case   :							Any						  =token_type_ids_example[:, 2].tolist()[i] - 1

                   if row_id >= 0 and col_id >= 0 and segment_id == 1:
                      snake_case   :							Any						  =int(prev_answers[(col_id, row_id)]     )

                snake_case   :							Union[str, Any]						  =model_labels

             snake_case   :							List[Any]						  =input_ids[index]
             snake_case   :							Tuple						  =attention_mask[index]  # shape (seq_len,)
             snake_case   :							Optional[Any]						  =token_type_ids[index]  # shape (seq_len, 7)
             snake_case   :							Optional[Any]						  =self.model(
                 input_ids=np.expand_dims(_snake_case,    axis=0     ),    attention_mask=np.expand_dims(_snake_case,    axis=0     ),    token_type_ids=np.expand_dims(_snake_case,    axis=0     ),    )
             snake_case   :							List[str]						  =outputs.logits

             if self.aggregate:
                all_aggregations.append(outputs.logits_aggregation     )

             all_logits.append(_snake_case     )

             snake_case   :							int						  =tfp.distributions.Bernoulli(logits=_snake_case     )
             snake_case   :							Union[str, Any]						  =dist_per_token.probs_parameter() * tf.cast(_snake_case,    tf.floataa     )

             snake_case   :							List[Any]						  =collections.defaultdict(_snake_case     )
             snake_case   :							Dict						  =token_type_ids_example
             for i, p in enumerate(tf.squeeze(_snake_case     ).numpy().tolist()     ):
                snake_case   :							Optional[Any]						  =token_type_ids_example[:, 0].tolist()[i]
                snake_case   :							Optional[Any]						  =token_type_ids_example[:, 1].tolist()[i] - 1
                snake_case   :							str						  =token_type_ids_example[:, 2].tolist()[i] - 1
                if col >= 0 and row >= 0 and segment_id == 1:
                   coords_to_probs[(col, row)].append(_snake_case     )

             snake_case   :							Optional[int]						  ={key: np.array(coords_to_probs[key]     ).mean() > 0.5 for key in coords_to_probs}

          snake_case   :							Dict						  =tf.concat(tuple(_snake_case     ),    0     )

          return (logits_batch,) if not self.aggregate else (logits_batch, tf.concat(tuple(_snake_case     ),    0     ))





    def __call__(    self    :    List[str],    *_snake_case    :    Any,    **_snake_case    :    List[Any]     ):



       '''simple docstring'''







       snake_case   :							Tuple						  =self._args_parser(*_snake_case,    **_snake_case     )

       snake_case   :							Dict						  =super().__call__(_snake_case,    **_snake_case     )
       if len(_snake_case     ) == 1:
          return results[0]
       return results





    def 					__snake_case       (    self    :    Dict,    _snake_case    :    Union[str, Any]=None,    _snake_case    :    Union[str, Any]=None,    _snake_case    :    List[Any]=None,    **_snake_case    :    List[str]     ):



       '''simple docstring'''







       snake_case   :							Optional[Any]						  ={}
       if padding is not None:
          snake_case   :							List[Any]						  =padding
       if truncation is not None:
          snake_case   :							int						  =truncation

       snake_case   :							Any						  ={}
       if sequential is not None:
          snake_case   :							int						  =sequential
       return preprocess_params, forward_params, {}





    def 					__snake_case       (    self    :    Dict,    _snake_case    :    Dict,    _snake_case    :    List[Any]=None,    _snake_case    :    Optional[int]=True,    _snake_case    :    Dict=None     ):



       '''simple docstring'''







       if truncation is None:
          if self.type == "tapas":
             snake_case   :							Union[str, Any]						  ='''drop_rows_to_fit'''
          else:
             snake_case   :							Dict						  ='''do_not_truncate'''

       snake_case  ,    snake_case   :							Optional[int]						  =pipeline_input['''table'''], pipeline_input['''query''']
       if table.empty:
          raise ValueError('''table is empty'''     )
       if query is None or query == "":
          raise ValueError('''query is empty'''     )
       snake_case   :							Dict						  =self.tokenizer(_snake_case,    _snake_case,    return_tensors=self.framework,    truncation=_snake_case,    padding=_snake_case     )
       snake_case   :							List[Any]						  =table
       return inputs





    def 					__snake_case       (    self    :    List[str],    _snake_case    :    List[Any],    _snake_case    :    Any=False     ):



       '''simple docstring'''







       snake_case   :							Union[str, Any]						  =model_inputs.pop('''table'''     )

       if self.type == "tapas":
          if sequential:
             snake_case   :							Dict						  =self.sequential_inference(**_snake_case     )
          else:
             snake_case   :							Optional[int]						  =self.batch_inference(**_snake_case     )
       else:
          snake_case   :							str						  =self.model.generate(**_snake_case     )
       snake_case   :							int						  ={'''model_inputs''': model_inputs, '''table''': table, '''outputs''': outputs}
       return model_outputs






    def 					__snake_case       (    self    :    Optional[int],    _snake_case    :    Optional[Any]     ):



       '''simple docstring'''







       snake_case   :							str						  =model_outputs['''model_inputs''']
       snake_case   :							Tuple						  =model_outputs['''table''']
       snake_case   :							Union[str, Any]						  =model_outputs['''outputs''']
       if self.type == "tapas":
          if self.aggregate:
             snake_case  ,    snake_case   :							Union[str, Any]						  =outputs[:2]
             snake_case   :							Tuple						  =self.tokenizer.convert_logits_to_predictions(_snake_case,    _snake_case,    _snake_case     )
             snake_case  ,    snake_case   :							Any						  =predictions
             snake_case   :							int						  ={i: self.model.config.aggregation_labels[pred] for i, pred in enumerate(_snake_case     )}

             snake_case   :							List[str]						  =self.model.config.no_aggregation_label_index
             snake_case   :							Dict						  ={
                 i: aggregators[i] + ''' > ''' for i, pred in enumerate(_snake_case     ) if pred != no_agg_label_index
             }
          else:
             snake_case   :							List[str]						  =outputs[0]
             snake_case   :							List[Any]						  =self.tokenizer.convert_logits_to_predictions(_snake_case,    _snake_case     )
             snake_case   :							str						  =predictions[0]
             snake_case   :							str						  ={}
             snake_case   :							str						  ={}
          snake_case   :							Optional[int]						  =[]
          for index, coordinates in enumerate(_snake_case     ):
             snake_case   :							int						  =[table.iat[coordinate] for coordinate in coordinates]
             snake_case   :							Dict						  =aggregators.get(_snake_case,    ''''''     )
             snake_case   :							Any						  =aggregators_prefix.get(_snake_case,    ''''''     )
             snake_case   :							int						  ={
                 '''answer''': aggregator_prefix + ''', '''.join(_snake_case     ),
                 '''coordinates''': coordinates,
                 '''cells''': [table.iat[coordinate] for coordinate in coordinates],
             }
             if aggregator:
                snake_case   :							Tuple						  =aggregator

             answers.append(_snake_case     )
          if len(_snake_case     ) == 0:
             raise PipelineException('''Empty answer'''     )
       else:
          snake_case   :							Optional[int]						  =[{'''answer''': answer} for answer in self.tokenizer.batch_decode(_snake_case,    skip_special_tokens=_snake_case     )]

       return answers if len(_snake_case     ) > 1 else answers[0]





