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

from transformers import BigBirdTokenizerFast


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




def 							__SCREAMING_SNAKE_CASE       (		SCREAMING_SNAKE_CASE	:				Optional[Any]   ,							SCREAMING_SNAKE_CASE	:				Optional[int]=0   ,							SCREAMING_SNAKE_CASE	:				Dict=None      )					->						List[Any]:
				return [" ".join(x.split(' '      )[begin:end]      ) for x in answers if len(x.split(' '      )      ) > 1]




def 							__SCREAMING_SNAKE_CASE       (		SCREAMING_SNAKE_CASE	:				str   ,							SCREAMING_SNAKE_CASE	:				int=False      )					->						int:
				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"
								__lowercase								= (
								    given_answers + get_sub_answers(SCREAMING_SNAKE_CASE   ,							begin=1      ) + get_sub_answers(SCREAMING_SNAKE_CASE   ,							end=-1      )
								)
				__lowercase								= []
				for answer in given_answers:
								__lowercase								= answer.replace('_'   ,							' '      ).lower()
								__lowercase								= ''.join(c if c not in PUNCTUATION_SET_TO_EXCLUDE else ' ' for c in alias      )
								answers.append(' '.join(alias.split()      ).strip()      )
				return set(SCREAMING_SNAKE_CASE      )




def 							__SCREAMING_SNAKE_CASE       (		SCREAMING_SNAKE_CASE	:				Union[str, Any]   ,							SCREAMING_SNAKE_CASE	:				List[str]   ,							SCREAMING_SNAKE_CASE	:				int=1   ,							SCREAMING_SNAKE_CASE	:				Tuple=100      )					->						List[Any]:
				__lowercase		, __lowercase								= jax.lax.top_k(SCREAMING_SNAKE_CASE   ,							SCREAMING_SNAKE_CASE      )
				__lowercase		, __lowercase								= jax.lax.top_k(SCREAMING_SNAKE_CASE   ,							SCREAMING_SNAKE_CASE      )

				__lowercase								= best_end_idx[:, None] - best_start_idx[None, :]
				__lowercase								= jnp.logical_or(widths < 0   ,							widths > max_size      )
				__lowercase								= (best_end_scores[:, None] + best_start_scores[None, :]) - (1E8 * mask)
				__lowercase								= jnp.argmax(SCREAMING_SNAKE_CASE      ).item()

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




