


'''simple docstring'''



import os

import jsonlines
import numpy as np
from tqdm import tqdm


__lowercase	:						Any       =	20_48
__lowercase	:						Optional[int]       =	40_96
__lowercase	:						int       =	42
__lowercase	:						Tuple       =	os.environ.pop('PROCESS_TRAIN', 'false')
__lowercase	:						List[Any]       =	{'null': 0, 'short': 1, 'long': 2, 'yes': 3, 'no': 4}






def   lowerCamelCase							(_SCREAMING_SNAKE_CASE      :   Dict   ):
					def choose_first(_SCREAMING_SNAKE_CASE      :   Optional[Any]						,      _SCREAMING_SNAKE_CASE      :   List[str]=False   ):
										assert isinstance(_SCREAMING_SNAKE_CASE						,      _SCREAMING_SNAKE_CASE   )
										if len(_SCREAMING_SNAKE_CASE   ) == 1:
															__a			:       Optional[int]					    =  answer[0]
															return {k: [answer[k]] for k in answer} if is_long_answer else answer
										for a in answer:
															if is_long_answer:
																				__a			:       Any					    =  {k: [a[k]] for k in a}
															if len(a['start_token']   ) > 0:
																				break
										return a

					__a			:       int					    =  {'id': example['id']}
					__a			:       str					    =  example['annotations']
					__a			:       str					    =  annotation['yes_no_answer']
					if 0 in yes_no_answer or 1 in yes_no_answer:
										__a			:       Any					    =  ['yes'] if 1 in yes_no_answer else ['no']
										__a			:       List[Any]					    =  []
										__a			:       Optional[Any]					    =  []
										__a			:       Tuple					    =  ['<cls>']
					else:
										__a			:       Optional[int]					    =  ['short']
										__a			:       List[Any]					    =  choose_first(annotation['short_answers']   )
										if len(out['start_token']   ) == 0:
															# answer will be long if short is not available
															__a			:       Optional[Any]					    =  ['long']
															__a			:       Any					    =  choose_first(annotation['long_answer']						,      is_long_answer=_SCREAMING_SNAKE_CASE   )
															__a			:       List[str]					    =  []
										answer.update(_SCREAMING_SNAKE_CASE   )

					# disregard some samples
					if len(answer['start_token']   ) > 1 or answer["start_token"] == answer["end_token"]:
										__a			:       List[str]					    =  True
					else:
										__a			:       Optional[Any]					    =  False

					__a			:       List[str]					    =  ['start_token', 'end_token', 'start_byte', 'end_byte', 'text']
					if not all(isinstance(answer[k]						,      _SCREAMING_SNAKE_CASE   ) for k in cols   ):
										raise ValueError('Issue in ID'						,      example['id']   )

					return answer






def   lowerCamelCase							(_SCREAMING_SNAKE_CASE      :   Optional[int]						,      _SCREAMING_SNAKE_CASE      :   int=False   ):
					__a			:       Optional[int]					    =  _get_single_answer(_SCREAMING_SNAKE_CASE   )
					# bytes are of no use
					del answer["start_byte"]
					del answer["end_byte"]

					# handle yes_no answers explicitly
					if answer["category"][0] in ["yes", "no"]:  # category is list with one element
										__a			:       int					    =  example['document']['tokens']
										__a			:       Tuple					    =  []
										for i in range(len(doc['token']   )   ):
															if not doc["is_html"][i]:
																				context.append(doc['token'][i]   )
										return {
										    "context": " ".join(_SCREAMING_SNAKE_CASE   ),
										    "answer": {
										        "start_token": -100,  # ignore index in cross-entropy
										        "end_token": -100,  # ignore index in cross-entropy
										        "category": answer["category"],
										        "span": answer["category"],  # extra
										    },
										}

					# later, help in removing all no answers
					if answer["start_token"] == [-1]:
										return {
										    "context": "None",
										    "answer": {
										        "start_token": -1,
										        "end_token": -1,
										        "category": "null",
										        "span": "None",  # extra
										    },
										}

					# handling normal samples

					__a			:       List[Any]					    =  ['start_token', 'end_token']
					answer.update({k: answer[k][0] if len(answer[k]   ) > 0 else answer[k] for k in cols}   )  # e.g. [10] == 10

					__a			:       Tuple					    =  example['document']['tokens']
					__a			:       int					    =  answer['start_token']
					__a			:       str					    =  answer['end_token']

					__a			:       List[Any]					    =  []
					for i in range(len(doc['token']   )   ):
										if not doc["is_html"][i]:
															context.append(doc['token'][i]   )
										else:
															if answer["start_token"] > i:
																				start_token -= 1
															if answer["end_token"] > i:
																				end_token -= 1
					__a			:       List[str]					    =  ' '.join(context[start_token:end_token]   )

					# checking above code
					if assertion:
										__a			:       Dict					    =  doc['is_html'][answer['start_token'] : answer['end_token']]
										__a			:       str					    =  doc['token'][answer['start_token'] : answer['end_token']]
										__a			:       Optional[int]					    =  ' '.join([old[i] for i in range(len(_SCREAMING_SNAKE_CASE   )   ) if not is_html[i]]   )
										if new != old:
															print('ID:'						,      example['id']   )
															print('New:'						,      _SCREAMING_SNAKE_CASE						,      end='\n'   )
															print('Old:'						,      _SCREAMING_SNAKE_CASE						,      end='\n\n'   )

					return {
					    "context": " ".join(_SCREAMING_SNAKE_CASE   ),
					    "answer": {
					        "start_token": start_token,
					        "end_token": end_token - 1,  # this makes it inclusive
					        "category": answer["category"],  # either long or short
					        "span": new,  # extra
					    },
					}






