


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

from transformers import BigBirdTokenizerFast


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




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




def   snake_case_	(    snake_case     ,				snake_case=False	)    ->							List[Any]:
       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__:  Optional[Any]         =  (
                  given_answers + get_sub_answers(snake_case     ,				begin=1	) + get_sub_answers(snake_case     ,				end=-1	)
              )
       lowercase__:  Optional[int]         =  []
       for answer in given_answers:
              lowercase__:  Dict         =  answer.replace('_'     ,				' '	).lower()
              lowercase__:  Union[str, Any]         =  ''.join(c if c not in PUNCTUATION_SET_TO_EXCLUDE else ' ' for c in alias	)
              answers.append(' '.join(alias.split()	).strip()	)
       return set(snake_case	)




def   snake_case_	(    snake_case     ,				snake_case     ,				snake_case=1     ,				snake_case=1_00	)    ->							str:
       lowercase__     ,		lowercase__:  str         =  jax.lax.top_k(snake_case     ,				snake_case	)
       lowercase__     ,		lowercase__:  Dict         =  jax.lax.top_k(snake_case     ,				snake_case	)

       lowercase__:  Dict         =  best_end_idx[:, None] - best_start_idx[None, :]
       lowercase__:  List[str]         =  jnp.logical_or(widths < 0     ,				widths > max_size	)
       lowercase__:  int         =  (best_end_scores[:, None] + best_start_scores[None, :]) - (1e8 * mask)
       lowercase__:  List[Any]         =  jnp.argmax(snake_case	).item()

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




def   snake_case_	(    snake_case	)    ->							Union[str, Any]:
       lowercase__:  int         =  sample['question']['text']
       lowercase__:  Union[str, Any]         =  sample['document']['tokens']['token']
       lowercase__:  Union[str, Any]         =  sample['document']['tokens']['is_html']
       lowercase__:  Optional[Any]         =  sample['annotations']['long_answer']
       lowercase__:  List[str]         =  sample['annotations']['short_answers']

       lowercase__:  Any         =  ' '.join([context[i] for i in range(len(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__:  Dict         =  []
       for s in short_answers:
              short_targets.extend(s['text']	)
       lowercase__:  Optional[Any]         =  list(set(snake_case	)	)

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

       lowercase__:  str         =  '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   snake_case_	(    )    ->							int:
       lowercase__:  Optional[Any]         =  load_from_disk('natural-questions-validation'	)
       lowercase__:  int         =  dataset.map(snake_case	).remove_columns(['annotations', 'document', 'id']	)
       print(snake_case	)

       lowercase__:  List[str]         =  dataset.filter(lambda snake_case	: (len(x['question']	) + len(x['context']	)) < 4 * 40_96	)
       lowercase__:  Any         =  short_validation_dataset.filter(lambda snake_case	: x["category"] != "null"	)
       short_validation_dataset

       lowercase__:  str         =  'vasudevgupta/flax-bigbird-natural-questions'
       lowercase__:  Any         =  FlaxBigBirdForNaturalQuestions.from_pretrained(snake_case	)
       lowercase__:  Tuple         =  BigBirdTokenizerFast.from_pretrained(snake_case	)

       @jax.jit
       def forward(*snake_case     ,				**snake_case	):
              lowercase__     ,		lowercase__     ,		lowercase__:  Dict         =  model(*snake_case     ,				**snake_case	)
              return start_logits, end_logits, jnp.argmax(snake_case     ,				axis=-1	)

       def evaluate(snake_case	):
              # encode question and context so that they are separated by a tokenizer.sep_token and cut at max_length
              lowercase__:  Tuple         =  tokenizer(
                  example['question']     ,				example['context']     ,				return_tensors='np'     ,				max_length=40_96     ,				padding='max_length'     ,				truncation=snake_case     ,				)

              lowercase__     ,		lowercase__     ,		lowercase__:  Union[str, Any]         =  forward(**snake_case	)

              lowercase__:  Optional[Any]         =  CATEGORY_MAPPING[category.item()]

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

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

              lowercase__:  int         =  38 if predicted_category == 'short' else 10_24
              lowercase__     ,		lowercase__:  Optional[Any]         =  get_best_valid_start_end_idx(
                  start_scores[0]     ,				end_scores[0]     ,				top_k=8     ,				max_size=snake_case	)

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

              lowercase__:  Optional[int]         =  expand_to_aliases(example['targets']     ,				make_sub_answers=snake_case	)
              lowercase__:  List[str]         =  expand_to_aliases(example['output']	)

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

              # if there is a common element, it's a exact match
              lowercase__:  str         =  len(list(answers & predictions	)	) > 0
              lowercase__:  List[Any]         =  predicted_category != 'null' and len(snake_case	) > 0

              return example

       lowercase__:  int         =  short_validation_dataset.map(snake_case	)

       lowercase__:  Tuple         =  len(snake_case	)
       lowercase__:  List[str]         =  len(short_validation_dataset.filter(lambda snake_case	: x["match"] == 1	)	)
       print('EM score:'     ,				(matched / total) * 1_00     ,				'%'	)


if __name__ == "__main__":
 main()






def   snake_case_	(    snake_case	)    ->							str:
       lowercase__:  Optional[Any]         =  1
       lowercase__:  Optional[Any]         =  0
       lowercase__:  int         =  1
       lowercase__:  Any         =  1
       lowercase__:  Optional[Any]         =  3
       lowercase__:  List[str]         =  3
       lowercase__:  Optional[Any]         =  limit
       lowercase__:  List[Any]         =  0

       lowercase__:  int         =  ''
       while counter != decimal + 1:
              if 4 * q + r - t < n * t:
                     result += str(snake_case	)
                     if counter == 0:
                            result += "."

                     if decimal == counter:
                            break

                     counter += 1
                     lowercase__:  List[Any]         =  10 * (r - n * t)
                     lowercase__:  int         =  ((10 * (3 * q + r)) // t) - 10 * n
                     q *= 10
                     lowercase__:  Any         =  nr
              else:
                     lowercase__:  List[str]         =  (2 * q + r) * l
                     lowercase__:  Tuple         =  (q * (7 * k) + 2 + (r * l)) // (t * l)
                     q *= k
                     t *= l
                     l += 2
                     k += 1
                     lowercase__:  Dict         =  nn
                     lowercase__:  Optional[int]         =  nr
       return result




def   snake_case_	(    )    ->							None:
       print(f'{calculate_pi(50	) = }'	)
       import doctest

       doctest.testmod()


if __name__ == "__main__":
 main()


