


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

from transformers import BigBirdTokenizerFast


snake_case__     							= {0: 'null', 1: 'short', 2: 'long', 3: 'yes', 4: 'no'}
snake_case__     							= set(''.join(['‘', '’', '´', '`', '.', ',', '-', '"']))
def  __SCREAMING_SNAKE_CASE       (				lowercase__  :Union[str, Any]      ,						lowercase__  :str=0      ,						lowercase__  :str=None				)       ->  Any:
						return [" ".join(x.split(' '				)[begin:end]				) for x in answers if len(x.split(' '				)				) > 1]
def  __SCREAMING_SNAKE_CASE       (				lowercase__  :List[Any]      ,						lowercase__  :List[Any]=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"
												__magic_name__				:							List[str]				=      (
												    given_answers + get_sub_answers(lowercase__      ,						begin=1				) + get_sub_answers(lowercase__      ,						end=-1				)
												)
						__magic_name__				:							Union[str, Any]				=      []
						for answer in given_answers:
												__magic_name__				:							List[Any]				=      answer.replace('_'      ,						' '				).lower()
												__magic_name__				:							int				=      ''.join(c if c not in PUNCTUATION_SET_TO_EXCLUDE else ' ' for c in alias				)
												answers.append(' '.join(alias.split()				).strip()				)
						return set(lowercase__				)
def  __SCREAMING_SNAKE_CASE       (				lowercase__  :int      ,						lowercase__  :List[Any]      ,						lowercase__  :List[Any]=1      ,						lowercase__  :List[str]=1_0_0				)       ->  List[Any]:
						__magic_name__		, __magic_name__				:							Any				=      jax.lax.top_k(lowercase__      ,						lowercase__				)
						__magic_name__		, __magic_name__				:							Tuple				=      jax.lax.top_k(lowercase__      ,						lowercase__				)

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

						return best_start_idx[best_score % top_k], best_end_idx[best_score // top_k]
def  __SCREAMING_SNAKE_CASE       (				lowercase__  :Tuple				)       ->  Optional[Any]:
						__magic_name__				:							Any				=      sample['question']['text']
						__magic_name__				:							Union[str, Any]				=      sample['document']['tokens']['token']
						__magic_name__				:							Dict				=      sample['document']['tokens']['is_html']
						__magic_name__				:							Optional[Any]				=      sample['annotations']['long_answer']
						__magic_name__				:							List[str]				=      sample['annotations']['short_answers']

						__magic_name__				:							Union[str, Any]				=      ' '.join([context[i] for i in range(len(lowercase__				)				) 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",
																		}

						__magic_name__				:							Union[str, Any]				=      []
						for s in short_answers:
												short_targets.extend(s['text']				)
						__magic_name__				:							int				=      list(set(lowercase__				)				)

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

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

						__magic_name__				:							int				=      dataset.filter(lambda lowercase__				: (len(x['question']				) + len(x['context']				)) < 4 * 4_0_9_6				)
						__magic_name__				:							str				=      short_validation_dataset.filter(lambda lowercase__				: x["category"] != "null"				)
						short_validation_dataset

						__magic_name__				:							int				=      'vasudevgupta/flax-bigbird-natural-questions'
						__magic_name__				:							int				=      FlaxBigBirdForNaturalQuestions.from_pretrained(lowercase__				)
						__magic_name__				:							Union[str, Any]				=      BigBirdTokenizerFast.from_pretrained(lowercase__				)

						@jax.jit
						def forward(*lowercase__  :Optional[int]      ,						**lowercase__  :Any				):
												__magic_name__		, __magic_name__		, __magic_name__				:							Optional[Any]				=      model(*lowercase__      ,						**lowercase__				)
												return start_logits, end_logits, jnp.argmax(lowercase__      ,						axis=-1				)

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

												__magic_name__		, __magic_name__		, __magic_name__				:							List[str]				=      forward(**lowercase__				)

												__magic_name__				:							int				=      CATEGORY_MAPPING[category.item()]

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

												if predicted_category in ["yes", "no", "null"]:
																		__magic_name__				:							Tuple				=      [predicted_category]
																		__magic_name__				:							int				=      example['output'] == example['targets']
																		__magic_name__				:							Any				=      predicted_category != 'null'
																		return example

												__magic_name__				:							Any				=      3_8 if predicted_category == 'short' else 1_0_2_4
												__magic_name__		, __magic_name__				:							str				=      get_best_valid_start_end_idx(
												    start_scores[0]      ,						end_scores[0]      ,						top_k=8      ,						max_size=lowercase__				)

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

												__magic_name__				:							Tuple				=      expand_to_aliases(example['targets']      ,						make_sub_answers=lowercase__				)
												__magic_name__				:							Tuple				=      expand_to_aliases(example['output']				)

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

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

												return example

						__magic_name__				:							Union[str, Any]				=      short_validation_dataset.map(lowercase__				)

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


if __name__ == "__main__":
			main()






import numpy as np
import torch
import torch.nn as nn
from transformers import CLIPConfig, CLIPVisionModel, PreTrainedModel

from ...utils import logging


snake_case__     							= logging.get_logger(__name__)
def  __SCREAMING_SNAKE_CASE       (				lowercase__  :Optional[Any]      ,						lowercase__  :List[str]				)       ->  Tuple:
						__magic_name__				:							List[str]				=      nn.functional.normalize(lowercase__				)
						__magic_name__				:							Union[str, Any]				=      nn.functional.normalize(lowercase__				)
						return torch.mm(lowercase__      ,						normalized_text_embeds.t()				)



class     snake_case__  (      _A  ):
				__lowerCAmelCase		=      CLIPConfig

				__lowerCAmelCase		=      ['''CLIPEncoderLayer''']
				def __init__(  self     ,      _UpperCAmelCase      ):




										"""simple docstring"""




										super().__init__(_UpperCAmelCase      )

										__magic_name__				:							List[Any]				=      CLIPVisionModel(config.vision_config      )
										__magic_name__				:							Tuple				=      nn.Linear(config.vision_config.hidden_size     ,      config.projection_dim     ,      bias=_UpperCAmelCase      )

										__magic_name__				:							str				=      nn.Parameter(torch.ones(17     ,      config.projection_dim      )     ,      requires_grad=_UpperCAmelCase      )
										__magic_name__				:							Optional[Any]				=      nn.Parameter(torch.ones(3     ,      config.projection_dim      )     ,      requires_grad=_UpperCAmelCase      )

										__magic_name__				:							Optional[Any]				=      nn.Parameter(torch.ones(17      )     ,      requires_grad=_UpperCAmelCase      )
										__magic_name__				:							str				=      nn.Parameter(torch.ones(3      )     ,      requires_grad=_UpperCAmelCase      )
				@torch.no_grad()
				def  UpperCamelCase_						(  self     ,      _UpperCAmelCase     ,      _UpperCAmelCase      ):




										"""simple docstring"""




										__magic_name__				:							Union[str, Any]				=      self.vision_model(_UpperCAmelCase      )[1]  # pooled_output
										__magic_name__				:							Any				=      self.visual_projection(_UpperCAmelCase      )

										# we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
										__magic_name__				:							Union[str, Any]				=      cosine_distance(_UpperCAmelCase     ,      self.special_care_embeds      ).cpu().float().numpy()
										__magic_name__				:							int				=      cosine_distance(_UpperCAmelCase     ,      self.concept_embeds      ).cpu().float().numpy()

										__magic_name__				:							str				=      []
										__magic_name__				:							List[Any]				=      image_embeds.shape[0]
										for i in range(_UpperCAmelCase      ):
																__magic_name__				:							Optional[Any]				=      {'special_scores': {}, 'special_care': [], 'concept_scores': {}, 'bad_concepts': []}

																# increase this value to create a stronger `nfsw` filter
																# at the cost of increasing the possibility of filtering benign images
																__magic_name__				:							Optional[int]				=      0.0

																for concept_idx in range(len(special_cos_dist[0]      )      ):
																						__magic_name__				:							Any				=      special_cos_dist[i][concept_idx]
																						__magic_name__				:							Tuple				=      self.special_care_embeds_weights[concept_idx].item()
																						__magic_name__				:							Dict				=      round(concept_cos - concept_threshold + adjustment     ,      3      )
																						if result_img["special_scores"][concept_idx] > 0:
																												result_img["special_care"].append({concept_idx, result_img['special_scores'][concept_idx]}      )
																												__magic_name__				:							str				=      0.01

																for concept_idx in range(len(cos_dist[0]      )      ):
																						__magic_name__				:							Optional[Any]				=      cos_dist[i][concept_idx]
																						__magic_name__				:							Tuple				=      self.concept_embeds_weights[concept_idx].item()
																						__magic_name__				:							Optional[int]				=      round(concept_cos - concept_threshold + adjustment     ,      3      )
																						if result_img["concept_scores"][concept_idx] > 0:
																												result_img["bad_concepts"].append(_UpperCAmelCase      )

																result.append(_UpperCAmelCase      )

										__magic_name__				:							Union[str, Any]				=      [len(res['bad_concepts']      ) > 0 for res in result]

										for idx, has_nsfw_concept in enumerate(_UpperCAmelCase      ):
																if has_nsfw_concept:
																						if torch.is_tensor(_UpperCAmelCase      ) or torch.is_tensor(images[0]      ):
																												__magic_name__				:							Union[str, Any]				=      torch.zeros_like(images[idx]      )  # black image
																						else:
																												__magic_name__				:							Optional[int]				=      np.zeros(images[idx].shape      )  # black image

										if any(_UpperCAmelCase      ):
																logger.warning(
																    'Potential NSFW content was detected in one or more images. A black image will be returned instead.'
																    ' Try again with a different prompt and/or seed.'      )

										return images, has_nsfw_concepts





				@torch.no_grad()
				def  UpperCamelCase_						(  self     ,      _UpperCAmelCase     ,      _UpperCAmelCase      ):




										"""simple docstring"""




										__magic_name__				:							List[str]				=      self.vision_model(_UpperCAmelCase      )[1]  # pooled_output
										__magic_name__				:							List[str]				=      self.visual_projection(_UpperCAmelCase      )

										__magic_name__				:							Union[str, Any]				=      cosine_distance(_UpperCAmelCase     ,      self.special_care_embeds      )
										__magic_name__				:							Optional[int]				=      cosine_distance(_UpperCAmelCase     ,      self.concept_embeds      )

										# increase this value to create a stronger `nsfw` filter
										# at the cost of increasing the possibility of filtering benign images
										__magic_name__				:							int				=      0.0

										__magic_name__				:							Optional[int]				=      special_cos_dist - self.special_care_embeds_weights + adjustment
										# special_scores = special_scores.round(decimals=3)
										__magic_name__				:							int				=      torch.any(special_scores > 0     ,      dim=1      )
										__magic_name__				:							Optional[Any]				=      special_care * 0.01
										__magic_name__				:							Any				=      special_adjustment.unsqueeze(1      ).expand(-1     ,      cos_dist.shape[1]      )

										__magic_name__				:							Optional[int]				=      (cos_dist - self.concept_embeds_weights) + special_adjustment
										# concept_scores = concept_scores.round(decimals=3)
										__magic_name__				:							Union[str, Any]				=      torch.any(concept_scores > 0     ,      dim=1      )

										__magic_name__				:							Dict				=      0.0  # black image

										return images, has_nsfw_concepts


