
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     __A(				a    ):




				def __call__(  self	,  _snake_case=None	,  _snake_case=None	,  **_snake_case       )					->   int:


								'''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:
																__a    							=					[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:
																				__a    							=					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:
												__a    							=					[{'''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.'''       )

																__a    							=					pd.DataFrame(tqa_pipeline_input['''table''']       )

								return tqa_pipeline_inputs







@add_end_docstrings(a    )
class     __A(				a    ):
				snake_case_		     =					'''table,query'''




				def __init__(  self	,  _snake_case=TableQuestionAnsweringArgumentHandler()	,  *_snake_case	,  **_snake_case       )					->   Optional[Any]:


								'''simple docstring'''




								super().__init__(*_snake_case	,  **_snake_case       )
								__a    							=					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()       )       )

								__a    							=					bool(getattr(self.model.config	,  '''aggregation_labels'''	,  _snake_case       )       ) and bool(
								    getattr(self.model.config	,  '''num_aggregation_labels'''	,  _snake_case       )       )
								__a    							=					'''tapas''' if hasattr(self.model.config	,  '''aggregation_labels'''       ) else None




				def        SCREAMING_SNAKE_CASE_	(  self	,  **_snake_case       )					->   Tuple:


								'''simple docstring'''




								return self.model(**_snake_case       )




				def        SCREAMING_SNAKE_CASE_	(  self	,  **_snake_case       )					->   Any:


								'''simple docstring'''




								if self.framework == "pt":
												__a    							=					[]
												__a    							=					[]
												__a    							=					None
												__a    							=					inputs['''input_ids'''].shape[0]

												__a    							=					inputs['''input_ids'''].to(self.device       )
												__a    							=					inputs['''attention_mask'''].to(self.device       )
												__a    							=					inputs['''token_type_ids'''].to(self.device       )
												__a    							=					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:
																				__a    							=					token_type_ids_example[:, 3]  # shape (seq_len,)
																				__a    							=					np.zeros_like(prev_labels_example.cpu().numpy()       )  # shape (seq_len,)

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

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

																				__a    							=					torch.from_numpy(_snake_case       ).type(torch.long       ).to(self.device       )

																__a    							=					input_ids[index]
																__a    							=					attention_mask[index]  # shape (seq_len,)
																__a    							=					token_type_ids[index]  # shape (seq_len, 7)
																__a    							=					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       )	,  )
																__a    							=					outputs.logits

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

																all_logits.append(_snake_case       )

																__a    							=					torch.distributions.Bernoulli(logits=_snake_case       )
																__a    							=					dist_per_token.probs * attention_mask_example.type(torch.floataa       ).to(
																    dist_per_token.probs.device       )

																__a    							=					collections.defaultdict(_snake_case       )
																for i, p in enumerate(probabilities.squeeze().tolist()       ):
																				__a    							=					token_type_ids_example[:, 0].tolist()[i]
																				__a    							=					token_type_ids_example[:, 1].tolist()[i] - 1
																				__a    							=					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       )

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

												__a    							=					torch.cat(tuple(_snake_case       )	,  0       )

												return (logits_batch,) if not self.aggregate else (logits_batch, torch.cat(tuple(_snake_case       )	,  0       ))
								else:
												__a    							=					[]
												__a    							=					[]
												__a    							=					None
												__a    							=					inputs['''input_ids'''].shape[0]

												__a    							=					inputs['''input_ids''']
												__a    							=					inputs['''attention_mask''']
												__a    							=					inputs['''token_type_ids'''].numpy()
												__a    							=					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:
																				__a    							=					token_type_ids_example[:, 3]  # shape (seq_len,)
																				__a    							=					np.zeros_like(_snake_case	,  dtype=np.intaa       )  # shape (seq_len,)

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

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

																				__a    							=					model_labels

																__a    							=					input_ids[index]
																__a    							=					attention_mask[index]  # shape (seq_len,)
																__a    							=					token_type_ids[index]  # shape (seq_len, 7)
																__a    							=					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       )	,  )
																__a    							=					outputs.logits

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

																all_logits.append(_snake_case       )

																__a    							=					tfp.distributions.Bernoulli(logits=_snake_case       )
																__a    							=					dist_per_token.probs_parameter() * tf.cast(_snake_case	,  tf.floataa       )

																__a    							=					collections.defaultdict(_snake_case       )
																__a    							=					token_type_ids_example
																for i, p in enumerate(tf.squeeze(_snake_case       ).numpy().tolist()       ):
																				__a    							=					token_type_ids_example[:, 0].tolist()[i]
																				__a    							=					token_type_ids_example[:, 1].tolist()[i] - 1
																				__a    							=					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       )

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

												__a    							=					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	,  *_snake_case	,  **_snake_case       )					->   Dict:


								'''simple docstring'''




								__a    							=					self._args_parser(*_snake_case	,  **_snake_case       )

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




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case=None	,  _snake_case=None	,  _snake_case=None	,  **_snake_case       )					->   Optional[Any]:


								'''simple docstring'''




								__a    							=					{}
								if padding is not None:
												__a    							=					padding
								if truncation is not None:
												__a    							=					truncation

								__a    							=					{}
								if sequential is not None:
												__a    							=					sequential
								return preprocess_params, forward_params, {}




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case	,  _snake_case=None	,  _snake_case=True	,  _snake_case=None       )					->   Union[str, Any]:


								'''simple docstring'''




								if truncation is None:
												if self.type == "tapas":
																__a    							=					'''drop_rows_to_fit'''
												else:
																__a    							=					'''do_not_truncate'''

								__a							,  __a    							=					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'''       )
								__a    							=					self.tokenizer(_snake_case	,  _snake_case	,  return_tensors=self.framework	,  truncation=_snake_case	,  padding=_snake_case       )
								__a    							=					table
								return inputs




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case	,  _snake_case=False       )					->   Optional[Any]:


								'''simple docstring'''




								__a    							=					model_inputs.pop('''table'''       )

								if self.type == "tapas":
												if sequential:
																__a    							=					self.sequential_inference(**_snake_case       )
												else:
																__a    							=					self.batch_inference(**_snake_case       )
								else:
												__a    							=					self.model.generate(**_snake_case       )
								__a    							=					{'''model_inputs''': model_inputs, '''table''': table, '''outputs''': outputs}
								return model_outputs




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case       )					->   Any:


								'''simple docstring'''




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

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

																answers.append(_snake_case       )
												if len(_snake_case       ) == 0:
																raise PipelineException('''Empty answer'''       )
								else:
												__a    							=					[{'''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]

import os

from apache_beam.io.filesystems import FileSystems
from apache_beam.pipeline import Pipeline

from .logging import get_logger


A  :			Optional[int]     				=				2 << 2_0  # 2mb
A  :			Optional[int]     				=				get_logger(__name__)







class     __A(				a    ):




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


								'''simple docstring'''




								__a    							=					self._options.get_all_options().get('''runner'''       )
								return runner in [None, "DirectRunner", "PortableRunner"]



def 					__lowerCAmelCase    (    a__		,					a__		,					a__=False						) ->						Union[str, Any]:
				__a    							=					FileSystems
				if fs.exists(a__						):
								if force_upload:
												logger.info(F"""Remote path already exist: {remote_file_path}. Overwriting it as force_upload=True."""						)
								else:
												logger.info(F"""Remote path already exist: {remote_file_path}. Skipping it as force_upload=False."""						)
												return
				with fs.create(a__						) as remote_file:
								with open(a__		,					'''rb'''						) as local_file:
												__a    							=					local_file.read(a__						)
												while chunk:
																remote_file.write(a__						)
																__a    							=					local_file.read(a__						)






def 					__lowerCAmelCase    (    a__		,					a__		,					a__=False						) ->						Dict:
				__a    							=					FileSystems
				if os.path.exists(a__						):
								if force_download:
												logger.info(F"""Local path already exist: {remote_file_path}. Overwriting it as force_upload=True."""						)
								else:
												logger.info(F"""Local path already exist: {remote_file_path}. Skipping it as force_upload=False."""						)
												return
				with fs.open(a__						) as remote_file:
								with open(a__		,					'''wb'''						) as local_file:
												__a    							=					remote_file.read(a__						)
												while chunk:
																local_file.write(a__						)
																__a    							=					remote_file.read(a__						)