def   lowerCamelCase							(_SCREAMING_SNAKE_CASE      :   Optional[int]						,      _SCREAMING_SNAKE_CASE      :   Dict						,      _SCREAMING_SNAKE_CASE      :   Tuple=2_048						,      _SCREAMING_SNAKE_CASE      :   Optional[Any]=4_096						,      _SCREAMING_SNAKE_CASE      :   int=True   ):
					# overlap will be of doc_stride - q_len

					__a			:       int					    =  get_context_and_ans(_SCREAMING_SNAKE_CASE						,      assertion=_SCREAMING_SNAKE_CASE   )
					__a			:       int					    =  out['answer']

					# later, removing these samples
					if answer["start_token"] == -1:
										return {
										    "example_id": example["id"],
										    "input_ids": [[-1]],
										    "labels": {
										        "start_token": [-1],
										        "end_token": [-1],
										        "category": ["null"],
										    },
										}

					__a			:       Optional[Any]					    =  tokenizer(example['question']['text']						,      out['context']   ).input_ids
					__a			:       Any					    =  input_ids.index(tokenizer.sep_token_id   ) + 1

					# return yes/no
					if answer["category"][0] in ["yes", "no"]:  # category is list with one element
										__a			:       Dict					    =  []
										__a			:       Tuple					    =  []
										__a			:       Optional[Any]					    =  input_ids[:q_len]
										__a			:       Tuple					    =  range(_SCREAMING_SNAKE_CASE						,      len(_SCREAMING_SNAKE_CASE   )						,      max_length - doc_stride   )
										for i in doc_start_indices:
															__a			:       Dict					    =  i + max_length - q_len
															__a			:       int					    =  input_ids[i:end_index]
															inputs.append(q_indices + slice   )
															category.append(answer['category'][0]   )
															if slice[-1] == tokenizer.sep_token_id:
																				break

										return {
										    "example_id": example["id"],
										    "input_ids": inputs,
										    "labels": {
										        "start_token": [-100] * len(_SCREAMING_SNAKE_CASE   ),
										        "end_token": [-100] * len(_SCREAMING_SNAKE_CASE   ),
										        "category": category,
										    },
										}

					__a			:       Dict					    =  out['context'].split()
					__a			:       Optional[Any]					    =  splitted_context[answer['end_token']]
					__a			:       Optional[Any]					    =  len(
					    tokenizer(
					        ' '.join(splitted_context[: answer['start_token']]   )						,      add_special_tokens=_SCREAMING_SNAKE_CASE						,      ).input_ids   )
					__a			:       Optional[int]					    =  len(
					    tokenizer(' '.join(splitted_context[: answer['end_token']]   )						,      add_special_tokens=_SCREAMING_SNAKE_CASE   ).input_ids   )

					answer["start_token"] += q_len
					answer["end_token"] += q_len

					# fixing end token
					__a			:       Optional[Any]					    =  len(tokenizer(_SCREAMING_SNAKE_CASE						,      add_special_tokens=_SCREAMING_SNAKE_CASE   ).input_ids   )
					if num_sub_tokens > 1:
										answer["end_token"] += num_sub_tokens - 1

					__a			:       List[str]					    =  input_ids[answer['start_token'] : answer['end_token'] + 1]  # right & left are inclusive
					__a			:       List[Any]					    =  answer['start_token']
					__a			:       Dict					    =  answer['end_token']

					if assertion:
										__a			:       List[str]					    =  tokenizer.decode(_SCREAMING_SNAKE_CASE   )
										if answer["span"] != new:
															print('ISSUE IN TOKENIZATION'   )
															print('OLD:'						,      answer['span']   )
															print('NEW:'						,      _SCREAMING_SNAKE_CASE						,      end='\n\n'   )

					if len(_SCREAMING_SNAKE_CASE   ) <= max_length:
										return {
										    "example_id": example["id"],
										    "input_ids": [input_ids],
										    "labels": {
										        "start_token": [answer["start_token"]],
										        "end_token": [answer["end_token"]],
										        "category": answer["category"],
										    },
										}

					__a			:       Optional[Any]					    =  input_ids[:q_len]
					__a			:       Optional[int]					    =  range(_SCREAMING_SNAKE_CASE						,      len(_SCREAMING_SNAKE_CASE   )						,      max_length - doc_stride   )

					__a			:       Optional[Any]					    =  []
					__a			:       Optional[int]					    =  []
					__a			:       int					    =  []
					__a			:       int					    =  []  # null, yes, no, long, short
					for i in doc_start_indices:
										__a			:       Tuple					    =  i + max_length - q_len
										__a			:       Dict					    =  input_ids[i:end_index]
										inputs.append(q_indices + slice   )
										assert len(inputs[-1]   ) <= max_length, "Issue in truncating length"

										if start_token >= i and end_token <= end_index - 1:
															__a			:       Optional[int]					    =  start_token - i + q_len
															__a			:       Dict					    =  end_token - i + q_len
															answers_category.append(answer['category'][0]   )  # ["short"] -> "short"
										else:
															__a			:       int					    =  -100
															__a			:       Tuple					    =  -100
															answers_category.append('null'   )
										__a			:       Union[str, Any]					    =  inputs[-1][start_token : end_token + 1]

										answers_start_token.append(_SCREAMING_SNAKE_CASE   )
										answers_end_token.append(_SCREAMING_SNAKE_CASE   )
										if assertion:
															if new != old and new != [tokenizer.cls_token_id]:
																				print('ISSUE in strided for ID:'						,      example['id']   )
																				print('New:'						,      tokenizer.decode(_SCREAMING_SNAKE_CASE   )   )
																				print('Old:'						,      tokenizer.decode(_SCREAMING_SNAKE_CASE   )						,      end='\n\n'   )
										if slice[-1] == tokenizer.sep_token_id:
															break

					return {
					    "example_id": example["id"],
					    "input_ids": inputs,
					    "labels": {
					        "start_token": answers_start_token,
					        "end_token": answers_end_token,
					        "category": answers_category,
					    },
					}






