






import jax
import jax.numpy as jnp
from bigbird_flax import FlaxBigBirdForNaturalQuestions
from datasets import load_from_disk

from transformers import BigBirdTokenizerFast


lowercase     :							List[Any]	     =			{0: '''null''', 1: '''short''', 2: '''long''', 3: '''yes''', 4: '''no'''}
lowercase     :							int	     =			set(''''''.join(['''‘''', '''’''', '''´''', '''`''', '''.''', ''',''', '''-''', '''"''']))

def    lowerCAmelCase__		(			_a				:	Optional[Any]     ,					_a				:	Optional[Any]=0     ,					_a				:	Tuple=None			):
		return [" ".join(x.split(" "			)[begin:end]			) for x in answers if len(x.split(" "			)			) > 1]

def    lowerCAmelCase__		(			_a				:	Optional[Any]     ,					_a				:	Optional[Any]=False			):
		if make_sub_answers:
				# if answers are longer than one word, make sure a predictions is correct if it coresponds to the complete 1: or :-1 sub word
				# *e.g.* if the correct answer contains a prefix such as "the", or "a"
				snake_case_ :							List[str]         =				(
				    given_answers + get_sub_answers(_a     ,					begin=1			) + get_sub_answers(_a     ,					end=-1			)
				)
		snake_case_ :							Optional[int]         =				[]
		for answer in given_answers:
				snake_case_ :							str         =				answer.replace("_"     ,					" "			).lower()
				snake_case_ :							int         =				"".join(c if c not in PUNCTUATION_SET_TO_EXCLUDE else " " for c in alias			)
				answers.append(" ".join(alias.split()			).strip()			)
		return set(_a			)