def 							__SCREAMING_SNAKE_CASE       (		SCREAMING_SNAKE_CASE	:				Dict      )					->						Dict:
				__lowercase								= sample['question']['text']
				__lowercase								= sample['document']['tokens']['token']
				__lowercase								= sample['document']['tokens']['is_html']
				__lowercase								= sample['annotations']['long_answer']
				__lowercase								= sample['annotations']['short_answers']

				__lowercase								= ' '.join([context[i] for i in range(len(SCREAMING_SNAKE_CASE      )      ) 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",
												}

				__lowercase								= []
				for s in short_answers:
								short_targets.extend(s['text']      )
				__lowercase								= list(set(SCREAMING_SNAKE_CASE      )      )

				__lowercase								= []
				for s in long_answers:
								if s["start_token"] == -1:
												continue
								__lowercase								= context[s['start_token'] : s['end_token']]
								__lowercase								= is_html[s['start_token'] : s['end_token']]
								__lowercase								= ' '.join([answer[i] for i in range(len(SCREAMING_SNAKE_CASE      )      ) if not html[i]]      )
								if new_answer not in long_targets:
												long_targets.append(SCREAMING_SNAKE_CASE      )

				__lowercase								= '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 							__SCREAMING_SNAKE_CASE       (		)					->						Tuple:
				__lowercase								= load_from_disk('natural-questions-validation'      )
				__lowercase								= dataset.map(SCREAMING_SNAKE_CASE      ).remove_columns(['annotations', 'document', 'id']      )
				print(SCREAMING_SNAKE_CASE      )

				__lowercase								= dataset.filter(lambda SCREAMING_SNAKE_CASE      : (len(x['question']      ) + len(x['context']      )) < 4 * 4096      )
				__lowercase								= short_validation_dataset.filter(lambda SCREAMING_SNAKE_CASE      : x["category"] != "null"      )
				short_validation_dataset

				__lowercase								= 'vasudevgupta/flax-bigbird-natural-questions'
				__lowercase								= FlaxBigBirdForNaturalQuestions.from_pretrained(SCREAMING_SNAKE_CASE      )
				__lowercase								= BigBirdTokenizerFast.from_pretrained(SCREAMING_SNAKE_CASE      )

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

				def evaluate(SCREAMING_SNAKE_CASE	:				List[Any]      ):
								# encode question and context so that they are separated by a tokenizer.sep_token and cut at max_length
								__lowercase								= tokenizer(
								    example['question']   ,							example['context']   ,							return_tensors='np'   ,							max_length=4096   ,							padding='max_length'   ,							truncation=SCREAMING_SNAKE_CASE   ,							)

								__lowercase		, __lowercase		, __lowercase								= forward(**SCREAMING_SNAKE_CASE      )

								__lowercase								= CATEGORY_MAPPING[category.item()]

								__lowercase								= example['long'] + example['short']
								if example["category"] in ["yes", "no", "null"]:
												__lowercase								= [example['category']]
								__lowercase								= example['category'] != 'null'
								# Now target can be: "yes", "no", "null", "list of long & short answers"

								if predicted_category in ["yes", "no", "null"]:
												__lowercase								= [predicted_category]
												__lowercase								= example['output'] == example['targets']
												__lowercase								= predicted_category != 'null'
												return example

								__lowercase								= 38 if predicted_category == 'short' else 1024
								__lowercase		, __lowercase								= get_best_valid_start_end_idx(
								    start_scores[0]   ,							end_scores[0]   ,							top_k=8   ,							max_size=SCREAMING_SNAKE_CASE      )

								__lowercase								= inputs['input_ids'][0].tolist()
								__lowercase								= [tokenizer.decode(input_ids[start_score : end_score + 1]      )]

								__lowercase								= expand_to_aliases(example['targets']   ,							make_sub_answers=SCREAMING_SNAKE_CASE      )
								__lowercase								= expand_to_aliases(example['output']      )

								# some preprocessing to both prediction and answer
								__lowercase								= {''.join(a.split()      ) for a in answers}
								__lowercase								= {''.join(p.split()      ) for p in predictions}
								__lowercase								= {s for s in predictions if s not in ['``', '\'\'', '`', '\'']}

								# if there is a common element, it's a exact match
								__lowercase								= len(list(answers & predictions      )      ) > 0
								__lowercase								= predicted_category != 'null' and len(SCREAMING_SNAKE_CASE      ) > 0

								return example

				__lowercase								= short_validation_dataset.map(SCREAMING_SNAKE_CASE      )

				__lowercase								= len(SCREAMING_SNAKE_CASE      )
				__lowercase								= len(short_validation_dataset.filter(lambda SCREAMING_SNAKE_CASE      : x["match"] == 1      )      )
				print('EM score:'   ,							(matched / total) * 100   ,							'%'      )


if __name__ == "__main__":
							main()







from __future__ import annotations

from enum import Enum
class      A__       (						lowerCAmelCase__    ):
							lowerCAmelCase__		:						Dict         =							24
							lowerCAmelCase__		:						Dict         =							21
							lowerCAmelCase__		:						Union[str, Any]         =							18
							lowerCAmelCase__		:						Dict         =							15
							lowerCAmelCase__		:						Any         =							12
							lowerCAmelCase__		:						Union[str, Any]         =							9
							lowerCAmelCase__		:						Any         =							6
							lowerCAmelCase__		:						Optional[Any]         =							3
							lowerCAmelCase__		:						Union[str, Any]         =							2
							lowerCAmelCase__		:						List[Any]         =							1
							lowerCAmelCase__		:						Optional[Any]         =							-1
							lowerCAmelCase__		:						Tuple         =							-2
							lowerCAmelCase__		:						Dict         =							-3
							lowerCAmelCase__		:						Union[str, Any]         =							-6
							lowerCAmelCase__		:						Any         =							-9
							lowerCAmelCase__		:						Union[str, Any]         =							-12
							lowerCAmelCase__		:						str         =							-15
							lowerCAmelCase__		:						Optional[int]         =							-18
							lowerCAmelCase__		:						Any         =							-21
							lowerCAmelCase__		:						Any         =							-24






class      A__       (						lowerCAmelCase__    ):
							lowerCAmelCase__		:						Tuple         =							8
							lowerCAmelCase__		:						Union[str, Any]         =							7
							lowerCAmelCase__		:						Optional[Any]         =							6
							lowerCAmelCase__		:						Tuple         =							5
							lowerCAmelCase__		:						Optional[int]         =							4
							lowerCAmelCase__		:						Optional[int]         =							3
							lowerCAmelCase__		:						Optional[int]         =							2
							lowerCAmelCase__		:						Any         =							1






def 							__SCREAMING_SNAKE_CASE       (		SCREAMING_SNAKE_CASE	:				float   ,							SCREAMING_SNAKE_CASE	:				str | SIUnit   ,							SCREAMING_SNAKE_CASE	:				str | SIUnit   ,							)					->						float:
				if isinstance(SCREAMING_SNAKE_CASE   ,							SCREAMING_SNAKE_CASE      ):
								__lowercase								= SIUnit[known_prefix.lower()]
				if isinstance(SCREAMING_SNAKE_CASE   ,							SCREAMING_SNAKE_CASE      ):
								__lowercase								= SIUnit[unknown_prefix.lower()]
				__lowercase								= known_amount * (
				    10 ** (known_prefix.value - unknown_prefix.value)
				)
				return unknown_amount




def 							__SCREAMING_SNAKE_CASE       (		SCREAMING_SNAKE_CASE	:				float   ,							SCREAMING_SNAKE_CASE	:				str | BinaryUnit   ,							SCREAMING_SNAKE_CASE	:				str | BinaryUnit   ,							)					->						float:
				if isinstance(SCREAMING_SNAKE_CASE   ,							SCREAMING_SNAKE_CASE      ):
								__lowercase								= BinaryUnit[known_prefix.lower()]
				if isinstance(SCREAMING_SNAKE_CASE   ,							SCREAMING_SNAKE_CASE      ):
								__lowercase								= BinaryUnit[unknown_prefix.lower()]
				__lowercase								= known_amount * (
				    2 ** ((known_prefix.value - unknown_prefix.value) * 10)
				)
				return unknown_amount


if __name__ == "__main__":
							import doctest

							doctest.testmod()