def   lowerCamelCase							(_SCREAMING_SNAKE_CASE      :   Dict						,      _SCREAMING_SNAKE_CASE      :   Any						,      _SCREAMING_SNAKE_CASE      :   int=2_048						,      _SCREAMING_SNAKE_CASE      :   Optional[Any]=4_096						,      _SCREAMING_SNAKE_CASE      :   Tuple=False   ):
					__a			:       Dict					    =  get_strided_contexts_and_ans(
					    _SCREAMING_SNAKE_CASE						,      _SCREAMING_SNAKE_CASE						,      doc_stride=_SCREAMING_SNAKE_CASE						,      max_length=_SCREAMING_SNAKE_CASE						,      assertion=_SCREAMING_SNAKE_CASE						,      )

					return example






def   lowerCamelCase							(_SCREAMING_SNAKE_CASE      :   Optional[Any]						,      _SCREAMING_SNAKE_CASE      :   Any   ):
					with jsonlines.open(_SCREAMING_SNAKE_CASE						,      'a'   ) as writer:
										for example in tqdm(_SCREAMING_SNAKE_CASE						,      total=len(_SCREAMING_SNAKE_CASE   )						,      desc='Saving samples ... '   ):
															__a			:       Optional[Any]					    =  example['labels']
															for ids, start, end, cat in zip(
															    example['input_ids']						,      labels['start_token']						,      labels['end_token']						,      labels['category']						,      ):
																				if start == -1 and end == -1:
																									continue  # leave waste samples with no answer
																				if cat == "null" and np.random.rand() < 0.6:
																									continue  # removing 50 % samples
																				writer.write(
																				    {
																				        'input_ids': ids,
																				        'start_token': start,
																				        'end_token': end,
																				        'category': CATEGORY_MAPPING[cat],
																				    }   )