def    lowerCAmelCase__		(			_a				:	Union[str, Any]     ,					_a				:	Optional[int]     ,					_a				:	Any=1     ,					_a				:	List[Any]=1_00			):
		snake_case_       ,				snake_case_ :							Dict         =				jax.lax.top_k(_a     ,					_a			)
		snake_case_       ,				snake_case_ :							List[Any]         =				jax.lax.top_k(_a     ,					_a			)

		snake_case_ :							Optional[int]         =				best_end_idx[:, None] - best_start_idx[None, :]
		snake_case_ :							List[str]         =				jnp.logical_or(widths < 0     ,					widths > max_size			)
		snake_case_ :							Optional[Any]         =				(best_end_scores[:, None] + best_start_scores[None, :]) - (1E8 * mask)
		snake_case_ :							Union[str, Any]         =				jnp.argmax(_a			).item()

		return best_start_idx[best_score % top_k], best_end_idx[best_score // top_k]

def    lowerCAmelCase__		(			_a				:	Union[str, Any]			):
		snake_case_ :							List[str]         =				sample["question"]["text"]
		snake_case_ :							Optional[Any]         =				sample["document"]["tokens"]["token"]
		snake_case_ :							Any         =				sample["document"]["tokens"]["is_html"]
		snake_case_ :							Optional[int]         =				sample["annotations"]["long_answer"]
		snake_case_ :							List[Any]         =				sample["annotations"]["short_answers"]

		snake_case_ :							List[Any]         =				" ".join([context[i] for i in range(len(_a			)			) if not is_html[i]]			)

		# 0 - No ; 1 - Yes
		for answer in sample["annotations"]["yes_no_answer"]:
				if answer == 0 or answer == 1:
						return {
						    "question": question,
						    "context": context_string,
						    "short": [],
						    "long": [],
						    "category": "no" if answer == 0 else "yes",
						}

		snake_case_ :							Dict         =				[]
		for s in short_answers:
				short_targets.extend(s["text"]			)
		snake_case_ :							Optional[Any]         =				list(set(_a			)			)

		snake_case_ :							List[str]         =				[]
		for s in long_answers:
				if s["start_token"] == -1:
						continue
				snake_case_ :							int         =				context[s["start_token"] : s["end_token"]]
				snake_case_ :							Union[str, Any]         =				is_html[s["start_token"] : s["end_token"]]
				snake_case_ :							List[Any]         =				" ".join([answer[i] for i in range(len(_a			)			) if not html[i]]			)
				if new_answer not in long_targets:
						long_targets.append(_a			)

		snake_case_ :							Dict         =				"long_short" if len(short_targets + long_targets			) > 0 else "null"

		return {
		    "question": question,
		    "context": context_string,
		    "short": short_targets,
		    "long": long_targets,
		    "category": category,
		}

def    lowerCAmelCase__		(			):
		snake_case_ :							List[Any]         =				load_from_disk("natural-questions-validation"			)
		snake_case_ :							List[str]         =				dataset.map(_a			).remove_columns(["annotations", "document", "id"]			)
		print(_a			)

		snake_case_ :							Optional[int]         =				dataset.filter(lambda _a			: (len(x["question"]			) + len(x["context"]			)) < 4 * 40_96			)
		snake_case_ :							Tuple         =				short_validation_dataset.filter(lambda _a			: x["category"] != "null"			)
		short_validation_dataset

		snake_case_ :							List[Any]         =				"vasudevgupta/flax-bigbird-natural-questions"
		snake_case_ :							List[Any]         =				FlaxBigBirdForNaturalQuestions.from_pretrained(_a			)
		snake_case_ :							int         =				BigBirdTokenizerFast.from_pretrained(_a			)

		@jax.jit
		def forward(*_a				:	str     ,					**_a				:	List[str]			):
				snake_case_       ,				snake_case_       ,				snake_case_ :							Optional[int]         =				model(*_a     ,					**_a			)
				return start_logits, end_logits, jnp.argmax(_a     ,					axis=-1			)

		def evaluate(_a				:	Optional[int]			):
				# encode question and context so that they are separated by a tokenizer.sep_token and cut at max_length
				snake_case_ :							int         =				tokenizer(
				    example["question"]     ,					example["context"]     ,					return_tensors="np"     ,					max_length=40_96     ,					padding="max_length"     ,					truncation=_a     ,					)

				snake_case_       ,				snake_case_       ,				snake_case_ :							Tuple         =				forward(**_a			)

				snake_case_ :							Union[str, Any]         =				CATEGORY_MAPPING[category.item()]

				snake_case_ :							Union[str, Any]         =				example["long"] + example["short"]
				if example["category"] in ["yes", "no", "null"]:
						snake_case_ :							Dict         =				[example["category"]]
				snake_case_ :							Optional[Any]         =				example["category"] != "null"
				# Now target can be: "yes", "no", "null", "list of long & short answers"

				if predicted_category in ["yes", "no", "null"]:
						snake_case_ :							List[str]         =				[predicted_category]
						snake_case_ :							str         =				example["output"] == example["targets"]
						snake_case_ :							List[Any]         =				predicted_category != "null"
						return example

				snake_case_ :							List[str]         =				38 if predicted_category == "short" else 10_24
				snake_case_       ,				snake_case_ :							Tuple         =				get_best_valid_start_end_idx(
				    start_scores[0]     ,					end_scores[0]     ,					top_k=8     ,					max_size=_a			)

				snake_case_ :							List[Any]         =				inputs["input_ids"][0].tolist()
				snake_case_ :							str         =				[tokenizer.decode(input_ids[start_score : end_score + 1]			)]

				snake_case_ :							int         =				expand_to_aliases(example["targets"]     ,					make_sub_answers=_a			)
				snake_case_ :							str         =				expand_to_aliases(example["output"]			)

				# some preprocessing to both prediction and answer
				snake_case_ :							Optional[Any]         =				{"".join(a.split()			) for a in answers}
				snake_case_ :							int         =				{"".join(p.split()			) for p in predictions}
				snake_case_ :							List[Any]         =				{s for s in predictions if s not in ["``", "''", "`", "'"]}

				# if there is a common element, it's a exact match
				snake_case_ :							str         =				len(list(answers & predictions			)			) > 0
				snake_case_ :							str         =				predicted_category != "null" and len(_a			) > 0

				return example

		snake_case_ :							Tuple         =				short_validation_dataset.map(_a			)

		snake_case_ :							Union[str, Any]         =				len(_a			)
		snake_case_ :							int         =				len(short_validation_dataset.filter(lambda _a			: x["match"] == 1			)			)
		print("EM score:"     ,					(matched / total) * 1_00     ,					"%"			)


if __name__ == "__main__":
							main()








def    lowerCAmelCase__		(			_a				:	int			):
		if not isinstance(_a     ,					_a			):
				snake_case_ :							Optional[int]         =				F'''Input value of [number={number}] must be an integer'''
				raise TypeError(_a			)
		if number < 1:
				raise ValueError("Input must be a positive integer"			)
		return number * (2 * number - 1)


if __name__ == "__main__":
							import doctest

							doctest.testmod()