if __name__ == "__main__":
	from datasets import load_dataset

	from transformers import BigBirdTokenizer

	__lowercase	:						Union[str, Any]       =	load_dataset('natural_questions')
	__lowercase	:						int       =	BigBirdTokenizer.from_pretrained('google/bigbird-roberta-base')

	__lowercase	:						List[str]       =	data['train' if PROCESS_TRAIN == 'true' else 'validation']

	__lowercase	:						Any       =	{
	    'tokenizer': tokenizer,
	    'doc_stride': DOC_STRIDE,
	    'max_length': MAX_LENGTH,
	    'assertion': False,
	}
	__lowercase	:						List[str]       =	data.map(prepare_inputs, fn_kwargs=fn_kwargs)
	__lowercase	:						Dict       =	data.remove_columns(['annotations', 'document', 'id', 'question'])
	print(data)

	np.random.seed(SEED)
	__lowercase	:						Union[str, Any]       =	'nq-training.jsonl' if PROCESS_TRAIN == 'true' else 'nq-validation.jsonl'
	save_to_disk(data, file_name=cache_file_name)







'''simple docstring'''



import torch

from transformers import AutoModel





class 							__UpperCamelCase				(		torch.nn.Module					):




	def __init__(   self						,				__a="sayef/fsner-bert-base-uncased"							):

						'''simple docstring'''





						super(__a						,				self							).__init__()

						__a			:       Tuple					    =  AutoModel.from_pretrained(__a						,				return_dict=__a							)
						__a			:       int					    =  torch.nn.CosineSimilarity(3						,				1E-0_8							)
						__a			:       Union[str, Any]					    =  torch.nn.Softmax(dim=1							)




	def       __UpperCAmelCase							(   self						,				**__a							):

						'''simple docstring'''





						return self.bert(**__a							).last_hidden_state




	def       __UpperCAmelCase							(   self						,				__a							):

						'''simple docstring'''





						return token_embeddings.sum(2						,				keepdim=__a							)




	def       __UpperCAmelCase							(   self						,				__a						,				__a						,				__a=1							):

						'''simple docstring'''





						return self.softmax(T * self.cos(__a						,				__a							)							)





	def       __UpperCAmelCase							(   self						,				__a						,				__a							):

						'''simple docstring'''





						__a			:       str					    =  W_supports['sizes'].tolist()
						__a			:       Union[str, Any]					    =  W_supports['start_token_id'].item()
						__a			:       Any					    =  W_supports['end_token_id'].item()

						del W_supports["sizes"]
						del W_supports["start_token_id"]
						del W_supports["end_token_id"]

						__a			:       Tuple					    =  self.BERT(**__a							)
						__a			:       str					    =  self.BERT(**__a							)

						__a			:       Any					    =  None
						__a			:       Dict					    =  None

						__a			:       Dict					    =  W_supports['input_ids'] == start_token_id
						__a			:       Union[str, Any]					    =  W_supports['input_ids'] == end_token_id

						for i, size in enumerate(__a							):
											if i == 0:
																__a			:       Optional[int]					    =  0
											else:
																__a			:       Union[str, Any]					    =  support_sizes[i - 1]

											__a			:       int					    =  S[s : s + size][start_token_masks[s : s + size]]
											__a			:       Union[str, Any]					    =  S[s : s + size][end_token_masks[s : s + size]]

											__a			:       Tuple					    =  torch.matmul(q[i]						,				s_start.T							).sum(1							).softmax(0							)
											__a			:       Dict					    =  torch.matmul(q[i]						,				s_end.T							).sum(1							).softmax(0							)

											if p_starts is not None:
																__a			:       str					    =  torch.vstack((p_starts, p_start)							)
																__a			:       str					    =  torch.vstack((p_ends, p_end)							)
											else:
																__a			:       List[str]					    =  p_start
																__a			:       int					    =  p_end

						return p_starts, p_ends



